Merge pull request 'added PL PCDU HK parsing' (#87) from mueller/plpcdu-hk-parsing into develop

Reviewed-on: #87

CHANGELOG.md

@@ -12,7 +12,12 @@ list yields a list of all related PRs for each release.
 
 # [v1.12.0]
 
+- Update GPS HK Parsing
+  PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/86
+- PL PCDU HK Parsing
+  PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/87
 - Add full regular reboot command
+  PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/85
 - Add Rad Sensor HK parsing
   PR: https://egit.irs.uni-stuttgart.de/eive/eive-tmtc/pulls/81
 - Add procedures, parser functions and general application functionalities

pus_tc/devs/plpcdu.py

@@ -11,6 +11,11 @@ from tmtccmd.config import (
     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_11_tc_sched import (
     generate_enable_tc_sched_cmd,
     generate_time_tagged_cmd,
@@ -39,12 +44,14 @@ class OpCodes:
     NORMAL_MPA = ["8", "nml-mpa"]
     NORMAL_HPA = ["9", "nml-hpa"]
 
-    INJECT_SSR_TO_DRO_FAILURE = ["15", "inject-ssr-dro-fault"]
-    INJECT_DRO_TO_X8_FAILURE = ["16", "inject-dro-x8-fault"]
-    INJECT_X8_TO_TX_FAILURE = ["17", "inject-x8-tx-fault"]
-    INJECT_TX_TO_MPA_FAILURE = ["18", "inject-tx-mpa-fault"]
-    INJECT_MPA_TO_HPA_FAILURE = ["19", "inject-mpa-hpa-fault"]
-    INJECT_ALL_ON_FAILURE = ["20", "inject-all-on-fault"]
+    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"]
+    INJECT_TX_TO_MPA_FAILURE = ["13", "inject-tx-mpa-fault"]
+    INJECT_MPA_TO_HPA_FAILURE = ["14", "inject-mpa-hpa-fault"]
+    INJECT_ALL_ON_FAILURE = ["15", "inject-all-on-fault"]
 
 
 class Info:
@@ -57,6 +64,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
 
     SWITCH_HPA_ON_PROC = "Full Procedure to switch HPA on"
 
@@ -122,6 +134,7 @@ def add_pl_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
         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,
@@ -151,6 +164,9 @@ def add_pl_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
         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.INJECT_SSR_TO_DRO_FAILURE,
@@ -242,6 +258,12 @@ def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
             mode=Modes.NORMAL,
             submode=submode_mask_to_submode(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.SWITCH_HPA_ON_PROC:
         hpa_on_procedure(tc_queue)
     if op_code in OpCodes.INJECT_ALL_ON_FAILURE:

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.devs.sus import handle_sus_hk
 from pus_tm.system.tcs import handle_thermal_controller_hk_data, TM_TCP_SERVER
@@ -155,9 +156,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
         )