eive-tmtc/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 tmtccmd.tm.service_8_fsfw_functional_cmd import Service8FsfwTm
from spacepackets.ecss.tm import PusTelemetry
from tmtccmd.logging import get_console_logger
from tmtccmd.logging.pus import (
    log_raw_pus_tm,
    log_raw_unknown_packet,
    PacketTypes,
    create_tmtc_logger,
)

from tmtccmd.pus.service_17_test import Service17TMExtended
from tmtccmd.tm.service_20_fsfw_parameters import Service20FsfwTm
from tmtccmd.tm.service_200_fsfw_mode import Service200FsfwTm
from tmtccmd.utility.tmtc_printer import PrintFormats, FsfwTmTcPrinter

from config.definitions import PUS_APID
from config.object_ids import get_object_ids
from .event_handler import handle_event_packet
from .verification_handler import handle_service_1_packet
from .hk_handling import handle_hk_packet

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()
    try:
        if service_type == 1:
            handle_service_1_packet(printer=FSFW_PRINTER, raw_tm=raw_tm_packet)
        elif service_type == 3:
            handle_hk_packet(
                printer=FSFW_PRINTER, raw_tm=raw_tm_packet, 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 == 8:
            tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm_packet)
            FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
        elif service_type == 17:
            tm_packet = Service17TMExtended.unpack(raw_telemetry=raw_tm_packet)
            FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
        elif service_type == 20:
            tm_packet = Service20FsfwTm.unpack(raw_telemetry=raw_tm_packet)
            FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
        elif service_type == 200:
            tm_packet = Service200FsfwTm.unpack(raw_telemetry=raw_tm_packet)
            FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
        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)
        log_raw_pus_tm(
            packet=raw_tm_packet, srv_subservice=(service_type, subservice_type)
        )
    except ValueError:
        # TODO: Log faulty packet
        LOGGER.warning("Invalid packet format detected")
        log_raw_unknown_packet(packet=raw_tm_packet, packet_type=PacketTypes.TM)