diff --git a/eive_tmtc/pus_tm/action_reply_handler.py b/eive_tmtc/pus_tm/action_reply_handler.py index 7d23419..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,15 +34,13 @@ 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: return handle_core_ctrl_action_replies(action_id, printer, custom_data) elif object_id.as_bytes == STAR_TRACKER_ID: return handle_startracker_replies(action_id, printer, custom_data) - elif object_id.as_bytes == PLOC_MPSOC_ID: - return handle_mpsoc_data_reply(action_id, printer, custom_data) elif object_id.as_bytes in [ ACU_HANDLER_ID, PDU_1_HANDLER_ID, @@ -73,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 8533075..b2f2c95 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() @@ -616,8 +615,72 @@ def handle_ploc_mpsoc_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes, set_id: pass +@dataclasses.dataclass +class DirElement: + name: str + attr: int + size: int + + def handle_mpsoc_data_reply( - action_id: int, printer: FsfwTmTcPrinter, custom_data: bytes + action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray ): - print(f"Received MPSoC data reply with action ID {action_id}") - pass + 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: + print(custom_data.hex(sep=",")) + 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] + current_idx += 12 + num_elements = struct.unpack("!I", custom_data[current_idx : current_idx + 4])[ + 0 + ] + current_idx += 4 + elems = [] + 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" + ) + for i in range(num_elements): + elem_name = custom_data[current_idx : current_idx + 12].decode("utf-8") + current_idx += 12 + elem_attr = custom_data[current_idx] + current_idx += 1 + elem_size = current_idx[current_idx : current_idx + 4] + current_idx += 4 + elems.append(DirElement(elem_name, elem_attr, elem_size)) + for elem in elems: + pw.dlog(f"{elem}")