import logging import struct 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, ) 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 handle_mpsoc_data_reply from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId from eive_tmtc.tmtc.acs.star_tracker import handle_star_tracker_action_replies from eive_tmtc.tmtc.power.tm import handle_get_param_data_reply from tmtccmd.tm import Service8FsfwTm from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter from spacepackets.ccsds.time import CdsShortTimestamp from tmtccmd.util import ObjectIdDictT _LOGGER = logging.getLogger(__name__) 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, time_reader=CdsShortTimestamp.empty() ) object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes) pw = PrintWrapper(printer.file_logger) 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, pw, custom_data) elif object_id.as_bytes == PLOC_MPSOC_ID: return handle_mpsoc_data_reply(action_id, pw, custom_data) elif object_id.as_bytes == PLOC_SUPV_ID: return handle_supervisor_replies(action_id, pw, custom_data) elif object_id.as_bytes == CORE_CONTROLLER_ID: return handle_core_ctrl_action_replies(action_id, pw, custom_data) elif object_id.as_bytes == STAR_TRACKER_ID: return handle_star_tracker_action_replies(action_id, pw, 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, pw: PrintWrapper, custom_data: bytearray): if action_id == struct.unpack("!I", ImtqActionId.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] pw.dlog(f"{header_list}") pw.dlog(f"{content_list}") def handle_supervisor_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray): if action_id == SupvActionId.DUMP_MRAM: header_list = ["MRAM Dump"] content_list = [custom_data[: len(custom_data)]] pw.dlog(f"{header_list}") pw.dlog(f"{content_list}") elif action_id == SupvActionId.READ_GPIO: header_list = ["GPIO state"] content_list = [struct.unpack("!H", custom_data[:2])[0]] pw.dlog(f"{header_list}") pw.dlog(f"{content_list}")