#!/usr/bin/env python3 """Example client for the sat-rs example application""" import enum import logging import struct import sys import time from typing import Optional import datetime import json import pprint import tmtccmd from spacepackets.ecss import PusTelemetry, PusTelecommand, PusVerificator from spacepackets.ecss.pus_17_test import Service17Tm from spacepackets.ecss.pus_1_verification import UnpackParams, Service1Tm from spacepackets.ccsds.time import CdsShortTimestamp from tmtccmd import CcsdsTmtcBackend, TcHandlerBase, ProcedureParamsWrapper from tmtccmd.tc.pus_3_fsfw_hk import generate_one_hk_command, make_sid from tmtccmd.tc.pus_11_tc_sched import create_time_tagged_cmd from tmtccmd.core.base import BackendRequest from tmtccmd.pus import VerificationWrapper from tmtccmd.tm import CcsdsTmHandler, SpecificApidHandlerBase from tmtccmd.com import ComInterface from tmtccmd.config import ( default_json_path, SetupParams, HookBase, TmtcDefinitionWrapper, CoreServiceList, OpCodeEntry, params_to_procedure_conversion, ) from tmtccmd.config import PreArgsParsingWrapper, SetupWrapper from tmtccmd.logging import add_colorlog_console_logger from tmtccmd.logging.pus import ( RegularTmtcLogWrapper, RawTmtcTimedLogWrapper, TimedLogWhen, ) from tmtccmd.tc import ( TcQueueEntryType, ProcedureWrapper, TcProcedureType, FeedWrapper, SendCbParams, DefaultPusQueueHelper, QueueWrapper, ) from tmtccmd.tm.pus_5_fsfw_event import Service5Tm from tmtccmd.util import FileSeqCountProvider, PusFileSeqCountProvider from tmtccmd.util.obj_id import ObjectIdDictT from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter _LOGGER = logging.getLogger(__name__) EXAMPLE_PUS_APID = 0x02 class SatRsConfigHook(HookBase): def __init__(self, json_cfg_path: str): super().__init__(json_cfg_path=json_cfg_path) def assign_communication_interface(self, com_if_key: str) -> Optional[ComInterface]: from tmtccmd.config.com import ( create_com_interface_default, create_com_interface_cfg_default, ) cfg = create_com_interface_cfg_default( com_if_key=com_if_key, json_cfg_path=self.cfg_path, space_packet_ids=None, ) return create_com_interface_default(cfg) def get_tmtc_definitions(self) -> TmtcDefinitionWrapper: from tmtccmd.config.globals import get_default_tmtc_defs defs = get_default_tmtc_defs() srv_5 = OpCodeEntry() srv_5.add("0", "Event Test") defs.add_service( name=CoreServiceList.SERVICE_5.value, info="PUS Service 5 Event", op_code_entry=srv_5, ) srv_17 = OpCodeEntry() srv_17.add("0", "Ping Test") defs.add_service( name=CoreServiceList.SERVICE_17_ALT, info="PUS Service 17 Test", op_code_entry=srv_17, ) srv_3 = OpCodeEntry() srv_3.add(HkOpCodes.GENERATE_ONE_SHOT, "Generate AOCS one shot HK") defs.add_service( name=CoreServiceList.SERVICE_3, info="PUS Service 3 Housekeeping", op_code_entry=srv_3, ) srv_8 = OpCodeEntry() srv_8.add("1", "Camera Image Request") defs.add_service( name=CoreServiceList.SERVICE_8, info="PUS Service 8 Action", op_code_entry=srv_8, ) srv_11 = OpCodeEntry() srv_11.add("0", "Scheduled TC Test") srv_11.add("1", "Scheduled Camera Request TC") defs.add_service( name=CoreServiceList.SERVICE_11, info="PUS Service 11 TC Scheduling", op_code_entry=srv_11, ) return defs def perform_mode_operation(self, tmtc_backend: CcsdsTmtcBackend, mode: int): _LOGGER.info("Mode operation hook was called") pass def get_object_ids(self) -> ObjectIdDictT: from tmtccmd.config.objects import get_core_object_ids return get_core_object_ids() class PusHandler(SpecificApidHandlerBase): def __init__( self, verif_wrapper: VerificationWrapper, printer: FsfwTmTcPrinter, raw_logger: RawTmtcTimedLogWrapper, ): super().__init__(EXAMPLE_PUS_APID, None) self.printer = printer self.raw_logger = raw_logger self.verif_wrapper = verif_wrapper def handle_tm(self, packet: bytes, _user_args: any): try: tm_packet = PusTelemetry.unpack( packet, time_reader=CdsShortTimestamp.empty() ) except ValueError as e: _LOGGER.warning("Could not generate PUS TM object from raw data") _LOGGER.warning(f"Raw Packet: [{packet.hex(sep=',')}], REPR: {packet!r}") raise e service = tm_packet.service dedicated_handler = False if service == 1: tm_packet = Service1Tm.unpack( data=packet, params=UnpackParams(CdsShortTimestamp.empty(), 1, 2) ) res = self.verif_wrapper.add_tm(tm_packet) if res is None: _LOGGER.info( f"Received Verification TM[{tm_packet.service}, {tm_packet.subservice}] " f"with Request ID {tm_packet.tc_req_id.as_u32():#08x}" ) _LOGGER.warning( f"No matching telecommand found for {tm_packet.tc_req_id}" ) else: self.verif_wrapper.log_to_console(tm_packet, res) self.verif_wrapper.log_to_file(tm_packet, res) dedicated_handler = True if service == 3: # _LOGGER.info("No handling for HK packets implemented") # _LOGGER.info(f"Raw packet: 0x[{packet.hex(sep=',')}]") pus_tm = PusTelemetry.unpack(packet, time_reader=CdsShortTimestamp.empty()) if pus_tm.subservice == 25: if len(pus_tm.source_data) < 8: raise ValueError("No addressable ID in HK packet") json_str = str(pus_tm.source_data[8:].decode('utf8')) json_object = json.loads(json_str) pprint.pprint(json_object) dedicated_handler = True if service == 5: tm_packet = Service5Tm.unpack(packet, time_reader=CdsShortTimestamp.empty()) if service == 17: tm_packet = Service17Tm.unpack( packet, time_reader=CdsShortTimestamp.empty() ) dedicated_handler = True if tm_packet.subservice == 2: self.printer.file_logger.info("Received Ping Reply TM[17,2]") _LOGGER.info("Received Ping Reply TM[17,2]") else: self.printer.file_logger.info( f"Received Test Packet with unknown subservice {tm_packet.subservice}" ) _LOGGER.info( f"Received Test Packet with unknown subservice {tm_packet.subservice}" ) if tm_packet is None: _LOGGER.info( f"The service {service} is not implemented in Telemetry Factory" ) tm_packet = PusTelemetry.unpack( packet, time_reader=CdsShortTimestamp.empty() ) self.raw_logger.log_tm(tm_packet) if not dedicated_handler and tm_packet is not None: pass # self.printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet) def make_addressable_id(target_id: int, unique_id: int) -> bytes: byte_string = bytearray(struct.pack("!I", target_id)) byte_string.extend(struct.pack("!I", unique_id)) return byte_string def read_addressable_id(data: bytes) -> tuple[int, int]: target_id = struct.unpack("!I", data[0:4])[0] set_id = struct.unpack("!I", data[4:8])[0] return (target_id, set_id) class RequestTargetId(enum.IntEnum): ACS = 1 PLD = 2 class AcsHkIds(enum.IntEnum): MGM_SET = 1 class HkOpCodes: GENERATE_ONE_SHOT = ["0", "oneshot"] def make_target_id(target_id: int) -> bytes: byte_string = bytearray(struct.pack("!I", target_id)) return byte_string class TcHandler(TcHandlerBase): def __init__( self, seq_count_provider: FileSeqCountProvider, verif_wrapper: VerificationWrapper, ): super(TcHandler, self).__init__() self.seq_count_provider = seq_count_provider self.verif_wrapper = verif_wrapper self.queue_helper = DefaultPusQueueHelper( queue_wrapper=QueueWrapper.empty(), tc_sched_timestamp_len=CdsShortTimestamp.TIMESTAMP_SIZE, seq_cnt_provider=seq_count_provider, pus_verificator=self.verif_wrapper.pus_verificator, default_pus_apid=EXAMPLE_PUS_APID, ) 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() raw_tc = pus_tc_wrapper.pus_tc.pack() _LOGGER.info(f"Sending {pus_tc_wrapper.pus_tc}") send_params.com_if.send(raw_tc) elif entry_helper.entry_type == TcQueueEntryType.LOG: log_entry = entry_helper.to_log_entry() _LOGGER.info(log_entry.log_str) def queue_finished_cb(self, helper: ProcedureWrapper): if helper.proc_type == TcProcedureType.DEFAULT: def_proc = helper.to_def_procedure() _LOGGER.info( f"Queue handling finished for service {def_proc.service} and " f"op code {def_proc.op_code}" ) def feed_cb(self, helper: ProcedureWrapper, wrapper: FeedWrapper): q = self.queue_helper q.queue_wrapper = wrapper.queue_wrapper if helper.proc_type == TcProcedureType.DEFAULT: def_proc = helper.to_def_procedure() service = def_proc.service op_code = def_proc.op_code if ( service == CoreServiceList.SERVICE_17 or service == CoreServiceList.SERVICE_17_ALT ): q.add_log_cmd("Sending PUS ping telecommand") return q.add_pus_tc(PusTelecommand(service=17, subservice=1)) if service == CoreServiceList.SERVICE_11: if op_code == "0": q.add_log_cmd("Sending PUS scheduled TC telecommand") crt_time = CdsShortTimestamp.from_now() time_stamp = crt_time + datetime.timedelta(seconds=10) time_stamp = time_stamp.pack() return q.add_pus_tc( create_time_tagged_cmd( time_stamp, PusTelecommand(service=17, subservice=1), apid=EXAMPLE_PUS_APID, ) ) if op_code == "1": q.add_log_cmd("Sending PUS scheduled TC telecommand") crt_time = CdsShortTimestamp.from_now() time_stamp = crt_time + datetime.timedelta(seconds=10) time_stamp = time_stamp.pack() return q.add_pus_tc( create_time_tagged_cmd( time_stamp, PusTelecommand(service=8, subservice=1, app_data=make_target_id(RequestTargetId.PLD)), apid=EXAMPLE_PUS_APID, ) ) if service == CoreServiceList.SERVICE_8: q.add_log_cmd("Sending PUS action request telecommand") return q.add_pus_tc( PusTelecommand(service=8, subservice=1, app_data=make_target_id(RequestTargetId.PLD)) ) if service == CoreServiceList.SERVICE_3: if op_code in HkOpCodes.GENERATE_ONE_SHOT: q.add_log_cmd("Sending HK one shot request") q.add_pus_tc( generate_one_hk_command( make_addressable_id(RequestTargetId.ACS, AcsHkIds.MGM_SET) ) ) pass def main(): add_colorlog_console_logger(_LOGGER) tmtccmd.init_printout(False) hook_obj = SatRsConfigHook(json_cfg_path=default_json_path()) parser_wrapper = PreArgsParsingWrapper() parser_wrapper.create_default_parent_parser() parser_wrapper.create_default_parser() parser_wrapper.add_def_proc_args() params = SetupParams() post_args_wrapper = parser_wrapper.parse(hook_obj, params) proc_wrapper = ProcedureParamsWrapper() if post_args_wrapper.use_gui: post_args_wrapper.set_params_without_prompts(proc_wrapper) else: post_args_wrapper.set_params_with_prompts(proc_wrapper) params.apid = EXAMPLE_PUS_APID setup_args = SetupWrapper( hook_obj=hook_obj, setup_params=params, proc_param_wrapper=proc_wrapper ) # Create console logger helper and file loggers tmtc_logger = RegularTmtcLogWrapper() printer = FsfwTmTcPrinter(tmtc_logger.logger) raw_logger = RawTmtcTimedLogWrapper(when=TimedLogWhen.PER_HOUR, interval=1) verificator = PusVerificator() verification_wrapper = VerificationWrapper( verificator, _LOGGER, printer.file_logger ) # Create primary TM handler and add it to the CCSDS Packet Handler tm_handler = PusHandler(verification_wrapper, printer, raw_logger) ccsds_handler = CcsdsTmHandler(generic_handler=None) ccsds_handler.add_apid_handler(tm_handler) # Create TC handler seq_count_provider = PusFileSeqCountProvider() tc_handler = TcHandler(seq_count_provider, verification_wrapper) tmtccmd.setup(setup_args=setup_args) init_proc = params_to_procedure_conversion(setup_args.proc_param_wrapper) tmtc_backend = tmtccmd.create_default_tmtc_backend( setup_wrapper=setup_args, tm_handler=ccsds_handler, tc_handler=tc_handler, init_procedure=init_proc, ) tmtccmd.start(tmtc_backend=tmtc_backend, hook_obj=hook_obj) try: while True: state = tmtc_backend.periodic_op(None) if state.request == BackendRequest.TERMINATION_NO_ERROR: sys.exit(0) elif state.request == BackendRequest.DELAY_IDLE: _LOGGER.info("TMTC Client in IDLE mode") time.sleep(3.0) elif state.request == BackendRequest.DELAY_LISTENER: time.sleep(0.8) elif state.request == BackendRequest.DELAY_CUSTOM: if state.next_delay.total_seconds() <= 0.4: time.sleep(state.next_delay.total_seconds()) else: time.sleep(0.4) elif state.request == BackendRequest.CALL_NEXT: pass except KeyboardInterrupt: sys.exit(0) if __name__ == "__main__": main()