# -*- coding: utf-8 -*- """ @file ploc_supervisor.py @brief Tests for commanding the supervisor of the PLOC. The supervisor is programmed by Thales. @author J. Meier @date 10.07.2021 """ import enum import struct from config.object_ids import PLOC_SUPV_ID, get_object_ids from config.definitions import CustomServiceList from pus_tm.defs import PrintWrapper from spacepackets.ecss.tc import PusTelecommand from tmtccmd.tc.pus_3_fsfw_hk import generate_one_hk_command, make_sid from tmtccmd.config import TmtcDefinitionWrapper from tmtccmd.config.tmtc import tmtc_definitions_provider, OpCodeEntry from tmtccmd.logging import get_console_logger from tmtccmd.tc import service_provider from tmtccmd.tc.decorator import ServiceProviderParams from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes from tmtccmd.tc.pus_8_funccmd import make_fsfw_action_cmd from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter from utility.input_helper import InputHelper LOGGER = get_console_logger() latchup_id_dict = { "0": "0.85V", "1": "1.8V", "2": "MISC", "3": "3.3V", "4": "NVM_4XO", "5": "MISSION", "6": "SAFECOTS", } HARDCODED = "0" MANUAL_INPUT = "1" HARDCODED_FILE = "/home/rmueller/EIVE/eive-obsw/tmtc/filetest/mpsoc_boot.bin" update_file_dict = { HARDCODED: ["hardcoded", ""], MANUAL_INPUT: ["manual input", ""], "2": ["/mnt/sd0/ploc/supervisor/update.bin", "/mnt/sd0/ploc/supervisor/update.bin"], "3": [ "/mnt/sd0/ploc/supervisor/update-large.bin", "/mnt/sd0/ploc/supervisor/update-large.bin", ], "4": [ "/mnt/sd0/ploc/supervisor/update-small.bin", "/mnt/sd0/ploc/supervisor/update-small.bin", ], "5": [ "/mnt/sd0/ploc/supervisor/mpsoc-uart-working.bin", "/mnt/sd0/ploc/supervisor/mpsoc-uart-working.bin", ], } event_buffer_path_dict = { MANUAL_INPUT: ["manual input", ""], "2": ["/mnt/sd0/ploc/supervisor", "/mnt/sd0/ploc/supervisor"], } class SupvActionIds: HK_REPORT = 1 START_MPSOC = 3 SHUTWOWN_MPSOC = 4 SEL_MPSOC_BOOT_IMAGE = 5 SET_BOOT_TIMEOUT = 6 SET_MAX_RESTART_TRIES = 7 RESET_MPSOC = 8 SET_TIME_REF = 9 DISABLE_HK = 10 GET_BOOT_STATUS_REPORT = 11 UPDATE_AVAILABLE = 12 ENABLE_LATCHUP_ALERT = 15 DISABLE_LATCHUP_ALERT = 16 SET_ALERT_LIMIT = 18 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 SELECT_NVM = 27 RUN_AUTO_EM_TESTS = 28 WIPE_MRAM = 29 DUMP_MRAM = 30 SET_GPIO = 34 READ_GPIO = 35 RESTART_SUPERVISOR = 36 FACTORY_RESET_CLEAR_ALL = 37 LOGGING_REQUEST_COUNTERS = 38 UPDATE_IMAGE_DATA = 39 FACTORY_RESET_CLEAR_MIRROR = 40 FACTORY_RESET_CLEAR_CIRCULAR = 41 START_MPSOC_QUIET = 45 SET_SHUTDOWN_TIMEOUT = 46 FACTORY_FLASH = 47 PERFORM_UPDATE = 48 TERMINATE_SUPV_HELPER = 49 ENABLE_AUTO_TM = 50 DISABLE_AUTO_TM = 51 LOGGING_REQUEST_EVENT_BUFFERS = 54 LOGGING_CLEAR_COUNTERS = 55 LOGGING_SET_TOPIC = 56 REQUEST_ADC_REPORT = 57 RESET_PL = 58 ENABLE_NVMS = 59 CONTINUE_UPDATE = 60 MEM_CHECK = 61 class SetIds: HK_REPORT = 102 BOOT_STATUS_REPORT = 103 class OpCodes: OFF = ["0", "off"] ON = ["1", "on"] NORMAL = ["2", "nml"] HK_TO_OBC = ["3", "hk_to_obc"] REQUEST_HK = ["4", "req_hk"] START_MPSOC = ["5", "start_mpsoc"] SHUTDOWN_MPSOC = ["6", "stop_mpsoc"] SEL_NVM = ["7", "sel_nvm"] SET_TIME_REF = ["set_time_ref"] FACTORY_FLASH = ["factory_flash"] REQ_BOOT_STATUS_REPORT = ["13", "boot_report"] START_UPDATE = ["42", "start_update"] PERFORM_UPDATE = ["update"] MEM_CHECK = ["mem_check"] class Info(str, enum.Enum): value: str OFF = "Switch Off" ON = "Switch On" NML = "Switch Normal" HK_TO_OBC = "Request HK from PLOC SUPV" REQUEST_HK = "Request HK set from PLOC Handler" SET_TIME_REF = "Set time reference" FACTORY_FLASH = "Factory Flash Mode" PERFORM_UPDATE = "Start or continue MPSoC SW update at starting bytes" START_UPDATE = "Start new MPSoC SW update" REQ_BOOT_STATUS_REPORT = "Request boot status report and HK" MEM_CHECK = "Memory Check" SEL_NVM = "Select NVM" @tmtc_definitions_provider def add_ploc_supv_cmds(defs: TmtcDefinitionWrapper): oce = OpCodeEntry() oce.add(OpCodes.OFF, Info.OFF) oce.add(OpCodes.ON, Info.ON) oce.add(OpCodes.NORMAL, Info.NML) oce.add(OpCodes.HK_TO_OBC, Info.HK_TO_OBC) oce.add(OpCodes.REQUEST_HK, Info.REQUEST_HK) oce.add(OpCodes.START_MPSOC, "PLOC Supervisor: Start MPSoC") oce.add(OpCodes.SHUTDOWN_MPSOC, "PLOC Supervisor: Shutdown MPSoC") oce.add(OpCodes.SEL_NVM, Info.SEL_NVM) oce.add(OpCodes.SET_TIME_REF, Info.SET_TIME_REF) oce.add("8", "PLOC Supervisor: Set max restart tries") oce.add("9", "PLOC Supervisor: Reset MPSoC") oce.add("11", "PLOC Supervisor: Set boot timeout") oce.add("12", "PLOC Supervisor: Disable Hk") oce.add(OpCodes.REQ_BOOT_STATUS_REPORT, Info.REQ_BOOT_STATUS_REPORT) oce.add("17", "PLOC Supervisor: Enable latchup alert") oce.add("18", "PLOC Supervisor: Disable latchup alert") oce.add("20", "PLOC Supervisor: Set alert limit") oce.add("23", "PLOC Supervisor: Set ADC enabled channels") oce.add("24", "PLOC Supervisor: Set ADC window and stride") oce.add("25", "PLOC Supervisor: Set ADC threshold") oce.add("26", "PLOC Supervisor: Request latchup status report") oce.add("27", "PLOC Supervisor: Copy ADC data to MRAM") oce.add("30", "PLOC Supervisor: Run auto EM tests") oce.add("31", "PLOC Supervisor: MRAM Wipe") oce.add("35", "PLOC Supervisor: Set GPIO") oce.add("36", "PLOC Supervisor: Read GPIO") oce.add("37", "PLOC Supervisor: Restart supervisor") oce.add("38", "PLOC Supervisor: Factory reset clear all") oce.add("39", "PLOC Supervisor: Factory reset clear mirror entries") oce.add("40", "PLOC Supervisor: Factory reset clear circular entries") oce.add(OpCodes.PERFORM_UPDATE, Info.PERFORM_UPDATE) oce.add(OpCodes.START_UPDATE, Info.START_UPDATE) oce.add("43", "PLOC Supervisor: Terminate supervisor process") oce.add("44", "PLOC Supervisor: Start MPSoC quiet") oce.add("45", "PLOC Supervisor: Set shutdown timeout") oce.add(OpCodes.FACTORY_FLASH, Info.FACTORY_FLASH) oce.add("47", "PLOC Supervisor: Enable auto TM") oce.add("48", "PLOC Supervisor: Disable auto TM") oce.add("51", "PLOC Supervisor: Logging request event buffers") oce.add("52", "PLOC Supervisor: Logging clear counters") oce.add("53", "PLOC Supervisor: Logging set topic") oce.add("54", "PLOC Supervisor: Logging request counters") oce.add("55", "PLOC Supervisor: Request ADC Report") oce.add("56", "PLOC Supervisor: Reset PL") oce.add("57", "PLOC Supervisor: Enable NVMs") oce.add("58", "PLOC Supervisor: Continue update") oce.add(OpCodes.MEM_CHECK, Info.MEM_CHECK) defs.add_service(CustomServiceList.PLOC_SUPV.value, "PLOC Supervisor", oce) @service_provider(CustomServiceList.PLOC_SUPV) def pack_ploc_supv_commands(p: ServiceProviderParams): q = p.queue_helper op_code = p.op_code object_id = get_object_ids().get(PLOC_SUPV_ID) q.add_log_cmd(f"Testing PLOC Supervisor with object id: {object_id.as_hex_string}") obyt = object_id.as_bytes prefix = "PLOC Supervisor" if op_code in OpCodes.OFF: q.add_log_cmd(f"{prefix}: {Info.OFF}") command = pack_mode_data(object_id.as_bytes, Modes.OFF, 0) q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command)) if op_code in OpCodes.ON: q.add_log_cmd(f"{prefix}: {Info.ON}") command = pack_mode_data(object_id.as_bytes, Modes.ON, 0) q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command)) if op_code in OpCodes.NORMAL: q.add_log_cmd(f"{prefix}: {Info.NML}") command = pack_mode_data(object_id.as_bytes, Modes.NORMAL, 0) q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command)) if op_code in OpCodes.HK_TO_OBC: q.add_log_cmd(f"{prefix}: {Info.HK_TO_OBC}") command = obyt + struct.pack("!I", SupvActionIds.HK_REPORT) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.REQUEST_HK: q.add_log_cmd(f"{prefix}: {Info.REQUEST_HK}") sid = make_sid(object_id.as_bytes, SetIds.HK_REPORT) cmd = generate_one_hk_command(sid) q.add_pus_tc(cmd) elif op_code in OpCodes.START_MPSOC: q.add_log_cmd("PLOC Supervisor: Start MPSoC") command = obyt + struct.pack("!I", SupvActionIds.START_MPSOC) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.SHUTDOWN_MPSOC: q.add_log_cmd("PLOC Supervisor: Shutdown MPSoC") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.SHUTWOWN_MPSOC) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.SEL_NVM: q.add_log_cmd("PLOC Supervisor: Select MPSoC boot image") mem = int(input("MEM (NVM0 - 0 or NVM1 - 1): ")) bp0 = int(input("BP0 (0 or 1): ")) bp1 = int(input("BP1 (0 or 1): ")) bp2 = int(input("BP2 (0 or 1): ")) command = pack_sel_boot_image_cmd(object_id.as_bytes, mem, bp0, bp1, bp2) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "8": q.add_log_cmd("PLOC Supervisor: Set max restart tries") restart_tries = int(input("Specify maximum restart tries: ")) command = ( object_id.as_bytes + struct.pack("!I", SupvActionIds.SET_MAX_RESTART_TRIES) + struct.pack("!B", restart_tries) ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "9": q.add_log_cmd("PLOC Supervisor: Reset MPSoC") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.RESET_MPSOC) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.SET_TIME_REF: q.add_log_cmd("PLOC Supervisor: Set time reference") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.SET_TIME_REF) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "11": q.add_log_cmd("PLOC Supervisor: Set boot timeout") boot_timeout = int(input("Specify boot timeout [ms]: ")) command = ( object_id.as_bytes + struct.pack("!I", SupvActionIds.SET_BOOT_TIMEOUT) + struct.pack("!I", boot_timeout) ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "12": q.add_log_cmd("PLOC Supervisor: Disable HK") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.DISABLE_HK) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.REQ_BOOT_STATUS_REPORT: q.add_log_cmd(f"{prefix}: {Info.REQ_BOOT_STATUS_REPORT}") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.GET_BOOT_STATUS_REPORT ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) q.add_wait_seconds(2.0) sid = make_sid(object_id.as_bytes, SetIds.BOOT_STATUS_REPORT) req_hk = generate_one_hk_command(sid) q.add_pus_tc(req_hk) if op_code == "17": q.add_log_cmd("PLOC Supervisor: Enable latchup alert") command = pack_lachtup_alert_cmd(object_id.as_bytes, True) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "18": q.add_log_cmd("PLOC Supervisor: Disable latchup alert") command = pack_lachtup_alert_cmd(object_id.as_bytes, False) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "20": q.add_log_cmd("PLOC Supervisor: Set alert limit") command = pack_set_alert_limit_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "23": q.add_log_cmd("PLOC Supervisor: Set ADC enabled channels") command = pack_set_adc_enabled_channels_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "24": q.add_log_cmd("PLOC Supervisor: Set ADC window and stride") command = pack_set_adc_window_and_stride_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "25": q.add_log_cmd("PLOC Supervisor: Set ADC threshold") command = pack_set_adc_threshold_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "26": q.add_log_cmd("PLOC Supervisor: Request latchup status report") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.GET_LATCHUP_STATUS_REPORT ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "27": q.add_log_cmd("PLOC Supervisor: Copy ADC data to MRAM") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.COPY_ADC_DATA_TO_MRAM ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "30": q.add_log_cmd("PLOC Supervisor: Run auto EM tests") command = pack_auto_em_tests_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "31": q.add_log_cmd("PLOC Supervisor: Wipe MRAM") command = pack_mram_wipe_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "35": q.add_log_cmd("PLOC Supervisor: Set GPIO command") command = pack_set_gpio_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "36": q.add_log_cmd("PLOC Supervisor: Read GPIO command") command = pack_read_gpio_cmd(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "37": q.add_log_cmd("PLOC Supervisor: Restart supervisor") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.RESTART_SUPERVISOR ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "38": q.add_log_cmd("PLOC Supervisor: Factory reset clear all") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.FACTORY_RESET_CLEAR_ALL ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "39": q.add_log_cmd("PLOC Supervisor: Factory reset clear mirror entries") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.FACTORY_RESET_CLEAR_MIRROR ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "40": q.add_log_cmd("PLOC Supervisor: Factory reset clear circular entries") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.FACTORY_RESET_CLEAR_CIRCULAR ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.START_UPDATE: q.add_log_cmd("PLOC Supversior: Start new MPSoC SW update") command = pack_update_command(object_id.as_bytes, True) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.PERFORM_UPDATE: q.add_log_cmd("PLOC Supervisor: Perform MPSoC SW update") command = pack_update_command(object_id.as_bytes, False) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "43": q.add_log_cmd("PLOC Supervisor: Terminate supervisor process") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.TERMINATE_SUPV_HELPER ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "44": q.add_log_cmd("PLOC Supervisor: Start MPSoC quiet") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.START_MPSOC_QUIET ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "45": q.add_log_cmd("PLOC Supervisor: Set shutdown timeout") command = pack_set_shutdown_timeout_command(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.FACTORY_FLASH: q.add_log_cmd(f"{prefix}: {Info.FACTORY_FLASH}") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.FACTORY_FLASH) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "47": q.add_log_cmd("PLOC Supervisor: Enable auto TM") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.ENABLE_AUTO_TM) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "48": q.add_log_cmd("PLOC Supervisor: Disable auto TM") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.DISABLE_AUTO_TM) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "51": q.add_log_cmd("PLOC Supervisor: Logging request event buffers") command = pack_logging_buffer_request(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "52": q.add_log_cmd("PLOC Supervisor: Logging clear counters") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.LOGGING_CLEAR_COUNTERS ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "53": q.add_log_cmd("PLOC Supervisor: Logging set topic") command = pack_logging_set_topic(object_id.as_bytes) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "54": q.add_log_cmd("PLOC Supervisor: Logging request counters") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.LOGGING_REQUEST_COUNTERS ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "55": q.add_log_cmd("PLOC Supervisor: Request ADC report") command = object_id.as_bytes + struct.pack( "!I", SupvActionIds.REQUEST_ADC_REPORT ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "56": q.add_log_cmd("PLOC Supervisor: Reset PL") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.RESET_PL) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "57": q.add_log_cmd("PLOC Supervisor: Enable NVMs") nvm01 = int(input("Enable (1) or disable(0) NVM 0 and 1: ")) nvm3 = int(input("Enable (1) or disable(0) NVM 3: ")) command = ( object_id.as_bytes + struct.pack("!I", SupvActionIds.ENABLE_NVMS) + struct.pack("B", nvm01) + struct.pack("B", nvm3) ) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code == "58": q.add_log_cmd("PLOC Supervisor: Continue update") command = object_id.as_bytes + struct.pack("!I", SupvActionIds.CONTINUE_UPDATE) q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command)) if op_code in OpCodes.MEM_CHECK: custom_data = bytearray() update_file = get_update_file() memory_id = int(input("Specify memory ID: ")) start_address = int(input("Specify start address: 0x"), 16) custom_data.extend(update_file.encode("utf-8")) custom_data.append(0) custom_data.extend(struct.pack("!B", memory_id)) custom_data.extend(struct.pack("!I", start_address)) q.add_log_cmd( f"{prefix}: {Info.MEM_CHECK} for file {update_file} at memory ID {memory_id} at start " f"address {start_address}" ) command = make_fsfw_action_cmd( object_id.as_bytes, SupvActionIds.MEM_CHECK, custom_data ) q.add_pus_tc(command) def pack_sel_boot_image_cmd( object_id: bytes, mem: int, bp0: int, bp1: int, bp2: int ) -> bytearray: """This function can be used to generate the command to select the image from which the MPSoC will boot @param object_id The object id of the PLOC supervisor handler. @param mem The memory from which the MPSoC shall boot (NVM0 - 0, NVM1 - 1) @param bp0 Partition pin 0 @param bp1 Partition pin 1 @param bp2 Partition pin 2 """ command = object_id + struct.pack("!I", SupvActionIds.SEL_MPSOC_BOOT_IMAGE) command = command + struct.pack("!B", mem) command = command + struct.pack("!B", bp0) command = command + struct.pack("!B", bp1) command = command + struct.pack("!B", bp2) return bytearray(command) def pack_update_available_cmd(object_id: bytes) -> 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 = 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 bytearray(command) def pack_lachtup_alert_cmd(object_id: bytes, 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 bytearray(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: bytes) -> 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.SET_ALERT_LIMIT) command = command + struct.pack("!B", latchup_id) command = command + struct.pack("!I", dutycycle) return bytearray(command) def pack_set_adc_enabled_channels_cmd(object_id: bytes) -> 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) cmd = object_id + struct.pack("!I", SupvActionIds.SET_ADC_ENABLED_CHANNELS) cmd = cmd + struct.pack("!H", ch) return bytearray(cmd) def pack_set_adc_window_and_stride_cmd(object_id: bytes) -> bytearray: window_size = int(input("Specify window size: ")) striding_step_size = int(input("Specify striding step size: ")) 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 bytearray(command) def pack_set_adc_threshold_cmd(object_id: bytes) -> bytearray: threshold = int(input("Specify threshold: ")) command = object_id + struct.pack("!I", SupvActionIds.SET_ADC_THRESHOLD) command = command + struct.pack("!I", threshold) return bytearray(command) def pack_select_nvm_cmd(object_id: bytes) -> bytearray: mem = int(input("Specify NVM (0 - NVM0, 1 - MVM1): ")) command = object_id + struct.pack("!I", SupvActionIds.SELECT_NVM) command = command + struct.pack("!B", mem) return bytearray(command) def pack_auto_em_tests_cmd(object_id: bytes) -> bytearray: test = int(input("Specify test (1 - complete, 2 - short): ")) command = object_id + struct.pack("!I", SupvActionIds.RUN_AUTO_EM_TESTS) command = command + struct.pack("!B", test) return bytearray(command) def pack_mram_wipe_cmd(object_id: bytes) -> bytearray: start = int(input("Start address: 0x"), 16) stop = int(input("Stop address: 0x"), 16) command = object_id + struct.pack("!I", SupvActionIds.WIPE_MRAM) command = command + struct.pack("!I", start) command = command + struct.pack("!I", stop) return bytearray(command) def pack_update_command(object_id: bytes, new_update: bool) -> bytearray: command = bytearray() memory_id = int(input("Specify memory ID: ")) start_address = int(input("Specify start address: 0x"), 16) update_file = get_update_file() if new_update: init_bytes_written = 0 init_seq_count = 1 del_mem = True else: init_bytes_written = input("Specify bytes to start from [0 default]: ") if init_bytes_written == "": init_bytes_written = 0 init_bytes_written = int(init_bytes_written) init_seq_count = input("Specify initial sequence count [1 default]: ") if init_seq_count == "": init_seq_count = 1 init_seq_count = int(init_seq_count) del_mem = input("Delete memory? [y/n, y default]: ") if del_mem.lower() in ["y", "1"]: del_mem = 1 elif del_mem.lower() in ["n", "0"]: del_mem = 0 else: raise ValueError("Invalid input, use y or n") command += object_id command += struct.pack("!I", SupvActionIds.PERFORM_UPDATE) command += bytearray(update_file, "utf-8") # Adding null terminator command += struct.pack("!B", 0) command += struct.pack("!B", memory_id) command += struct.pack("!I", start_address) command.extend(struct.pack("!I", init_bytes_written)) command.extend(struct.pack("!H", init_seq_count)) command.append(del_mem) return bytearray(command) def pack_set_shutdown_timeout_command(object_id: bytes) -> bytearray: command = bytearray() command += object_id command += struct.pack("!I", SupvActionIds.SET_SHUTDOWN_TIMEOUT) timeout = int(input("Specify shutdown timeout (ms): ")) command += struct.pack("!I", timeout) return command def pack_logging_buffer_request(object_id: bytes) -> bytearray: command = bytearray() command += object_id command += struct.pack("!I", SupvActionIds.LOGGING_REQUEST_EVENT_BUFFERS) path = get_event_buffer_path() command += bytearray(path, "utf-8") return command def pack_set_gpio_cmd(object_id: bytes) -> bytearray: port = int(input("Specify port: 0x"), 16) pin = int(input("Specify pin: 0x"), 16) val = int(input("Specify val: 0x"), 16) command = object_id + struct.pack("!I", SupvActionIds.SET_GPIO) command = command + struct.pack("!B", port) command = command + struct.pack("!B", pin) command = command + struct.pack("!B", val) return bytearray(command) def pack_read_gpio_cmd(object_id: bytes) -> bytearray: port = int(input("Specify port: 0x"), 16) pin = int(input("Specify pin: 0x"), 16) command = object_id + struct.pack("!I", SupvActionIds.READ_GPIO) command = command + struct.pack("!B", port) command = command + struct.pack("!B", pin) return bytearray(command) def pack_logging_set_topic(object_id: bytes) -> bytearray: command = object_id + struct.pack("!I", SupvActionIds.LOGGING_SET_TOPIC) tpc = int(input("Specify logging topic: ")) command += struct.pack("!B", tpc) return bytearray(command) def get_update_file() -> str: LOGGER.info("Specify update file ") input_helper = InputHelper(update_file_dict) key = input_helper.get_key() if key == HARDCODED: file = HARDCODED_FILE elif key == MANUAL_INPUT: file = input("Ploc Supervisor: Specify absolute name of update file: ") else: file = update_file_dict[key][1] return file def get_event_buffer_path() -> str: LOGGER.info("Specify path where to store event buffer file ") input_helper = InputHelper(event_buffer_path_dict) key = input_helper.get_key() if key == MANUAL_INPUT: file = input("Ploc Supervisor: Specify path: ") else: file = event_buffer_path_dict[key][1] return file def handle_supv_hk_data(set_id: int, hk_data: bytes, printer: FsfwTmTcPrinter): pw = PrintWrapper(printer) current_idx = 0 if set_id == SetIds.HK_REPORT: pass elif set_id == SetIds.BOOT_STATUS_REPORT: fmt_str = "!BBIIBBBBBB" inc_len = struct.calcsize(fmt_str) ( soc_state, power_cycles, boot_after_ms, boot_timeout_ms, active_nvm, bp_0_state, bp_1_state, bp_2_state, boot_state, boot_cycles, ) = struct.unpack(fmt_str, hk_data[0 : 0 + inc_len]) current_idx += inc_len pw.dlog( f"SoC state (0:off, 1:booting, 2:update, 3:operating, 4:shutdown, 5:reset): {soc_state}" ) pw.dlog(f"Power Cycles {power_cycles}") pw.dlog(f"Boot after {boot_after_ms} ms | Boot timeout {boot_timeout_ms} ms") pw.dlog(f"Active NVM: {active_nvm}") pw.dlog( f"BP0 State {bp_0_state} | BP1 State {bp_1_state} | BP2 State {bp_2_state}" ) pw.dlog(f"Boot State {boot_state} | Boot Cycles {boot_cycles}") pw.printer.print_validity_buffer(hk_data[current_idx:], 10) else: pw.dlog(f"PLOC SUPV: HK handling not implemented for set ID {set_id}") pw.dlog(f"Raw Data: 0x[{hk_data.hex(sep=',')}]")