Merge branch 'v6.0.0-dev' into add-supv-latchup-report

eive_tmtc/tmtc/payload/ploc_supervisor.py

@@ -11,19 +11,15 @@ import enum
 import logging
 import struct
 
-from spacepackets.ecss.tc import PusTelecommand
-from tmtccmd.config import TmtcDefinitionWrapper
-from tmtccmd.config.tmtc import OpCodeEntry, tmtc_definitions_provider
-from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
-from tmtccmd.pus.s8_fsfw_action import create_action_cmd
-from tmtccmd.pus.s200_fsfw_mode import Mode, pack_mode_data
-from tmtccmd.pus.tc.s3_fsfw_hk import generate_one_hk_command, make_sid
-from tmtccmd.tmtc import service_provider
-from tmtccmd.tmtc.decorator import ServiceProviderParams
-
-from eive_tmtc.config.definitions import CustomServiceList
 from eive_tmtc.config.object_ids import PLOC_SUPV_ID, get_object_ids
 from eive_tmtc.pus_tm.defs import PrintWrapper
+from spacepackets.ecss.tc import PusTelecommand
+from tmtccmd.pus.tc.s3_fsfw_hk import generate_one_hk_command, make_sid
+from tmtccmd.config.tmtc import CmdTreeNode
+from tmtccmd.tmtc import DefaultPusQueueHelper
+from tmtccmd.pus.s200_fsfw_mode import pack_mode_data, Mode
+from tmtccmd.pus.s8_fsfw_action import create_action_cmd
+from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
 from eive_tmtc.utility.input_helper import InputHelper
 
 _LOGGER = logging.getLogger(__name__)
@@ -153,9 +149,11 @@ class OpCode:
 class Info(str, enum.Enum):
     OFF = "Switch Off"
     ON = "Switch On"
-    NML = "Switch Normal"
+    NORMAL = "Switch Normal"
     HK_TO_OBC = "Request  HK from PLOC SUPV"
     REQUEST_GENERIC_HK_SET = "Request prompted HK set from PLOC Handler"
+    START_MPSOC = "Start MPSoC"
+    SHUTDOWN_MPSOC = "Shutdown MPSoC"
     SET_TIME_REF = "Set time reference"
     FACTORY_FLASH = "Factory Flash Mode"
     PERFORM_UPDATE = "Start or continue MPSoC SW update at starting bytes"
@@ -169,79 +167,49 @@ class Info(str, enum.Enum):
     READ_STATUS_REPORT = "Read HK status report"
 
 
-@tmtc_definitions_provider
-def add_ploc_supv_cmds(defs: TmtcDefinitionWrapper):
-    oce = OpCodeEntry()
-    oce.add(OpCode.OFF, Info.OFF)
-    oce.add(OpCode.ON, Info.ON)
-    oce.add(OpCode.NORMAL, Info.NML)
-    oce.add(OpCode.HK_TO_OBC, Info.HK_TO_OBC)
-    oce.add(OpCode.REQUEST_GENERIC_HK_SET, Info.REQUEST_GENERIC_HK_SET)
-    oce.add(OpCode.START_MPSOC, "PLOC Supervisor: Start MPSoC")
-    oce.add(OpCode.SHUTDOWN_MPSOC, "PLOC Supervisor: Shutdown MPSoC")
-    oce.add(OpCode.SEL_NVM, Info.SEL_NVM)
-    oce.add(OpCode.SET_TIME_REF, Info.SET_TIME_REF)
-    oce.add(OpCode.FACTORY_RESET, Info.FACTORY_RESET)
-    oce.add(OpCode.RESET_MPSOC, Info.RESET_MPSOC)
-    oce.add("8", "PLOC Supervisor: Set max restart tries")
-    oce.add("11", "PLOC Supervisor: Set boot timeout")
-    oce.add(OpCode.READ_STATUS_REPORT, Info.READ_STATUS_REPORT)
-    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(OpCode.SET_GPIO, Info.SET_GPIO)
-    oce.add(OpCode.READ_GPIO, Info.READ_GPIO)
-    oce.add("37", "PLOC Supervisor: Restart supervisor")
-    oce.add(OpCode.PERFORM_UPDATE, Info.PERFORM_UPDATE)
-    oce.add(OpCode.START_UPDATE, Info.START_UPDATE)
-    oce.add("43", "PLOC Supervisor: Terminate supervisor process")
-    oce.add("45", "PLOC Supervisor: Set shutdown timeout")
-    oce.add(OpCode.FACTORY_FLASH, Info.FACTORY_FLASH)
-    oce.add("52", "PLOC Supervisor: Logging clear counters")
-    oce.add("53", "PLOC Supervisor: Logging set topic")
-    oce.add("56", "PLOC Supervisor: Reset PL")
-    oce.add("57", "PLOC Supervisor: Enable NVMs")
-    oce.add("58", "PLOC Supervisor: Continue update")
-    oce.add(OpCode.MEM_CHECK, Info.MEM_CHECK)
-
-    defs.add_service(CustomServiceList.PLOC_SUPV.value, "PLOC Supervisor", oce)
+def create_ploc_supv_node() -> CmdTreeNode:
+    op_code_strs = [
+        getattr(OpCode, key) for key in dir(OpCode) if not key.startswith("__")
+    ]
+    info_strs = [getattr(Info, key) for key in dir(OpCode) if not key.startswith("__")]
+    combined_dict = dict(zip(op_code_strs, info_strs))
+    node = CmdTreeNode("ploc_supv", "PLOC Supervisor", hide_children_for_print=True)
+    for op_code, info in combined_dict.items():
+        node.add_child(CmdTreeNode(op_code, info))
+    return node
 
 
-@service_provider(CustomServiceList.PLOC_SUPV)
-def pack_ploc_supv_commands(p: ServiceProviderParams):  # noqa C901
-    q = p.queue_helper
-    op_code = p.op_code
+def pack_ploc_supv_commands(q: DefaultPusQueueHelper, cmd_str: str):  # noqa C901
     object_id = get_object_ids().get(PLOC_SUPV_ID)
     assert object_id is not None
     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 == OpCode.OFF:
