eive/eive-tmtc
eive/eive-tmtc
14
0
Fork 0
Code Issues Pull Requests 2 Projects Releases 97 Wiki Activity

MPSoC flash content reporter #190

Merged
meggert merged 4 commits from mpsoc_action_reply_handler into main 2023-05-15 09:10:42 +02:00
Conversation 0 Commits 4 Files Changed 3 +73 -43
2 changed files with 73 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit e05a54b076 - Show all commits

37
eive_tmtc/pus_tm/action_reply_handler.py
View File

@ -3,7 +3,7 @@ from eive_tmtc.config.object_ids import *
from eive_tmtc.tmtc.acs.imtq import ImtqActionId from eive_tmtc.tmtc.acs.imtq import ImtqActionId
from eive_tmtc.pus_tm.defs import PrintWrapper from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.tmtc.core import handle_core_ctrl_action_replies 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.payload.ploc_supervisor import SupvActionId
from eive_tmtc.tmtc.acs.star_tracker import StarTrackerActionId from eive_tmtc.tmtc.acs.star_tracker import StarTrackerActionId
from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
@ -23,7 +23,6 @@ def handle_action_reply(
tm_packet = Service8FsfwTm.unpack( tm_packet = Service8FsfwTm.unpack(
raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty() 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) object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
pw = PrintWrapper(printer) pw = PrintWrapper(printer)
custom_data = tm_packet.custom_data custom_data = tm_packet.custom_data
@ -35,15 +34,13 @@ def handle_action_reply(
if object_id.as_bytes == IMTQ_HANDLER_ID: if object_id.as_bytes == IMTQ_HANDLER_ID:
return handle_imtq_replies(action_id, printer, custom_data) return handle_imtq_replies(action_id, printer, custom_data)
elif object_id.as_bytes == PLOC_MPSOC_ID: 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: elif object_id.as_bytes == PLOC_SUPV_ID:
return handle_supervisor_replies(action_id, printer, custom_data) return handle_supervisor_replies(action_id, printer, custom_data)
elif object_id.as_bytes == CORE_CONTROLLER_ID: elif object_id.as_bytes == CORE_CONTROLLER_ID:
return handle_core_ctrl_action_replies(action_id, printer, custom_data) return handle_core_ctrl_action_replies(action_id, printer, custom_data)
elif object_id.as_bytes == STAR_TRACKER_ID: elif object_id.as_bytes == STAR_TRACKER_ID:
return handle_startracker_replies(action_id, printer, custom_data) 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 [ elif object_id.as_bytes in [
ACU_HANDLER_ID, ACU_HANDLER_ID,
PDU_1_HANDLER_ID, PDU_1_HANDLER_ID,
@ -73,36 +70,6 @@ def handle_imtq_replies(
printer.file_logger.info(content_list) 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( def handle_supervisor_replies(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
): ):

79
eive_tmtc/tmtc/payload/ploc_mpsoc.py
View File

@ -6,6 +6,7 @@
@author J. Meier @author J. Meier
@date 06.03.2021 @date 06.03.2021
""" """
import dataclasses
import logging import logging
import struct import struct
import enum import enum
@ -59,6 +60,7 @@ class SetId(enum.IntEnum):
class ActionId(enum.IntEnum): class ActionId(enum.IntEnum):
TC_MEM_WRITE = 1 TC_MEM_WRITE = 1
TC_MEM_READ = 2 TC_MEM_READ = 2
TM_MEM_READ_RPT = 6
TC_FLASH_WRITE_FULL_FILE = 9 TC_FLASH_WRITE_FULL_FILE = 9
TC_FLASH_DELETE = 10 TC_FLASH_DELETE = 10
TC_REPLAY_START = 11 TC_REPLAY_START = 11
@ -70,6 +72,7 @@ class ActionId(enum.IntEnum):
TC_MODE_REPLAY = 16 TC_MODE_REPLAY = 16
TC_CAM_CMD_SEND = 17 TC_CAM_CMD_SEND = 17
TC_MODE_IDLE = 18 TC_MODE_IDLE = 18
TM_CAM_CMD_RPT = 19
SET_UART_TX_TRISTATE = 20 SET_UART_TX_TRISTATE = 20
RELEASE_UART_TX = 21 RELEASE_UART_TX = 21
TC_CAM_TAKE_PIC = 22 TC_CAM_TAKE_PIC = 22
@ -77,6 +80,7 @@ class ActionId(enum.IntEnum):
TC_DOWNLINK_DATA_MODULATE = 24 TC_DOWNLINK_DATA_MODULATE = 24
TC_MODE_SNAPSHOT = 25 TC_MODE_SNAPSHOT = 25
TC_FLASH_DIR_GET_CONTENT = 28 TC_FLASH_DIR_GET_CONTENT = 28
TM_FLASH_DIRECTORY_CONTENT = 29
TC_FLASH_READ_FULL_FILE = 30 TC_FLASH_READ_FULL_FILE = 30
@ -130,11 +134,6 @@ class MemAddresses(enum.IntEnum):
DEADBEEF = 0x40000004 DEADBEEF = 0x40000004
class PlocReplyIds(enum.IntEnum):
TM_MEM_READ_RPT = 6
TM_CAM_CMD_RPT = 19
@tmtc_definitions_provider @tmtc_definitions_provider
def add_ploc_mpsoc_cmds(defs: TmtcDefinitionWrapper): def add_ploc_mpsoc_cmds(defs: TmtcDefinitionWrapper):
oce = OpCodeEntry() oce = OpCodeEntry()
@ -616,8 +615,72 @@ def handle_ploc_mpsoc_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes, set_id:
pass pass
@dataclasses.dataclass
class DirElement:
name: str
attr: int
size: int
def handle_mpsoc_data_reply( 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}") pw = PrintWrapper(printer)
pass 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}")