Merge remote-tracking branch 'origin/develop' into mueller/plpcdu-hk-parsing

pus_tc/cmd_definitions.py

@@ -1,8 +1,3 @@
-from pus_tc.devs.gps import GpsOpCodes
-from pus_tc.devs.pcdu import add_pcdu_cmds
-from pus_tc.devs.plpcdu import add_pl_pcdu_cmds
-from pus_tc.devs.rad_sensor import add_rad_sens_cmds
-from pus_tc.system.core import add_core_controller_definitions
 from tmtccmd.config import (
     add_op_code_entry,
     add_service_op_code_entry,
@@ -10,12 +5,19 @@ from tmtccmd.config import (
     ServiceOpCodeDictT,
     OpCodeDictKeys,
 )
-from config.definitions import CustomServiceList
+from tmtccmd.config.globals import get_default_service_op_code_dict
+
+from pus_tc.devs.gps import GpsOpCodes
+from pus_tc.devs.pcdu import add_pcdu_cmds
+from pus_tc.devs.plpcdu import add_pl_pcdu_cmds
+from pus_tc.devs.rad_sensor import add_rad_sens_cmds
+from pus_tc.system.core import add_core_controller_definitions
 from pus_tc.devs.heater import add_heater_cmds
 from pus_tc.devs.rtd import specify_rtd_cmds
 from pus_tc.devs.reaction_wheels import add_rw_cmds
 from pus_tc.devs.bpx_batt import BpxOpCodes
-from tmtccmd.config.globals import get_default_service_op_code_dict
+
+from config.definitions import CustomServiceList
 
 
 def get_eive_service_op_code_dict() -> ServiceOpCodeDictT:
@@ -342,15 +344,20 @@ def add_time_cmds(cmd_dict: ServiceOpCodeDictT):
 
 def add_imtq_cmds(cmd_dict: ServiceOpCodeDictT):
     op_code_dict_srv_imtq = {
-        "0": ("IMTQ Tests All", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "0": ("Mode off", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "1": ("IMTQ perform pos X self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "1": ("Mode on", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "2": ("IMTQ perform neg X self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "2": ("Mode normal", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "3": ("IMTQ perform pos Y self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "3": ("IMTQ perform pos X self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "4": ("IMTQ perform neg Y self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "4": ("IMTQ perform neg X self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "5": ("IMTQ perform pos Z self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "5": ("IMTQ perform pos Y self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "6": ("IMTQ perform neg Z self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "6": ("IMTQ perform neg Y self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "7": ("IMTQ command dipole", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "7": ("IMTQ perform pos Z self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "8": ("IMTQ get commanded dipole", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "8": ("IMTQ perform neg Z self test", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "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}),
+        "13": ("IMTQ get raw MTM measurement one shot", {OpCodeDictKeys.TIMEOUT: 2.0}),
     }
     service_imtq_tuple = ("IMTQ Device", op_code_dict_srv_imtq)
     cmd_dict[CustomServiceList.IMTQ.value] = service_imtq_tuple

pus_tc/devs/imtq.py

@@ -9,7 +9,8 @@ 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_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
 
 
 class ImtqSetIds:
@@ -48,7 +49,22 @@ def pack_imtq_test_into(
         )
     )
 
-    if op_code == "0" or op_code == "1":
+    if op_code == "0":
+        tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Set mode off"))
+        command = pack_mode_data(object_id, Modes.OFF, 0)
+        command = PusTelecommand(service=200, subservice=1, ssc=9, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "1":
+        tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Set mode on"))
+        command = pack_mode_data(object_id, Modes.ON, 0)
+        command = PusTelecommand(service=200, subservice=1, ssc=10, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "2":
+        tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Mode Normal"))
+        command = pack_mode_data(object_id, Modes.NORMAL, 0)
+        command = PusTelecommand(service=200, subservice=1, ssc=11, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "3":
         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)
@@ -74,7 +90,7 @@ def pack_imtq_test_into(
         command = generate_one_hk_command(sid, 24)
         tc_queue.appendleft(command.pack_command_tuple())
 
-    if op_code == "0" or op_code == "2":
+    if op_code == "4":
         tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Perform negative x self test"))
         command = object_id + ImtqActionIds.perform_negative_x_test
         command = PusTelecommand(service=8, subservice=128, ssc=25, app_data=command)
@@ -100,7 +116,7 @@ def pack_imtq_test_into(
         command = generate_one_hk_command(sid, 27)
         tc_queue.appendleft(command.pack_command_tuple())
 
-    if op_code == "0" or op_code == "3":
+    if op_code == "5":
         tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Perform positive y self test"))
         command = object_id + ImtqActionIds.perform_positive_y_test
         command = PusTelecommand(service=8, subservice=128, ssc=28, app_data=command)
@@ -126,7 +142,7 @@ def pack_imtq_test_into(
         command = generate_one_hk_command(sid, 30)
         tc_queue.appendleft(command.pack_command_tuple())
 
-    if op_code == "0" or op_code == "4":
+    if op_code == "6":
         tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Perform negative y self test"))
         command = object_id + ImtqActionIds.perform_negative_y_test
         command = PusTelecommand(service=8, subservice=128, ssc=31, app_data=command)
@@ -152,7 +168,7 @@ def pack_imtq_test_into(
         command = generate_one_hk_command(sid, 33)
         tc_queue.appendleft(command.pack_command_tuple())
 
-    if op_code == "0" or op_code == "5":
+    if op_code == "7":
         tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Perform positive z self test"))
         command = object_id + ImtqActionIds.perform_positive_z_test
         command = PusTelecommand(service=8, subservice=128, ssc=34, app_data=command)
@@ -178,7 +194,7 @@ def pack_imtq_test_into(
         command = generate_one_hk_command(sid, 36)
         tc_queue.appendleft(command.pack_command_tuple())
 
-    if op_code == "0" or op_code == "6":
+    if op_code == "8":
         tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Perform negative z self test"))
         command = object_id + ImtqActionIds.perform_negative_z_test
         command = PusTelecommand(service=8, subservice=128, ssc=35, app_data=command)
@@ -204,7 +220,7 @@ def pack_imtq_test_into(
         command = generate_one_hk_command(sid, 37)
         tc_queue.appendleft(command.pack_command_tuple())
 
-    if op_code == "0" or op_code == "7":
+    if op_code == "9":
         tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Commanding dipole"))
         x_dipole = 0
         y_dipole = 0
@@ -213,12 +229,33 @@ def pack_imtq_test_into(
         command = pack_dipole_command(object_id, x_dipole, y_dipole, z_dipole, duration)
         tc_queue.appendleft(command.pack_command_tuple())
 
-    if op_code == "0" or op_code == "8":
+    if op_code == "10":
         tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Get commanded dipole"))
         command = object_id + ImtqActionIds.get_commanded_dipole
         command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
         tc_queue.appendleft(command.pack_command_tuple())
 
+    if op_code == "11":
+        tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Get engineering hk set"))
+        command = generate_one_diag_command(
+            sid=make_sid(object_id=object_id, set_id=ImtqSetIds.ENG_HK_SET), ssc=0
+        )
+        tc_queue.appendleft(command.pack_command_tuple())
+
+    if op_code == "12":
+        tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Get calibrated MTM hk set"))
+        command = generate_one_diag_command(
+            sid=make_sid(object_id=object_id, set_id=ImtqSetIds.CAL_MTM_SET), ssc=0
+        )
+        tc_queue.appendleft(command.pack_command_tuple())
+
+    if op_code == "13":
+        tc_queue.appendleft((QueueCommands.PRINT, "IMTQ: Get raw MTM hk set"))
+        command = generate_one_diag_command(
+            sid=make_sid(object_id=object_id, set_id=ImtqSetIds.RAW_MTM_SET), ssc=0
+        )
+        tc_queue.appendleft(command.pack_command_tuple())
+
     return tc_queue
 
 

pus_tc/devs/ploc_supervisor.py

@@ -126,7 +126,7 @@ def pack_ploc_supv_commands(
         command = PusTelecommand(service=200, subservice=1, ssc=10, app_data=command)
         tc_queue.appendleft(command.pack_command_tuple())
     if op_code == "2":
-        tc_queue.appendleft((QueueCommands.PRINT, "Star tracker: Mode Normal"))
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC Supervisor: Mode Normal"))
         command = pack_mode_data(object_id, Modes.NORMAL, 0)
         command = PusTelecommand(service=200, subservice=1, ssc=11, app_data=command)
         tc_queue.appendleft(command.pack_command_tuple())

pus_tc/devs/plpcdu.py

@@ -1,4 +1,6 @@
 import enum
+import struct
+import time
 from typing import Optional
 
 from config.definitions import CustomServiceList
@@ -14,6 +16,10 @@ from tmtccmd.tc.pus_3_fsfw_hk import (
     make_sid,
     generate_one_diag_command,
 )
+from tmtccmd.tc.pus_11_tc_sched import (
+    generate_enable_tc_sched_cmd,
+    generate_time_tagged_cmd,
+)
 from tmtccmd.tc.pus_200_fsfw_modes import pack_mode_data, Modes, Subservices
 from tmtccmd.tc.pus_20_params import (
     pack_scalar_double_param_app_data,
@@ -28,14 +34,15 @@ LOGGER = get_console_logger()
 
 
 class OpCodes:
-    SWITCH_ON = ["0", "on"]
+    SWITCH_HPA_ON_PROC = ["0", "proc-hpa"]
-    SWITCH_OFF = ["1", "off"]
+    SWITCH_ON = ["2", "on"]
-    NORMAL_SSR = ["2", "nml-ssr"]
+    SWITCH_OFF = ["3", "off"]
-    NORMAL_DRO = ["3", "nml-dro"]
+    NORMAL_SSR = ["4", "nml-ssr"]
-    NORMAL_X8 = ["4", "nml-x8"]
+    NORMAL_DRO = ["5", "nml-dro"]
-    NORMAL_TX = ["5", "nml-tx"]
+    NORMAL_X8 = ["6", "nml-x8"]
-    NORMAL_MPA = ["6", "nml-mpa"]
+    NORMAL_TX = ["7", "nml-tx"]
-    NORMAL_HPA = ["7", "nml-hpa"]
+    NORMAL_MPA = ["8", "nml-mpa"]
+    NORMAL_HPA = ["9", "nml-hpa"]
 
     REQ_OS_HK = ["8", "hk-os"]
 
@@ -46,12 +53,6 @@ class OpCodes:
     INJECT_MPA_TO_HPA_FAILURE = ["14", "inject-mpa-hpa-fault"]
     INJECT_ALL_ON_FAILURE = ["15", "inject-all-on-fault"]
 
-    # The following commands might become deprecated in the future
-    UPDATE_DRO_TO_X8_WAIT = ["128", "dro-to-x8-wait"]
-    UPDATE_X8_TO_TX_WAIT_TIME = ["129", "x8-to-tx-wait"]
-    UPDATE_TX_TO_MPA_WAIT_TIME = ["130", "tx-to-mpa-wait"]
-    UPDATE_MPA_TO_HPA_WAIT_TIME = ["131", "mpa-to-hpa-wait"]
-
 
 class Info:
     NORMAL = "PL PCDU ADC modules normal"
@@ -69,6 +70,8 @@ class Info:
 class SetIds(enum.IntEnum):
     ADC = 0
 
+    SWITCH_HPA_ON_PROC = "Full Procedure to switch HPA on"
+
 
 class NormalSubmodesMask(enum.IntEnum):
     SOLID_STATE_RELAYS_ADC_ON = 0
@@ -119,10 +122,19 @@ class ParamIds(enum.IntEnum):
 
 def add_pl_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
     op_code_dict = dict()
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.SWITCH_HPA_ON_PROC,
+        info=Info.SWITCH_HPA_ON_PROC,
+    )
     add_op_code_entry(
         op_code_dict=op_code_dict, keys=OpCodes.SWITCH_ON, info=Info.SWITCH_ON
     )
     add_op_code_entry(
+        op_code_dict=op_code_dict, keys=OpCodes.SWITCH_OFF, info=Info.SWITCH_OFF
+    )
+    add_op_code_entry(
+>>>>>>> origin/develop
         op_code_dict=op_code_dict,
         keys=OpCodes.NORMAL_SSR,
         info=Info.NORMAL_SSR,
@@ -155,14 +167,6 @@ def add_pl_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
     add_op_code_entry(
         op_code_dict=op_code_dict, keys=OpCodes.REQ_OS_HK, info=Info.REQ_OS_HK
     )
-    add_op_code_entry(
-        op_code_dict=op_code_dict, keys=OpCodes.SWITCH_OFF, info=Info.SWITCH_OFF
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.UPDATE_DRO_TO_X8_WAIT,
-        info="Update DRO to X8 wait time",
-    )
     add_op_code_entry(
         op_code_dict=op_code_dict,
         keys=OpCodes.INJECT_SSR_TO_DRO_FAILURE,
@@ -215,67 +219,44 @@ def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
             tc_queue=tc_queue,
             info=Info.NORMAL_SSR,
             mode=Modes.NORMAL,
-            submode=(1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON),
+            submode=submode_mask_to_submode(
+                NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON
+            ),
         )
     if op_code in OpCodes.NORMAL_DRO:
         pack_pl_pcdu_mode_cmd(
             tc_queue=tc_queue,
             info=Info.NORMAL_DRO,
             mode=Modes.NORMAL,
-            submode=(
+            submode=submode_mask_to_submode(NormalSubmodesMask.DRO_ON),
-                1 << NormalSubmodesMask.DRO_ON
-                | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
-            ),
         )
     if op_code in OpCodes.NORMAL_X8:
         pack_pl_pcdu_mode_cmd(
             tc_queue=tc_queue,
             info=Info.NORMAL_X8,
             mode=Modes.NORMAL,
-            submode=(
+            submode=submode_mask_to_submode(NormalSubmodesMask.X8_ON),
-                1 << NormalSubmodesMask.DRO_ON
-                | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
-                | (1 << NormalSubmodesMask.X8_ON)
-            ),
         )
     if op_code in OpCodes.NORMAL_TX:
         pack_pl_pcdu_mode_cmd(
             tc_queue=tc_queue,
             info=Info.NORMAL_TX,
             mode=Modes.NORMAL,
-            submode=(
+            submode=submode_mask_to_submode(NormalSubmodesMask.TX_ON),
-                1 << NormalSubmodesMask.DRO_ON
-                | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
-                | (1 << NormalSubmodesMask.X8_ON)
-                | (1 << NormalSubmodesMask.TX_ON)
-            ),
         )
     if op_code in OpCodes.NORMAL_MPA:
         pack_pl_pcdu_mode_cmd(
             tc_queue=tc_queue,
             info=Info.NORMAL_MPA,
             mode=Modes.NORMAL,
-            submode=(
+            submode=submode_mask_to_submode(NormalSubmodesMask.MPA_ON),
-                1 << NormalSubmodesMask.DRO_ON
-                | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
-                | (1 << NormalSubmodesMask.X8_ON)
-                | (1 << NormalSubmodesMask.TX_ON)
-                | (1 << NormalSubmodesMask.MPA_ON)
-            ),
         )
     if op_code in OpCodes.NORMAL_HPA:
         pack_pl_pcdu_mode_cmd(
             tc_queue=tc_queue,
             info=Info.NORMAL_HPA,
             mode=Modes.NORMAL,
-            submode=(
+            submode=submode_mask_to_submode(NormalSubmodesMask.HPA_ON),
-                1 << NormalSubmodesMask.DRO_ON
-                | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
-                | (1 << NormalSubmodesMask.X8_ON)
-                | (1 << NormalSubmodesMask.TX_ON)
-                | (1 << NormalSubmodesMask.MPA_ON)
-                | (1 << NormalSubmodesMask.HPA_ON)
-            ),
         )
     if op_code in OpCodes.REQ_OS_HK:
         tc_queue.appendleft((QueueCommands.PRINT, f"PL PCDU: {Info.REQ_OS_HK}"))
@@ -283,6 +264,8 @@ def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
             sid=make_sid(object_id=PL_PCDU_ID, set_id=SetIds.ADC), ssc=0
         )
         tc_queue.appendleft(cmd.pack_command_tuple())
+    if op_code in OpCodes.SWITCH_HPA_ON_PROC:
+        hpa_on_procedure(tc_queue)
     if op_code in OpCodes.INJECT_ALL_ON_FAILURE:
         pack_failure_injection_cmd(
             tc_queue=tc_queue,
@@ -291,6 +274,129 @@ def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
         )
 
 
+def hpa_on_procedure(tc_queue: TcQueueT):
+    delay_dro_to_x8 = request_wait_time()
+    if delay_dro_to_x8 is None:
+        delay_dro_to_x8 = 900
+    tc_queue.appendleft(
+        (
+            QueueCommands.PRINT,
+            f"Starting procedure to switch on PL PCDU HPA with DRO to X8 delay of {delay_dro_to_x8} seconds",
+        )
+    )
+    pl_pcdu_on = PusTelecommand(
+        service=200,
+        subservice=Subservices.TC_MODE_COMMAND,
+        app_data=pack_mode_data(object_id=PL_PCDU_ID, mode=Modes.ON, submode=0),
+    )
+    ssr_on = PusTelecommand(
+        service=200,
+        subservice=Subservices.TC_MODE_COMMAND,
+        app_data=pack_mode_data(
+            object_id=PL_PCDU_ID,
+            mode=Modes.NORMAL,
+            submode=submode_mask_to_submode(
+                NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON
+            ),
+        ),
+    )
+    dro_on = PusTelecommand(
+        service=200,
+        subservice=Subservices.TC_MODE_COMMAND,
+        app_data=pack_mode_data(
+            object_id=PL_PCDU_ID,
+            mode=Modes.NORMAL,
+            submode=submode_mask_to_submode(NormalSubmodesMask.DRO_ON),
+        ),
+    )
+    x8_on = PusTelecommand(
+        service=200,
+        subservice=Subservices.TC_MODE_COMMAND,
+        app_data=pack_mode_data(
+            object_id=PL_PCDU_ID,
+            mode=Modes.NORMAL,
+            submode=submode_mask_to_submode(NormalSubmodesMask.X8_ON),
+        ),
+    )
+    tx_on = PusTelecommand(
+        service=200,
+        subservice=Subservices.TC_MODE_COMMAND,
+        app_data=pack_mode_data(
+            object_id=PL_PCDU_ID,
+            mode=Modes.NORMAL,
+            submode=submode_mask_to_submode(NormalSubmodesMask.TX_ON),
+        ),
+    )
+    mpa_on = PusTelecommand(
+        service=200,
+        subservice=Subservices.TC_MODE_COMMAND,
+        app_data=pack_mode_data(
+            object_id=PL_PCDU_ID,
+            mode=Modes.NORMAL,
+            submode=submode_mask_to_submode(NormalSubmodesMask.MPA_ON),
+        ),
+    )
+    hpa_on = PusTelecommand(
+        service=200,
+        subservice=Subservices.TC_MODE_COMMAND,
+        app_data=pack_mode_data(
+            object_id=PL_PCDU_ID,
+            mode=Modes.NORMAL,
+            submode=submode_mask_to_submode(NormalSubmodesMask.HPA_ON),
+        ),
+    )
+    current_time = time.time()
+
+    enb_sched = generate_enable_tc_sched_cmd(ssc=0)
+
+    sched_time = current_time + 10
+    tc_queue.appendleft(enb_sched.pack_command_tuple())
+    tagged_on_cmd = generate_time_tagged_cmd(
+        release_time=struct.pack("!I", sched_time),
+        tc_to_insert=pl_pcdu_on,
+        ssc=1,
+    )
+    tc_queue.appendleft(tagged_on_cmd.pack_command_tuple())
+
+    sched_time += 5
+    tagged_ssr_cmd = generate_time_tagged_cmd(
+        release_time=struct.pack("!I", sched_time),
+        tc_to_insert=ssr_on,
+        ssc=2,
+    )
+    tc_queue.appendleft(tagged_ssr_cmd.pack_command_tuple())
+
+    sched_time += 5
+    tagged_dro_cmd = generate_time_tagged_cmd(
+        release_time=struct.pack("!I", sched_time), tc_to_insert=dro_on, ssc=3
+    )
+    tc_queue.appendleft(tagged_dro_cmd.pack_command_tuple())
+
+    sched_time += delay_dro_to_x8
+    tagged_x8_cmd = generate_time_tagged_cmd(
+        release_time=struct.pack("!I", sched_time), tc_to_insert=x8_on, ssc=4
+    )
+    tc_queue.appendleft(tagged_x8_cmd.pack_command_tuple())
+
+    sched_time += 5
+    tagged_tx_cmd = generate_time_tagged_cmd(
+        release_time=struct.pack("!I", sched_time), tc_to_insert=tx_on, ssc=5
+    )
+    tc_queue.appendleft(tagged_tx_cmd.pack_command_tuple())
+
+    sched_time += 5
+    tagged_mpa_cmd = generate_time_tagged_cmd(
+        release_time=struct.pack("!I", sched_time), tc_to_insert=mpa_on, ssc=6
+    )
+    tc_queue.appendleft(tagged_mpa_cmd.pack_command_tuple())
+
+    sched_time += 5
+    tagged_hpa_cmd = generate_time_tagged_cmd(
+        release_time=struct.pack("!I", sched_time), tc_to_insert=hpa_on, ssc=7
+    )
+    tc_queue.appendleft(tagged_hpa_cmd.pack_command_tuple())
+
+
 def request_wait_time() -> Optional[float]:
     while True:
         wait_time = input("Please enter DRO to X8 wait time in seconds, x to cancel: ")
@@ -307,6 +413,45 @@ def request_wait_time() -> Optional[float]:
             return wait_time
 
 
+def submode_mask_to_submode(on_tgt: NormalSubmodesMask) -> int:
+    if on_tgt == NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON:
+        return 1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON
+    if on_tgt == NormalSubmodesMask.DRO_ON:
+        return 1 << NormalSubmodesMask.DRO_ON | (
+            1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON
+        )
+    if on_tgt == NormalSubmodesMask.X8_ON:
+        return (
+            1 << NormalSubmodesMask.DRO_ON
+            | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
+            | (1 << NormalSubmodesMask.X8_ON)
+        )
+    if on_tgt == NormalSubmodesMask.TX_ON:
+        return (
+            1 << NormalSubmodesMask.DRO_ON
+            | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
+            | (1 << NormalSubmodesMask.X8_ON)
+            | (1 << NormalSubmodesMask.TX_ON)
+        )
+    if on_tgt == NormalSubmodesMask.MPA_ON:
+        return (
+            1 << NormalSubmodesMask.DRO_ON
+            | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
+            | (1 << NormalSubmodesMask.X8_ON)
+            | (1 << NormalSubmodesMask.TX_ON)
+            | (1 << NormalSubmodesMask.MPA_ON)
+        )
+    if on_tgt == NormalSubmodesMask.HPA_ON:
+        return (
+            1 << NormalSubmodesMask.DRO_ON
+            | (1 << NormalSubmodesMask.SOLID_STATE_RELAYS_ADC_ON)
+            | (1 << NormalSubmodesMask.X8_ON)
+            | (1 << NormalSubmodesMask.TX_ON)
+            | (1 << NormalSubmodesMask.MPA_ON)
+            | (1 << NormalSubmodesMask.HPA_ON)
+        )
+
+
 def pack_wait_time_cmd(tc_queue: TcQueueT, param_id: int, print_str: str):
     wait_time = request_wait_time()
     tc_queue.appendleft(

pus_tm/devs/imtq_mgt.py

@@ -4,6 +4,104 @@ from pus_tm.defs import PrintWrapper
 from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
 
 
+def handle_eng_set(printer: FsfwTmTcPrinter, hk_data: bytes):
+    pw = PrintWrapper(printer)
+    header_list = [
+        "Digital Voltage [mV]",
+        "Analog Voltage [mV]",
+        "Digital Current [mA]",
+        "Analog Current [mA]",
+        "Coil Current X [mA]",
+        "Coil Current Y [mA]",
+        "Coil Current Z [mA]",
+        "Coil X Temperature [°C]",
+        "Coil Y Temperature [°C]",
+        "Coil Z Temperature [°C]",
+        "Coil Z Temperature [°C]",
+        "MCU Temperature [°C]",
+    ]
+    digital_voltage = struct.unpack("!H", hk_data[0:2])[0]
+    analog_voltage = struct.unpack("!H", hk_data[2:4])[0]
+    digital_current = struct.unpack("!f", hk_data[4:8])[0]
+    analog_current = struct.unpack("!f", hk_data[8:12])[0]
+    coil_x_current = struct.unpack("!f", hk_data[12:16])[0]
+    coil_y_current = struct.unpack("!f", hk_data[16:20])[0]
+    coil_z_current = struct.unpack("!f", hk_data[20:24])[0]
+    coil_x_temperature = struct.unpack("!H", hk_data[24:26])[0]
+    coil_y_temperature = struct.unpack("!H", hk_data[26:28])[0]
+    coil_z_temperature = struct.unpack("!H", hk_data[30:32])[0]
+    mcu_temperature = struct.unpack("!H", hk_data[32:34])[0]
+
+    validity_buffer = hk_data[42:]
+    content_list = [
+        digital_voltage,
+        analog_voltage,
+        digital_current,
+        analog_current,
+        coil_x_current,
+        coil_y_current,
+        coil_z_current,
+        coil_x_temperature,
+        coil_y_temperature,
+        coil_z_temperature,
+        mcu_temperature
+    ]
+    num_of_vars = len(header_list)
+    pw.dlog(str(header_list))
+    pw.dlog(str(content_list))
+    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
+
+
+def handle_calibrated_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes):
+    pw = PrintWrapper(printer)
+    header_list = [
+        "Calibrated MTM X [nT]",
+        "Calibrated MTM Y [nT]",
+        "Calibrated MTM Z [nT]",
+        "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
+    ]
+    num_of_vars = len(header_list)
+    pw.dlog(str(header_list))
+    pw.dlog(str(content_list))
+    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
+
+
+def handle_raw_mtm_measurement(printer: FsfwTmTcPrinter, hk_data: bytes):
+    pw = PrintWrapper(printer)
+    header_list = [
+        "Raw MTM X [nT]",
+        "Raw MTM Y [nT]",
+        "Raw MTM Z [nT]",
+        "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
+    ]
+    num_of_vars = len(header_list)
+    pw.dlog(str(header_list))
+    pw.dlog(str(content_list))
+    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
+
+
 def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes):
     pw = PrintWrapper(printer)
     header_list = [

pus_tm/hk_handling.py

@@ -16,7 +16,8 @@ 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
+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
@@ -83,8 +84,14 @@ def handle_regular_hk_print(
             set_id <= ImtqSetIds.NEGATIVE_Z_TEST
         ):
             return handle_self_test_data(printer, hk_data)
+        elif set_id == ImtqSetIds.ENG_HK_SET:
+            return handle_eng_set(printer, hk_data)
+        elif set_id == ImtqSetIds.CAL_MTM_SET:
+            return handle_calibrated_mtm_measurement(printer, hk_data)
+        elif set_id == ImtqSetIds.RAW_MTM_SET:
+            return handle_raw_mtm_measurement(printer, hk_data)
         else:
-            LOGGER.info("Service 3 TM: Syrlinks handler reply with unknown set id")
+            LOGGER.info("Service 3 TM: IMTQ handler reply with unknown set id")
     if objb == obj_ids.GPS_HANDLER_0_ID or object_id == obj_ids.GPS_HANDLER_1_ID:
         handle_gps_data(printer=printer, hk_data=hk_data)
     if objb == obj_ids.BPX_HANDLER_ID:

spacepackets

@@ -1 +1 @@
-Subproject commit d0c3f4a802c3cddc5be2919af763b08fe6a6b05c
+Subproject commit 9ee7922bf7b7a678f8e5ebd5926001defac9a3d4