diff --git a/config/hook_implementations.py b/config/hook_implementations.py index 498f921..36dbe3b 100644 --- a/config/hook_implementations.py +++ b/config/hook_implementations.py @@ -50,6 +50,12 @@ class EiveHookObject(TmTcHookBase): } service_heater_tuple = ("Heater Device", op_code_dict_srv_heater) + op_code_dict_srv_imtq = { + "0": ("IMTQ Tests All", {OpCodeDictKeys.TIMEOUT: 2.0}), + "1": ("IMTQ perform pos X self test", {OpCodeDictKeys.TIMEOUT: 2.0}), + } + service_imtq_tuple = ("IMTQ Device", op_code_dict_srv_imtq) + service_op_code_dict[CustomServiceList.ACU.value] = service_acu_tuple service_op_code_dict[CustomServiceList.TMP1075_1.value] = service_tmp1075_1_tuple service_op_code_dict[CustomServiceList.TMP1075_2.value] = service_tmp1075_2_tuple @@ -58,6 +64,7 @@ class EiveHookObject(TmTcHookBase): service_op_code_dict[CustomServiceList.PDU1.value] = service_pdu2_tuple service_op_code_dict[CustomServiceList.PDU1.value] = service_pdu2_tuple service_op_code_dict[CustomServiceList.HEATER.value] = service_heater_tuple + service_op_code_dict[CustomServiceList.IMTQ.value] = service_imtq_tuple return service_op_code_dict def get_json_config_file_path(self) -> str: diff --git a/pus_tc/imtq.py b/pus_tc/imtq.py index c8ebafd..838e840 100644 --- a/pus_tc/imtq.py +++ b/pus_tc/imtq.py @@ -5,10 +5,11 @@ @author J. Meier @date 25.03.2021 """ -from tmtccmd.core.definitions import QueueCommands +from tmtccmd.config.definitions import QueueCommands from tmtccmd.pus_tc.packer import TcQueueT from tmtccmd.ecss.tc import PusTelecommand +from tmtccmd.pus_tc.service_3_housekeeping import make_sid, generate_one_hk_command class ImtqTestProcedure: @@ -18,16 +19,39 @@ class ImtqTestProcedure: Setting all to False will only run the tests set to True. """ all = False - command_dipole = True - get_commanded_dipole = True + command_dipole = False + get_commanded_dipole = False + positive_x_test = True + negative_x_test = False + positive_y_test = False + negative_y_test = False + positive_z_test = False + negative_z_test = False + +class SetIds: + POSITIVE_X_TEST = 1 + NEGATIVE_X_TEST = 2 + POSITIVE_Y_TEST = 3 + NEGATIVE_Y_TEST = 4 + POSITIVE_Z_TEST = 5 + NEGATIVE_Z_TEST = 6 class ImtqActionIds: start_actuation_dipole = bytearray([0x0, 0x0, 0x0, 0x02]) get_commanded_dipole = bytearray([0x0, 0x0, 0x0, 0x03]) + perform_positive_x_test = bytearray([0x0, 0x0, 0x0, 0x07]) + perform_negative_x_test = bytearray([0x0, 0x0, 0x0, 0x08]) + perform_positive_y_test = bytearray([0x0, 0x0, 0x0, 0x09]) + perform_negative_y_test = bytearray([0x0, 0x0, 0x0, 0x0A]) + perform_positive_z_test = bytearray([0x0, 0x0, 0x0, 0x0B]) + perform_negative_z_test = bytearray([0x0, 0x0, 0x0, 0x0C]) + # Initiates the reading of the last performed self test. After sending this command the results can be downlinked + # via the housekeeping service by using the appropriate set ids listed above. + read_self_test_results = bytearray([0x0, 0x0, 0x0, 0x0D]) -def pack_imtq_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT: +def pack_imtq_test_into(object_id: bytearray, tc_queue: TcQueueT, op_code: str) -> TcQueueT: tc_queue.appendleft( (QueueCommands.PRINT, "Testing ISIS IMTQ handler with object id: 0x" + object_id.hex()) @@ -42,9 +66,27 @@ def pack_imtq_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT: if ImtqTestProcedure.all or ImtqTestProcedure.get_commanded_dipole: tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Get commanded dipole")) command = object_id + ImtqActionIds.get_commanded_dipole - command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command) + command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command) tc_queue.appendleft(command.pack_command_tuple()) + if op_code == "0" or op_code == "1": + tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Perform positive x self test")) + command = object_id + ImtqActionIds.perform_positive_x_test + command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + + tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Initiate reading of positive x self test results")) + command = object_id + ImtqActionIds.read_self_test_results + command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command) + tc_queue.appendleft(command.pack_command_tuple()) + + tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Request dataset with positive x self test results")) + sid = make_sid(object_id, SetIds.POSITIVE_X_TEST) + command = generate_one_hk_command(sid, 24) + tc_queue.appendleft(command.pack_command_tuple()) + + return tc_queue + def pack_dipole_command(object_id: bytearray, x_dipole: int, y_dipole: int, z_dipole: int, duration: int) -> bytearray: """ This function packs the command causing the ISIS IMTQ to generate a dipole. diff --git a/pus_tc/ploc.py b/pus_tc/ploc.py index 0717f59..b8d593a 100644 --- a/pus_tc/ploc.py +++ b/pus_tc/ploc.py @@ -7,7 +7,7 @@ """ import struct -from tmtccmd.core.definitions import QueueCommands +from tmtccmd.config.definitions import QueueCommands from tmtccmd.pus_tc.packer import TcQueueT from tmtccmd.ecss.tc import PusTelecommand diff --git a/pus_tc/tc_packer_hook.py b/pus_tc/tc_packer_hook.py index 0156a5b..41b572b 100644 --- a/pus_tc/tc_packer_hook.py +++ b/pus_tc/tc_packer_hook.py @@ -64,7 +64,7 @@ def pack_service_queue_user(service: Union[str, int], op_code: str, service_queu return pack_heater_test_into(object_id=object_id, tc_queue=service_queue) if service == CustomServiceList.IMTQ.value: object_id = IMTQ_HANDLER_ID - return pack_imtq_test_into(object_id=object_id, tc_queue=service_queue) + return pack_imtq_test_into(object_id=object_id, tc_queue=service_queue, op_code=op_code) if service == CustomServiceList.PLOC.value: object_id = PLOC_ID return pack_ploc_test_into(object_id=object_id, tc_queue=service_queue) diff --git a/pus_tm/hk_handling_hook.py b/pus_tm/hk_handling_hook.py index 32e6cba..4846480 100644 --- a/pus_tm/hk_handling_hook.py +++ b/pus_tm/hk_handling_hook.py @@ -10,7 +10,7 @@ from typing import Tuple from tmtccmd.pus_tm.service_3_housekeeping import Service3Base from tmtccmd.utility.logger import get_logger from pus_tc.syrlinks_hk_handler import SetIds -from config.object_ids import SYRLINKS_HANDLER +from config.object_ids import SYRLINKS_HANDLER, IMTQ_HANDLER_ID LOGGER = get_logger() @@ -43,6 +43,12 @@ def handle_user_hk_packet(object_id: bytes, set_id: int, hk_data: bytearray, else: LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id") return [], [], bytearray(), 0 + if object_id == IMTQ_HANDLER_ID: + if set_id == SetIds.RX_REGISTERS_DATASET: + return imtq_positive_x_test(hk_data) + else: + LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id") + return [], [], bytearray(), 0 else: LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.") return [], [], bytearray(), 0 @@ -77,3 +83,30 @@ def handle_syrlinks_tx_registers_dataset(hk_data: bytearray) -> Tuple[list, list tx_agc_value = struct.unpack('!H', hk_data[2:4]) hk_content = [tx_status, tx_waveform, tx_agc_value] return hk_header, hk_content, validity_buffer, 3 + + +def imtq_positive_x_test(hk_data: bytearray) -> Tuple[list, list, bytearray, int]: + hk_header = [] + hk_content = [] + validity_buffer = bytearray() + hk_header = ["Init Err", "Init Raw Mag X [T]", "Init Raw Mag Y [T]", "Init Raw Mag Z [T]", "Init Cal Mag X [T]", + "Init Cal Mag Y [T]", "Init Cal Mag Z [T]", "Init Coil X Current [A]", "Init Coil Y Current [A]", + "Init Coil Z Current [A]", "Init Coil X Temperature [°C]", "Init Coil Y Temperature [°C]", + "Init Coil Z Temperature [°C]"] + init_err = hk_data[0] + init_raw_mag_x = struct.unpack('!f', hk_data[1:5]) + init_raw_mag_y = struct.unpack('!f', hk_data[5:9]) + init_raw_mag_z = struct.unpack('!f', hk_data[9:13]) + init_cal_mag_x = struct.unpack('!f', hk_data[13:17]) + init_cal_mag_y = struct.unpack('!f', hk_data[17:21]) + init_cal_mag_z = struct.unpack('!f', hk_data[21:25]) + init_coil_x_current = struct.unpack('!f', hk_data[25:29]) + init_coil_y_current = struct.unpack('!f', hk_data[29:33]) + init_coil_z_current = struct.unpack('!f', hk_data[33:37]) + init_coil_x_temperature = struct.unpack('!H', hk_data[37:39]) + init_coil_y_temperature = struct.unpack('!H', hk_data[39:41]) + init_coil_z_temperature = struct.unpack('!H', hk_data[41:43]) + hk_content = [init_err, init_raw_mag_x, init_raw_mag_y, init_raw_mag_z, init_cal_mag_x, init_cal_mag_y, + init_cal_mag_z, init_coil_x_current, init_coil_y_current, init_coil_z_current, + init_coil_x_temperature, init_coil_y_temperature, init_coil_z_temperature] + return hk_header, hk_content, len(hk_header)