import enum from typing import Optional from tmtccmd.config import QueueCommands from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices from tmtccmd.tc.pus_20_params import ( pack_scalar_double_param_app_data, pack_fsfw_load_param_cmd, pack_boolean_parameter_app_data, ) from tmtccmd.logging 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_ON = ["0", "on"] SWITCH_OFF = ["1", "off"] NORMAL_SSR = ["2", "nml-ssr"] NORMAL_DRO = ["3", "nml-dro"] NORMAL_X8 = ["4", "nml-x8"] NORMAL_TX = ["5", "nml-tx"] NORMAL_MPA = ["6", "nml-mpa"] NORMAL_HPA = ["7", "nml-hpa"] INJECT_SSR_TO_DRO_FAILURE = ["10", "inject-ssr-dro-fault"] INJECT_DRO_TO_X8_FAILURE = ["11", "inject-dro-x8-fault"] INJECT_X8_TO_TX_FAILURE = ["12", "inject-x8-tx-fault"] INJECT_TX_TO_MPA_FAILURE = ["13", "inject-tx-mpa-fault"] INJECT_MPA_TO_HPA_FAILURE = ["14", "inject-mpa-hpa-fault"] INJECT_ALL_ON_FAILURE = ["15", "inject-all-on-fault"] # The following commands might become deprecated in the future UPDATE_DRO_TO_X8_WAIT = ["128", "dro-to-x8-wait"] UPDATE_X8_TO_TX_WAIT_TIME = ["129", "x8-to-tx-wait"] UPDATE_TX_TO_MPA_WAIT_TIME = ["130", "tx-to-mpa-wait"] UPDATE_MPA_TO_HPA_WAIT_TIME = ["131", "mpa-to-hpa-wait"] class Info: NORMAL = "PL PCDU ADC modules normal" SWITCH_ON = "Switching PL PCDU on" SWITCH_OFF = "Switching PL PCDU off" NORMAL_SSR = f"{NORMAL}, SSR on" NORMAL_DRO = f"{NORMAL},DRO on" NORMAL_X8 = f"{NORMAL}, X8 on" NORMAL_TX = f"{NORMAL}, TX on" NORMAL_MPA = f"{NORMAL}, MPA on" NORMAL_HPA = f"{NORMAL}, HPA on" class NormalSubmodesMask(enum.IntEnum): SOLID_STATE_RELAYS_ADC_ON = 0 DRO_ON = 1 X8_ON = 2 TX_ON = 3 MPA_ON = 4 HPA_ON = 5 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_ON: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.SWITCH_ON, mode=Modes.ON, submode=0 ) if op_code in OpCodes.SWITCH_OFF: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.SWITCH_OFF, mode=Modes.OFF, submode=0 ) if op_code in OpCodes.NORMAL_SSR: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.NORMAL_SSR, mode=Modes.NORMAL, submode=(1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON), ) if op_code in OpCodes.NORMAL_DRO: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.NORMAL_DRO, mode=Modes.NORMAL, submode=( 1 << NormalSubmodesMask.DRO_ON | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) ), ) if op_code in OpCodes.NORMAL_X8: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.NORMAL_X8, mode=Modes.NORMAL, submode=( 1 << NormalSubmodesMask.DRO_ON | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) | (1 << NormalSubmodesMask.X8_ON) ), ) if op_code in OpCodes.NORMAL_TX: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.NORMAL_TX, mode=Modes.NORMAL, submode=( 1 << NormalSubmodesMask.DRO_ON | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) | (1 << NormalSubmodesMask.X8_ON) | (1 << NormalSubmodesMask.TX_ON) ), ) if op_code in OpCodes.NORMAL_MPA: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.NORMAL_MPA, mode=Modes.NORMAL, submode=( 1 << NormalSubmodesMask.DRO_ON | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) | (1 << NormalSubmodesMask.X8_ON) | (1 << NormalSubmodesMask.TX_ON) | (1 << NormalSubmodesMask.MPA_ON) ), ) if op_code in OpCodes.NORMAL_HPA: pack_pl_pcdu_mode_cmd( tc_queue=tc_queue, info=Info.NORMAL_HPA, mode=Modes.NORMAL, submode=( 1 << NormalSubmodesMask.DRO_ON | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON) | (1 << NormalSubmodesMask.X8_ON) | (1 << NormalSubmodesMask.TX_ON) | (1 << NormalSubmodesMask.MPA_ON) | (1 << NormalSubmodesMask.HPA_ON) ), ) if op_code in OpCodes.INJECT_ALL_ON_FAILURE: pack_failure_injection_cmd( tc_queue=tc_queue, param_id=ParamIds.INJECT_ALL_ON_FAILURE, print_str="All On", ) 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()) def pack_failure_injection_cmd(tc_queue: TcQueueT, param_id: int, print_str: str): tc_queue.appendleft((QueueCommands.PRINT, f"Inserting {print_str} error")) param_data = pack_boolean_parameter_app_data( object_id=PL_PCDU_ID, domain_id=0, unique_id=param_id, parameter=True ) cmd = pack_fsfw_load_param_cmd(ssc=0, app_data=param_data) tc_queue.appendleft(cmd.pack_command_tuple()) def pack_pl_pcdu_mode_cmd(tc_queue: TcQueueT, info: str, mode: Modes, submode: int): tc_queue.appendleft( ( QueueCommands.PRINT, info, ) ) mode_data = pack_mode_data(object_id=PL_PCDU_ID, mode=mode, submode=submode) mode_cmd = PusTelecommand( service=200, subservice=Subservices.TC_MODE_COMMAND, app_data=mode_data ) tc_queue.appendleft(mode_cmd.pack_command_tuple())