162 lines
7.1 KiB
Python
162 lines
7.1 KiB
Python
import logging
|
|
from typing import cast
|
|
|
|
from eive_tmtc.config.definitions import (
|
|
CFDP_REMOTE_ENTITY_ID,
|
|
PUS_APID,
|
|
CFDP_LOCAL_ENTITY_ID,
|
|
)
|
|
from eive_tmtc.pus_tc.cmd_demux import handle_pus_procedure
|
|
from eive_tmtc.cfdp.handler import CfdpInCcsdsHandler
|
|
from tmtccmd import TcHandlerBase, ProcedureWrapper
|
|
from cfdppy.defs import CfdpRequestType
|
|
from tmtccmd.logging import get_current_time_string
|
|
from tmtccmd.logging.pus import RawTmtcTimedLogWrapper
|
|
from tmtccmd.tmtc import (
|
|
DefaultPusQueueHelper,
|
|
QueueWrapper,
|
|
FeedWrapper,
|
|
TcProcedureType,
|
|
SendCbParams,
|
|
TcQueueEntryType,
|
|
)
|
|
from tmtccmd.config.cfdp import generic_cfdp_params_to_put_request
|
|
from spacepackets.ecss import PusVerificator
|
|
from spacepackets.seqcount import FileSeqCountProvider
|
|
from spacepackets.cfdp import PduHolder, DirectiveType
|
|
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class TcHandler(TcHandlerBase):
|
|
def __init__(
|
|
self,
|
|
seq_count_provider: FileSeqCountProvider,
|
|
cfdp_in_ccsds_handler: CfdpInCcsdsHandler,
|
|
pus_verificator: PusVerificator,
|
|
high_level_file_logger: logging.Logger,
|
|
raw_pus_logger: RawTmtcTimedLogWrapper,
|
|
gui: bool,
|
|
):
|
|
super().__init__()
|
|
self.cfdp_handler_started = False
|
|
self.seq_count_provider = seq_count_provider
|
|
self.pus_verificator = pus_verificator
|
|
self.high_level_file_logger = high_level_file_logger
|
|
self.pus_raw_logger = raw_pus_logger
|
|
self.gui = gui
|
|
self.proxy_op = True
|
|
self.queue_helper = DefaultPusQueueHelper(
|
|
queue_wrapper=QueueWrapper.empty(),
|
|
default_pus_apid=PUS_APID,
|
|
seq_cnt_provider=seq_count_provider,
|
|
pus_verificator=pus_verificator,
|
|
tc_sched_timestamp_len=4,
|
|
)
|
|
self.cfdp_in_ccsds_handler = cfdp_in_ccsds_handler
|
|
|
|
def cfdp_done(self) -> bool:
|
|
if self.cfdp_handler_started:
|
|
if (
|
|
not self.cfdp_in_ccsds_handler.put_request_pending()
|
|
and not self.proxy_op
|
|
):
|
|
self.cfdp_handler_started = False
|
|
return True
|
|
return False
|
|
|
|
def feed_cb(self, info: ProcedureWrapper, wrapper: FeedWrapper):
|
|
self.queue_helper.queue_wrapper = wrapper.queue_wrapper
|
|
if info.proc_type == TcProcedureType.DEFAULT:
|
|
handle_pus_procedure(info.to_def_procedure(), self.queue_helper)
|
|
elif info.proc_type == TcProcedureType.CFDP:
|
|
self.handle_cfdp_procedure(info)
|
|
|
|
def send_cb(self, send_params: SendCbParams):
|
|
entry_helper = send_params.entry
|
|
if entry_helper.is_tc:
|
|
if entry_helper.entry_type == TcQueueEntryType.PUS_TC:
|
|
pus_tc_wrapper = entry_helper.to_pus_tc_entry()
|
|
# pus_tc_wrapper.pus_tc.apid = PUS_APID
|
|
raw_tc = pus_tc_wrapper.pus_tc.pack()
|
|
self.pus_raw_logger.log_tc(pus_tc_wrapper.pus_tc)
|
|
tc_info_string = f"Sent {pus_tc_wrapper.pus_tc}"
|
|
_LOGGER.info(tc_info_string)
|
|
self.high_level_file_logger.info(
|
|
f"{get_current_time_string(True)}: {tc_info_string}"
|
|
)
|
|
# with open("tc.bin", "wb") as of:
|
|
# of.write(raw_tc)
|
|
send_params.com_if.send(raw_tc)
|
|
elif entry_helper.entry_type == TcQueueEntryType.CCSDS_TC:
|
|
cfdp_packet_in_ccsds = entry_helper.to_space_packet_entry()
|
|
send_params.com_if.send(cfdp_packet_in_ccsds.space_packet.pack())
|
|
# TODO: Log raw CFDP packets similarly to how PUS packets are logged.
|
|
# - Log full raw format including space packet wrapper
|
|
# - Log context information: Transaction ID, and PDU type and directive
|
|
# Could re-use file logger. Should probably do that
|
|
# print(f"sending packet: [{cfdp_packet_in_ccsds.space_packet.pack()}]")
|
|
# with open(f"cfdp_packet_{self.cfdp_counter}", "wb") as of:
|
|
# of.write(cfdp_packet_in_ccsds.space_packet.pack())
|
|
# self.cfdp_counter += 1
|
|
elif entry_helper.entry_type == TcQueueEntryType.LOG:
|
|
log_entry = entry_helper.to_log_entry()
|
|
_LOGGER.info(log_entry.log_str)
|
|
self.high_level_file_logger.info(log_entry.log_str)
|
|
|
|
def handle_cfdp_procedure(self, info: ProcedureWrapper):
|
|
cfdp_procedure = info.to_cfdp_procedure()
|
|
if cfdp_procedure.cfdp_request_type == CfdpRequestType.PUT:
|
|
if (
|
|
not self.cfdp_in_ccsds_handler.put_request_pending()
|
|
and not self.cfdp_handler_started
|
|
):
|
|
put_req_cfg_wrapper = cfdp_procedure.request_wrapper.to_put_request()
|
|
if put_req_cfg_wrapper.cfg.proxy_op:
|
|
self.proxy_op = True
|
|
put_req = generic_cfdp_params_to_put_request(
|
|
params=put_req_cfg_wrapper.cfg,
|
|
local_id=CFDP_LOCAL_ENTITY_ID,
|
|
remote_id=CFDP_REMOTE_ENTITY_ID,
|
|
dest_id_proxy_put_req=CFDP_LOCAL_ENTITY_ID,
|
|
)
|
|
_LOGGER.info(
|
|
f"CFDP: Starting file put request with parameters:\n{put_req}"
|
|
)
|
|
assert put_req is not None
|
|
self.cfdp_in_ccsds_handler.cfdp_handler.put_request(put_req)
|
|
self.cfdp_handler_started = True
|
|
|
|
for source_pair, _ in self.cfdp_in_ccsds_handler:
|
|
if source_pair is not None:
|
|
pdu, sp = source_pair
|
|
pdu = cast(PduHolder, pdu)
|
|
if pdu.is_file_directive:
|
|
if pdu.pdu_directive_type == DirectiveType.METADATA_PDU:
|
|
metadata = pdu.to_metadata_pdu()
|
|
self.queue_helper.add_log_cmd(
|
|
"CFDP Source: Sending Metadata PDU for file with size "
|
|
f"{metadata.file_size}"
|
|
)
|
|
elif pdu.pdu_directive_type == DirectiveType.EOF_PDU:
|
|
self.queue_helper.add_log_cmd(
|
|
"CFDP Source: Sending EOF PDU"
|
|
)
|
|
else:
|
|
fd_pdu = pdu.to_file_data_pdu()
|
|
self.queue_helper.add_log_cmd(
|
|
"CFDP Source: Sending File Data PDU for segment at offset "
|
|
f"{fd_pdu.offset} with length {len(fd_pdu.file_data)}"
|
|
)
|
|
self.queue_helper.add_ccsds_tc(sp)
|
|
self.cfdp_in_ccsds_handler.source_handler.state_machine()
|
|
|
|
def queue_finished_cb(self, info: ProcedureWrapper):
|
|
if info is not None:
|
|
if info.proc_type == TcQueueEntryType.PUS_TC:
|
|
def_proc = info.to_def_procedure()
|
|
_LOGGER.info(f"Finished queue for command {def_proc.cmd_path}")
|
|
elif info.proc_type == TcProcedureType.CFDP:
|
|
_LOGGER.info("Finished CFDP queue")
|