135 lines
4.8 KiB
Python
135 lines
4.8 KiB
Python
import struct
|
|
from config.object_ids import *
|
|
from tmtc.acs.imtq import ImtqActionIds
|
|
from pus_tm.defs import PrintWrapper
|
|
from tmtc.payload.ploc_mpsoc import PlocReplyIds
|
|
from tmtc.payload.ploc_supervisor import SupvActionIds
|
|
from pus_tc.devs.star_tracker import StarTrackerActionIds
|
|
from tmtc.power.tm import handle_get_param_data_reply
|
|
from tmtccmd.logging import get_console_logger
|
|
from tmtccmd.tm import Service8FsfwTm
|
|
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
|
|
|
|
LOGGER = get_console_logger()
|
|
|
|
|
|
def handle_action_reply(
|
|
raw_tm: bytes, printer: FsfwTmTcPrinter, obj_id_dict: ObjectIdDictT
|
|
):
|
|
"""Core Action reply handler
|
|
:return:
|
|
"""
|
|
tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm)
|
|
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
|
|
action_id = tm_packet.action_id
|
|
generic_print_str = printer.generic_action_packet_tm_print(
|
|
packet=tm_packet, obj_id=object_id
|
|
)
|
|
pw.dlog(generic_print_str)
|
|
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)
|
|
elif object_id.as_bytes == PLOC_SUPV_ID:
|
|
return handle_supervisor_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 in [
|
|
ACU_HANDLER_ID,
|
|
PDU_1_HANDLER_ID,
|
|
PDU_2_HANDLER_ID,
|
|
P60_DOCK_HANDLER,
|
|
]:
|
|
return handle_get_param_data_reply(object_id, action_id, pw, custom_data)
|
|
else:
|
|
pw.dlog(f"No dedicated action reply handler found for reply from {object_id}")
|
|
pw.dlog(f"Raw Data: {tm_packet.custom_data.hex(sep=',')}")
|
|
|
|
|
|
def handle_imtq_replies(
|
|
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
|
|
):
|
|
if action_id == struct.unpack("!I", ImtqActionIds.get_commanded_dipole)[0]:
|
|
header_list = [
|
|
"Commanded X-Dipole",
|
|
"Commanded Y-Dipole",
|
|
"Commanded Z-Dipole",
|
|
]
|
|
[x_dipole, y_dipole, z_dipole] = struct.unpack("!HHH", custom_data[0:6])
|
|
content_list = [x_dipole, y_dipole, z_dipole]
|
|
print(header_list)
|
|
print(content_list)
|
|
printer.file_logger.info(header_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(
|
|
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
|
|
):
|
|
if action_id == SupvActionIds.DUMP_MRAM:
|
|
header_list = ["MRAM Dump"]
|
|
content_list = [custom_data[: len(custom_data)]]
|
|
print(header_list)
|
|
print(content_list)
|
|
printer.file_logger.info(header_list)
|
|
printer.file_logger.info(content_list)
|
|
elif action_id == SupvActionIds.READ_GPIO:
|
|
header_list = ["GPIO state"]
|
|
content_list = [struct.unpack("!H", custom_data[:2])[0]]
|
|
print(header_list)
|
|
print(content_list)
|
|
printer.file_logger.info(header_list)
|
|
printer.file_logger.info(content_list)
|
|
|
|
|
|
def handle_startracker_replies(
|
|
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
|
|
):
|
|
if action_id == StarTrackerActionIds.CHECKSUM:
|
|
if len(custom_data) != 5:
|
|
LOGGER.warning(
|
|
"Star tracker reply has invalid length {0}".format(len(custom_data))
|
|
)
|
|
return
|
|
header_list = ["Checksum", "Checksum valid"]
|
|
print(custom_data[4])
|
|
checksum_valid_flag = custom_data[4] >> 8
|
|
content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
|
|
print(header_list)
|
|
print(content_list)
|
|
printer.file_logger.info(header_list)
|
|
printer.file_logger.info(content_list)
|