Merge pull request 'MPSoC flash content reporter' (#190) from mpsoc_action_reply_handler into main

Reviewed-on: #190
Reviewed-by: Marius Eggert <eggertm@irs.uni-stuttgart.de>

CHANGELOG.md

@@ -14,6 +14,7 @@ list yields a list of all related PRs for each release.
 
 - Event handling for reboot counter events.
 - Start adding new MPSoC commands, improve MPSoC commanding module a bit.
+- Handling for PLOC MPSoC flash content report.
 
 ## Fixed
 

eive_tmtc/pus_tm/action_reply_handler.py

@@ -3,7 +3,7 @@ from eive_tmtc.config.object_ids import *
 from eive_tmtc.tmtc.acs.imtq import ImtqActionId
 from eive_tmtc.pus_tm.defs import PrintWrapper
 from eive_tmtc.tmtc.core import handle_core_ctrl_action_replies
-from eive_tmtc.tmtc.payload.ploc_mpsoc import PlocReplyIds
+from eive_tmtc.tmtc.payload.ploc_mpsoc import handle_mpsoc_data_reply
 from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId
 from eive_tmtc.tmtc.acs.star_tracker import StarTrackerActionId
 from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
@@ -23,7 +23,6 @@ def handle_action_reply(
     tm_packet = Service8FsfwTm.unpack(
         raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty()
     )
-    printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
     object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
     pw = PrintWrapper(printer)
     custom_data = tm_packet.custom_data
@@ -35,7 +34,7 @@ def handle_action_reply(
     if object_id.as_bytes == IMTQ_HANDLER_ID:
         return handle_imtq_replies(action_id, printer, custom_data)
     elif object_id.as_bytes == PLOC_MPSOC_ID:
-        return handle_ploc_replies(action_id, printer, custom_data)
+        return handle_mpsoc_data_reply(action_id, printer, custom_data)
     elif object_id.as_bytes == PLOC_SUPV_ID:
         return handle_supervisor_replies(action_id, printer, custom_data)
     elif object_id.as_bytes == CORE_CONTROLLER_ID:
@@ -71,36 +70,6 @@ def handle_imtq_replies(
         printer.file_logger.info(content_list)
 
 
-def handle_ploc_replies(
-    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
-):
-    if action_id == PlocReplyIds.TM_MEM_READ_RPT:
-        header_list = [
-            "PLOC Memory Address",
-            "PLOC Mem Len",
-            "PLOC Read Memory Data",
-        ]
-        content_list = [
-            "0x" + custom_data[:4].hex(),
-            struct.unpack("!H", custom_data[4:6])[0],
-            "0x" + custom_data[6:10].hex(),
-        ]
-        print(header_list)
-        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(
     action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
 ):

eive_tmtc/tmtc/payload/ploc_mpsoc.py

@@ -6,6 +6,7 @@
 @author J. Meier
 @date   06.03.2021
 """
+import dataclasses
 import logging
 import struct
 import enum
@@ -59,6 +60,7 @@ class SetId(enum.IntEnum):
 class ActionId(enum.IntEnum):
     TC_MEM_WRITE = 1
     TC_MEM_READ = 2
+    TM_MEM_READ_RPT = 6
     TC_FLASH_WRITE_FULL_FILE = 9
     TC_FLASH_DELETE = 10
     TC_REPLAY_START = 11
@@ -70,6 +72,7 @@ class ActionId(enum.IntEnum):
     TC_MODE_REPLAY = 16
     TC_CAM_CMD_SEND = 17
     TC_MODE_IDLE = 18
+    TM_CAM_CMD_RPT = 19
     SET_UART_TX_TRISTATE = 20
     RELEASE_UART_TX = 21
     TC_CAM_TAKE_PIC = 22
@@ -77,6 +80,7 @@ class ActionId(enum.IntEnum):
     TC_DOWNLINK_DATA_MODULATE = 24
     TC_MODE_SNAPSHOT = 25
     TC_FLASH_DIR_GET_CONTENT = 28
+    TM_FLASH_DIRECTORY_CONTENT = 29
     TC_FLASH_READ_FULL_FILE = 30
 
 
@@ -130,11 +134,6 @@ class MemAddresses(enum.IntEnum):
     DEADBEEF = 0x40000004
 
 
-class PlocReplyIds(enum.IntEnum):
-    TM_MEM_READ_RPT = 6
-    TM_CAM_CMD_RPT = 19
-
-
 @tmtc_definitions_provider
 def add_ploc_mpsoc_cmds(defs: TmtcDefinitionWrapper):
     oce = OpCodeEntry()
@@ -233,15 +232,17 @@ def pack_ploc_mpsoc_commands(p: ServiceProviderParams):
         data = object_id.as_bytes + struct.pack("!I", ActionId.TC_DOWNLINK_PWR_OFF)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
     if op_code == OpCode.FLASH_GET_DIR_CONTENT:
-        q.add_log_cmd(f"{prefix}: {OpCode.FLASH_GET_DIR_CONTENT}")
+        q.add_log_cmd(f"{prefix}: {Info.FLASH_GET_DIR_CONTENT}")
-        dir_name = input("Please specify MPSoC directory name to get information for")
+        dir_name = input("Please specify MPSoC directory name to get information for: ")
         dir_name = bytearray(dir_name.encode("utf-8"))
         dir_name.append(0)
-        return create_action_cmd(
+        q.add_pus_tc(
+            create_action_cmd(
                 object_id=object_id.as_bytes,
                 action_id=ActionId.TC_FLASH_DIR_GET_CONTENT,
                 user_data=dir_name,
             )
+        )
     if op_code == OpCode.REPLAY_WRITE_SEQ:
         q.add_log_cmd(f"{prefix}: {Info.REPLAY_WRITE_SEQ}")
         data = prepare_replay_write_sequence_cmd(object_id.as_bytes)
@@ -612,3 +613,83 @@ def handle_ploc_mpsoc_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes, set_id:
     else:
         _LOGGER.warning(f"Unknown set ID {set_id} for MPSoC HK")
     pass
+
+
+@dataclasses.dataclass
+class DirElement:
+    name: str
+    attr: int
+    size: int
+
+
+def handle_mpsoc_data_reply(
+    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
+):
+    pw = PrintWrapper(printer)
+    if action_id == ActionId.TM_MEM_READ_RPT:
+        header_list = [
+            "PLOC Memory Address",
+            "PLOC Mem Len",
+            "PLOC Read Memory Data",
+        ]
+        content_list = [
+            "0x" + custom_data[:4].hex(),
+            struct.unpack("!H", custom_data[4:6])[0],
+            "0x" + custom_data[6:10].hex(),
+        ]
+        print(header_list)
+        print(content_list)
+        printer.file_logger.info(header_list)
+        printer.file_logger.info(content_list)
+    elif action_id == ActionId.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)
+    elif action_id == ActionId.TM_FLASH_DIRECTORY_CONTENT:
+        if len(custom_data) < 16:
+            _LOGGER.warning(
+                "PLOC MPSoC flash directory data shorter than minimum 16 bytes"
+            )
+        current_idx = 0
+        dir_name_short = custom_data[current_idx : current_idx + 12].decode("utf-8")
+        current_idx += 12
+        num_elements = struct.unpack("!I", custom_data[current_idx : current_idx + 4])[
+            0
+        ]
+        current_idx += 4
+        elem_names = []
+        elem_attrs = []
+        elem_sizes = []
+        expected_size = 16 + num_elements * 17
+        if len(custom_data) < expected_size:
+            _LOGGER.warning(
+                f"PLOC MPSoC flash directory data shorter than expected {expected_size}"
+            )
+        pw.dlog(
+            f"Received PLOC MPSoC flash directory content for path {dir_name_short} "
+            f"with {num_elements} elements"
+        )
+        # It is as weird as it looks..
+        for _ in range(num_elements):
+            end_of_str = custom_data[current_idx : current_idx + 12].index(b"\x00")
+            elem_name = custom_data[current_idx : current_idx + end_of_str].decode(
+                "utf-8"
+            )
+            current_idx += 12
+            elem_names.append(elem_name)
+        for _ in range(num_elements):
+            elem_attrs.append(custom_data[current_idx])
+            current_idx += 1
+        for _ in range(num_elements):
+            elem_sizes.append(
+                struct.unpack("!I", custom_data[current_idx : current_idx + 4])[0]
+            )
+            current_idx += 4
+        for i in range(num_elements):
+            pw.dlog(f"{DirElement(elem_names[i], elem_attrs[i], elem_sizes[i])}")