supervisor commands

pus_tc/devs/ploc_supervisor.py

@@ -13,6 +13,7 @@ from tmtccmd.config.definitions import QueueCommands
 from tmtccmd.tc.packer import TcQueueT
 from tmtccmd.utility.logger import get_console_logger
 from tmtccmd.tc.service_200_mode import pack_mode_data, Modes
+from utility.input_helper import InputHelper
 
 LOGGER = get_console_logger()
 
@@ -26,6 +27,13 @@ latchup_id_dict = {
     "6": "SAFECOTS",
 }
 
+MANUAL_INPUT = "1"
+
+update_file_dict = {
+    MANUAL_INPUT: ["manual input", ""],
+    "2": ["/mnt/sd0/ploc/supervisor/update.bin", "/mnt/sd0/ploc/supervisor/update.bin"],
+}
+
 
 class SupvActionIds:
     HK_REPORT = 1
@@ -69,6 +77,13 @@ class SupvActionIds:
     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
 
 
 class SupvHkIds:
@@ -361,6 +376,51 @@ def pack_ploc_supv_commands(
         )
         command = PusTelecommand(service=8, subservice=128, ssc=56, app_data=command)
         tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "42":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Perform update"))
+        command = pack_update_command(object_id)
+        command = PusTelecommand(service=8, subservice=128, ssc=57, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "43":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Terminate supervisor process"))
+        command = object_id + struct.pack(
+            "!I", SupvActionIds.TERMINATE_SUPV_HELPER
+        )
+        command = PusTelecommand(service=8, subservice=128, ssc=58, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "44":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Start MPSoC quiet"))
+        command = object_id + struct.pack(
+            "!I", SupvActionIds.START_MPSOC_QUIET
+        )
+        command = PusTelecommand(service=8, subservice=128, ssc=59, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "45":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Set shutdown timeout"))
+        command = pack_set_shutdown_timeout_command(object_id)
+        command = PusTelecommand(service=8, subservice=128, ssc=60, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "46":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Factory flash"))
+        command = object_id + struct.pack(
+            "!I", SupvActionIds.FACTORY_FLASH
+        )
+        command = PusTelecommand(service=8, subservice=128, ssc=61, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "47":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Enable auto TM"))
+        command = object_id + struct.pack(
+            "!I", SupvActionIds.ENABLE_AUTO_TM
+        )
+        command = PusTelecommand(service=8, subservice=128, ssc=62, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "48":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Disable auto TM"))
+        command = object_id + struct.pack(
+            "!I", SupvActionIds.DISABLE_AUTO_TM
+        )
+        command = PusTelecommand(service=8, subservice=128, ssc=63, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
 
     return tc_queue
 
@@ -605,6 +665,30 @@ def pack_set_debug_verbosity_cmd(object_id: bytearray) -> bytearray:
     return command
 
 
+def pack_update_command(object_id: bytearray) -> bytearray:
+    command = bytearray()
+    memory_id = int(input("Specify memory ID: "))
+    start_address = int(input("Specify start address: 0x"), 16)
+    update_file = get_update_file()
+    command += object_id
+    command += struct.pack('!I', SupvActionIds.PERFORM_UPDATE)
+    command += bytearray(update_file, 'utf-8')
+    # Adding null terminator
+    command += bytes(0)
+    command += struct.pack('!B', memory_id)
+    command += struct.pack('!I', start_address)
+    return command
+
+
+def pack_set_shutdown_timeout_command(object_id: bytearray) -> 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 get_debug_verbosity() -> int:
     tries = 0
     while tries < 3:
@@ -650,3 +734,14 @@ def pack_read_gpio_cmd(object_id: bytearray) -> bytearray:
     command = command + struct.pack("!B", port)
     command = command + struct.pack("!B", pin)
     return command
+
+
+def get_update_file() -> str:
+    LOGGER.info("Specify update file ")
+    input_helper = InputHelper(update_file_dict)
+    key = input_helper.get_key()
+    if key == MANUAL_INPUT:
+        file = input("Ploc MPSoC: Specify absolute name of update file: ")
+    else:
+        file = update_file_dict[key][1]
+    return file