diff --git a/config/hook_implementations.py b/config/hook_implementations.py index 52fa233..c5bad08 100644 --- a/config/hook_implementations.py +++ b/config/hook_implementations.py @@ -95,6 +95,28 @@ class EiveHookObject(TmTcHookBase): "10": ("PLOC Supervisor: Set time reference", {OpCodeDictKeys.TIMEOUT: 2.0}), "11": ("PLOC Supervisor: Set boot timeout", {OpCodeDictKeys.TIMEOUT: 2.0}), "12": ("PLOC Supervisor: Disable Hk", {OpCodeDictKeys.TIMEOUT: 2.0}), + "13": ("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}), + "24": ("PLOC Supervisor: Set ADC window and stride", {OpCodeDictKeys.TIMEOUT: 2.0}), + "25": ("PLOC Supervisor: Set ADC threshold", {OpCodeDictKeys.TIMEOUT: 2.0}), + "26": ("PLOC Supervisor: Request latchup status report", {OpCodeDictKeys.TIMEOUT: 2.0}), + "27": ("PLOC Supervisor: Copy ADC data to MRAM", {OpCodeDictKeys.TIMEOUT: 2.0}), + "28": ("PLOC Supervisor: Enable/Disable NVM0/1/3", {OpCodeDictKeys.TIMEOUT: 2.0}), + "29": ("PLOC Supervisor: Select NVM", {OpCodeDictKeys.TIMEOUT: 2.0}), + "30": ("PLOC Supervisor: Run auto EM tests", {OpCodeDictKeys.TIMEOUT: 2.0}), + "31": ("PLOC Supervisor: MRAM Wipe", {OpCodeDictKeys.TIMEOUT: 2.0}), + "32": ("PLOC Supervisor: MRAM Dump", {OpCodeDictKeys.TIMEOUT: 2.0}), + "33": ("PLOC Supervisor: Print CPU stats", {OpCodeDictKeys.TIMEOUT: 2.0}), + "34": ("PLOC Supervisor: Set debug verbosity", {OpCodeDictKeys.TIMEOUT: 2.0}), } service_ploc_supv_tuple = ("PLOC Supervisor", op_code_dict_srv_ploc_supv) diff --git a/pus_tc/ploc_supervisor.py b/pus_tc/ploc_supervisor.py index a1e8523..65aec36 100644 --- a/pus_tc/ploc_supervisor.py +++ b/pus_tc/ploc_supervisor.py @@ -13,6 +13,15 @@ from tmtccmd.config.definitions import QueueCommands from tmtccmd.tc.packer import TcQueueT from tmtccmd.ecss.tc import PusTelecommand +latchup_id_dict = { + "0": "0.85V", + "1": "1.8V", + "2": "MISC", + "3": "3.3V", + "4": "NVM_4XO", + "5": "MISSION", + "6": "SAFECOTS" + } class SupvActionIds: HK_REPORT = 1 @@ -25,10 +34,40 @@ class SupvActionIds: RESET_MPSOC = 8 SET_TIME_REF = 9 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 + FACTORY_RESET = 37 + REQUEST_LOGGING_DATA = 38 + UPDATE_IMAGE_DATA = 39 class SupvHkIds: HK_REPORT = 52 + BOOT_STATUS_REPORT = 53 def pack_ploc_supv_test_into(object_id: bytearray, tc_queue: TcQueueT, op_code: str) -> TcQueueT: @@ -91,7 +130,117 @@ def pack_ploc_supv_test_into(object_id: bytearray, tc_queue: TcQueueT, op_code: elif op_code == "12": tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable HK")) command = object_id + struct.pack('!I', SupvActionIds.DISABLE_HK) - command = PusTelecommand(service=8, subservice=128, ssc=28, app_data=command) + command = PusTelecommand(service=8, subservice=128, ssc=29, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "13": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Request boot status report")) + 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")) + command = pack_lachtup_alert_cmd(object_id, True) + command = PusTelecommand(service=8, subservice=128, ssc=34, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "18": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable latchup alert")) + 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")) + command = pack_set_adc_enabled_channels_cmd(object_id) + command = PusTelecommand(service=8, subservice=128, ssc=40, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "24": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set ADC window and stride")) + command = pack_set_adc_window_and_stride_cmd(object_id) + command = PusTelecommand(service=8, subservice=128, ssc=41, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "25": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set ADC threshold")) + command = pack_set_adc_threshold_cmd(object_id) + command = PusTelecommand(service=8, subservice=128, ssc=42, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "26": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Request latchup status report")) + command = object_id + struct.pack('!I', SupvActionIds.GET_LATCHUP_STATUS_REPORT) + command = PusTelecommand(service=8, subservice=128, ssc=43, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "27": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Copy ADC data to MRAM")) + command = object_id + struct.pack('!I', SupvActionIds.COPY_ADC_DATA_TO_MRAM) + command = PusTelecommand(service=8, subservice=128, ssc=44, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "28": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enalbe/Disable NVMs")) + command = pack_enable_nvms_cmd(object_id) + command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "29": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Select NVM")) + command = pack_select_nvm_cmd(object_id) + command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "30": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Run auto EM tests")) + command = pack_auto_em_tests_cmd(object_id) + command = PusTelecommand(service=8, subservice=128, ssc=45, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + elif op_code == "31": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Wipe MRAM")) + 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 == "32": + tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Dump MRAM")) + command = pack_mram_dump_cmd(object_id) + command = PusTelecommand(service=8, subservice=128, ssc=47, 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()) return tc_queue @@ -112,3 +261,244 @@ def pack_sel_boot_image_cmd(object_id: bytearray, mem: int, bp0: int, bp1: int, command = command + struct.pack('!B', bp1) command = command + struct.pack('!B', bp2) return command + + +def pack_update_available_cmd(object_id: bytearray) -> bytearray: + """ + @brief This function packs the udpate availabe command. + @param object_id The object id of the PLOC supervisor handler. + """ + image_select = 1 + image_partition = 0 + image_size = 222 + image_crc = 0x0 + number_of_packets = 150 + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.UPDATE_AVAILABLE) + command = command + struct.pack('!B', image_select) + command = command + struct.pack('!B', image_partition) + command = command + struct.pack('!I', image_size) + command = command + struct.pack('!I', image_crc) + command = command + struct.pack('!I', number_of_packets) + return command + + +def pack_watchdogs_enable_cmd(object_id: bytearray) -> bytearray: + """ + @brief This function packs the command to enable or disable watchdogs on the PLOC. + @param object_id The object id of the PLOC supervisor handler. + @note Enable = 1, Disable = 0 + """ + watchdog_ps = 1 + watchdog_pl = 1 + watchdog_int = 0 + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.WATCHDOGS_ENABLE) + command = command + struct.pack('!B', watchdog_ps) + command = command + struct.pack('!B', watchdog_pl) + command = command + struct.pack('!B', watchdog_int) + return command + + +def pack_watchdog_config_timeout_cmd(object_id: bytearray) -> bytearray: + """ + @brief This function packs the command set the timeout of one of the three watchdogs of the PLOC. + @param object_id The object id of the PLOC supervisor handler. + """ + watchdog = int(input("Specify watchdog (0 - PS, 1 - PL, 2 - INT):")) + timeout = int(input("Specify timeout (1000 ms - 360000 ms):")) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.WATCHDOGS_CONFIG_TIMEOUT) + command = command + struct.pack('!B', watchdog) + command = command + struct.pack('!I', timeout) + return command + + +def pack_lachtup_alert_cmd(object_id: bytearray, state: bool) -> bytearray: + """ + @brief This function packs the command to enable or disable a certain latchup alerts. + @param object_id The object id of the PLOC supervisor handler. + @param state True - enable latchup alert, False - disable latchup alert + """ + latchup_id = get_latchup_id() + command = bytearray() + if state: + command = object_id + struct.pack('!I', SupvActionIds.ENABLE_LATCHUP_ALERT) + else: + command = object_id + struct.pack('!I', SupvActionIds.DISABLE_LATCHUP_ALERT) + command = command + struct.pack('!B', latchup_id) + return command + + +def pack_auto_calibrate_alert_cmd(object_id: bytearray) -> bytearray: + """ + @brief This function packs the command to auto calibrate a latchup alert. + @param object_id The object id of the PLOC supervisor handler. + """ + latchup_id = get_latchup_id() + mg = int(input("Specify MG:")) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.AUTO_CALIBRATE_ALERT) + command = command + struct.pack('!B', latchup_id) + command = command + struct.pack('!I', mg) + return command + + +def get_latchup_id() -> int: + key_column_width = 10 + description_column_width = 50 + separator_width = key_column_width + description_column_width + 3 + separator_string = separator_width * "-" + key_string = "Latchup ID".ljust(key_column_width) + description_string = "Description".ljust(description_column_width) + print(f"{key_string} | {description_string}") + print(separator_string) + for key in latchup_id_dict: + 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:")) + + +def pack_set_alert_limit_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. + """ + latchup_id = get_latchup_id() + dutycycle = int(input("Specify dutycycle:")) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.AUTO_CALIBRATE_ALERT) + command = command + struct.pack('!B', latchup_id) + command = command + struct.pack('!I', dutycycle) + 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 TP:")) + div = int(input("Specify 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:")) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.SET_ADC_ENABLED_CHANNELS) + command = command + struct.pack('!H', ch) + return command + + +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:")) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.SET_ADC_WINDOW_AND_STRIDE) + command = command + struct.pack('!H', window_size) + command = command + struct.pack('!H', striding_step_size) + return command + + +def pack_set_adc_threshold_cmd(object_id: bytearray) -> bytearray: + threshold = int(input("Specify threshold:")) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.SET_ADC_THRESHOLD) + command = command + struct.pack('!I', threshold) + 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):")) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.SELECT_NVM) + command = command + struct.pack('!B', mem) + return command + + +def pack_auto_em_tests_cmd(object_id: bytearray) -> bytearray: + 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) + return command + + +def pack_mram_wipe_cmd(object_id: bytearray) -> bytearray: + start = int(input("Start address: 0x"), 16) + stop = int(input("Stop address: 0x"), 16) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.WIPE_MRAM) + command = command + struct.pack('!I', start) + command = command + struct.pack('!I', stop) + return command + + +def pack_mram_dump_cmd(object_id: bytearray) -> bytearray: + start = int(input("Start address: 0x"), 16) + stop = int(input("Stop address: 0x"), 16) + command = bytearray() + command = object_id + struct.pack('!I', SupvActionIds.DUMP_MRAM) + command = command + struct.pack('!I', start) + command = command + struct.pack('!I', stop) + 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() + print("Specify debug verbosity") + 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: ")) + command = object_id + struct.pack('!I', SupvActionIds.SET_DBG_VERBOSITY) + command = command + struct.pack('!B', verbosity) + return command \ No newline at end of file diff --git a/pus_tm/service_8_hook.py b/pus_tm/service_8_hook.py index 4d50c4a..358be86 100644 --- a/pus_tm/service_8_hook.py +++ b/pus_tm/service_8_hook.py @@ -3,6 +3,7 @@ from typing import Tuple from config.object_ids import * from pus_tc.imtq import ImtqActionIds from pus_tc.ploc_mpsoc import PlocReplyIds +from pus_tc.ploc_supervisor import SupvActionIds def user_analyze_service_8_data( @@ -32,6 +33,8 @@ def user_analyze_service_8_data( return handle_imtq_replies(action_id, custom_data) elif object_id == PLOC_MPSOC_ID: return handle_ploc_replies(action_id, custom_data) + elif object_id == PLOC_SUPV_ID: + return handle_supervisor_replies(action_id, custom_data) else: header_list = [] content_list = [] @@ -56,3 +59,12 @@ def handle_ploc_replies(action_id: int, custom_data: bytearray) -> Tuple[list, l header_list = ['PLOC Memory Address', 'PLOC Mem Len', 'PLOC Read Memory Data'] content_list = [custom_data[:4], custom_data[4:6], custom_data[6:10]] return header_list, content_list + + +def handle_supervisor_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]: + header_list = [] + content_list = [] + if action_id == SupvActionIds.DUMP_MRAM: + header_list = ['MRAM Dump'] + content_list = [custom_data[:len(custom_data)]] + return header_list, content_list