fsfw-example-tmtc-common/pus_tm/factory_hook.py

"""
@brief      This file transfers control of TM parsing to the user
@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
            it to your needs.
"""
from spacepackets.ecss.tm import PusTelemetry
from spacepackets.util import PrintFormats
from spacepackets.ccsds.spacepacket import PacketTypes

from tmtccmd.tm.pus_2_rawcmd import Service2Tm
from tmtccmd.tm.pus_17_test import Service17TMExtended
from tmtccmd.tm.pus_20_fsfw_parameters import Service20FsfwTm
from tmtccmd.tm.pus_200_fsfw_modes import Service200FsfwTm
from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
from tmtccmd.logging import get_console_logger
from tmtccmd.logging.pus import (
    create_tmtc_logger,
    log_raw_pus_tm,
    log_raw_unknown_packet,
)

from common_tmtc.config.object_ids import get_object_ids
from common_tmtc.pus_tm.action_reply_handling import handle_action_reply
from common_tmtc.pus_tm.event_handler import handle_event_packet
from common_tmtc.pus_tm.verification_handler import handle_service_1_packet
from common_tmtc.pus_tm.hk_handling import handle_hk_packet
from common_tmtc.config.definitions import PUS_APID


LOGGER = get_console_logger()


FSFW_PRINTER = FsfwTmTcPrinter(file_logger=create_tmtc_logger())


def ccsds_tm_handler(apid: int, raw_tm_packet: bytes, _user_args: any) -> None:
    if apid == PUS_APID:
        pus_factory_hook(raw_tm_packet=raw_tm_packet)


def pus_factory_hook(raw_tm_packet: bytes):
    if len(raw_tm_packet) < 8:
        LOGGER.warning("Detected packet shorter than 8 bytes!")
        return
    service_type = raw_tm_packet[7]
    subservice_type = raw_tm_packet[8]
    file_logger = FSFW_PRINTER.file_logger
    obj_id_dict = get_object_ids()
    dedicated_handler = True
    try:
        tm_packet = None
        if service_type == 1:
            handle_service_1_packet(printer=FSFW_PRINTER, raw_tm=raw_tm_packet)
        elif service_type == 2:
            tm_packet = Service2Tm.unpack(raw_tm_packet)
            dedicated_handler = False
        elif service_type == 3:
            handle_hk_packet(
                printer=FSFW_PRINTER, raw_tm=raw_tm_packet, obj_id_dict=obj_id_dict
            )
        elif service_type == 8:
            handle_action_reply(
                raw_tm=raw_tm_packet, printer=FSFW_PRINTER, obj_id_dict=obj_id_dict
            )
        elif service_type == 5:
            handle_event_packet(
                raw_tm=raw_tm_packet, printer=FSFW_PRINTER, file_logger=file_logger
            )
        elif service_type == 17:
            tm_packet = Service17TMExtended.unpack(raw_telemetry=raw_tm_packet)
            dedicated_handler = False
        elif service_type == 20:
            tm_packet = Service20FsfwTm.unpack(raw_telemetry=raw_tm_packet)
            dedicated_handler = False
        elif service_type == 200:
            tm_packet = Service200FsfwTm.unpack(raw_telemetry=raw_tm_packet)
            dedicated_handler = False
        else:
            LOGGER.info(
                f"The service {service_type} is not implemented in Telemetry Factory"
            )
            tm_packet = PusTelemetry.unpack(raw_telemetry=raw_tm_packet)
            tm_packet.print_source_data(PrintFormats.HEX)
            dedicated_handler = True
        if not dedicated_handler and tm_packet is not None:
            FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
        log_raw_pus_tm(
            packet=raw_tm_packet, srv_subservice=(service_type, subservice_type)
        )
    except ValueError:
        LOGGER.warning("Invalid packet format detected")
        log_raw_unknown_packet(packet=raw_tm_packet, packet_type=PacketTypes.TM)