+    if cmd_str == OpCode.OFF:
         q.add_log_cmd(f"{prefix}: {Info.OFF}")
         command = pack_mode_data(object_id.as_bytes, Mode.OFF, 0)
         q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
-    if op_code == OpCode.ON:
+    if cmd_str == OpCode.ON:
         q.add_log_cmd(f"{prefix}: {Info.ON}")
         command = pack_mode_data(object_id.as_bytes, Mode.ON, 0)
         q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
-    if op_code == OpCode.NORMAL:
-        q.add_log_cmd(f"{prefix}: {Info.NML}")
+    if cmd_str == OpCode.NORMAL:
+        q.add_log_cmd(f"{prefix}: {Info.NORMAL}")
         command = pack_mode_data(object_id.as_bytes, Mode.NORMAL, 0)
         q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=command))
-    if op_code == OpCode.HK_TO_OBC:
+    if cmd_str == OpCode.HK_TO_OBC:
         q.add_log_cmd(f"{prefix}: {Info.HK_TO_OBC}")
         command = obyt + struct.pack("!I", SupvActionId.REQUEST_HK_REPORT)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == OpCode.REQUEST_GENERIC_HK_SET:
-        q.add_log_cmd(f"{prefix}: {Info.REQUEST_GENERIC_HK_SET}")
-        sid = make_sid(object_id.as_bytes, prompt_set_id())
-        cmd = generate_one_hk_command(sid)
-        q.add_pus_tc(cmd)
-    if op_code == OpCode.READ_STATUS_REPORT:
+    if cmd_str == OpCode.START_MPSOC:
+        q.add_log_cmd("PLOC Supervisor: Start MPSoC")
+        command = obyt + struct.pack("!I", SupvActionId.START_MPSOC)
+        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
+    if cmd_str == OpCode.SHUTDOWN_MPSOC:
+        q.add_log_cmd("PLOC Supervisor: Shutdown MPSoC")
+        command = object_id.as_bytes + struct.pack("!I", SupvActionId.SHUTWOWN_MPSOC)
+        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
+    if cmd_str == OpCode.READ_STATUS_REPORT:
         q.add_log_cmd(f"{prefix}: {Info.READ_STATUS_REPORT}")
         set_id = prompt_set_id()
         action_cmd = None
@@ -274,15 +242,15 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):  # noqa C901
         q.add_wait_seconds(2.0)
         q.add_pus_tc(req_hk)
         assert action_cmd is not None
-    elif op_code == OpCode.START_MPSOC:
+    elif cmd_str == OpCode.START_MPSOC:
         q.add_log_cmd("PLOC Supervisor: Start MPSoC")
         command = obyt + struct.pack("!I", SupvActionId.START_MPSOC)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == OpCode.SHUTDOWN_MPSOC:
