eive-tmtc/pus_tm/action_reply_handler.py

150 lines
5.2 KiB
Python
Raw Normal View History

import struct
2021-06-11 13:51:56 +02:00
from config.object_ids import *
2022-03-04 10:44:55 +01:00
from pus_tc.devs.imtq import ImtqActionIds
2022-08-18 15:52:06 +02:00
from tmtc.ploc_mpsoc import PlocReplyIds
2022-08-18 14:08:05 +02:00
from tmtc.ploc_supervisor import SupvActionIds
2022-03-04 10:44:55 +01:00
from pus_tc.devs.star_tracker import StarTrackerActionIds
from gomspace.gomspace_common import GomspaceDeviceActionIds
2022-04-05 00:51:52 +02:00
from tmtccmd.logging import get_console_logger
2022-04-06 16:41:46 +02:00
from tmtccmd.tm import Service8FsfwTm
2022-07-08 16:25:46 +02:00
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
2021-12-24 07:32:35 +01:00
LOGGER = get_console_logger()
2021-02-09 12:35:10 +01:00
2022-04-06 16:41:46 +02:00
def handle_action_reply(
raw_tm: bytes, printer: FsfwTmTcPrinter, obj_id_dict: ObjectIdDictT
):
"""Core Action reply handler
:return:
2021-02-09 12:35:10 +01:00
"""
2022-04-06 16:41:46 +02:00
tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm)
printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
2022-04-30 09:26:19 +02:00
object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
2022-04-06 16:41:46 +02:00
custom_data = tm_packet.custom_data
action_id = tm_packet.action_id
2022-04-06 17:01:01 +02:00
generic_print_str = printer.generic_action_packet_tm_print(
packet=tm_packet, obj_id=object_id
)
2022-04-06 16:41:46 +02:00
print(generic_print_str)
printer.file_logger.info(generic_print_str)
2022-04-30 09:26:19 +02:00
if object_id.as_bytes == IMTQ_HANDLER_ID:
2022-04-06 16:41:46 +02:00
return handle_imtq_replies(action_id, printer, custom_data)
2022-04-30 09:26:19 +02:00
elif object_id.as_bytes == PLOC_MPSOC_ID:
2022-04-06 16:41:46 +02:00
return handle_ploc_replies(action_id, printer, custom_data)
2022-04-30 09:26:19 +02:00
elif object_id.as_bytes == PLOC_SUPV_ID:
2022-04-06 16:41:46 +02:00
return handle_supervisor_replies(action_id, printer, custom_data)
2022-04-30 09:26:19 +02:00
elif object_id.as_bytes == STAR_TRACKER_ID:
2022-04-06 16:41:46 +02:00
return handle_startracker_replies(action_id, printer, custom_data)
elif object_id.as_bytes == ACU_HANDLER_ID:
return handle_acu_replies(action_id, printer, custom_data)
2022-04-06 16:41:46 +02:00
def handle_imtq_replies(
2022-04-06 17:01:01 +02:00
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
2022-04-06 16:41:46 +02:00
):
2022-01-18 14:03:56 +01:00
if action_id == struct.unpack("!I", ImtqActionIds.get_commanded_dipole)[0]:
2022-04-06 16:41:46 +02:00
header_list = [
2022-02-03 16:02:55 +01:00
"Commanded X-Dipole",
"Commanded Y-Dipole",
"Commanded Z-Dipole",
]
2022-04-06 17:01:01 +02:00
[x_dipole, y_dipole, z_dipole] = struct.unpack("!HHH", custom_data[0:6])
2022-04-06 16:41:46 +02:00
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)
2021-04-26 15:01:22 +02:00
2022-04-06 16:41:46 +02:00
def handle_ploc_replies(
2022-04-06 17:01:01 +02:00
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
2022-04-06 16:41:46 +02:00
):
2022-05-04 19:08:54 +02:00
if action_id == PlocReplyIds.TM_MEM_READ_RPT:
2022-04-06 16:41:46 +02:00
header_list = [
2022-02-03 16:02:55 +01:00
"PLOC Memory Address",
"PLOC Mem Len",
"PLOC Read Memory Data",
]
2022-04-06 16:41:46 +02:00
content_list = [
2022-03-22 19:29:55 +01:00
"0x" + custom_data[:4].hex(),
struct.unpack("!H", custom_data[4:6])[0],
"0x" + custom_data[6:10].hex(),
]
2022-04-06 16:41:46 +02:00
print(header_list)
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
2022-05-04 19:08:54 +02:00
elif action_id == PlocReplyIds.TM_CAM_CMD_RPT:
header_list = ["Camera reply string", "ACK"]
2022-05-05 14:30:28 +02:00
content_list = [
custom_data[: len(custom_data) - 1].decode("utf-8"),
hex(custom_data[-1]),
]
2022-05-04 19:08:54 +02:00
print(header_list)
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
2022-01-18 14:03:56 +01:00
def handle_supervisor_replies(
2022-04-06 17:01:01 +02:00
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
2022-04-06 16:41:46 +02:00
):
if action_id == SupvActionIds.DUMP_MRAM:
2022-04-06 16:41:46 +02:00
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)
2022-05-03 19:09:23 +02:00
elif action_id == SupvActionIds.READ_GPIO:
header_list = ["GPIO state"]
2022-05-05 14:30:28 +02:00
content_list = [struct.unpack("!H", custom_data[:2])[0]]
2022-05-03 19:09:23 +02:00
print(header_list)
print(content_list)
printer.file_logger.info(header_list)
printer.file_logger.info(content_list)
2022-01-18 14:03:56 +01:00
def handle_startracker_replies(
2022-04-06 17:01:01 +02:00
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
2022-04-06 16:41:46 +02:00
):
2021-12-24 07:32:35 +01:00
if action_id == StarTrackerActionIds.CHECKSUM:
if len(custom_data) != 5:
2022-01-18 14:03:56 +01:00
LOGGER.warning(
"Star tracker reply has invalid length {0}".format(len(custom_data))
)
2022-04-06 16:41:46 +02:00
return
header_list = ["Checksum", "Checksum valid"]
2021-12-24 07:32:35 +01:00
print(custom_data[4])
checksum_valid_flag = custom_data[4] >> 8
2022-04-06 16:41:46 +02:00
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)
def handle_acu_replies(
action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
):
if action_id == GomspaceDeviceActionIds.PARAM_GET:
header_list = [
2022-07-04 15:22:53 +02:00
"Gomspace action ID" "Table ID",
"Memory Address",
2022-07-04 15:22:53 +02:00
"Payload length" "Payload",
]
fmt_str = "!BBHH"
2022-07-04 15:22:53 +02:00
(action, table_id, address, payload_length) = struct.unpack(
fmt_str, custom_data[:6]
)
content_list = [
action,
table_id,
"0x" + hex(address),
payload_length,
"0x" + custom_data[6:].hex(),
]
print(header_list)
print(content_list)