from typing import Optional import struct from eive_tmtc.config.definitions import CustomServiceList from eive_tmtc.pus_tc.devs.pdec_handler import CommandId from spacepackets.ecss import PusTelecommand from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry from tmtccmd.config.tmtc import tmtc_definitions_provider from tmtccmd.tc import DefaultPusQueueHelper from tmtccmd.util import ObjectIdU32 from tmtccmd.tc.pus_200_fsfw_modes import Mode, pack_mode_data, Subservice import eive_tmtc.config.object_ids as oids from eive_tmtc.config.object_ids import get_object_ids RTD_IDS = [ oids.RTD_0_PLOC_HSPD, oids.RTD_1_PLOC_MISSIONBRD, oids.RTD_2_4K_CAM, oids.RTD_3_DAC_HSPD, oids.RTD_4_STR, oids.RTD_5_RW1_MX_MY, oids.RTD_6_DRO, oids.RTD_7_SCEX, oids.RTD_8_X8, oids.RTD_9_HPA, oids.RTD_10_PL_TX, oids.RTD_11_MPA, oids.RTD_12_ACU, oids.RTD_13_PLPCDU_HSPD, oids.RTD_14_TCS_BRD, oids.RTD_15_IMTQ, ] class CommandId: WRITE_CONFIG = 6 class OpCode: ON = ["0", "on"] OFF = ["1", "off"] NORMAL = ["2", "normal"] WRITE_CONFIG = ["3", "Write config"] class Info: ON = "Switch handler on" OFF = "Switch handler off" NORMAL = "Switch handler normal" WRITE_CONFIG = "Write config" @tmtc_definitions_provider def specify_rtd_cmds(defs: TmtcDefinitionWrapper): oce = OpCodeEntry() oce.add(keys=OpCode.ON, info=Info.ON) oce.add(keys=OpCode.NORMAL, info=Info.NORMAL) oce.add(keys=OpCode.OFF, info=Info.OFF) oce.add(keys=OpCode.WRITE_CONFIG, info=Info.WRITE_CONFIG) defs.add_service( name=CustomServiceList.RTD.value, info="RTD commands", op_code_entry=oce ) def pack_rtd_commands( op_code: str, object_id: Optional[ObjectIdU32], q: DefaultPusQueueHelper ): if object_id is not None and object_id not in RTD_IDS: print("Specified object ID not a valid RTD ID") object_id = None if object_id is None: tgt_rtd_idx = prompt_rtd_idx() object_id_dict = get_object_ids() object_id = object_id_dict.get(RTD_IDS[tgt_rtd_idx]) if op_code in OpCode.ON: app_data = pack_mode_data(object_id=object_id.as_bytes, mode=Mode.ON, submode=0) q.add_pus_tc( PusTelecommand( service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=app_data ) ) if op_code in OpCode.NORMAL: app_data = pack_mode_data( object_id=object_id.as_bytes, mode=Mode.NORMAL, submode=0 ) q.add_pus_tc( PusTelecommand( service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=app_data ) ) if op_code in OpCode.OFF: app_data = pack_mode_data( object_id=object_id.as_bytes, mode=Mode.OFF, submode=0 ) q.add_pus_tc( PusTelecommand( service=200, subservice=Subservice.TC_MODE_COMMAND, app_data=app_data ) ) if op_code in OpCode.WRITE_CONFIG: command = object_id.as_bytes + struct.pack('!I', CommandId.WRITE_CONFIG) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) def prompt_rtd_idx(): while True: rtd_idx = input("Please specify RTD index [0-15]: ") if not rtd_idx.isdigit(): print("Invalid input") continue rtd_idx = int(rtd_idx) if rtd_idx < 0 or rtd_idx > 15: print("Invalid device index") continue return rtd_idx