diff --git a/pus_tc/cmd_definitions.py b/pus_tc/cmd_definitions.py index 31ab58d..e8f8266 100644 --- a/pus_tc/cmd_definitions.py +++ b/pus_tc/cmd_definitions.py @@ -357,7 +357,10 @@ def add_imtq_cmds(cmd_dict: ServiceOpCodeDictT): "9": ("IMTQ command dipole", {OpCodeDictKeys.TIMEOUT: 2.0}), "10": ("IMTQ get commanded dipole", {OpCodeDictKeys.TIMEOUT: 2.0}), "11": ("IMTQ get engineering hk set", {OpCodeDictKeys.TIMEOUT: 2.0}), - "12": ("IMTQ get calibrated MTM measurement one shot", {OpCodeDictKeys.TIMEOUT: 2.0}), + "12": ( + "IMTQ get calibrated MTM measurement one shot", + {OpCodeDictKeys.TIMEOUT: 2.0}, + ), "13": ("IMTQ get raw MTM measurement one shot", {OpCodeDictKeys.TIMEOUT: 2.0}), } service_imtq_tuple = ("IMTQ Device", op_code_dict_srv_imtq) diff --git a/pus_tc/devs/imtq.py b/pus_tc/devs/imtq.py index 754c8a6..f71de6b 100644 --- a/pus_tc/devs/imtq.py +++ b/pus_tc/devs/imtq.py @@ -9,7 +9,11 @@ from tmtccmd.config.definitions import QueueCommands from tmtccmd.tc.packer import TcQueueT from spacepackets.ecss.tc import PusTelecommand -from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_diag_command, generate_one_hk_command +from tmtccmd.tc.pus_3_fsfw_hk import ( + make_sid, + generate_one_diag_command, + generate_one_hk_command, +) from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes diff --git a/pus_tc/system/proc.py b/pus_tc/system/proc.py index 1790416..7ad725b 100644 --- a/pus_tc/system/proc.py +++ b/pus_tc/system/proc.py @@ -1,3 +1,5 @@ +from typing import List + from config.definitions import CustomServiceList from tmtccmd.config import ( QueueCommands, @@ -146,12 +148,12 @@ def pack_generic_hk_listening_cmds( ) elif one_rw is True: - activate_one_rw_in_sequence( + activate_all_rws_in_sequence( tc_queue=tc_queue, ) elif two_rws is True: - activate_two_rws_in_sequence( + activate_all_rws_two_consecutively( tc_queue=tc_queue, ) @@ -652,128 +654,89 @@ def activate_mgts_alternately( tc_queue.appendleft((QueueCommands.WAIT, 40.0)) -def activate_one_rw_in_sequence( - tc_queue: TcQueueT, -): +def rw_speed_cmd_single( + tc_queue: TcQueueT, oid: bytes, init_ssc: int, speed: int, ramp_time: int +) -> int: + command = pack_set_speed_command( + object_id=oid, speed=speed, ramp_time=ramp_time, ssc=init_ssc + ) + init_ssc += 1 + tc_queue.appendleft(command.pack_command_tuple()) + tc_queue.appendleft((QueueCommands.WAIT, 70.0)) + command = pack_set_speed_command( + object_id=oids.RW1_ID, speed=0, ramp_time=ramp_time, ssc=init_ssc + ) + tc_queue.appendleft(command.pack_command_tuple()) + return init_ssc + 1 + + +def rw_speed_up_cmd_consec( + tc_queue: TcQueueT, obids: List[bytes], init_ssc: int, speed: int, ramp_time: int +) -> int: + for oid in obids: + command = pack_set_speed_command( + object_id=oid, speed=speed, ramp_time=ramp_time, ssc=init_ssc + ) + tc_queue.appendleft(command.pack_command_tuple()) + init_ssc += 1 + return init_ssc + + +def rw_speed_down_cmd_consec( + tc_queue: TcQueueT, obids: List[bytes], init_ssc: int, ramp_time: int +) -> int: + for oid in obids: + command = pack_set_speed_command( + object_id=oid, speed=0, ramp_time=ramp_time, ssc=init_ssc + ) + tc_queue.appendleft(command.pack_command_tuple()) + init_ssc += 1 + return init_ssc + + +def activate_all_rws_in_sequence( + tc_queue: TcQueueT, init_ssc: int, test_speed: int, test_ramp_time: int +) -> int: + new_ssc = init_ssc # RW1 speed cmd tc_queue.appendleft((QueueCommands.WAIT, 5.0)) - command = pack_set_speed_command( - object_id=oids.RW1_ID, - speed=20000, - ramp_time=10000, + new_ssc = rw_speed_cmd_single( + tc_queue, oids.RW1_ID, new_ssc, test_speed, test_ramp_time ) - tc_queue.appendleft(command.pack_command_tuple()) - tc_queue.appendleft((QueueCommands.WAIT, 70.0)) - command = pack_set_speed_command( - object_id=oids.RW1_ID, - speed=0, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft((QueueCommands.WAIT, 10.0)) - - # RW2 speed cmd - command = pack_set_speed_command( - object_id=oids.RW2_ID, - speed=20000, - ramp_time=10000, + new_ssc = rw_speed_cmd_single( + tc_queue, oids.RW2_ID, new_ssc, test_speed, test_ramp_time ) - tc_queue.appendleft(command.pack_command_tuple()) - tc_queue.appendleft((QueueCommands.WAIT, 70.0)) - command = pack_set_speed_command( - object_id=oids.RW2_ID, - speed=0, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft((QueueCommands.WAIT, 10.0)) - - # RW3 speed cmd - command = pack_set_speed_command( - object_id=oids.RW3_ID, - speed=20000, - ramp_time=10000, + new_ssc = rw_speed_cmd_single( + tc_queue, oids.RW3_ID, new_ssc, test_speed, test_ramp_time ) - tc_queue.appendleft(command.pack_command_tuple()) - tc_queue.appendleft((QueueCommands.WAIT, 70.0)) - command = pack_set_speed_command( - object_id=oids.RW3_ID, - speed=0, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft((QueueCommands.WAIT, 10.0)) - - # RW4 speed cmd - command = pack_set_speed_command( - object_id=oids.RW4_ID, - speed=20000, - ramp_time=10000, + new_ssc = rw_speed_cmd_single( + tc_queue, oids.RW4_ID, new_ssc, test_speed, test_ramp_time ) - tc_queue.appendleft(command.pack_command_tuple()) - tc_queue.appendleft((QueueCommands.WAIT, 70.0)) - command = pack_set_speed_command( - object_id=oids.RW4_ID, - speed=0, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft((QueueCommands.WAIT, 5.0)) + return new_ssc -def activate_two_rws_in_sequence(tc_queue: TcQueueT): +def activate_all_rws_two_consecutively(tc_queue: TcQueueT, init_ssc: int) -> int: + new_ssc = init_ssc # RW1+3 speed cmd tc_queue.appendleft((QueueCommands.WAIT, 5.0)) - command = pack_set_speed_command( - object_id=oids.RW1_ID, - speed=-20000, - ramp_time=10000, + new_ssc = rw_speed_up_cmd_consec( + tc_queue, [oids.RW1_ID, oids.RW3_ID], new_ssc, -20000, 10000 ) - tc_queue.appendleft(command.pack_command_tuple()) - command = pack_set_speed_command( - object_id=oids.RW3_ID, - speed=-20000, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) - tc_queue.appendleft((QueueCommands.WAIT, 70.0)) - command = pack_set_speed_command( - object_id=oids.RW1_ID, - speed=0, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) - command = pack_set_speed_command( - object_id=oids.RW3_ID, - speed=0, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft((QueueCommands.WAIT, 10.0)) # RW2+4 speed cmd - command = pack_set_speed_command( - object_id=oids.RW2_ID, - speed=-20000, - ramp_time=10000, + new_ssc = rw_speed_up_cmd_consec( + tc_queue, [oids.RW2_ID, oids.RW4_ID], new_ssc, -20000, 10000 ) - tc_queue.appendleft(command.pack_command_tuple()) - command = pack_set_speed_command( - object_id=oids.RW4_ID, - speed=-20000, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft((QueueCommands.WAIT, 70.0)) - command = pack_set_speed_command( - object_id=oids.RW2_ID, - speed=0, - ramp_time=10000, + new_ssc = rw_speed_down_cmd_consec( + tc_queue, [oids.RW1_ID, oids.RW3_ID], new_ssc, 10000 ) - tc_queue.appendleft(command.pack_command_tuple()) - command = pack_set_speed_command( - object_id=oids.RW4_ID, - speed=0, - ramp_time=10000, - ) - tc_queue.appendleft(command.pack_command_tuple()) tc_queue.appendleft((QueueCommands.WAIT, 5.0)) + new_ssc = rw_speed_down_cmd_consec( + tc_queue, [oids.RW2_ID, oids.RW4_ID], new_ssc, 10000 + ) + return new_ssc diff --git a/pus_tm/devs/imtq_mgt.py b/pus_tm/devs/imtq_mgt.py index 6277f2d..6a3aae4 100644 --- a/pus_tm/devs/imtq_mgt.py +++ b/pus_tm/devs/imtq_mgt.py @@ -44,7 +44,7 @@ def handle_eng_set(printer: FsfwTmTcPrinter, hk_data: bytes): coil_x_temperature, coil_y_temperature, coil_z_temperature, - mcu_temperature + mcu_temperature, ] num_of_vars = len(header_list) pw.dlog(str(header_list)) @@ -58,19 +58,14 @@ def handle_calibrated_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes): "Calibrated MTM X [nT]", "Calibrated MTM Y [nT]", "Calibrated MTM Z [nT]", - "Coild actuation status" + "Coild actuation status", ] mtm_x = struct.unpack("!I", hk_data[0:4])[0] mtm_y = struct.unpack("!I", hk_data[4:8])[0] mtm_z = struct.unpack("!I", hk_data[8:12])[0] coil_actuation_status = hk_data[12] validity_buffer = hk_data[12:] - content_list = [ - mtm_x, - mtm_y, - mtm_z, - coil_actuation_status - ] + content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status] num_of_vars = len(header_list) pw.dlog(str(header_list)) pw.dlog(str(content_list)) @@ -83,19 +78,14 @@ def handle_raw_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes): "Raw MTM X [nT]", "Raw MTM Y [nT]", "Raw MTM Z [nT]", - "Coild actuation status" + "Coild actuation status", ] mtm_x = struct.unpack("!f", hk_data[0:4])[0] mtm_y = struct.unpack("!f", hk_data[4:8])[0] mtm_z = struct.unpack("!f", hk_data[8:12])[0] coil_actuation_status = hk_data[12] validity_buffer = hk_data[12:] - content_list = [ - mtm_x, - mtm_y, - mtm_z, - coil_actuation_status - ] + content_list = [mtm_x, mtm_y, mtm_z, coil_actuation_status] num_of_vars = len(header_list) pw.dlog(str(header_list)) pw.dlog(str(content_list)) diff --git a/pus_tm/hk_handling.py b/pus_tm/hk_handling.py index 09ab28e..cad39ab 100644 --- a/pus_tm/hk_handling.py +++ b/pus_tm/hk_handling.py @@ -15,8 +15,12 @@ from tmtccmd.logging import get_console_logger from pus_tm.devs.bpx_bat import handle_bpx_hk_data from pus_tm.devs.gps import handle_gps_data from pus_tm.devs.gyros import handle_gyros_hk_data -from pus_tm.devs.imtq_mgt import handle_self_test_data, handle_eng_set, handle_calibrated_mtm_measurement, \ - handle_raw_mtm_measurement +from pus_tm.devs.imtq_mgt import ( + handle_self_test_data, + handle_eng_set, + handle_calibrated_mtm_measurement, + handle_raw_mtm_measurement, +) from pus_tm.devs.pcdu import handle_pdu_data, handle_p60_hk_data, handle_acu_hk_data from pus_tm.devs.syrlinks import handle_syrlinks_hk_data from pus_tc.devs.imtq import ImtqSetIds