import struct from config.object_ids import * from pus_tc.devs.imtq import ImtqActionIds from pus_tc.devs.ploc_mpsoc import PlocReplyIds from pus_tc.devs.ploc_supervisor import SupvActionIds from pus_tc.devs.star_tracker import StarTrackerActionIds from tmtccmd.logging import get_console_logger from tmtccmd.config.definitions import DataReplyUnpacked from tmtccmd.tm import Service8FsfwTm from tmtccmd.utility.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) 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 ) print(generic_print_str) printer.file_logger.info(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) 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 ): reply = DataReplyUnpacked() 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)