fsfw-example-tmtc-common/pus_tm/factory_hook.py
2022-05-20 00:19:50 +02:00

93 lines
3.6 KiB
Python

"""
@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)