32 lines
1.3 KiB
Python
32 lines
1.3 KiB
Python
|
import logging
|
||
|
|
|
||
|
|
from eive_tmtc.config.definitions import CFDP_APID
|
||
|
|
from spacepackets.ccsds import SPACE_PACKET_HEADER_SIZE
|
||
|
|
from spacepackets.cfdp import PduFactory, PduType, DirectiveType
|
||
|
|
from tmtccmd.cfdp.handler import CfdpInCcsdsHandler
|
||
|
|
from tmtccmd.tm import SpecificApidHandlerBase
|
||
|
|
|
||
|
|
_LOGGER = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
|
||
|
|
class CfdpInCcsdsWrapper(SpecificApidHandlerBase):
|
||
|
|
def __init__(self, cfdp_in_ccsds_handler: CfdpInCcsdsHandler):
|
||
|
|
super().__init__(CFDP_APID, None)
|
||
|
|
self.handler = cfdp_in_ccsds_handler
|
||
|
|
|
||
|
|
def handle_tm(self, packet: bytes, _user_args: any):
|
||
|
|
# Ignore the space packet header. Its only purpose is to use the same protocol and
|
||
|
|
# have a seaprate APID for space packets. If this function is called, the APID is correct.
|
||
|
|
pdu = packet[SPACE_PACKET_HEADER_SIZE:]
|
||
|
|
pdu_base = PduFactory.from_raw(pdu)
|
||
|
|
if pdu_base.pdu_type == PduType.FILE_DATA:
|
||
|
|
_LOGGER.info("Received File Data PDU TM")
|
||
|
|
else:
|
||
|
|
if pdu_base.directive_type == DirectiveType.FINISHED_PDU:
|
||
|
|
_LOGGER.info("Received Finished PDU TM")
|
||
|
|
else:
|
||
|
|
_LOGGER.info(
|
||
|
|
f"Received File Directive PDU with type {pdu_base.directive_type!r} TM"
|
||
|
|
)
|
||
|
|
self.handler.pass_pdu_packet(pdu_base)
|