added PL PCDU HK parsing

pus_tc/cmd_definitions.py

@@ -1,5 +1,6 @@
 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 tmtccmd.config import (
     add_op_code_entry,
@@ -380,89 +381,6 @@ def add_core_controller_definitions(cmd_dict: ServiceOpCodeDictT):
     )
 
 
-def add_pl_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
-    from pus_tc.devs.plpcdu import OpCodes, Info
-
-    op_code_dict = dict()
-    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.NORMAL_SSR,
-        info=Info.NORMAL_SSR,
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.NORMAL_DRO,
-        info=Info.NORMAL_DRO,
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.NORMAL_X8,
-        info=Info.NORMAL_X8,
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.NORMAL_TX,
-        info=Info.NORMAL_TX,
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.NORMAL_MPA,
-        info=Info.NORMAL_MPA,
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.NORMAL_HPA,
-        info=Info.NORMAL_HPA,
-    )
-    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,
-        info="Inject failure SSR to DRO transition",
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.INJECT_DRO_TO_X8_FAILURE,
-        info="Inject failure in DRO to X8 transition",
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.INJECT_X8_TO_TX_FAILURE,
-        info="Inject failure in X8 to TX transition",
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.INJECT_TX_TO_MPA_FAILURE,
-        info="Inject failure in TX to MPA transition",
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.INJECT_MPA_TO_HPA_FAILURE,
-        info="Inject failure in MPA to HPA transition",
-    )
-    add_op_code_entry(
-        op_code_dict=op_code_dict,
-        keys=OpCodes.INJECT_ALL_ON_FAILURE,
-        info="Inject failure in all on mode",
-    )
-    add_service_op_code_entry(
-        srv_op_code_dict=cmd_dict,
-        name=CustomServiceList.PL_PCDU.value,
-        info="PL PCDU",
-        op_code_entry=op_code_dict,
-    )
-
-
 def add_time_cmds(cmd_dict: ServiceOpCodeDictT):
     from pus_tc.system.time import OpCodes, Info
 

pus_tc/devs/plpcdu.py

@@ -1,8 +1,19 @@
 import enum
 from typing import Optional
 
-from tmtccmd.config import QueueCommands
+from config.definitions import CustomServiceList
+from tmtccmd.config import (
+    QueueCommands,
+    ServiceOpCodeDictT,
+    add_op_code_entry,
+    add_service_op_code_entry,
+)
 from tmtccmd.tc.definitions import TcQueueT
+from tmtccmd.tc.pus_3_fsfw_hk import (
+    generate_one_hk_command,
+    make_sid,
+    generate_one_diag_command,
+)
 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,
@@ -26,6 +37,8 @@ class OpCodes:
     NORMAL_MPA = ["6", "nml-mpa"]
     NORMAL_HPA = ["7", "nml-hpa"]
 
+    REQ_OS_HK = ["8", "hk-os"]
+
     INJECT_SSR_TO_DRO_FAILURE = ["10", "inject-ssr-dro-fault"]
     INJECT_DRO_TO_X8_FAILURE = ["11", "inject-dro-x8-fault"]
     INJECT_X8_TO_TX_FAILURE = ["12", "inject-x8-tx-fault"]
@@ -50,6 +63,11 @@ class Info:
     NORMAL_TX = f"{NORMAL}, TX on"
     NORMAL_MPA = f"{NORMAL}, MPA on"
     NORMAL_HPA = f"{NORMAL}, HPA on"
+    REQ_OS_HK = "Request One Shot HK"
+
+
+class SetIds(enum.IntEnum):
+    ADC = 0
 
 
 class NormalSubmodesMask(enum.IntEnum):
@@ -99,6 +117,90 @@ class ParamIds(enum.IntEnum):
     INJECT_ALL_ON_FAILURE = 35
 
 
