import enum from typing import Optional from tmtccmd.config import QueueCommands from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.service_200_mode import pack_mode_data, Modes, Subservices from tmtccmd.tc.service_20_parameter import ( pack_scalar_double_param_app_data, pack_fsfw_load_param_cmd, ) from tmtccmd.utility.logger import get_console_logger from spacepackets.ecss.tc import PusTelecommand from config.object_ids import PL_PCDU_ID LOGGER = get_console_logger() class OpCodes: SWITCH_ADC_ON = ["0", "switch-adc-on"] SWITCH_ALL_ON = ["1", "switch-all-on"] UPDATE_DRO_TO_X8_WAIT = ["2", "dro-to-x8-wait"] UPDATE_X8_TO_TX_WAIT_TIME = ["3", "x8-to-tx-wait"] TX_TO_MPA_WAIT_TIME = ["4", "tx-to-mpa-wait"] MPA_TO_HPA_WAIT_TIME = ["5", "mpa-to-hpa-wait"] INJECT_SSR_TO_DRO_FAILURE = ["6", "inject-ssr-dro-fault"] INJECT_DRO_TO_X8_FAILURE = ["7", "inject-dro-x8-fault"] INJECT_X8_TO_TX_FAILURE = ["8", "inject-x8-tx-fault"] INJECT_TX_TO_MPA_FAILURE = ["9", "inject-tx-mpa-fault"] INJECT_MPA_TO_HPA_FAILURE = ["10", "inject-mpa-hpa-fault"] INJECT_ALL_ON_FAILURE = ["11", "inject-all-on-fault"] class Submodes(enum.IntEnum): ADC_ON = 0 ALL_ON = 1 class ParamIds(enum.IntEnum): NEG_V_LOWER_BOUND = 0 NEG_V_UPPER_BOUND = 1 DRO_U_LOWER_BOUND = 2 DRO_U_UPPER_BOUND = 3 DRO_I_UPPER_BOUND = 4 X8_U_LOWER_BOUND = 5 X8_U_UPPER_BOUND = 6 X8_I_UPPER_BOUND = 7 TX_U_LOWER_BOUND = 8 TX_U_UPPER_BOUND = 9 TX_I_UPPER_BOUND = 10 MPA_U_LOWER_BOUND = 11 MPA_U_UPPER_BOUND = 12 MPA_I_UPPER_BOUND = 13 HPA_U_LOWER_BOUND = 14 HPA_U_UPPER_BOUND = 15 HPA_I_UPPER_BOUND = 16 SSR_TO_DRO_WAIT_TIME = 17 DRO_TO_X8_WAIT_TIME = 18 X8_TO_TX_WAIT_TIME = 19 TX_TO_MPA_WAIT_TIME = 20 MPA_TO_HPA_WAIT_TIME = 21 INJECT_SSR_TO_DRO_FAILURE = 30 INJECT_DRO_TO_X8_FAILURE = 31 INJECT_X8_TO_TX_FAILURE = 32 INJECT_TX_TO_MPA_FAILURE = 33 INJECT_MPA_TO_HPA_FAILURE = 34 INJECT_ALL_ON_FAILURE = 35 def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str): if op_code in OpCodes.SWITCH_ADC_ON: tc_queue.appendleft((QueueCommands.PRINT, "Switching PL PCDU ADC module on")) mode_data = pack_mode_data( object_id=PL_PCDU_ID, mode=Modes.NORMAL, submode=Submodes.ADC_ON ) mode_cmd = PusTelecommand( service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data ) tc_queue.appendleft(mode_cmd.pack_command_tuple()) if op_code in OpCodes.SWITCH_ALL_ON: tc_queue.appendleft((QueueCommands.PRINT, "Switching all PL PCDU modules on")) mode_data = pack_mode_data( object_id=PL_PCDU_ID, mode=Modes.NORMAL, submode=Submodes.ALL_ON ) mode_cmd = PusTelecommand( service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data ) tc_queue.appendleft(mode_cmd.pack_command_tuple()) if op_code in OpCodes.UPDATE_DRO_TO_X8_WAIT: pack_wait_time_cmd( tc_queue=tc_queue, param_id=ParamIds.DRO_TO_X8_WAIT_TIME, print_str="DRO to X8", ) if op_code in OpCodes.UPDATE_X8_TO_TX_WAIT_TIME: pass def request_wait_time() -> Optional[float]: while True: wait_time = input("Please enter DRO to X8 wait time in seconds, x to cancel: ") if wait_time.lower() == "x": return None try: wait_time = float(wait_time) except ValueError: LOGGER.warning("Invalid input") continue if wait_time <= 0: LOGGER.warning("Invalid input") else: return wait_time def pack_wait_time_cmd(tc_queue: TcQueueT, param_id: int, print_str: str): wait_time = request_wait_time() tc_queue.appendleft( (QueueCommands.PRINT, f"Updating {print_str} wait time to {wait_time}") ) if wait_time is None: return param_data = pack_scalar_double_param_app_data( object_id=PL_PCDU_ID, domain_id=0, unique_id=param_id, parameter=wait_time, ) cmd = pack_fsfw_load_param_cmd(ssc=0, app_data=param_data) tc_queue.appendleft(cmd.pack_command_tuple())