diff --git a/CHANGELOG.md b/CHANGELOG.md index 0e5c84a..a57d1d4 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/eive_tmtc/pus_tm/action_reply_handler.py b/eive_tmtc/pus_tm/action_reply_handler.py index bc556d4..65334af 100644 --- a/eive_tmtc/pus_tm/action_reply_handler.py +++ b/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 ): diff --git a/eive_tmtc/tmtc/payload/ploc_mpsoc.py b/eive_tmtc/tmtc/payload/ploc_mpsoc.py index deeb41d..255669c 100644 --- a/eive_tmtc/tmtc/payload/ploc_mpsoc.py +++ b/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,14 +232,16 @@ 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}") - dir_name = input("Please specify MPSoC directory name to get information for") + 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 = bytearray(dir_name.encode("utf-8")) dir_name.append(0) - return create_action_cmd( - object_id=object_id.as_bytes, - action_id=ActionId.TC_FLASH_DIR_GET_CONTENT, - user_data=dir_name, + 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}") @@ -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])}")