eive-tmtc/eive_tmtc/pus_tm/action_reply_handler.py

96 lines
3.7 KiB
Python
Raw Normal View History

2023-06-19 17:16:00 +02:00
import logging
import struct
2023-06-19 17:16:00 +02:00
from eive_tmtc.config.object_ids import (
ACU_HANDLER_ID,
PDU_1_HANDLER_ID,
PDU_2_HANDLER_ID,
IMTQ_HANDLER_ID,
PLOC_MPSOC_ID,
PLOC_SUPV_ID,
CORE_CONTROLLER_ID,
STAR_TRACKER_ID,
P60_DOCK_HANDLER,
2023-12-01 13:02:23 +01:00
ACS_CONTROLLER,
2023-06-19 17:16:00 +02:00
)
2023-01-16 14:13:06 +01:00
from eive_tmtc.tmtc.acs.imtq import ImtqActionId
2022-11-29 16:53:29 +01:00
from eive_tmtc.pus_tm.defs import PrintWrapper
2023-04-16 02:56:14 +02:00
from eive_tmtc.tmtc.core import handle_core_ctrl_action_replies
2023-05-12 16:08:57 +02:00
from eive_tmtc.tmtc.payload.ploc_mpsoc import handle_mpsoc_data_reply
2023-01-16 14:13:06 +01:00
from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId
2023-10-19 16:52:08 +02:00
from eive_tmtc.tmtc.acs.star_tracker import handle_star_tracker_action_replies
2023-12-01 13:02:23 +01:00
from eive_tmtc.tmtc.acs.acs_ctrl import handle_acs_ctrl_action_replies
2022-11-29 16:53:29 +01:00
from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply
2023-11-10 19:23:06 +01:00
from tmtccmd.pus.s8_fsfw_action import Service8FsfwTm
2023-05-25 11:31:06 +02:00
from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
2023-01-16 17:45:46 +01:00
from spacepackets.ccsds.time import CdsShortTimestamp
2023-06-19 17:16:00 +02:00
from tmtccmd.util import ObjectIdDictT
2021-12-24 07:32:35 +01:00
2023-02-01 11:17:04 +01:00
_LOGGER = logging.getLogger(__name__)
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
"""
2023-01-16 17:45:46 +01:00
tm_packet = Service8FsfwTm.unpack(
raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty()
)
2022-04-30 09:26:19 +02:00
object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
2023-05-24 13:50:37 +02:00
pw = PrintWrapper(printer.file_logger)
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-08-24 11:19:49 +02:00
pw.dlog(generic_print_str)
2022-04-30 09:26:19 +02:00
if object_id.as_bytes == IMTQ_HANDLER_ID:
2023-05-24 13:50:37 +02:00
return handle_imtq_replies(action_id, pw, custom_data)
2022-04-30 09:26:19 +02:00
elif object_id.as_bytes == PLOC_MPSOC_ID:
2023-05-24 13:50:37 +02:00
return handle_mpsoc_data_reply(action_id, pw, custom_data)
2022-04-30 09:26:19 +02:00
elif object_id.as_bytes == PLOC_SUPV_ID:
2023-05-24 13:50:37 +02:00
return handle_supervisor_replies(action_id, pw, custom_data)
2023-04-16 02:00:50 +02:00
elif object_id.as_bytes == CORE_CONTROLLER_ID:
2023-05-24 13:50:37 +02:00
return handle_core_ctrl_action_replies(action_id, pw, custom_data)
2022-04-30 09:26:19 +02:00
elif object_id.as_bytes == STAR_TRACKER_ID:
2023-10-19 16:52:08 +02:00
return handle_star_tracker_action_replies(action_id, pw, custom_data)
2023-12-01 13:02:23 +01:00
elif object_id.as_bytes == ACS_CONTROLLER:
return handle_acs_ctrl_action_replies(action_id, pw, custom_data)
2022-08-24 11:24:25 +02:00
elif object_id.as_bytes in [
ACU_HANDLER_ID,
PDU_1_HANDLER_ID,
PDU_2_HANDLER_ID,
P60_DOCK_HANDLER,
]:
2022-08-25 18:13:37 +02:00
return handle_get_param_data_reply(object_id, action_id, pw, custom_data)
2022-08-24 11:19:49 +02:00
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=',')}")
2023-05-24 13:50:37 +02:00
def handle_imtq_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray):
2023-01-16 14:13:06 +01:00
if action_id == struct.unpack("!I", ImtqActionId.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]
2023-05-24 13:50:37 +02:00
pw.dlog(f"{header_list}")
pw.dlog(f"{content_list}")
2021-04-26 15:01:22 +02:00
2023-05-24 13:50:37 +02:00
def handle_supervisor_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray):
2023-01-16 14:13:06 +01:00
if action_id == SupvActionId.DUMP_MRAM:
2022-04-06 16:41:46 +02:00
header_list = ["MRAM Dump"]
content_list = [custom_data[: len(custom_data)]]
2023-05-24 13:50:37 +02:00
pw.dlog(f"{header_list}")
pw.dlog(f"{content_list}")
2023-01-16 14:13:06 +01:00
elif action_id == SupvActionId.READ_GPIO:
2022-05-03 19:09:23 +02:00
header_list = ["GPIO state"]
2022-05-05 14:30:28 +02:00
content_list = [struct.unpack("!H", custom_data[:2])[0]]
2023-05-24 13:50:37 +02:00
pw.dlog(f"{header_list}")
pw.dlog(f"{content_list}")