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/devs/reaction_wheels.py b/pus_tc/devs/reaction_wheels.py index 970f964..f127830 100644 --- a/pus_tc/devs/reaction_wheels.py +++ b/pus_tc/devs/reaction_wheels.py @@ -196,20 +196,36 @@ def pack_rw_ass_cmds(tc_queue: TcQueueT, object_id: bytes, op_code: str): def pack_set_speed_command( - object_id: bytes, speed: int, ramp_time: int, ssc: int + object_id: bytes, speed: int, ramp_time_ms: int, ssc: int ) -> PusTelecommand: """With this function a command is packed to set the speed of a reaction wheel :param object_id: The object id of the reaction wheel handler. :param speed: Valid speeds are [-65000, -1000] and [1000, 65000]. Values are specified in 0.1 * RPM - :param ramp_time: The time after which the reaction wheel will reach the commanded speed. + :param ramp_time_ms: The time after which the reaction wheel will reach the commanded speed. Valid times are 10 - 10000 ms :param ssc: Source sequence count """ + if speed > 0: + if speed < 1000 or speed > 65000: + raise ValueError( + "Invalid RW speed specified. " + "Allowed range is [1000, 65000] 0.1 * RPM" + ) + else: + if speed < -65000 or speed > -1000: + raise ValueError( + "Invalid RW speed specified. " + "Allowed range is [-65000, -1000] 0.1 * RPM" + ) + if ramp_time_ms < 0 or ( + ramp_time_ms > 0 and (ramp_time_ms > 10000 or ramp_time_ms < 10) + ): + raise ValueError("Invalid Ramp Speed time. Allowed range is [10-10000] ms") command_id = RwCommandIds.SET_SPEED command = bytearray() command += object_id + command_id command = command + struct.pack("!i", speed) - command = command + ramp_time.to_bytes(length=2, byteorder="big") + command = command + ramp_time_ms.to_bytes(length=2, byteorder="big") command = PusTelecommand(service=8, subservice=128, ssc=ssc, app_data=command) return command diff --git a/pus_tc/system/proc.py b/pus_tc/system/proc.py index 1790416..79e1e24 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, @@ -8,16 +10,6 @@ from tmtccmd.config import ( from tmtccmd.tc.definitions import TcQueueT from tmtccmd.tc.pus_3_fsfw_hk import * -""" -from config.object_ids import ( - BPX_HANDLER_ID, - P60_DOCK_HANDLER, - PDU_1_HANDLER_ID, - PDU_2_HANDLER_ID, - ACU_HANDLER_ID, - CORE_CONTROLLER_ID, -) -""" import config.object_ids as oids from pus_tc.devs.bpx_batt import BpxSetIds from pus_tc.system.core import SetIds as CoreSetIds @@ -146,14 +138,11 @@ def pack_generic_hk_listening_cmds( ) elif one_rw is True: - activate_one_rw_in_sequence( - tc_queue=tc_queue, + activate_all_rws_in_sequence( + tc_queue=tc_queue, test_speed=20000, test_ramp_time=10000, init_ssc=0 ) - elif two_rws is True: - activate_two_rws_in_sequence( - tc_queue=tc_queue, - ) + activate_all_rws_two_consecutively(tc_queue=tc_queue, init_ssc=0) else: pass @@ -221,7 +210,7 @@ def pack_proc_commands(tc_queue: TcQueueT, op_code: str): if op_code in OpCodes.RAD_SEN_FT: key = KAI.RAD_SEN_FT[0] - sid_list.append(make_sid(oids.RAD_SENSOR_ID, RadSetIds.RAD_SEN_CORE)) + sid_list.append(make_sid(oids.RAD_SENSOR_ID, RadSetIds.HK)) pack_generic_hk_listening_cmds( tc_queue=tc_queue, proc_key=key, @@ -652,128 +641,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_ms=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_ms=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_ms=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_ms=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 2fafe4e..2fffd6e 100644 --- a/pus_tm/hk_handling.py +++ b/pus_tm/hk_handling.py @@ -16,8 +16,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