diff --git a/.gitignore b/.gitignore index 58ff4d8..47b4b46 100644 --- a/.gitignore +++ b/.gitignore @@ -5,4 +5,4 @@ __pycache__ log /gps_log.txt /config/*.json -tmtc_conf.json \ No newline at end of file +tmtc_conf.json diff --git a/pus_tc/cmd_definitions.py b/pus_tc/cmd_definitions.py index b4b4598..d83231e 100644 --- a/pus_tc/cmd_definitions.py +++ b/pus_tc/cmd_definitions.py @@ -560,7 +560,6 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT): "1": ("PLOC Supervisor: Set mode on", {OpCodeDictKeys.TIMEOUT: 2.0}), "2": ("PLOC Supervisor: Set mode normal", {OpCodeDictKeys.TIMEOUT: 2.0}), "3": ("PLOC Supervisor: Get HK Report", {OpCodeDictKeys.TIMEOUT: 2.0}), - "4": ("PLOC Supervisor: Restart MPSoC", {OpCodeDictKeys.TIMEOUT: 2.0}), "5": ("PLOC Supervisor: Start MPSoC", {OpCodeDictKeys.TIMEOUT: 2.0}), "6": ("PLOC Supervisor: Shutdown MPSoC", {OpCodeDictKeys.TIMEOUT: 2.0}), "7": ( @@ -576,18 +575,9 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT): "PLOC Supervisor: Request boot status report", {OpCodeDictKeys.TIMEOUT: 2.0}, ), - "14": ("PLOC Supervisor: Update available", {OpCodeDictKeys.TIMEOUT: 2.0}), - "15": ("PLOC Supervisor: Watchdogs enable", {OpCodeDictKeys.TIMEOUT: 2.0}), - "16": ( - "PLOC Supervisor: Watchdog Configure Timeout", - {OpCodeDictKeys.TIMEOUT: 2.0}, - ), "17": ("PLOC Supervisor: Enable latchup alert", {OpCodeDictKeys.TIMEOUT: 2.0}), "18": ("PLOC Supervisor: Disable latchup alert", {OpCodeDictKeys.TIMEOUT: 2.0}), - "19": ("PLOC Supervisor: Auto calibrate alert", {OpCodeDictKeys.TIMEOUT: 2.0}), "20": ("PLOC Supervisor: Set alert limit", {OpCodeDictKeys.TIMEOUT: 2.0}), - "21": ("PLOC Supervisor: Set alert irq filter", {OpCodeDictKeys.TIMEOUT: 2.0}), - "22": ("PLOC Supervisor: Set ADC sweep period", {OpCodeDictKeys.TIMEOUT: 2.0}), "23": ( "PLOC Supervisor: Set ADC enabled channels", {OpCodeDictKeys.TIMEOUT: 2.0}, @@ -604,8 +594,6 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT): "27": ("PLOC Supervisor: Copy ADC data to MRAM", {OpCodeDictKeys.TIMEOUT: 2.0}), "30": ("PLOC Supervisor: Run auto EM tests", {OpCodeDictKeys.TIMEOUT: 2.0}), "31": ("PLOC Supervisor: MRAM Wipe", {OpCodeDictKeys.TIMEOUT: 2.0}), - "33": ("PLOC Supervisor: Print CPU stats", {OpCodeDictKeys.TIMEOUT: 2.0}), - "34": ("PLOC Supervisor: Set debug verbosity", {OpCodeDictKeys.TIMEOUT: 2.0}), "35": ("PLOC Supervisor: Set GPIO", {OpCodeDictKeys.TIMEOUT: 2.0}), "36": ("PLOC Supervisor: Read GPIO", {OpCodeDictKeys.TIMEOUT: 2.0}), "37": ("PLOC Supervisor: Restart supervisor", {OpCodeDictKeys.TIMEOUT: 2.0}), @@ -628,8 +616,6 @@ def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT): "46": ("PLOC Supervisor: Factory flash", {OpCodeDictKeys.TIMEOUT: 2.0}), "47": ("PLOC Supervisor: Enable auto TM", {OpCodeDictKeys.TIMEOUT: 2.0}), "48": ("PLOC Supervisor: Disable auto TM", {OpCodeDictKeys.TIMEOUT: 2.0}), - "49": ("PLOC Supervisor: Enable ADC monitor task", {OpCodeDictKeys.TIMEOUT: 2.0}), - "50": ("PLOC Supervisor: Disable ADC monitor task", {OpCodeDictKeys.TIMEOUT: 2.0}), "51": ("PLOC Supervisor: Logging request event buffers", {OpCodeDictKeys.TIMEOUT: 2.0}), "52": ("PLOC Supervisor: Logging clear counters", {OpCodeDictKeys.TIMEOUT: 2.0}), "53": ("PLOC Supervisor: Logging set topic", {OpCodeDictKeys.TIMEOUT: 2.0}), diff --git a/pus_tc/devs/ploc_mpsoc.py b/pus_tc/devs/ploc_mpsoc.py index 9a39acb..bec0ac7 100644 --- a/pus_tc/devs/ploc_mpsoc.py +++ b/pus_tc/devs/ploc_mpsoc.py @@ -59,8 +59,10 @@ class MemAddresses(enum.IntEnum): DEADBEEF = 0x40000004 -class PlocReplyIds: - tm_mem_read_report = 6 +class PlocReplyIds(enum.IntEnum): + TM_MEM_READ_RPT = 6 + TM_CAM_CMD_RPT = 19 + def pack_ploc_mpsoc_commands( diff --git a/pus_tc/devs/ploc_supervisor.py b/pus_tc/devs/ploc_supervisor.py index 9b785e7..542c1d9 100644 --- a/pus_tc/devs/ploc_supervisor.py +++ b/pus_tc/devs/ploc_supervisor.py @@ -44,7 +44,6 @@ event_buffer_path_dict = { class SupvActionIds: HK_REPORT = 1 - RESTART_MPSOC = 2 START_MPSOC = 3 SHUTWOWN_MPSOC = 4 SEL_MPSOC_BOOT_IMAGE = 5 @@ -55,27 +54,19 @@ class SupvActionIds: DISABLE_HK = 10 GET_BOOT_STATUS_REPORT = 11 UPDATE_AVAILABLE = 12 - WATCHDOGS_ENABLE = 13 - WATCHDOGS_CONFIG_TIMEOUT = 14 ENABLE_LATCHUP_ALERT = 15 DISABLE_LATCHUP_ALERT = 16 - AUTO_CALIBRATE_ALERT = 17 SET_ALERT_LIMIT = 18 - SET_ALERT_IRQ_FILTER = 19 SET_ADC_SWEEP_PERIOD = 20 SET_ADC_ENABLED_CHANNELS = 21 SET_ADC_WINDOW_AND_STRIDE = 22 SET_ADC_THRESHOLD = 23 GET_LATCHUP_STATUS_REPORT = 24 COPY_ADC_DATA_TO_MRAM = 25 - ENABLE_NVMS = 26 SELECT_NVM = 27 RUN_AUTO_EM_TESTS = 28 WIPE_MRAM = 29 DUMP_MRAM = 30 - SET_DBG_VERBOSITY = 31 - CAN_LOOPBACK_TEST = 32 - PRINT_CPU_STATS = 33 SET_GPIO = 34 READ_GPIO = 35 RESTART_SUPERVISOR = 36 @@ -91,8 +82,6 @@ class SupvActionIds: TERMINATE_SUPV_HELPER = 49 ENABLE_AUTO_TM = 50 DISABLE_AUTO_TM = 51 - ENABLE_ADC_MONITOR_TASK = 52 - DISABLE_ADC_MONITOR_TASK = 53 LOGGING_REQUEST_EVENT_BUFFERS = 54 LOGGING_CLEAR_COUNTERS = 55 LOGGING_SET_TOPIC = 56 @@ -134,11 +123,6 @@ def pack_ploc_supv_commands( command = object_id + struct.pack("!I", SupvActionIds.HK_REPORT) command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "4": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Restart MPSoC")) - command = object_id + struct.pack("!I", SupvActionIds.RESTART_MPSOC) - command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "5": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Start MPSoC")) command = object_id + struct.pack("!I", SupvActionIds.START_MPSOC) @@ -162,9 +146,9 @@ def pack_ploc_supv_commands( tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "8": tc_queue.appendleft( - (QueueCommands.PRINT, "PLOC Supervisor: Set max restart tries: ") + (QueueCommands.PRINT, "PLOC Supervisor: Set max restart tries") ) - restart_tries = int(input("Set maximum restart tries:")) + restart_tries = int(input("Specify maximum restart tries: ")) command = ( object_id + struct.pack("!I", SupvActionIds.SET_MAX_RESTART_TRIES) @@ -186,7 +170,7 @@ def pack_ploc_supv_commands( tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "11": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set boot timeout")) - boot_timeout = int(input("Specify boot timeout [ms]:")) + boot_timeout = int(input("Specify boot timeout [ms]: ")) command = ( object_id + struct.pack("!I", SupvActionIds.SET_BOOT_TIMEOUT) @@ -206,23 +190,6 @@ def pack_ploc_supv_commands( command = object_id + struct.pack("!I", SupvActionIds.GET_BOOT_STATUS_REPORT) command = PusTelecommand(service=8, subservice=128, ssc=30, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "14": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Update available")) - command = pack_update_available_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=31, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "15": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Watchdogs Enable")) - command = pack_watchdogs_enable_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=32, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "16": - tc_queue.appendleft( - (QueueCommands.PRINT, "PLOC Supervisor: Watchdog configure timeout") - ) - command = pack_watchdog_config_timeout_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=33, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "17": tc_queue.appendleft( (QueueCommands.PRINT, "PLOC Supervisor: Enable latchup alert") @@ -237,32 +204,11 @@ def pack_ploc_supv_commands( command = pack_lachtup_alert_cmd(object_id, False) command = PusTelecommand(service=8, subservice=128, ssc=35, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "19": - tc_queue.appendleft( - (QueueCommands.PRINT, "PLOC Supervisor: Auto calibrate alert") - ) - command = pack_auto_calibrate_alert_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=36, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "20": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set alert limit")) command = pack_set_alert_limit_cmd(object_id) command = PusTelecommand(service=8, subservice=128, ssc=37, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "21": - tc_queue.appendleft( - (QueueCommands.PRINT, "PLOC Supervisor: Set alert irq filter") - ) - command = pack_set_alert_irq_filter_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=38, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "22": - tc_queue.appendleft( - (QueueCommands.PRINT, "PLOC Supervisor: Set ADC sweep period") - ) - command = pack_set_adc_sweep_period_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=39, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "23": tc_queue.appendleft( (QueueCommands.PRINT, "PLOC Supervisor: Set ADC enabled channels") @@ -306,18 +252,6 @@ def pack_ploc_supv_commands( command = pack_mram_wipe_cmd(object_id) command = PusTelecommand(service=8, subservice=128, ssc=46, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "33": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Print CPU stats")) - command = pack_print_cpu_stats_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=48, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "34": - tc_queue.appendleft( - (QueueCommands.PRINT, "PLOC Supervisor: Set debug verbosity") - ) - command = pack_set_debug_verbosity_cmd(object_id) - command = PusTelecommand(service=8, subservice=128, ssc=49, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "35": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set GPIO command")) command = pack_set_gpio_cmd(object_id) @@ -412,20 +346,6 @@ def pack_ploc_supv_commands( ) command = PusTelecommand(service=8, subservice=128, ssc=63, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "49": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable ADC monitor task")) - command = object_id + struct.pack( - "!I", SupvActionIds.ENABLE_ADC_MONITOR_TASK - ) - command = PusTelecommand(service=8, subservice=128, ssc=64, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) - elif op_code == "50": - tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable ADC monitor task")) - command = object_id + struct.pack( - "!I", SupvActionIds.DISABLE_ADC_MONITOR_TASK - ) - command = PusTelecommand(service=8, subservice=128, ssc=65, app_data=command) - tc_queue.appendleft(command.pack_command_tuple()) elif op_code == "51": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Logging request event buffers")) command = pack_logging_buffer_request(object_id) @@ -552,7 +472,7 @@ def pack_auto_calibrate_alert_cmd(object_id: bytearray) -> bytearray: @param object_id The object id of the PLOC supervisor handler. """ latchup_id = get_latchup_id() - mg = int(input("Specify MG:")) + mg = int(input("Specify MG: ")) command = bytearray() command = object_id + struct.pack("!I", SupvActionIds.AUTO_CALIBRATE_ALERT) command = command + struct.pack("!B", latchup_id) @@ -573,7 +493,7 @@ def get_latchup_id() -> int: key_string = key.ljust(key_column_width) description_string = latchup_id_dict[key].ljust(description_column_width) print(f"{key_string} | {description_string}") - return int(input("Specify latchup ID:")) + return int(input("Specify latchup ID: ")) def pack_set_alert_limit_cmd(object_id: bytearray) -> bytearray: @@ -582,7 +502,7 @@ def pack_set_alert_limit_cmd(object_id: bytearray) -> bytearray: @param object_id The object id of the PLOC supervisor handler. """ latchup_id = get_latchup_id() - dutycycle = int(input("Specify dutycycle:")) + dutycycle = int(input("Specify dutycycle: ")) command = bytearray() command = object_id + struct.pack("!I", SupvActionIds.SET_ALERT_LIMIT) command = command + struct.pack("!B", latchup_id) @@ -590,49 +510,20 @@ def pack_set_alert_limit_cmd(object_id: bytearray) -> bytearray: return command -def pack_set_alert_irq_filter_cmd(object_id: bytearray) -> bytearray: - """ - @brief This function packs the command to configure the latchup alert irq filter. - @param object_id The object id of the PLOC supervisor handler. - """ - latchup_id = get_latchup_id() - tp = int(input("Specify filter type (TP):")) - div = int(input("Specify clock divider (DIV):")) - command = bytearray() - command = object_id + struct.pack("!I", SupvActionIds.SET_ALERT_IRQ_FILTER) - command = command + struct.pack("!B", latchup_id) - command = command + struct.pack("!B", tp) - command = command + struct.pack("!B", div) - return command - - -def pack_set_adc_sweep_period_cmd(object_id: bytearray) -> bytearray: - """ - @brief This function packs the command to set the limit of a latchup alert. - @param object_id The object id of the PLOC supervisor handler. - """ - sweep_period = int(input("Specify sweep period (min 21 us):")) - command = bytearray() - command = object_id + struct.pack("!I", SupvActionIds.SET_ADC_SWEEP_PERIOD) - command = command + struct.pack("!I", sweep_period) - return command - - def pack_set_adc_enabled_channels_cmd(object_id: bytearray) -> bytearray: """ @brief This function packs the command to enable or disable channels of the ADC. @param object_id The object id of the PLOC supervisor handler. """ ch = int(input("Specify ch: 0x"), 16) - command = bytearray() - command = object_id + struct.pack("!I", SupvActionIds.SET_ADC_ENABLED_CHANNELS) - command = command + struct.pack("!H", ch) - return command + cmd = object_id + struct.pack("!I", SupvActionIds.SET_ADC_ENABLED_CHANNELS) + cmd = cmd + struct.pack("!H", ch) + return cmd def pack_set_adc_window_and_stride_cmd(object_id: bytearray) -> bytearray: - window_size = int(input("Specify window size:")) - striding_step_size = int(input("Specify striding step size:")) + window_size = int(input("Specify window size: ")) + striding_step_size = int(input("Specify striding step size: ")) command = bytearray() command = object_id + struct.pack("!I", SupvActionIds.SET_ADC_WINDOW_AND_STRIDE) command = command + struct.pack("!H", window_size) @@ -648,18 +539,8 @@ def pack_set_adc_threshold_cmd(object_id: bytearray) -> bytearray: return command -def pack_enable_nvms_cmd(object_id: bytearray) -> bytearray: - n01 = int(input("NVM0/1 (0 - off, 1 - on):")) - n3 = int(input("NVM3 (0 - off, 1 - on):")) - command = bytearray() - command = object_id + struct.pack("!I", SupvActionIds.ENABLE_NVMS) - command = command + struct.pack("!B", n01) - command = command + struct.pack("!B", n3) - return command - - def pack_select_nvm_cmd(object_id: bytearray) -> bytearray: - mem = int(input("Specify NVM (0 - NVM0, 1 - MVM1):")) + mem = int(input("Specify NVM (0 - NVM0, 1 - MVM1): ")) command = bytearray() command = object_id + struct.pack("!I", SupvActionIds.SELECT_NVM) command = command + struct.pack("!B", mem) @@ -667,7 +548,7 @@ def pack_select_nvm_cmd(object_id: bytearray) -> bytearray: def pack_auto_em_tests_cmd(object_id: bytearray) -> bytearray: - test = int(input("Specify test (1 - complete, 2 - short):")) + test = int(input("Specify test (1 - complete, 2 - short): ")) command = bytearray() command = object_id + struct.pack("!I", SupvActionIds.RUN_AUTO_EM_TESTS) command = command + struct.pack("!B", test) @@ -684,22 +565,6 @@ def pack_mram_wipe_cmd(object_id: bytearray) -> bytearray: return command -def pack_print_cpu_stats_cmd(object_id: bytearray) -> bytearray: - en = 1 - command = bytearray() - command = object_id + struct.pack("!I", SupvActionIds.PRINT_CPU_STATS) - command = command + struct.pack("!B", en) - return command - - -def pack_set_debug_verbosity_cmd(object_id: bytearray) -> bytearray: - command = bytearray() - verbosity = get_debug_verbosity() - command = object_id + struct.pack("!I", SupvActionIds.SET_DBG_VERBOSITY) - command = command + struct.pack("!B", verbosity) - return command - - def pack_update_command(object_id: bytearray) -> bytearray: command = bytearray() memory_id = int(input("Specify memory ID: ")) @@ -733,31 +598,6 @@ def pack_logging_buffer_request(object_id: bytearray) -> bytearray: return command -def get_debug_verbosity() -> int: - tries = 0 - while tries < 3: - try: - print("Debug verbosity options") - verbosity_options_dict = { - 0: "None", - 1: "Error", - 2: "Warn", - 3: "Info", - } - print("{:<6} | {}".format("Key", "Description")) - for entry in verbosity_options_dict.items(): - print("{:<6} | {}".format(entry[0], entry[1])) - verbosity = int(input("Specify verbosity key: ")) - if verbosity > len(verbosity_options_dict) - 1: - raise ValueError - return verbosity - except ValueError: - LOGGER.warning("Invalid verbosity key specified") - tries = tries + 1 - LOGGER.error("get_debug_verbosity: Exceeded max tries to input verbosity key") - quit() - - def pack_set_gpio_cmd(object_id: bytearray) -> bytearray: port = int(input("Specify port : 0x"), 16) pin = int(input("Specify pin: 0x"), 16) diff --git a/pus_tm/action_reply_handler.py b/pus_tm/action_reply_handler.py index 08fae82..17085de 100644 --- a/pus_tm/action_reply_handler.py +++ b/pus_tm/action_reply_handler.py @@ -58,7 +58,7 @@ def handle_imtq_replies( def handle_ploc_replies( action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray ): - if action_id == PlocReplyIds.tm_mem_read_report: + if action_id == PlocReplyIds.TM_MEM_READ_RPT: header_list = [ "PLOC Memory Address", "PLOC Mem Len", @@ -73,6 +73,13 @@ def handle_ploc_replies( print(content_list) printer.file_logger.info(header_list) printer.file_logger.info(content_list) + elif action_id == PlocReplyIds.TM_CAM_CMD_RPT: + header_list = ["Camera reply string", "ACK"] + content_list = [custom_data[:len(custom_data) - 1].decode('utf-8'), hex(custom_data[-1])] + print(header_list) + print(content_list) + printer.file_logger.info(header_list) + printer.file_logger.info(content_list) def handle_supervisor_replies( @@ -86,6 +93,13 @@ def handle_supervisor_replies( print(content_list) printer.file_logger.info(header_list) printer.file_logger.info(content_list) + elif action_id == SupvActionIds.READ_GPIO: + header_list = ["GPIO state"] + content_list = [struct.unpack('!H', custom_data[:2])[0]] + print(header_list) + print(content_list) + printer.file_logger.info(header_list) + printer.file_logger.info(content_list) def handle_startracker_replies( diff --git a/tmtccmd b/tmtccmd index ea29923..33a7308 160000 --- a/tmtccmd +++ b/tmtccmd @@ -1 +1 @@ -Subproject commit ea2992372f1ece7a9dab297ae390a5ab729e8f04 +Subproject commit 33a730815207c6f1dd6947e895e243d28d0b1e5a