eive/eive-tmtc
eive/eive-tmtc
14
0
Fork 0
Code Issues Pull Requests 2 Projects Releases 97 Wiki Activity

re-run black

Browse Source
This commit is contained in:
Robin Müller 2022-12-21 16:07:49 +01:00
parent 6385a4a7db
commit 00991b92f1
1 changed files with 203 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

263
eive_tmtc/tmtc/acs/acs_ctrl.py
View File

@ -163,115 +163,194 @@ def pack_acs_ctrl_command(p: ServiceProviderParams):
q = p.queue_helper
if 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, SetIds.MGM_RAW_SET)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.MGM_RAW_SET), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.MGM_RAW_SET)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.MGM_PROC_SET)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.MGM_PROC_SET), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.MGM_PROC_SET)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.SUS_RAW_SET)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.SUS_RAW_SET), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.SUS_RAW_SET)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.SUS_PROC_SET)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.SUS_PROC_SET), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.SUS_PROC_SET)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.GYR_RAW_SET)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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(False, make_sid(ACS_CONTROLLER, SetIds.GYR_RAW_SET), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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(False, make_sid(ACS_CONTROLLER, SetIds.GYR_RAW_SET)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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(generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.GYR_PROC_SET)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.GYR_PROC_SET), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.GYR_PROC_SET)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.GPS_PROC_SET)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.GPS_PROC_SET), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.GPS_PROC_SET)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.MEKF_DATA)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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(False, make_sid(ACS_CONTROLLER, SetIds.MEKF_DATA), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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(False, make_sid(ACS_CONTROLLER, SetIds.MEKF_DATA)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.CTRL_VAL_DATA)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.CTRL_VAL_DATA), 2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.CTRL_VAL_DATA)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.ACTUATOR_CMD_DATA)))
q.add_pus_tc(
generate_one_hk_command(make_sid(ACS_CONTROLLER, SetIds.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, SetIds.ACTUATOR_CMD_DATA),
2.0)
cmd_tuple = enable_periodic_hk_command_with_interval(
False, make_sid(ACS_CONTROLLER, SetIds.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, SetIds.ACTUATOR_CMD_DATA)))
q.add_pus_tc(
disable_periodic_hk_command(
False, make_sid(ACS_CONTROLLER, SetIds.ACTUATOR_CMD_DATA)
)
)
else:
LOGGER.info(f"Unknown op code {op_code}")
@ -309,11 +388,15 @@ def handle_acs_ctrl_sus_raw_data(pw: PrintWrapper, hk_data: 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 = [f"{val:#04x}" for val in sus_list]
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)
@ -458,26 +541,27 @@ def handle_gyr_data_raw(pw: PrintWrapper, hk_data: bytes):
pw.dlog("HK data too small")
return
current_idx = 0
gyr_0_adis = [
f"{val:8.3f}" for val in struct.unpack(double_fmt, hk_data[:inc_len_double])
]
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 = [
f"{val:8.3f}" for val in struct.unpack(float_fmt, hk_data[:inc_len_flt])
]
gyr_1_l3 = struct.unpack(
float_fmt, hk_data[current_idx : current_idx + inc_len_flt]
)
current_idx += inc_len_flt
gyr_2_adis = [
f"{val:8.3f}" for val in struct.unpack(double_fmt, hk_data[:inc_len_double])
]
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
gyr_3_l3 = [
f"{val:8.3f}" for val in struct.unpack(float_fmt, hk_data[:inc_len_flt])
]
current_idx += inc_len_flt
pw.dlog(f"GYR 0 ADIS: {gyr_0_adis}")
pw.dlog(f"GYR 1 L3: {gyr_1_l3}")
pw.dlog(f"GYR 2 ADIS: {gyr_2_adis}")
pw.dlog(f"GYR 3 L3: {gyr_3_l3}")
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)
@ -516,15 +600,31 @@ def handle_gps_data_processed(pw: PrintWrapper, hk_data: bytes):
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])]
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])]
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
velo = [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]
)
]
pw.dlog(f"GPS Latitude: {lat} [rad]")
pw.dlog(f"GPS Longitude: {long} [rad]")
pw.dlog(f"GPS Velocity: {velo} [m/s]")
def handle_mekf_data(pw: PrintWrapper, hk_data: bytes):
pw.dlog("Received MEKF Set")
fmt_quat = "!dddd"
@ -535,9 +635,19 @@ def handle_mekf_data(pw: PrintWrapper, hk_data: bytes):
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])]
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])]
rate = [
f"{val:8.3f}"
for val in struct.unpack(
fmt_vec, hk_data[current_idx : current_idx + inc_len_vec]
)
]
pw.dlog(f"MEKF Quaternion: {quat}")
pw.dlog(f"MEKF Rotational Rate: {rate}")
@ -552,11 +662,26 @@ def handle_ctrl_val_data(pw: PrintWrapper, hk_data: bytes):
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])]
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])]
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])]
err_ang = [
f"{val:8.3f}"
for val in struct.unpack(
fmt_scalar, hk_data[current_idx : 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]")
@ -574,14 +699,26 @@ def handle_act_cmd_data(pw: PrintWrapper, hk_data: bytes):
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])]
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])]
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])]
mtq_tgt_dipole = [
f"{val:d}"
for val in struct.unpack(
fmt_vec3_int16, hk_data[current_idx : 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}")