import datetime import enum import logging import math import socket import struct from socket import AF_INET from typing import Tuple from tmtccmd.config.tmtc import CmdTreeNode from tmtccmd.fsfw.tmtc_printer import get_validity_buffer_str from tmtccmd.pus.s8_fsfw_action import create_action_cmd from tmtccmd.pus.s20_fsfw_param import create_load_param_cmd from tmtccmd.pus.s20_fsfw_param_defs import ( create_matrix_double_parameter, create_matrix_float_parameter, create_scalar_double_parameter, create_scalar_float_parameter, create_scalar_i32_parameter, create_scalar_u8_parameter, create_scalar_u16_parameter, create_vector_double_parameter, create_vector_float_parameter, ) from tmtccmd.pus.s200_fsfw_mode import Mode, pack_mode_command from tmtccmd.pus.tc.s3_fsfw_hk import ( create_request_one_diag_command, create_request_one_hk_command, disable_periodic_hk_command, enable_periodic_hk_command_with_interval, make_sid, ) from tmtccmd.tmtc.queue import DefaultPusQueueHelper from eive_tmtc.config.object_ids import ACS_CONTROLLER from eive_tmtc.pus_tm.defs import PrintWrapper from eive_tmtc.tmtc.acs.defs import AcsMode, SafeSubmode _LOGGER = logging.getLogger(__name__) 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 ATTITUDE_ESTIMATION_DATA = 7 CTRL_VAL_DATA = 8 ACTUATOR_CMD_DATA = 9 FUSED_ROT_RATE_DATA = 10 FUSED_ROT_RATE_SOURCE_DATA = 11 class DataSetRequest(enum.IntEnum): ONESHOT = 0 ENABLE = 1 DISABLE = 2 class ActionId(enum.IntEnum): SOLAR_ARRAY_DEPLOYMENT_SUCCESSFUL = 0 RESET_MEKF = 1 RESTORE_MEKF_NONFINITE_RECOVERY = 2 UPDATE_TLE = 3 READ_TLE = 4 UPDATE_MEKF_STANDARD_DEVIATIONS = 5 CTRL_STRAT_DICT = { 0: "OFF", 1: "NO_MAG_FIELD_FOR_CONTROL", 2: "NO_SENSORS_FOR_CONTROL", # OBSW <= v6.1.0 10: "LEGACY_SAFE_MEKF", 11: "LEGACY_WITHOUT_MEKF", 12: "LEGACY_ECLIPSE_DAMPING", 13: "LEGACY_ECLIPSE_IDELING", # Added in OBSW v6.2.0 14: "SAFE_MEKF", 15: "SAFE_GYR", 16: "SAFE_SUSMGM", 17: "SAFE_ECLIPSE_DAMPING_GYR", 18: "SAFE_ECLIPSE_DAMPING_SUSMGM", 19: "SAFE_ECLIPSE_IDELING", 20: "DETUMBLE_FULL", 21: "DETUMBLE_DETERIORATED", 100: "PTG_MEKF", 101: "PTG_STR", 102: "PTG_QUEST", } GPS_SOURCE_DICT = { 0: "NONE", 1: "GPS", 2: "GPS_EXTRAPOLATED", 3: "SGP4", } FUSED_ROT_RATE_SOURCE_DICT = { 0: "NONE", 1: "SUSMGM", 2: "QUEST", 3: "STR", } class OpCodes: OFF = "off" SAFE = "safe" DTBL = "safe_detumble" IDLE = "ptg_idle" NADIR = "ptg_nadir" TARGET = "ptg_target" GS = "ptg_target_gs" INERTIAL = "ptg_inertial" SAFE_PTG = "confirm_deployment" RESET_MEKF = "reset_mekf" RESTORE_MEKF_NONFINITE_RECOVERY = "restore_mekf_nonfinite_recovery" UPDATE_TLE = "update_tle" READ_TLE = "read_tle" UPDATE_MEKF_STANDARD_DEVIATIONS = "update_mekf_standard_deviations" SET_PARAMETER_SCALAR = "set_scalar_param" SET_PARAMETER_VECTOR = "set_vector_param" SET_PARAMETER_MATRIX = "set_matrix_param" ONE_SHOOT_HK = "one_shot_hk" ENABLE_HK = "enable_hk" DISABLE_HK = "disable_hk" class Info: OFF = "Switch ACS CTRL off" SAFE = "Switch ACS CTRL - safe" DTBL = "Switch ACS CTRL - safe with detumble submode" IDLE = "Switch ACS CTRL - pointing idle" NADIR = "Switch ACS CTRL normal - pointing nadir" TARGET = "Switch ACS CTRL normal - pointing target" GS = "Switch ACS CTRL normal - pointing target groundstation" INERTIAL = "Switch ACS CTRL normal - pointing inertial" SAFE_PTG = "Confirm deployment of both solar arrays" RESET_MEKF = "Reset the MEKF" RESTORE_MEKF_NONFINITE_RECOVERY = "Restore MEKF non-finite recovery" UPDATE_TLE = "Update TLE" READ_TLE = "Read the currently stored TLE" UPDATE_MEKF_STANDARD_DEVIATIONS = ( "Update the Standard Deviations within the MEKF to the current ACS Parameter " "Values" ) SET_PARAMETER_SCALAR = "Set Scalar Parameter" SET_PARAMETER_VECTOR = "Set Vector Parameter" SET_PARAMETER_MATRIX = "Set Matrix Parameter" ONE_SHOOT_HK = "One shoot HK Set" ENABLE_HK = "Enable Periodic HK" DISABLE_HK = "Disable Periodic HK" 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) def create_acs_ctrl_node() -> CmdTreeNode: # Zip the two classes together into a dictionary op_code_strs = [ getattr(OpCodes, key) for key in dir(OpCodes) if not key.startswith("__") ] info_strs = [getattr(Info, key) for key in dir(OpCodes) if not key.startswith("__")] combined_dict = dict(zip(op_code_strs, info_strs)) acs_ctrl = CmdTreeNode( "acs_ctrl", "ACS Controller", hide_children_which_are_leaves=True ) for op_code, info in combined_dict.items(): acs_ctrl.add_child(CmdTreeNode(op_code, info)) return acs_ctrl def pack_acs_ctrl_command(q: DefaultPusQueueHelper, cmd_str: str): # noqa C901 if cmd_str in OpCodes.OFF: q.add_log_cmd(f"{Info.OFF}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, Mode.OFF, 0)) elif cmd_str in OpCodes.SAFE: q.add_log_cmd(f"{Info.SAFE}") q.add_pus_tc( pack_mode_command(ACS_CONTROLLER, AcsMode.SAFE, SafeSubmode.DEFAULT) ) elif cmd_str in OpCodes.DTBL: q.add_log_cmd(f"{Info.DTBL}") q.add_pus_tc( pack_mode_command(ACS_CONTROLLER, AcsMode.SAFE, SafeSubmode.DETUMBLE) ) elif cmd_str in OpCodes.IDLE: q.add_log_cmd(f"{Info.IDLE}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, AcsMode.IDLE, 0)) elif cmd_str in OpCodes.NADIR: q.add_log_cmd(f"{Info.NADIR}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, AcsMode.PTG_NADIR, 0)) elif cmd_str in OpCodes.TARGET: q.add_log_cmd(f"{Info.TARGET}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, AcsMode.PTG_TARGET, 0)) elif cmd_str in OpCodes.GS: q.add_log_cmd(f"{Info.GS}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, AcsMode.PTG_TARGET_GS, 0)) elif cmd_str in OpCodes.INERTIAL: q.add_log_cmd(f"{Info.INERTIAL}") q.add_pus_tc(pack_mode_command(ACS_CONTROLLER, AcsMode.PTG_INERTIAL, 0)) elif cmd_str in OpCodes.SAFE_PTG: q.add_log_cmd(f"{Info.SAFE_PTG}") q.add_pus_tc( create_action_cmd( ACS_CONTROLLER, ActionId.SOLAR_ARRAY_DEPLOYMENT_SUCCESSFUL ) ) elif cmd_str in OpCodes.RESET_MEKF: q.add_log_cmd(f"{Info.RESET_MEKF}") q.add_pus_tc(create_action_cmd(ACS_CONTROLLER, ActionId.RESET_MEKF)) elif cmd_str in OpCodes.RESTORE_MEKF_NONFINITE_RECOVERY: q.add_log_cmd(f"{Info.RESTORE_MEKF_NONFINITE_RECOVERY}") q.add_pus_tc( create_action_cmd(ACS_CONTROLLER, ActionId.RESTORE_MEKF_NONFINITE_RECOVERY) ) elif cmd_str in OpCodes.UPDATE_TLE: q.add_log_cmd(f"{Info.UPDATE_TLE}") while True: line1 = input("Please input the first line of the TLE: ") if len(line1) == 69: break else: print("The line does not have the required length of 69 characters") while True: line2 = input("Please input the second line of the TLE: ") if len(line2) == 69: break else: print("The line does not have the required length of 69 characters") tle = line1.encode() + line2.encode() q.add_pus_tc(create_action_cmd(ACS_CONTROLLER, ActionId.UPDATE_TLE, tle)) elif cmd_str == OpCodes.READ_TLE: q.add_log_cmd(f"{Info.READ_TLE}") q.add_pus_tc(create_action_cmd(ACS_CONTROLLER, ActionId.READ_TLE)) elif cmd_str == OpCodes.UPDATE_MEKF_STANDARD_DEVIATIONS: q.add_log_cmd(f"{Info.UPDATE_MEKF_STANDARD_DEVIATIONS}") q.add_pus_tc( create_action_cmd(ACS_CONTROLLER, ActionId.UPDATE_MEKF_STANDARD_DEVIATIONS) ) elif cmd_str == OpCodes.SET_PARAMETER_SCALAR: q.add_log_cmd(f"{Info.SET_PARAMETER_SCALAR}") set_acs_ctrl_param_scalar(q) elif cmd_str == OpCodes.SET_PARAMETER_VECTOR: q.add_log_cmd(f"{Info.SET_PARAMETER_VECTOR}") set_acs_ctrl_param_vector(q) elif cmd_str == OpCodes.SET_PARAMETER_MATRIX: q.add_log_cmd(f"{Info.SET_PARAMETER_MATRIX}") set_acs_ctrl_param_matrix(q) elif cmd_str == OpCodes.ONE_SHOOT_HK: q.add_log_cmd(Info.ONE_SHOOT_HK) request_dataset(q, DataSetRequest.ONESHOT) elif cmd_str == OpCodes.ENABLE_HK: q.add_log_cmd(Info.ENABLE_HK) request_dataset(q, DataSetRequest.ENABLE) elif cmd_str == OpCodes.DISABLE_HK: q.add_log_cmd(Info.DISABLE_HK) request_dataset(q, DataSetRequest.DISABLE) else: logging.getLogger(__name__).info(f"Unknown op code {cmd_str}") def request_dataset(q: DefaultPusQueueHelper, req_type: DataSetRequest): for val in SetId: print("{:<2}: {:<20}".format(val, val.name)) set_id = int(input("Specify the dataset \n" "")) if set_id in [ SetId.GYR_RAW_SET, SetId.GPS_PROC_SET, SetId.ATTITUDE_ESTIMATION_DATA, ]: is_diag = True else: is_diag = False match req_type: case DataSetRequest.ONESHOT: if is_diag: q.add_pus_tc( create_request_one_diag_command(make_sid(ACS_CONTROLLER, set_id)) ) else: q.add_pus_tc( create_request_one_hk_command(make_sid(ACS_CONTROLLER, set_id)) ) case DataSetRequest.ENABLE: interval = float( input("Please specify interval in floating point seconds: ") ) if is_diag: cmd_tuple = enable_periodic_hk_command_with_interval( True, make_sid(ACS_CONTROLLER, set_id), interval ) else: cmd_tuple = enable_periodic_hk_command_with_interval( False, make_sid(ACS_CONTROLLER, set_id), interval ) q.add_pus_tc(cmd_tuple[0]) q.add_pus_tc(cmd_tuple[1]) case DataSetRequest.DISABLE: if is_diag: q.add_pus_tc( disable_periodic_hk_command(True, make_sid(ACS_CONTROLLER, set_id)) ) else: q.add_pus_tc( disable_periodic_hk_command(False, make_sid(ACS_CONTROLLER, set_id)) ) def set_acs_ctrl_param_scalar(q: DefaultPusQueueHelper): pt = int( input( 'Specify parameter type to set {0: "uint8", 1: "uint16", 2: "int32", 3:' ' "float", 4: "double"}: ' ) ) sid = int(input("Specify parameter struct ID to set: ")) pid = int(input("Specify parameter ID to set: ")) match pt: case 0: param = int(input("Specify parameter value to set: ")) q.add_pus_tc( create_load_param_cmd( create_scalar_u8_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameter=param, ) ) ) case 1: param = int(input("Specify parameter value to set: ")) q.add_pus_tc( create_load_param_cmd( create_scalar_u16_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameter=param, ) ) ) case 2: param = int(input("Specify parameter value to set: ")) q.add_pus_tc( create_load_param_cmd( create_scalar_i32_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameter=param, ) ) ) case 3: param = float(input("Specify parameter value to set: ")) q.add_pus_tc( create_load_param_cmd( create_scalar_float_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameter=param, ) ) ) case 4: param = float(input("Specify parameter value to set: ")) q.add_pus_tc( create_load_param_cmd( create_scalar_double_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameter=param, ) ) ) def set_acs_ctrl_param_vector(q: DefaultPusQueueHelper): pt = int(input('Specify parameter type to set {0: "float", 1: "double"}: ')) sid = int(input("Specify parameter struct ID to set: ")) pid = int(input("Specify parameter ID to set: ")) match pt: case 0: elms = int(input("Specify number of elements in vector to set: ")) param = [] for _ in range(elms): param.append( float(input("Specify parameter vector entry value to set: ")) ) print(param) if input("Confirm selected parameter values (Y/N): ") == "Y": q.add_pus_tc( create_load_param_cmd( create_vector_float_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameters=param, ) ) ) else: q.add_log_cmd("Aborted by user input") return case 1: elms = int(input("Specify number of elements in vector to set: ")) param = [] for _ in range(elms): param.append( float(input("Specify parameter vector entry value to set: ")) ) print(param) if input("Confirm selected parameter values (Y/N): ") == "Y": q.add_pus_tc( create_load_param_cmd( create_vector_double_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameters=param, ) ) ) else: q.add_log_cmd("Aborted by user input") return def set_acs_ctrl_param_matrix(q: DefaultPusQueueHelper): pt = int(input('Specify parameter type to set {0: "float", 1: "double"}: ')) sid = int(input("Specify parameter struct ID to set: ")) pid = int(input("Specify parameter ID to set: ")) match pt: case 0: rows = int(input("Specify number of rows in matrix to set: ")) cols = int(input("Specify number of columns in matrix to set: ")) row = [] param = [] for _ in range(rows): for _ in range(cols): row.append( float(input("Specify parameter vector entry value to set: ")) ) param.append(row) print(param) if input("Confirm selected parameter values (Y/N): ") == "Y": q.add_pus_tc( create_load_param_cmd( create_matrix_float_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameters=param, ) ) ) else: q.add_log_cmd("Aborted by user input") return case 1: rows = int(input("Specify number of rows in matrix to set: ")) cols = int(input("Specify number of columns in matrix to set: ")) row = [] param = [] for _ in range(rows): for _ in range(cols): row.append( float(input("Specify parameter vector entry value to set: ")) ) param.append(row) row = [] print(param) if input("Confirm selected parameter values (Y/N): ") == "Y": q.add_pus_tc( create_load_param_cmd( create_matrix_double_parameter( object_id=ACS_CONTROLLER, domain_id=sid, unique_id=pid, parameters=param, ) ) ) else: q.add_log_cmd("Aborted by user input") return def handle_acs_ctrl_hk_data( pw: PrintWrapper, set_id: int, hk_data: bytes, packet_time: datetime.datetime, ): pw.ilog(_LOGGER, f"Received ACS CTRL HK with packet time {packet_time}") 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.ATTITUDE_ESTIMATION_DATA: handle_attitude_estimation_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) case SetId.FUSED_ROT_RATE_DATA: handle_fused_rot_rate_data(pw, hk_data) case SetId.FUSED_ROT_RATE_SOURCE_DATA: handle_fused_rot_rate_source_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.dlog(get_validity_buffer_str(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" f" size of {3 * 4 * 12 + 3 * 8 * 3} bytes" ) return current_idx = 0 vec_fmt = "[{:8.3f}, {:8.3f}, {:8.3f}]" 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 = vec_fmt.format(*sus_list) current_idx += length pw.dlog(f"{f'SUS {idx} CALIB'.ljust(25)}: {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 = vec_fmt.format(*sus_vec_tot) current_idx += inc_len pw.dlog(f"{'SUS Vector Total'.ljust(25)}: {sus_vec_tot}") sus_vec_tot_deriv = struct.unpack( fmt_str, hk_data[current_idx : current_idx + inc_len] ) sus_vec_tot_deriv = vec_fmt.format(*sus_vec_tot_deriv) current_idx += inc_len pw.dlog(f"{'SUS Vector Derivative'.ljust(25)}: {sus_vec_tot_deriv}") sun_ijk_model = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) sun_ijk_model = vec_fmt.format(*sun_ijk_model) current_idx += inc_len pw.dlog(f"{'SUS ijk Model'.ljust(25)}: {sun_ijk_model}") pw.dlog(get_validity_buffer_str(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[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.dlog(get_validity_buffer_str(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" vec_fmt = "[{:8.3f}, {:8.3f}, {:8.3f}]" inc_len = struct.calcsize(fmt_str) mgm_0 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_0_str = vec_fmt.format(*mgm_0) pw.dlog(f"{'MGM 0 Vec'.ljust(25)}: {mgm_0_str}") current_idx += inc_len mgm_1 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_1_str = vec_fmt.format(*mgm_1) pw.dlog(f"{'MGM 1 Vec'.ljust(25)}: {mgm_1_str}") current_idx += inc_len mgm_2 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_2_str = vec_fmt.format(*mgm_2) pw.dlog(f"{'MGM 2 Vec'.ljust(25)}: {mgm_2_str}") current_idx += inc_len mgm_3 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_3_str = vec_fmt.format(*mgm_3) pw.dlog(f"{'MGM 3 Vec'.ljust(25)}: {mgm_3_str}") current_idx += inc_len mgm_4 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len]) mgm_4_str = vec_fmt.format(*mgm_4) pw.dlog(f"{'MGM 4 Vec'.ljust(25)}: {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 = vec_fmt.format(*mgm_vec_tot) current_idx += inc_len pw.dlog(f"{'MGM Total Vec'.ljust(25)}: {mgm_vec_tot}") mgm_vec_tot_deriv = struct.unpack( fmt_str, hk_data[current_idx : current_idx + inc_len] ) mgm_vec_tot_deriv = vec_fmt.format(*mgm_vec_tot_deriv) pw.dlog(f"{'MGM Total Vec Deriv'.ljust(25)}: {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 = vec_fmt.format(*mag_igrf_model) pw.dlog(f"{'MAG IGRF Model'.ljust(25)}: {mag_igrf_model}") current_idx += inc_len if PERFORM_MGM_CALIBRATION: perform_mgm_calibration(pw, mgm_3) pw.dlog(get_validity_buffer_str(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.dlog(get_validity_buffer_str(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*180/math.pi: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*180/math.pi: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.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=5)) def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received GPS Processed Set") fmt_source = "!B" fmt_scalar = "!d" fmt_vec = "!ddd" inc_len_source = struct.calcsize(fmt_source) inc_len_scalar = struct.calcsize(fmt_scalar) inc_len_vec = struct.calcsize(fmt_vec) if len(hk_data) < 3 * inc_len_scalar + 2 * inc_len_vec + inc_len_source: pw.dlog("Received HK set too small") return current_idx = 0 lat = [ f"{val*180/math.pi: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*180/math.pi:8.3f}" for val in struct.unpack( fmt_scalar, hk_data[current_idx : current_idx + inc_len_scalar] ) ] current_idx += inc_len_scalar alt = [ 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] ) ] 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 source = struct.unpack( fmt_source, hk_data[current_idx : current_idx + inc_len_source] )[0] current_idx += inc_len_source if GPS_SOURCE_DICT.get(source) is not None: pw.dlog(f"GPS Source: {GPS_SOURCE_DICT[source]}") else: pw.dlog(f"'GPS Source (key unknown)': {source}") pw.dlog(f"GPS Latitude: {lat} [deg]") pw.dlog(f"GPS Longitude: {long} [deg]") pw.dlog(f"GPS Altitude: {alt} [m]") pw.dlog(f"GPS Position: {pos} [m]") pw.dlog(f"GPS Velocity: {velo} [m/s]") pw.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=6)) def handle_attitude_estimation_data(pw: PrintWrapper, hk_data: bytes): mekf_status = { 0: "UNINITIALIZED", 1: "NO_GYR_DATA", 2: "NO_MODEL_VECTORS", 3: "NO_SUS_MGM_STR_DATA", 4: "COVARIANCE_INVERSION_FAILED", 5: "NOT_FINITE", 10: "INITIALIZED", 11: "RUNNING", } pw.dlog("Received Attitude Estimation Set") fmt_quat = "!dddd" fmt_str_4 = "[{:8.3f}, {:8.3f}, {:8.3f}, {:8.3f}]" fmt_str_3 = "[{:8.3f}, {:8.3f}, {:8.3f}]" fmt_vec = "!ddd" fmt_sts = "!B" inc_len_quat = struct.calcsize(fmt_quat) inc_len_vec = struct.calcsize(fmt_vec) inc_len_sts = struct.calcsize(fmt_sts) old_size = inc_len_quat + inc_len_vec + inc_len_sts + 1 new_size = 2 * inc_len_quat + inc_len_vec + inc_len_sts + 1 size = len(hk_data) if size not in [old_size, new_size]: pw.dlog(f"Received Attitude Estimation HK Set of unexpected size: {size}") return current_idx = 0 mekf_quat = struct.unpack( fmt_quat, hk_data[current_idx : current_idx + inc_len_quat] ) current_idx += inc_len_quat rates = [ rate * 180 / math.pi for rate in struct.unpack( fmt_vec, hk_data[current_idx : current_idx + inc_len_vec] ) ] current_idx += inc_len_vec status = struct.unpack(fmt_sts, hk_data[current_idx : current_idx + inc_len_sts])[0] current_idx += inc_len_sts if mekf_status.get(status) is not None: pw.dlog(f"{'MEKF Status'.ljust(25)}: {mekf_status[status]}") else: pw.dlog(f"{'MEKF Raw Status (key unknown)'.ljust(25)}: {status}") pw.dlog(f"{'MEKF Quaternion'.ljust(25)}: {fmt_str_4.format(*mekf_quat)}") pw.dlog(f"{'MEKF Rotational Rate'.ljust(25)}: {fmt_str_3.format(*rates)}") if size == new_size: quest_quat = struct.unpack( fmt_quat, hk_data[current_idx : current_idx + inc_len_quat] ) current_idx += inc_len_quat pw.dlog(f"{'QUEST Quaternion'.ljust(25)}: {fmt_str_4.format(*quest_quat)}") pw.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=4)) return pw.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=3)) def handle_ctrl_val_data(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received CTRL Values Set") fmt_strat = "!B" fmt_quat = "!dddd" fmt_scalar = "!d" fmt_vec = "!ddd" inc_len_strat = struct.calcsize(fmt_strat) inc_len_quat = struct.calcsize(fmt_quat) inc_len_scalar = struct.calcsize(fmt_scalar) inc_len_vec = struct.calcsize(fmt_vec) if len(hk_data) < inc_len_strat + 2 * inc_len_quat + inc_len_scalar + inc_len_vec: pw.dlog("Received HK set too small") return current_idx = 0 strat = struct.unpack( fmt_strat, hk_data[current_idx : current_idx + inc_len_strat] )[0] current_idx += inc_len_strat 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*180/math.pi:8.3f}" for val in struct.unpack( fmt_scalar, hk_data[current_idx : current_idx + inc_len_scalar] ) ] current_idx += inc_len_scalar tgt_rot = [ f"{val*180/math.pi:8.3f}" for val in struct.unpack( fmt_vec, hk_data[current_idx : current_idx + inc_len_vec] ) ] current_idx += inc_len_vec if CTRL_STRAT_DICT.get(strat) is not None: pw.dlog(f"{'Ctrl Strategy'.ljust(25)}: {CTRL_STRAT_DICT[strat]}") else: pw.dlog(f"{'Ctrl Strategy (key unknown)'.ljust(25)}: {strat}") 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} [deg]") pw.dlog(f"Control Values Target Rotational Rate: {tgt_rot} [deg/s]") pw.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=5)) 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.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=3)) def handle_fused_rot_rate_data(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received Fused Rotation Rates Data Set") fmt_vec3_double = "!ddd" inc_len_vec3_double = struct.calcsize(fmt_vec3_double) fmt_source = "!B" inc_len_source = struct.calcsize(fmt_source) old_size = 3 * inc_len_vec3_double + 1 new_size = 3 * inc_len_vec3_double + inc_len_source + 1 size = len(hk_data) if size not in [old_size, new_size]: pw.dlog(f"Received Fused Rot Rate HK set of unexpected size: {len(hk_data)}") return current_idx = 0 rot_rate_orthogonal = [ f"{val*180/math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double rot_rate_parallel = [ f"{val*180/math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double rot_rate_total = [ f"{val*180/math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double pw.dlog(f"Fused Rotational Rate Orthogonal: {rot_rate_orthogonal} [deg/s]") pw.dlog(f"Fused Rotational Rate Parallel: {rot_rate_parallel} [deg/s]") pw.dlog(f"Fused Rotational Rate Total: {rot_rate_total} [deg/s]") if size == new_size: rot_rate_source = struct.unpack( fmt_source, hk_data[current_idx : current_idx + inc_len_source] )[0] current_idx += inc_len_source if FUSED_ROT_RATE_SOURCE_DICT.get(rot_rate_source) is not None: pw.dlog( f"Fused Rotational Rate Source: {FUSED_ROT_RATE_SOURCE_DICT[rot_rate_source]}" ) else: pw.dlog(f"Ctrl Strategy (key unknown): {rot_rate_source}") pw.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=4)) return pw.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=3)) def handle_fused_rot_rate_source_data(pw: PrintWrapper, hk_data: bytes): pw.dlog("Received Fused Rotation Rates Sources Data Set") fmt_vec3_double = "!ddd" inc_len_vec3_double = struct.calcsize(fmt_vec3_double) if len(hk_data) < 5 * inc_len_vec3_double: pw.dlog("Received HK set too small") return current_idx = 0 rot_rate_orthogonal_susmgm = [ f"{val*180/math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double rot_rate_parallel_susmgm = [ f"{val*180/math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double rot_rate_total_susmgm = [ f"{val*180/math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double rot_rate_total_quest = [ f"{val * 180 / math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double rot_rate_total_str = [ f"{val * 180 / math.pi:8.3f}" for val in struct.unpack( fmt_vec3_double, hk_data[current_idx : current_idx + inc_len_vec3_double] ) ] current_idx += inc_len_vec3_double pw.dlog( f"Fused Rotational Rate Orthogonal SUSMGM: {rot_rate_orthogonal_susmgm} [deg/s]" ) pw.dlog( f"Fused Rotational Rate Parallel SUSMGM: {rot_rate_parallel_susmgm} [deg/s]" ) pw.dlog(f"Fused Rotational Rate Total SUSMGM: {rot_rate_total_susmgm} [deg/s]") pw.dlog(f"Fused Rotational Rate Total QUEST: {rot_rate_total_quest} [deg/s]") pw.dlog(f"Fused Rotational Rate Total STR: {rot_rate_total_str} [deg/s]") pw.dlog(get_validity_buffer_str(hk_data[current_idx:], num_vars=5)) def handle_acs_ctrl_action_replies( action_id: int, pw: PrintWrapper, custom_data: bytes ): if action_id == ActionId.READ_TLE: handle_read_tle(pw, custom_data) def handle_read_tle(pw: PrintWrapper, custom_data: bytes): pw.dlog("Received TLE") data_length = 69 * 2 if len(custom_data) != data_length: raise ValueError(f"Received data of unexpected length {len(custom_data)}") tle = custom_data.decode() pw.dlog(f"{tle[0:69]}\n{tle[69:69*2]}") def perform_mgm_calibration( # noqa C901: Complexity okay pw: PrintWrapper, mgm_tuple: Tuple ): # noqa C901: Complexity okay global CALIBR_SOCKET, CALIBRATION_ADDR if not PERFORM_MGM_CALIBRATION: return assert CALIBR_SOCKET is not None 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("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( "MGM calibration: Reply received command magnetometer_field has" f" invalid length {len(reply)}" ) return else: if str(reply[0]) == "0": pw.dlog("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(f"Socket was closed: {e}") except ConnectionRefusedError or OSError: pw.dlog("Connecting to Calibration Socket on addrss {} failed")