From 451c2c07ee749cac52eb1f69b0f64cbf3b5dbdbe Mon Sep 17 00:00:00 2001 From: meggert Date: Tue, 5 Dec 2023 16:41:29 +0100 Subject: [PATCH] stuff --- eive_tmtc/tmtc/acs/acs_ctrl.py | 144 +++++++++++++++++++++++++++------ 1 file changed, 119 insertions(+), 25 deletions(-) diff --git a/eive_tmtc/tmtc/acs/acs_ctrl.py b/eive_tmtc/tmtc/acs/acs_ctrl.py index f281b54..7b8dbcb 100644 --- a/eive_tmtc/tmtc/acs/acs_ctrl.py +++ b/eive_tmtc/tmtc/acs/acs_ctrl.py @@ -97,13 +97,20 @@ CTRL_STRAT_DICT = { 31: "PTG_RAW", } -GPS_COURCE_DICT = { +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" @@ -533,13 +540,15 @@ def handle_acs_ctrl_hk_data( case SetId.GPS_PROC_SET: handle_gps_data_processed(pw, hk_data) case SetId.ATTITUDE_ESTIMATION_DATA: - handle_mekf_data(pw, hk_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): @@ -561,7 +570,7 @@ def handle_acs_ctrl_sus_raw_data(pw: PrintWrapper, hk_data: bytes): sus_list_formatted = vec_fmt.format(*sus_list) current_idx += length pw.dlog(f"SUS {idx} RAW: {sus_list_formatted}") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=12) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=12)) def handle_acs_ctrl_sus_processed_data(pw: PrintWrapper, hk_data: bytes): @@ -597,7 +606,7 @@ def handle_acs_ctrl_sus_processed_data(pw: PrintWrapper, hk_data: bytes): sun_ijk_model = vec_fmt.format(*sun_ijk_model) current_idx += inc_len pw.dlog(f"{'SUS ijk Model'.ljust(25)}: {sun_ijk_model}") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=15) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=15)) def handle_raw_mgm_data(pw: PrintWrapper, hk_data: bytes): @@ -653,7 +662,7 @@ def handle_raw_mgm_data(pw: PrintWrapper, hk_data: bytes): pw.dlog(f"{entry[0].ljust(28)}: {entry[1]}") current_idx += 1 assert current_idx == 61 - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=6) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=6)) def handle_mgm_data_processed(pw: PrintWrapper, hk_data: bytes): @@ -707,7 +716,7 @@ def handle_mgm_data_processed(pw: PrintWrapper, hk_data: bytes): current_idx += inc_len if PERFORM_MGM_CALIBRATION: perform_mgm_calibration(pw, mgm_3) - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=8) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=8)) def handle_gyr_data_raw(pw: PrintWrapper, hk_data: bytes): @@ -741,7 +750,7 @@ def handle_gyr_data_raw(pw: PrintWrapper, hk_data: bytes): 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)}") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], 4) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], 4)) GYR_NAMES = ["GYR 0 ADIS", "GYR 1 L3", "GYR 2 ADIS", "GYR 3 L3"] @@ -767,7 +776,7 @@ def handle_gyr_data_processed(pw: PrintWrapper, hk_data: bytes): ] pw.dlog(f"GYR Vec Total: {gyr_vec_tot}") current_idx += inc_len - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=5) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=5)) def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes): @@ -821,8 +830,8 @@ def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes): fmt_source, hk_data[current_idx : current_idx + inc_len_source] )[0] current_idx += inc_len_source - if GPS_COURCE_DICT.get(source) is not None: - pw.dlog(f"GPS Source: {GPS_COURCE_DICT[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]") @@ -830,10 +839,10 @@ def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes): pw.dlog(f"GPS Altitude: {alt} [m]") pw.dlog(f"GPS Position: {pos} [m]") pw.dlog(f"GPS Velocity: {velo} [m/s]") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=6) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=6)) -def handle_mekf_data(pw: PrintWrapper, hk_data: bytes): +def handle_attitude_estimation_data(pw: PrintWrapper, hk_data: bytes): mekf_status = { 0: "UNINITIALIZED", 1: "NO_GYR_DATA", @@ -844,7 +853,7 @@ def handle_mekf_data(pw: PrintWrapper, hk_data: bytes): 10: "INITIALIZED", 11: "RUNNING", } - pw.dlog("Received MEKF Set") + 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}]" @@ -853,11 +862,16 @@ def handle_mekf_data(pw: PrintWrapper, hk_data: bytes): inc_len_quat = struct.calcsize(fmt_quat) inc_len_vec = struct.calcsize(fmt_vec) inc_len_sts = struct.calcsize(fmt_sts) - if len(hk_data) < inc_len_quat + inc_len_vec + inc_len_sts: - pw.dlog("Received HK set too small") + 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 - quat = struct.unpack(fmt_quat, hk_data[current_idx : current_idx + inc_len_quat]) + 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 @@ -872,9 +886,16 @@ def handle_mekf_data(pw: PrintWrapper, hk_data: bytes): 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(*quat)}") + 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)}") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3) + if size == new_size: + quest_quat = struct.unpack( + fmt_quat, hk_data[current_idx : current_idx + inc_len_quat] + ) + pw.dlog(f"{'QUEST Quaternion'.ljust(25)}: {fmt_str_4.format(*quest_quat)}") + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=4)) + return + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3)) def handle_ctrl_val_data(pw: PrintWrapper, hk_data: bytes): @@ -924,14 +945,14 @@ def handle_ctrl_val_data(pw: PrintWrapper, hk_data: bytes): ] current_idx += inc_len_vec if CTRL_STRAT_DICT.get(strat) is not None: - pw.dlog(f"{'Safe Ctrl Strategy'.ljust(25)}: {CTRL_STRAT_DICT[strat]}") + pw.dlog(f"{'Ctrl Strategy'.ljust(25)}: {CTRL_STRAT_DICT[strat]}") else: - pw.dlog(f"{'Safe Ctrl Strategy (key unknown)'.ljust(25)}: {strat}") + 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]") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=5) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=5)) def handle_act_cmd_data(pw: PrintWrapper, hk_data: bytes): @@ -970,15 +991,20 @@ def handle_act_cmd_data(pw: PrintWrapper, hk_data: bytes): 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}") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3) + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(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) - if len(hk_data) < 3 * inc_len_vec3_double: - pw.dlog("Received HK set too small") + 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 = [ @@ -1005,7 +1031,75 @@ def handle_fused_rot_rate_data(pw: PrintWrapper, hk_data: bytes): 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]") - FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=3) + 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(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=4)) + return + pw.dlog(FsfwTmTcPrinter.get_validity_buffer(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(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=5)) def handle_acs_ctrl_action_replies(