import struct from typing import Tuple from config.object_ids import * from pus_tc.imtq import ImtqActionIds from pus_tc.ploc_mpsoc import PlocReplyIds from pus_tc.ploc_supervisor import SupvActionIds from pus_tc.star_tracker import StarTrackerActionIds from tmtccmd.utility.logger import get_console_logger LOGGER = get_console_logger() def user_analyze_service_8_data( object_id: bytes, action_id: int, custom_data: bytearray ) -> Tuple[list, list]: """ This function is called by the TMTC core if a Service 8 data reply (subservice 130) is received. The user can return a tuple of two lists, where the first list is a list of header strings to print and the second list is a list of values to print. The TMTC core will take care of printing both lists and logging them. @param object_id: @param action_id: @param custom_data: @return: """ if object_id == PDU_2_HANDLER_ID: header_list = ["PDU2 Service 8 Reply"] data_string = str() for index in range(len(custom_data)): data_string += str(hex(custom_data[index])) + " , " data_string = data_string.rstrip() data_string = data_string.rstrip(",") data_string = data_string.rstrip() content_list = [data_string] elif object_id == IMTQ_HANDLER_ID: return handle_imtq_replies(action_id, custom_data) elif object_id == PLOC_MPSOC_ID: return handle_ploc_replies(action_id, custom_data) elif object_id == PLOC_SUPV_ID: return handle_supervisor_replies(action_id, custom_data) elif object_id == STAR_TRACKER_ID: return handle_startracker_replies(action_id, custom_data) else: header_list = [] content_list = [] return header_list, content_list def handle_imtq_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]: header_list = [] content_list = [] if action_id == struct.unpack("!I", ImtqActionIds.get_commanded_dipole)[0]: header_list = ["Commanded X-Dipole", "Commanded Y-Dipole", "Commanded Z-Dipole"] x_dipole = struct.unpack("!H", custom_data[:2]) y_dipole = struct.unpack("!H", custom_data[2:4]) z_dipole = struct.unpack("!H", custom_data[4:6]) content_list = [x_dipole[0], y_dipole[0], z_dipole[0]] def handle_ploc_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]: header_list = [] content_list = [] if action_id == PlocReplyIds.tm_mem_read_report: header_list = ["PLOC Memory Address", "PLOC Mem Len", "PLOC Read Memory Data"] content_list = [custom_data[:4], custom_data[4:6], custom_data[6:10]] return header_list, content_list def handle_supervisor_replies( action_id: int, custom_data: bytearray ) -> Tuple[list, list]: header_list = [] content_list = [] if action_id == SupvActionIds.DUMP_MRAM: header_list = ["MRAM Dump"] content_list = [custom_data[: len(custom_data)]] return header_list, content_list def handle_startracker_replies( action_id: int, custom_data: bytearray ) -> Tuple[list, list]: header_list = [] content_list = [] 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, content_list 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] return header_list, content_list