+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_ON, info=Info.SWITCH_ON
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.NORMAL_SSR,
+        info=Info.NORMAL_SSR,
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.NORMAL_DRO,
+        info=Info.NORMAL_DRO,
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.NORMAL_X8,
+        info=Info.NORMAL_X8,
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.NORMAL_TX,
+        info=Info.NORMAL_TX,
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.NORMAL_MPA,
+        info=Info.NORMAL_MPA,
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.NORMAL_HPA,
+        info=Info.NORMAL_HPA,
+    )
+    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,
+        info="Inject failure SSR to DRO transition",
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.INJECT_DRO_TO_X8_FAILURE,
+        info="Inject failure in DRO to X8 transition",
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.INJECT_X8_TO_TX_FAILURE,
+        info="Inject failure in X8 to TX transition",
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.INJECT_TX_TO_MPA_FAILURE,
+        info="Inject failure in TX to MPA transition",
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.INJECT_MPA_TO_HPA_FAILURE,
+        info="Inject failure in MPA to HPA transition",
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=OpCodes.INJECT_ALL_ON_FAILURE,
+        info="Inject failure in all on mode",
+    )
+    add_service_op_code_entry(
+        srv_op_code_dict=cmd_dict,
+        name=CustomServiceList.PL_PCDU.value,
+        info="PL PCDU",
+        op_code_entry=op_code_dict,
+    )
+
+
 def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
     if op_code in OpCodes.SWITCH_ON:
         pack_pl_pcdu_mode_cmd(
@@ -175,6 +277,12 @@ def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
                 | (1 << NormalSubmodesMask.HPA_ON)
             ),
         )
+    if op_code in OpCodes.REQ_OS_HK:
+        tc_queue.appendleft((QueueCommands.PRINT, f"PL PCDU: {Info.REQ_OS_HK}"))
+        cmd = generate_one_diag_command(
+            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.INJECT_ALL_ON_FAILURE:
         pack_failure_injection_cmd(
             tc_queue=tc_queue,

pus_tm/devs/plpcdu.py

@@ -0,0 +1,52 @@
+import struct
+
+from pus_tm.defs import PrintWrapper
+from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
+from pus_tc.devs.plpcdu import SetIds
+
+
+ADC_CHANNELS_NAMED = [
+    "U BAT DIV 6",
+    "U NEG V FB",
+    "I HPA",
+    "U HPA DIV 6",
+    "I MPA",
+    "U MPA DIV 6",
+    "I TX",
+    "U TX DIV 6",
+    "I X8",
+    "U X8 DIV 6",
+    "I DRO",
+    "U DRO DIV 6",
+]
+
+
+def handle_plpcdu_hk(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
+    if set_id == SetIds.ADC:
+        pw = PrintWrapper(printer)
+        current_idx = 0
+        pw.dlog("Received PL PCDU ADC HK data")
+        channels = []
+        ch_print = "Channels Raw (hex): ["
+        for i in range(12):
+            channels.append(
+                struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
+            )
+            if i < 11:
+                ch_print += f"{channels[i]:06x},"
+            else:
+                ch_print += f"{channels[i]:06x}]"
+            current_idx += 2
+        processed_vals = []
+        for i in range(12):
+            processed_vals.append(
+                struct.unpack("!f", hk_data[current_idx : current_idx + 4])[0]
+            )
+            current_idx += 4
+        temp = struct.unpack("!f", hk_data[current_idx : current_idx + 4])[0]
+        current_idx += 4
+        pw.dlog(f"Temperature: {temp} C")
+        pw.dlog(ch_print)
+        for i in range(12):
+            pw.dlog(f"{ADC_CHANNELS_NAMED[i].ljust(24)} | {processed_vals[i]}")
+        printer.print_validity_buffer(validity_buffer=hk_data[current_idx:], num_vars=3)

pus_tm/hk_handling.py

@@ -1,6 +1,7 @@
 """HK Handling for EIVE OBSW"""
 import struct
 
+from pus_tm.devs.plpcdu import handle_plpcdu_hk
 from pus_tm.devs.rad_sensor import handle_rad_sensor_data
 from pus_tm.system.tcs import handle_thermal_controller_hk_data, TM_TCP_SERVER
 from tmtccmd.config.definitions import HkReplyUnpacked
@@ -126,9 +127,9 @@ def handle_regular_hk_print(
         handle_mgm_hk_data(
             object_id=object_id, hk_data=hk_data, printer=printer, set_id=set_id
         )
-    if objb == obj_ids.PL_PCDU_ID:
-        log_to_both(printer, "Received PL PCDU HK data")
-    if objb == obj_ids.THERMAL_CONTROLLER_ID:
+    elif objb == obj_ids.PL_PCDU_ID:
+        handle_plpcdu_hk(set_id=set_id, hk_data=hk_data, printer=printer)
+    elif objb == obj_ids.THERMAL_CONTROLLER_ID:
         handle_thermal_controller_hk_data(
             object_id=object_id, printer=printer, set_id=set_id, hk_data=hk_data
         )