import enum import logging import socket import struct from socket import AF_INET from typing import Tuple from eive_tmtc.config.definitions import CustomServiceList from eive_tmtc.config.object_ids import ACS_CONTROLLER from eive_tmtc.pus_tm.defs import PrintWrapper from tmtccmd.config.tmtc import ( tmtc_definitions_provider, TmtcDefinitionWrapper, OpCodeEntry, ) from tmtccmd.tc import service_provider from tmtccmd.tc.pus_200_fsfw_mode import Mode, pack_mode_command from tmtccmd.tc.decorator import ServiceProviderParams from tmtccmd.tc.pus_3_fsfw_hk import ( generate_one_hk_command, make_sid, enable_periodic_hk_command_with_interval, disable_periodic_hk_command, create_request_one_diag_command, ) from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter class SetId(enum.IntEnum): MGM_RAW_SET = 0 MGM_PROC_SET = 1 SUS_RAW_SET = 2 SUS_PROC_SET = 3 GYR_RAW_SET = 4 GYR_PROC_SET = 5 GPS_PROC_SET = 6 MEKF_DATA = 7 CTRL_VAL_DATA = 8 ACTUATOR_CMD_DATA = 9 class Submode(enum.IntEnum): SAFE = 2 DETUMBLE = 3 IDLE = 4 class OpCodes: OFF = ["off"] SAFE = ["normal_safe"] DTBL = ["normal_detumble"] IDLE = ["normal_idle"] REQUEST_RAW_MGM_HK = ["0", "mgm_raw_hk"] ENABLE_RAW_MGM_HK = ["1", "mgm_raw_enable_hk"] DISABLE_RAW_MGM_HK = ["2", "mgm_raw_disable_hk"] REQUEST_PROC_MGM_HK = ["3", "mgm_proc_hk"] ENABLE_PROC_MGM_HK = ["4", "mgm_proc_enable_hk"] DISABLE_PROC_MGM_HK = ["5", "mgm_proc_disable_hk"] REQUEST_RAW_SUS_HK = ["6", "sus_raw_hk"] ENABLE_RAW_SUS_HK = ["7", "sus_raw_enable_hk"] DISABLE_RAW_SUS_HK = ["8", "sus_raw_disable_hk"] REQUEST_PROC_SUS_HK = ["9", "sus_proc_hk"] ENABLE_PROC_SUS_HK = ["10", "sus_proc_enable_hk"] DISABLE_PROC_SUS_HK = ["11", "sus_proc_disable_hk"] REQUEST_RAW_GYR_HK = ["12", "gyr_raw_hk"] ENABLE_RAW_GYR_HK = ["13", "gyr_raw_enable_hk"] DISABLE_RAW_GYR_HK = ["14", "gyr_raw_disable_hk"] REQUEST_PROC_GYR_HK = ["15", "gyr_proc_hk"] ENABLE_PROC_GYR_HK = ["16", "gyr_proc_enable_hk"] DISABLE_PROC_GYR_HK = ["17", "gyr_proc_disable_hk"] REQUEST_PROC_GPS_HK = ["18", "gps_proc_hk"] ENABLE_PROC_GPS_HK = ["19", "gps_proc_enable_hk"] DISABLE_PROC_GPS_HK = ["20", "gps_proc_disable_hk"] REQUEST_MEKF_HK = ["21", "mekf_hk"] ENABLE_MEKF_HK = ["22", "mekf_enable_hk"] DISABLE_MEKF_HK = ["23", "mekf_disable_hk"] REQUEST_CTRL_VAL_HK = ["24", "ctrl_val_hk"] ENABLE_CTRL_VAL_HK = ["25", "ctrl_val_enable_hk"] DISABLE_CTRL_VAL_HK = ["26", "ctrl_val_disable_hk"] REQUEST_ACT_CMD_HK = ["27", "act_cmd_hk"] ENABLE_ACT_CMD_HK = ["28", "act_cmd_enable_hk"] DISABLE_ACT_CMD_HK = ["29", "act_cmd_disable_hk"] class Info: OFF = "Switch ACS CTRL off" SAFE = "Switch ACS CTRL normal safe" DTBL = "Switch ACS CTRL normal detumble" IDLE = "Switch ACS CTRL normal idle" REQUEST_RAW_MGM_HK = "Request Raw MGM HK once" ENABLE_RAW_MGM_HK = "Enable Raw MGM HK data generation" DISABLE_RAW_MGM_HK = "Disable Raw MGM HK data generation" REQUEST_PROC_MGM_HK = "Request Processed MGM HK" ENABLE_PROC_MGM_HK = "Enable Processed MGM HK data generation" DISABLE_PROC_MGM_HK = "Disable Processed MGM HK data generation" REQUEST_RAW_SUS_HK = "Request Raw SUS HK" ENABLE_RAW_SUS_HK = "Enable Raw SUS HK data generation" DISABLE_RAW_SUS_HK = "Disable Raw SUS HK data generation" REQUEST_PROC_SUS_HK = "Request Processed SUS HK" ENABLE_PROC_SUS_HK = "Enable Processed SUS HK data generation" DISABLE_PROC_SUS_HK = "Disable Processed MGM HK data generation" REQUEST_RAW_GYR_HK = "Request Raw GYR HK" ENABLE_RAW_GYR_HK = "Enable Raw GYR HK data generation" DISABLE_RAW_GYR_HK = "Disable Raw GYR HK data generation" REQUEST_PROC_GYR_HK = "Request Processed GYR HK" ENABLE_PROC_GYR_HK = "Enable Processed GYR HK data generation" DISABLE_PROC_GYR_HK = "Disable Processed GYR HK data generation" REQUEST_PROC_GPS_HK = "Request Processed GPS HK" ENABLE_PROC_GPS_HK = "Enable Processed GPS HK data generation" DISABLE_PROC_GPS_HK = "Disable Processed GPS HK data generation" REQUEST_MEKF_HK = "Request MEKF HK" ENABLE_MEKF_HK = "Enable MEKF HK data generation" DISABLE_MEKF_HK = "Disable MEKF HK data generation" REQUEST_CTRL_VAL_HK = "Request Control Values HK" ENABLE_CTRL_VAL_HK = "Enable Control Values HK data generation" DISABLE_CTRL_VAL_HK = "Disable Control Values HK data generation" REQUEST_ACT_CMD_HK = "Request Actuator Commands HK" ENABLE_ACT_CMD_HK = "Enable Actuator Commands HK data generation" DISABLE_ACT_CMD_HK = "Disable Actuator Commands HK data generation" PERFORM_MGM_CALIBRATION = False CALIBRATION_SOCKET_HOST = "localhost" CALIBRATION_SOCKET_PORT = 6677 CALIBRATION_ADDR = (CALIBRATION_SOCKET_HOST, CALIBRATION_SOCKET_PORT) if PERFORM_MGM_CALIBRATION: CALIBR_SOCKET = socket.socket(AF_INET, socket.SOCK_STREAM) CALIBR_SOCKET.setblocking(False) CALIBR_SOCKET.settimeout(0.2) CALIBR_SOCKET.connect(CALIBRATION_ADDR) @tmtc_definitions_provider def acs_cmd_defs(defs: TmtcDefinitionWrapper): oce = OpCodeEntry() oce.add(keys=OpCodes.OFF, info=Info.OFF) oce.add(keys=OpCodes.SAFE, info=Info.SAFE) oce.add(keys=OpCodes.DTBL, info=Info.DTBL) oce.add(keys=OpCodes.IDLE, info=Info.IDLE) oce.add(keys=OpCodes.REQUEST_RAW_MGM_HK, info=Info.REQUEST_RAW_MGM_HK) oce.add(keys=OpCodes.ENABLE_RAW_MGM_HK, info=Info.ENABLE_RAW_MGM_HK) oce.add(keys=OpCodes.DISABLE_RAW_MGM_HK, info=Info.DISABLE_RAW_MGM_HK) oce.add(keys=OpCodes.REQUEST_PROC_MGM_HK, info=Info.REQUEST_PROC_MGM_HK) oce.add(keys=OpCodes.ENABLE_PROC_MGM_HK, info=Info.ENABLE_PROC_MGM_HK) oce.add(keys=OpCodes.DISABLE_PROC_MGM_HK, info=Info.DISABLE_PROC_MGM_HK) oce.add(keys=OpCodes.REQUEST_RAW_SUS_HK, info=Info.REQUEST_RAW_SUS_HK) oce.add(keys=OpCodes.ENABLE_RAW_SUS_HK, info=Info.ENABLE_RAW_SUS_HK) oce.add(keys=OpCodes.DISABLE_RAW_SUS_HK, info=Info.DISABLE_RAW_SUS_HK) oce.add(keys=OpCodes.REQUEST_PROC_SUS_HK, info=Info.REQUEST_PROC_SUS_HK) oce.add(keys=OpCodes.ENABLE_PROC_SUS_HK, info=Info.ENABLE_PROC_SUS_HK) oce.add(keys=OpCodes.DISABLE_PROC_SUS_HK, info=Info.DISABLE_PROC_SUS_HK) oce.add(keys=OpCodes.REQUEST_RAW_GYR_HK, info=Info.REQUEST_RAW_GYR_HK) oce.add(keys=OpCodes.ENABLE_RAW_GYR_HK, info=Info.ENABLE_RAW_GYR_HK) oce.add(keys=OpCodes.DISABLE_RAW_GYR_HK, info=Info.DISABLE_RAW_GYR_HK) oce.add(keys=OpCodes.REQUEST_PROC_GYR_HK, info=Info.REQUEST_PROC_GYR_HK) oce.add(keys=OpCodes.ENABLE_PROC_GYR_HK, info=Info.ENABLE_PROC_GYR_HK) oce.add(keys=OpCodes.DISABLE_PROC_GYR_HK, info=Info.DISABLE_PROC_GYR_HK) oce.add(keys=OpCodes.REQUEST_PROC_GPS_HK, info=Info.REQUEST_PROC_GPS_HK) oce.add(keys=OpCodes.ENABLE_PROC_GPS_HK, info=Info.ENABLE_PROC_GPS_HK) oce.add(keys=OpCodes.DISABLE_PROC_GPS_HK, info=Info.DISABLE_PROC_GPS_HK) oce.add(keys=OpCodes.REQUEST_MEKF_HK, info=Info.REQUEST_MEKF_HK) oce.add(keys=OpCodes.ENABLE_MEKF_HK, info=Info.ENABLE_MEKF_HK) oce.add(keys=OpCodes.DISABLE_MEKF_HK, info=Info.DISABLE_MEKF_HK) oce.add(keys=OpCodes.REQUEST_CTRL_VAL_HK, info=Info.REQUEST_CTRL_VAL_HK) oce.add(keys=OpCodes.ENABLE_CTRL_VAL_HK, info=Info.ENABLE_CTRL_VAL_HK) oce.add(keys=OpCodes.DISABLE_CTRL_VAL_HK, info=Info.DISABLE_CTRL_VAL_HK) oce.add(keys=OpCodes.REQUEST_ACT_CMD_HK, info=Info.REQUEST_ACT_CMD_HK) oce.add(keys=OpCodes.ENABLE_ACT_CMD_HK, info=Info.ENABLE_ACT_CMD_HK) oce.add(keys=OpCodes.DISABLE_ACT_CMD_HK, info=Info.DISABLE_ACT_CMD_HK) defs.add_service( name=CustomServiceList.ACS_CTRL.value, info="ACS Controller", op_code_entry=oce ) @service_provider(CustomServiceList.ACS_CTRL.value) def pack_acs_ctrl_command(p: ServiceProviderParams): op_code = p.op_code q = p.queue_helper if op_code in OpCodes.OFF: q.add_log_cmd(f"{Info.OFF}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, Mode.OFF, 0)) elif op_code in OpCodes.SAFE: q.add_log_cmd(f"{Info.SAFE}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, Mode.NORMAL, Submode.SAFE)) elif op_code in OpCodes.DTBL: q.add_log_cmd(f"{Info.DTBL}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, Mode.NORMAL, Submode.DETUMBLE)) elif op_code in OpCodes.IDLE: q.add_log_cmd(f"{Info.IDLE}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, Mode.NORMAL, Submode.IDLE)) elif op_code in OpCodes.REQUEST_RAW_MGM_HK: q.add_log_cmd(Info.REQUEST_RAW_MGM_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.MGM_RAW_SET)) ) elif op_code in OpCodes.ENABLE_RAW_MGM_HK: q.add_log_cmd(Info.ENABLE_RAW_MGM_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.MGM_RAW_SET), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_RAW_MGM_HK: q.add_log_cmd(Info.DISABLE_RAW_MGM_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.MGM_RAW_SET) ) ) elif op_code in OpCodes.REQUEST_PROC_MGM_HK: q.add_log_cmd(Info.REQUEST_PROC_MGM_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.MGM_PROC_SET)) ) elif op_code in OpCodes.ENABLE_PROC_MGM_HK: q.add_log_cmd(Info.ENABLE_PROC_MGM_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.MGM_PROC_SET), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_PROC_MGM_HK: q.add_log_cmd(Info.DISABLE_PROC_MGM_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.MGM_PROC_SET) ) ) elif op_code in OpCodes.REQUEST_RAW_SUS_HK: q.add_log_cmd(Info.REQUEST_RAW_SUS_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.SUS_RAW_SET)) ) elif op_code in OpCodes.ENABLE_RAW_SUS_HK: q.add_log_cmd(Info.ENABLE_RAW_SUS_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.SUS_RAW_SET), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_RAW_SUS_HK: q.add_log_cmd(Info.DISABLE_RAW_SUS_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.SUS_RAW_SET) ) ) elif op_code in OpCodes.REQUEST_PROC_SUS_HK: q.add_log_cmd(Info.REQUEST_PROC_SUS_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.SUS_PROC_SET)) ) elif op_code in OpCodes.ENABLE_PROC_SUS_HK: q.add_log_cmd(Info.ENABLE_PROC_SUS_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.SUS_PROC_SET), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_PROC_SUS_HK: q.add_log_cmd(Info.DISABLE_PROC_SUS_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.SUS_PROC_SET) ) ) elif op_code in OpCodes.REQUEST_RAW_GYR_HK: q.add_log_cmd(Info.REQUEST_RAW_GYR_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.GYR_RAW_SET)) ) elif op_code in OpCodes.ENABLE_RAW_GYR_HK: q.add_log_cmd(Info.ENABLE_RAW_GYR_HK) cmd_tuple = enable_periodic_hk_command_with_interval( True, make_sid(ACS_CONTROLLER, SetId.GYR_RAW_SET), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_RAW_GYR_HK: q.add_log_cmd(Info.DISABLE_RAW_GYR_HK) q.add_pus_tc( disable_periodic_hk_command( True, make_sid(ACS_CONTROLLER, SetId.GYR_RAW_SET) ) ) elif op_code in OpCodes.REQUEST_PROC_GYR_HK: q.add_log_cmd(Info.REQUEST_PROC_GYR_HK) q.add_pus_tc( create_request_one_diag_command( make_sid(ACS_CONTROLLER, SetId.GYR_PROC_SET) ) ) elif op_code in OpCodes.ENABLE_PROC_GYR_HK: q.add_log_cmd(Info.ENABLE_PROC_GYR_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.GYR_PROC_SET), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_PROC_GYR_HK: q.add_log_cmd(Info.DISABLE_PROC_GYR_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.GYR_PROC_SET) ) ) elif op_code in OpCodes.REQUEST_PROC_GPS_HK: q.add_log_cmd(Info.REQUEST_PROC_GPS_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.GPS_PROC_SET)) ) elif op_code in OpCodes.ENABLE_PROC_GPS_HK: q.add_log_cmd(Info.ENABLE_PROC_GPS_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.GPS_PROC_SET), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_PROC_GPS_HK: q.add_log_cmd(Info.DISABLE_PROC_GPS_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.GPS_PROC_SET) ) ) elif op_code in OpCodes.REQUEST_MEKF_HK: q.add_log_cmd(Info.REQUEST_MEKF_HK) q.add_pus_tc(generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.MEKF_DATA))) elif op_code in OpCodes.ENABLE_MEKF_HK: q.add_log_cmd(Info.ENABLE_MEKF_HK) cmd_tuple = enable_periodic_hk_command_with_interval( True, make_sid(ACS_CONTROLLER, SetId.MEKF_DATA), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_MEKF_HK: q.add_log_cmd(Info.DISABLE_MEKF_HK) q.add_pus_tc( disable_periodic_hk_command(True, make_sid(ACS_CONTROLLER, SetId.MEKF_DATA)) ) elif op_code in OpCodes.REQUEST_CTRL_VAL_HK: q.add_log_cmd(Info.REQUEST_CTRL_VAL_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.CTRL_VAL_DATA)) ) elif op_code in OpCodes.ENABLE_CTRL_VAL_HK: q.add_log_cmd(Info.ENABLE_CTRL_VAL_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.CTRL_VAL_DATA), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_CTRL_VAL_HK: q.add_log_cmd(Info.DISABLE_CTRL_VAL_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.CTRL_VAL_DATA) ) ) elif op_code in OpCodes.REQUEST_ACT_CMD_HK: q.add_log_cmd(Info.REQUEST_ACT_CMD_HK) q.add_pus_tc( generate_one_hk_command(make_sid(ACS_CONTROLLER, SetId.ACTUATOR_CMD_DATA)) ) elif op_code in OpCodes.ENABLE_ACT_CMD_HK: q.add_log_cmd(Info.ENABLE_ACT_CMD_HK) cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, SetId.ACTUATOR_CMD_DATA), 2.0 ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) elif op_code in OpCodes.DISABLE_ACT_CMD_HK: q.add_log_cmd(Info.DISABLE_ACT_CMD_HK) q.add_pus_tc( disable_periodic_hk_command( False, make_sid(ACS_CONTROLLER, SetId.ACTUATOR_CMD_DATA) ) ) else: logging.getLogger(__name__).info(f"Unknown op code {op_code}") def handle_acs_ctrl_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes): pw = PrintWrapper(printer) match set_id: case SetId.MGM_RAW_SET: handle_raw_mgm_data(pw, hk_data) case SetId.MGM_PROC_SET: handle_mgm_data_processed(pw, hk_data) case SetId.SUS_RAW_SET: handle_acs_ctrl_sus_raw_data(pw, hk_data) case SetId.SUS_PROC_SET: handle_acs_ctrl_sus_processed_data(pw, hk_data) case SetId.GYR_RAW_SET: handle_gyr_data_raw(pw, hk_data) case SetId.GYR_PROC_SET: handle_gyr_data_processed(pw, hk_data) case SetId.GPS_PROC_SET: handle_gps_data_processed(pw, hk_data) case SetId.MEKF_DATA: handle_mekf_data(pw, hk_data) case SetId.CTRL_VAL_DATA: handle_ctrl_val_data(pw, hk_data) case SetId.ACTUATOR_CMD_DATA: handle_act_cmd_data(pw, hk_data) def handle_acs_ctrl_sus_raw_data(pw: PrintWrapper, hk_data: bytes): if len(hk_data) < 6 * 2 * 12: pw.dlog( f"SUS Raw dataset with size {len(hk_data)} does not have expected size" f" of {6 * 2 * 12} bytes" ) return current_idx = 0 vec_fmt = "[" for _ in range(5): vec_fmt += "{:#06x}, " vec_fmt += "{:#06x}]" for idx in range(12): fmt_str = "!HHHHHH" length = struct.calcsize(fmt_str) sus_list = struct.unpack(fmt_str, hk_data[current_idx : current_idx + length]) sus_list_formatted = vec_fmt.format(*sus_list) current_idx += length pw.dlog(f"SUS {idx} RAW: {sus_list_formatted}") pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=12) def handle_acs_ctrl_sus_processed_data(pw: PrintWrapper, hk_data: bytes): if len(hk_data) < 3 * 4 * 12 + 3 * 8 * 3: pw.dlog( f"SUS Processed dataset with size {len(hk_data)} does not have expected size" f" of {3 * 4 * 12 + 3 * 8 * 3} bytes" ) return current_idx = 0 for idx in range(12): fmt_str = "!fff" length = struct.calcsize(fmt_str) sus_list = struct.unpack(fmt_str, hk_data[current_idx : current_idx + length]) sus_list_formatted = [f"{val:8.3f}" for val in sus_list] current_idx += length pw.dlog(f"SUS {idx} CALIB: {sus_list_formatted}") fmt_str = "!ddd" inc_len = struct.calcsize(fmt_str) sus_vec_tot = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) sus_vec_tot = [f"{val:8.3f}" for val in sus_vec_tot] current_idx += inc_len pw.dlog(f"SUS Vector Total: {sus_vec_tot}") sus_vec_tot_deriv = struct.unpack( fmt_str, hk_data[current_idx : current_idx + inc_len] ) sus_vec_tot_deriv = [f"{val:8.3f}" for val in {sus_vec_tot_deriv}] current_idx += inc_len pw.dlog(f"SUS Vector Derivative: {sus_vec_tot_deriv}") sun_ijk_model = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) sun_ijk_model = [f"{val:8.3f}" for val in sun_ijk_model] current_idx += inc_len pw.dlog(f"SUS ijk Model: {sun_ijk_model}") pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=15) def handle_raw_mgm_data(pw: PrintWrapper, hk_data: bytes): current_idx = 0 if len(hk_data) < 61: pw.dlog( f"ACS CTRL HK: MGM HK data with length {len(hk_data)} shorter than expected 61 bytes" ) pw.dlog(f"Raw Data: {hk_data.hex(sep=',')}") return def unpack_float_tuple(idx: int) -> (tuple, int): f_tuple = struct.unpack( float_tuple_fmt_str, hk_data[idx : idx + struct.calcsize(float_tuple_fmt_str)], ) idx += struct.calcsize(float_tuple_fmt_str) return f_tuple, idx float_tuple_fmt_str = "!fff" mgm_0_lis3_floats_ut, current_idx = unpack_float_tuple(current_idx) mgm_1_rm3100_floats_ut, current_idx = unpack_float_tuple(current_idx) mgm_2_lis3_floats_ut, current_idx = unpack_float_tuple(current_idx) mgm_3_rm3100_floats_ut, current_idx = unpack_float_tuple(current_idx) isis_floats_nt, current_idx = unpack_float_tuple(current_idx) imtq_mgm_ut = tuple(val / 1000.0 for val in isis_floats_nt) pw.dlog("ACS CTRL HK: MGM values [X,Y,Z] in floating point uT: ") mgm_lists = [ mgm_0_lis3_floats_ut, mgm_1_rm3100_floats_ut, mgm_2_lis3_floats_ut, mgm_3_rm3100_floats_ut, imtq_mgm_ut, ] formatted_list = [] # Reserve 8 decimal digits, use precision 3 float_str_fmt = "[{:8.3f}, {:8.3f}, {:8.3f}]" for mgm_entry in mgm_lists[0:4]: formatted_list.append(float_str_fmt.format(*mgm_entry)) formatted_list.append(float_str_fmt.format(*mgm_lists[4])) formatted_list.append(hk_data[current_idx]) print_str_list = [ "ACS Board MGM 0 LIS3MDL", "ACS Board MGM 1 RM3100", "ACS Board MGM 2 LIS3MDL", "ACS Board MGM 3 RM3100", "IMTQ MGM:", "IMTQ Actuation Status:", ] for entry in zip(print_str_list, formatted_list): pw.dlog(f"{entry[0].ljust(28)}: {entry[1]}") current_idx += 1 assert current_idx == 61 pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=6) def handle_mgm_data_processed(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received Processed MGM Set") fmt_str = "!fffffddd" inc_len = struct.calcsize(fmt_str) if len(hk_data) < inc_len: pw.dlog("Recieved HK set too small") return current_idx = 0 fmt_str = "!fff" inc_len = struct.calcsize(fmt_str) mgm_0 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_0_str = [f"{val:8.3f}" for val in mgm_0] pw.dlog(f"MGM 0 Vec: {mgm_0_str}") current_idx += inc_len mgm_1 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_1_str = [f"{val:8.3f}" for val in mgm_1] pw.dlog(f"MGM 1 Vec: {mgm_1_str}") current_idx += inc_len mgm_2 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_2_str = [f"{val:8.3f}" for val in mgm_2] pw.dlog(f"MGM 2 Vec: {mgm_2_str}") current_idx += inc_len mgm_3 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_3_str = [f"{val:8.3f}" for val in mgm_3] pw.dlog(f"MGM 3 Vec: {mgm_3_str}") current_idx += inc_len mgm_4 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_4_str = [f"{val:8.3f}" for val in mgm_4] pw.dlog(f"MGM 4 Vec: {mgm_4_str}") current_idx += inc_len fmt_str = "!ddd" inc_len = struct.calcsize(fmt_str) mgm_vec_tot = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_vec_tot = [f"{val:8.3f}" for val in mgm_vec_tot] current_idx += inc_len pw.dlog(f"MGM Total Vec: {mgm_vec_tot}") mgm_vec_tot_deriv = struct.unpack( fmt_str, hk_data[current_idx : current_idx + inc_len] ) mgm_vec_tot_deriv = [f"{val:8.3f}" for val in mgm_vec_tot_deriv] pw.dlog(f"MGM Total Vec Deriv: {mgm_vec_tot_deriv}") current_idx += inc_len mag_igrf_model = struct.unpack( fmt_str, hk_data[current_idx : current_idx + inc_len] ) mag_igrf_model = [f"{val:8.3f}" for val in mag_igrf_model] pw.dlog(f"MAG IGRF Model: {mag_igrf_model}") current_idx += inc_len if PERFORM_MGM_CALIBRATION: perform_mgm_calibration(pw, mgm_3) pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=8) def handle_gyr_data_raw(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received GYR Raw Set with rotation rates in deg per second") float_fmt = "!fff" double_fmt = "!ddd" inc_len_flt = struct.calcsize(float_fmt) inc_len_double = struct.calcsize(double_fmt) if len(hk_data) < 2 * inc_len_double + 2 * inc_len_flt: pw.dlog("HK data too small") return current_idx = 0 float_str_fmt = "[{:8.3f}, {:8.3f}, {:8.3f}]" gyr_0_adis = struct.unpack( double_fmt, hk_data[current_idx : current_idx + inc_len_double] ) current_idx += inc_len_double gyr_1_l3 = struct.unpack( float_fmt, hk_data[current_idx : current_idx + inc_len_flt] ) current_idx += inc_len_flt gyr_2_adis = struct.unpack( double_fmt, hk_data[current_idx : current_idx + inc_len_double] ) current_idx += inc_len_double gyr_3_l3 = struct.unpack( float_fmt, hk_data[current_idx : current_idx + inc_len_flt] ) current_idx += inc_len_flt pw.dlog(f"{'GYR 0 ADIS'.ljust(15)}: {float_str_fmt.format(*gyr_0_adis)}") pw.dlog(f"{'GYR 1 L3'.ljust(15)}: {float_str_fmt.format(*gyr_1_l3)}") pw.dlog(f"{'GYR 2 ADIS'.ljust(15)}: {float_str_fmt.format(*gyr_2_adis)}") pw.dlog(f"{'GYR 3 L3'.ljust(15)}: {float_str_fmt.format(*gyr_3_l3)}") pw.printer.print_validity_buffer(hk_data[current_idx:], 4) GYR_NAMES = ["GYR 0 ADIS", "GYR 1 L3", "GYR 2 ADIS", "GYR 3 L3"] def handle_gyr_data_processed(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received GYR Processed Set with rotation rates in deg per second") fmt_str = "!ddd" inc_len = struct.calcsize(fmt_str) current_idx = 0 for i in range(4): gyr_vec = [ f"{val:8.3f}" for val in struct.unpack( fmt_str, hk_data[current_idx : current_idx + inc_len] ) ] pw.dlog(f"{GYR_NAMES[i]}: {gyr_vec}") current_idx += inc_len gyr_vec_tot = [ f"{val:8.3f}" for val in struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) ] pw.dlog(f"GYR Vec Total: {gyr_vec_tot}") current_idx += inc_len pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=5) def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received GPS Processed Set") fmt_scalar = "!d" fmt_vec = "!ddd" inc_len_scalar = struct.calcsize(fmt_scalar) inc_len_vec = struct.calcsize(fmt_vec) if len(hk_data) < 2 * inc_len_scalar + 2 * inc_len_vec: pw.dlog("Received HK set too small") return current_idx = 0 lat = [ f"{val:8.3f}" for val in struct.unpack( fmt_scalar, hk_data[current_idx : current_idx + inc_len_scalar] ) ] current_idx += inc_len_scalar long = [ f"{val:8.3f}" for val in struct.unpack( fmt_scalar, hk_data[current_idx : current_idx + inc_len_scalar] ) ] current_idx += inc_len_scalar pos = [ f"{val:8.3f}" for val in struct.unpack( fmt_vec, hk_data[current_idx : current_idx + inc_len_vec] ) ] velo = [ f"{val:8.3f}" for val in struct.unpack( fmt_vec, hk_data[current_idx : current_idx + inc_len_vec] ) ] current_idx += inc_len_vec pw.dlog(f"GPS Latitude: {lat} [rad]") pw.dlog(f"GPS Longitude: {long} [rad]") pw.dlog(f"GPS Position: {pos} [m]") pw.dlog(f"GPS Velocity: {velo} [m/s]") pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=3) def handle_mekf_data(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received MEKF Set") fmt_quat = "!dddd" fmt_vec = "!ddd" inc_len_quat = struct.calcsize(fmt_quat) inc_len_vec = struct.calcsize(fmt_vec) if len(hk_data) < inc_len_quat + inc_len_vec: pw.dlog("Received HK set too small") return current_idx = 0 quat = [ f"{val:8.3f}" for val in struct.unpack( fmt_quat, hk_data[current_idx : current_idx + inc_len_quat] ) ] current_idx += inc_len_quat rate = [ f"{val:8.3f}" for val in struct.unpack( fmt_vec, hk_data[current_idx : current_idx + inc_len_vec] ) ] current_idx += inc_len_vec pw.dlog(f"MEKF Quaternion: {quat}") pw.dlog(f"MEKF Rotational Rate: {rate}") pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=2) def handle_ctrl_val_data(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received CTRL Values Set") fmt_quat = "!dddd" fmt_scalar = "!d" inc_len_quat = struct.calcsize(fmt_quat) inc_len_scalar = struct.calcsize(fmt_scalar) if len(hk_data) < 2 * inc_len_quat + inc_len_scalar: pw.dlog("Received HK set too small") return current_idx = 0 tgt_quat = [ f"{val:8.3f}" for val in struct.unpack( fmt_quat, hk_data[current_idx : current_idx + inc_len_quat] ) ] current_idx += inc_len_quat err_quat = [ f"{val:8.3f}" for val in struct.unpack( fmt_quat, hk_data[current_idx : current_idx + inc_len_quat] ) ] current_idx += inc_len_quat err_ang = [ f"{val:8.3f}" for val in struct.unpack( fmt_scalar, hk_data[current_idx : current_idx + inc_len_scalar] ) ] current_idx += inc_len_scalar pw.dlog(f"Control Values Target Quaternion: {tgt_quat}") pw.dlog(f"Control Values Error Quaternion: {err_quat}") pw.dlog(f"Control Values Error Angle: {err_ang} [rad]") pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=3) def handle_act_cmd_data(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received Actuator Command Values Set") fmt_vec4_double = "!dddd" fmt_vec4_int32 = "!iiii" fmt_vec3_int16 = "!hhh" inc_len_vec4_double = struct.calcsize(fmt_vec4_double) inc_len_vec4_int32 = struct.calcsize(fmt_vec4_int32) inc_len_vec3_int16 = struct.calcsize(fmt_vec3_int16) if len(hk_data) < inc_len_vec4_double + inc_len_vec4_int32 + inc_len_vec3_int16: pw.dlog("Received HK set too small") return current_idx = 0 rw_tgt_torque = [ f"{val:8.3f}" for val in struct.unpack( fmt_vec4_double, hk_data[current_idx : current_idx + inc_len_vec4_double] ) ] current_idx += inc_len_vec4_double rw_tgt_speed = [ f"{val:d}" for val in struct.unpack( fmt_vec4_int32, hk_data[current_idx : current_idx + inc_len_vec4_int32] ) ] current_idx += inc_len_vec4_int32 mtq_tgt_dipole = [ f"{val:d}" for val in struct.unpack( fmt_vec3_int16, hk_data[current_idx : current_idx + inc_len_vec3_int16] ) ] current_idx += inc_len_vec3_int16 pw.dlog(f"Actuator Commands RW Target Torque: {rw_tgt_torque}") pw.dlog(f"Actuator Commands RW Target Speed: {rw_tgt_speed}") pw.dlog(f"Actuator Commands MTQ Target Dipole: {mtq_tgt_dipole}") pw.printer.print_validity_buffer(hk_data[current_idx:], num_vars=3) def perform_mgm_calibration(pw: PrintWrapper, mgm_tuple: Tuple): global CALIBR_SOCKET, CALIBRATION_ADDR try: declare_api_cmd = "declare_api_version 2" CALIBR_SOCKET.sendall(f"{declare_api_cmd}\n".encode()) reply = CALIBR_SOCKET.recv(1024) if len(reply) != 2: pw.dlog( f"MGM calibration: Reply received command {declare_api_cmd} has" f" invalid length {len(reply)}" ) return else: if str(reply[0]) == "0": pw.dlog(f"MGM calibration: API version 2 was not accepted") return if len(mgm_tuple) != 3: pw.dlog(f"MGM tuple has invalid length {len(mgm_tuple)}") mgm_list = [mgm / 1e6 for mgm in mgm_tuple] command = ( f"magnetometer_field {mgm_list[0]} {mgm_list[1]} {mgm_list[2]}\n".encode() ) CALIBR_SOCKET.sendall(command) reply = CALIBR_SOCKET.recv(1024) if len(reply) != 2: pw.dlog( f"MGM calibration: Reply received command magnetometer_field has invalid " f"length {len(reply)}" ) return else: if str(reply[0]) == "0": pw.dlog(f"MGM calibration: magnetmeter field format was not accepted") return pw.dlog(f"Sent data {mgm_list} to Helmholtz Testbench successfully") except socket.timeout: pw.dlog("Socket timeout") except BlockingIOError as e: pw.dlog(f"Error {e}") except ConnectionResetError as e: pw.dlog("Socket was closed") except ConnectionRefusedError or OSError: pw.dlog("Connecting to Calibration Socket on addrss {} failed")