+    if cmd_str == OpCode.SHUTDOWN_MPSOC:
         q.add_log_cmd("PLOC Supervisor: Shutdown MPSoC")
         command = object_id.as_bytes + struct.pack("!I", SupvActionId.SHUTWOWN_MPSOC)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == OpCode.SEL_NVM:
+    if cmd_str == OpCode.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): "))
@@ -290,7 +258,7 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):  # noqa C901
         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 == OpCode.FACTORY_RESET:
+    if cmd_str == OpCode.FACTORY_RESET:
         q.add_log_cmd(f"{prefix}: {Info.FACTORY_RESET}")
         while True:
             print("Please select the key for a factory reset operation")
@@ -307,7 +275,7 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):  # noqa C901
                 user_data=bytes([key]),
             )
         )
-    if op_code == "8":
+    if cmd_str == "8":
         q.add_log_cmd("PLOC Supervisor: Set max restart tries")
         restart_tries = int(input("Specify maximum restart tries: "))
         command = (
@@ -316,15 +284,15 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):  # noqa C901
             + struct.pack("!B", restart_tries)
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == OpCode.RESET_MPSOC:
+    if cmd_str == OpCode.RESET_MPSOC:
         q.add_log_cmd(Info.RESET_MPSOC)
         command = object_id.as_bytes + struct.pack("!I", SupvActionId.RESET_MPSOC)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == OpCode.SET_TIME_REF:
+    if cmd_str == OpCode.SET_TIME_REF:
         q.add_log_cmd("PLOC Supervisor: Set time reference")
         command = object_id.as_bytes + struct.pack("!I", SupvActionId.SET_TIME_REF)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == "11":
+    if cmd_str == "11":
         q.add_log_cmd("PLOC Supervisor: Set boot timeout")
         boot_timeout = int(input("Specify boot timeout [ms]: "))
         command = (
@@ -333,83 +301,63 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):  # noqa C901
             + struct.pack("!I", boot_timeout)
         )
         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":
+    if cmd_str == "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":
+    if cmd_str == "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":
+    if cmd_str == "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 == "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 == OpCode.SET_GPIO:
+    if cmd_str == OpCode.SET_GPIO:
         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 == OpCode.READ_GPIO:
+    if cmd_str == OpCode.READ_GPIO:
         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":
+    if cmd_str == "37":
         q.add_log_cmd("PLOC Supervisor: Restart supervisor")
         command = object_id.as_bytes + struct.pack(
             "!I", SupvActionId.RESTART_SUPERVISOR
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code in OpCode.START_UPDATE:
+    if cmd_str in OpCode.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 OpCode.PERFORM_UPDATE:
+    if cmd_str in OpCode.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":
+    if cmd_str == "43":
         q.add_log_cmd("PLOC Supervisor: Terminate supervisor process")
         command = object_id.as_bytes + struct.pack(
             "!I", SupvActionId.TERMINATE_SUPV_HELPER
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == "44":
+    if cmd_str == "44":
         q.add_log_cmd("PLOC Supervisor: Start MPSoC quiet")
         command = object_id.as_bytes + struct.pack("!I", SupvActionId.START_MPSOC_QUIET)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == "45":
+    if cmd_str == "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 OpCode.FACTORY_FLASH:
+    if cmd_str in OpCode.FACTORY_FLASH:
         q.add_log_cmd(f"{prefix}: {Info.FACTORY_FLASH}")
         command = object_id.as_bytes + struct.pack("!I", SupvActionId.FACTORY_FLASH)
         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", SupvActionId.LOGGING_CLEAR_COUNTERS
-        )
-        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", SupvActionId.REQUEST_LOGGING_COUNTERS
-        )
-        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == "56":
+    if cmd_str == "56":
         q.add_log_cmd("PLOC Supervisor: Reset PL")
         command = object_id.as_bytes + struct.pack("!I", SupvActionId.RESET_PL)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == "57":
+    if cmd_str == "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: "))
@@ -420,11 +368,11 @@ def pack_ploc_supv_commands(p: ServiceProviderParams):  # noqa C901
             + struct.pack("B", nvm3)
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == "58":
+    if cmd_str == "58":
         q.add_log_cmd("PLOC Supervisor: Continue update")
         command = object_id.as_bytes + struct.pack("!I", SupvActionId.CONTINUE_UPDATE)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=command))
-    if op_code == OpCode.MEM_CHECK:
+    if cmd_str == OpCode.MEM_CHECK:
         custom_data = bytearray()
         update_file = get_update_file()
         memory_id = int(input("Specify memory ID: "))