Merge pull request 'PDU1 and PDU2 HK Handling' (#57) from mueller/pdu1-pdu2-hk-handling into develop

Reviewed-on: #57

config/hook_implementations.py

@@ -122,7 +122,7 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
         "2": ("Star Tracker: Mode Normal", {OpCodeDictKeys.TIMEOUT: 2.0}),
         "3": ("Star Tracker: Mode Off", {OpCodeDictKeys.TIMEOUT: 2.0}),
         "4": ("Star Tracker: Mode Raw", {OpCodeDictKeys.TIMEOUT: 2.0}),
-        "5": ("Star Tracker: Ping", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "5": ("Star Tracker: Ping", {OpCodeDictKeys.TIMEOUT: 5.0}),
         "6": (
             "Star Tracker: Switch to bootloader program",
             {OpCodeDictKeys.TIMEOUT: 2.0},

gomspace/gomspace_common.py

@@ -37,11 +37,13 @@ class Info:
 
 
 class SetIds:
-    PDU_1 = 1
+    PDU_1_CORE = 1
-    PDU_2 = 2
+    PDU_1_AUX = 2
-    P60_CORE = 3
+    PDU_2_CORE = 3
-    P60_AUX = 4
+    PDU_2_AUX = 4
-    ACU = 5
+    P60_CORE = 5
+    P60_AUX = 6
+    ACU = 7
 
 
 class TableIds:

pus_tc/cmd_definitions.py

@@ -335,7 +335,12 @@ def add_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
     add_op_code_entry(
         op_code_dict=op_code_dict,
         keys=GomspaceOpCodes.REQUEST_CORE_HK_ONCE,
-        info="PDU1: Request HK once",
+        info=GsInfo.REQUEST_CORE_HK_ONCE,
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=GomspaceOpCodes.REQUEST_AUX_HK_ONCE,
+        info=GsInfo.REQUEST_AUX_HK_ONCE,
     )
     add_op_code_entry(
         op_code_dict=op_code_dict,
@@ -442,7 +447,12 @@ def add_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
     add_op_code_entry(
         op_code_dict=op_code_dict,
         keys=GomspaceOpCodes.REQUEST_CORE_HK_ONCE,
-        info="PDU2: Request HK once",
+        info=GsInfo.REQUEST_CORE_HK_ONCE,
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=GomspaceOpCodes.REQUEST_AUX_HK_ONCE,
+        info=GsInfo.REQUEST_AUX_HK_ONCE,
     )
     add_op_code_entry(
         op_code_dict=op_code_dict,

pus_tc/devs/pdu1.py

@@ -5,7 +5,11 @@
 """
 from tmtccmd.config.definitions import QueueCommands
 from tmtccmd.tc.packer import TcQueueT
-from tmtccmd.tc.service_3_housekeeping import generate_one_hk_command, make_sid
+from tmtccmd.tc.service_3_housekeeping import (
+    generate_one_hk_command,
+    make_sid,
+    generate_one_diag_command,
+)
 from gomspace.gomspace_common import *
 from gomspace.gomspace_pdu_definitions import *
 from config.object_ids import PDU_1_HANDLER_ID
@@ -211,8 +215,13 @@ def pack_pdu1_commands(object_id: bytearray, tc_queue: TcQueueT, op_code: str):
         )
         tc_queue.appendleft(command.pack_command_tuple())
     if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
-        tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Requesting HK Table Once"))
+        tc_queue.appendleft((QueueCommands.PRINT, f"PDU1: {Info.REQUEST_CORE_HK_ONCE}"))
-        hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1)
+        hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1_CORE)
+        command = generate_one_diag_command(sid=hk_sid, ssc=0)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code in GomspaceOpCodes.REQUEST_AUX_HK_ONCE:
+        tc_queue.appendleft((QueueCommands.PRINT, f"PDU1: {Info.REQUEST_AUX_HK_ONCE}"))
+        hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1_AUX)
         command = generate_one_hk_command(sid=hk_sid, ssc=0)
         tc_queue.appendleft(command.pack_command_tuple())
     if op_code in GomspaceOpCodes.PRINT_SWITCH_V_I:

pus_tc/devs/pdu2.py

@@ -8,7 +8,11 @@
 """
 from tmtccmd.config.definitions import QueueCommands
 from tmtccmd.tc.packer import TcQueueT
-from tmtccmd.tc.service_3_housekeeping import generate_one_hk_command, make_sid
+from tmtccmd.tc.service_3_housekeeping import (
+    generate_one_hk_command,
+    generate_one_diag_command,
+    make_sid,
+)
 from gomspace.gomspace_common import *
 from gomspace.gomspace_pdu_definitions import *
 from config.object_ids import PDU_2_HANDLER_ID
@@ -233,8 +237,13 @@ def pack_pdu2_commands(object_id: bytearray, tc_queue: TcQueueT, op_code: str):
         )
         tc_queue.appendleft(command.pack_command_tuple())
     if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
-        tc_queue.appendleft((QueueCommands.PRINT, "PDU2: Requesting HK Table Once"))
+        tc_queue.appendleft((QueueCommands.PRINT, f"PDU2: {Info.REQUEST_CORE_HK_ONCE}"))
-        hk_sid = make_sid(object_id=PDU_2_HANDLER_ID, set_id=SetIds.PDU_2)
+        hk_sid = make_sid(object_id=PDU_2_HANDLER_ID, set_id=SetIds.PDU_2_CORE)
+        command = generate_one_diag_command(sid=hk_sid, ssc=0)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code in GomspaceOpCodes.REQUEST_AUX_HK_ONCE:
+        tc_queue.appendleft((QueueCommands.PRINT, f"PDU2: {Info.REQUEST_AUX_HK_ONCE}"))
+        hk_sid = make_sid(object_id=PDU_2_HANDLER_ID, set_id=SetIds.PDU_2_AUX)
         command = generate_one_hk_command(sid=hk_sid, ssc=0)
         tc_queue.appendleft(command.pack_command_tuple())
     if op_code in GomspaceOpCodes.PRINT_SWITCH_V_I:

pus_tm/hk_handling.py

@@ -16,16 +16,8 @@ from pus_tc.devs.syrlinks_hk_handler import SetIds
 from pus_tc.devs.p60dock import SetIds
 from pus_tc.devs.imtq import ImtqSetIds
 from tmtccmd.pus.obj_id import ObjectId, ObjectIdDictT
-from config.object_ids import (
+import config.object_ids as obj_ids
-    SYRLINKS_HANDLER_ID,
+
-    IMTQ_HANDLER_ID,
-    GPS_HANDLER_0_ID,
-    GPS_HANDLER_1_ID,
-    BPX_HANDLER_ID,
-    CORE_CONTROLLER_ID,
-    P60_DOCK_HANDLER,
-    PL_PCDU_ID,
-)
 
 LOGGER = get_console_logger()
 
@@ -66,29 +58,37 @@ def handle_regular_hk_print(
     object_id = object_id.as_bytes
     set_id = hk_packet.set_id
     """This function is called when a Service 3 Housekeeping packet is received."""
-    if object_id == SYRLINKS_HANDLER_ID:
+    if object_id == obj_ids.SYRLINKS_HANDLER_ID:
         if set_id == SetIds.RX_REGISTERS_DATASET:
             return handle_syrlinks_rx_registers_dataset(printer, hk_data)
         elif set_id == SetIds.TX_REGISTERS_DATASET:
             return handle_syrlinks_tx_registers_dataset(printer, hk_data)
         else:
             LOGGER.info("Service 3 TM: Syrlinks handler reply with unknown set id")
-    elif object_id == IMTQ_HANDLER_ID:
+    elif object_id == obj_ids.IMTQ_HANDLER_ID:
         if (set_id >= ImtqSetIds.POSITIVE_X_TEST) and (
             set_id <= ImtqSetIds.NEGATIVE_Z_TEST
         ):
             return handle_self_test_data(printer, hk_data)
         else:
             LOGGER.info("Service 3 TM: Syrlinks handler reply with unknown set id")
-    elif object_id == GPS_HANDLER_0_ID or object_id == GPS_HANDLER_1_ID:
+    elif object_id == obj_ids.GPS_HANDLER_0_ID or object_id == obj_ids.GPS_HANDLER_1_ID:
         handle_gps_data(printer=printer, hk_data=hk_data)
-    elif object_id == BPX_HANDLER_ID:
+    elif object_id == obj_ids.BPX_HANDLER_ID:
         handle_bpx_hk_data(hk_data=hk_data, set_id=set_id, printer=printer)
-    elif object_id == CORE_CONTROLLER_ID:
+    elif object_id == obj_ids.CORE_CONTROLLER_ID:
         return handle_core_hk_data(printer=printer, hk_data=hk_data)
-    elif object_id == P60_DOCK_HANDLER:
+    elif object_id == obj_ids.PDU_1_HANDLER_ID:
+        return handle_pdu_data(
+            printer=printer, pdu_idx=1, set_id=set_id, hk_data=hk_data
+        )
+    elif object_id == obj_ids.PDU_2_HANDLER_ID:
+        return handle_pdu_data(
+            printer=printer, pdu_idx=2, set_id=set_id, hk_data=hk_data
+        )
+    elif object_id == obj_ids.P60_DOCK_HANDLER:
         handle_p60_hk_data(printer=printer, set_id=set_id, hk_data=hk_data)
-    elif object_id == PL_PCDU_ID:
+    elif object_id == obj_ids.PL_PCDU_ID:
         log_to_both(printer, "Received PL PCDU HK data")
     else:
         LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.")
@@ -429,6 +429,159 @@ P60_INDEX_LIST = [
 
 WDT_LIST = ["GND", "I2C", "CAN", "CSP0", "CSP1"]
 
+PDU1_CHANNELS_NAMES = [
+    "TCS Board",
+    "Syrlinks",
+    "Startracker",
+    "MGT",
+    "SUS Nominal",
+    "SCEX",
+    "PLOC",
+    "ACS A Side",
+    "Unused Channel 8",
+]
+
+PDU2_CHANNELS_NAMES = [
+    "Q7S",
+    "Payload PCDU CH1",
+    "RW",
+    "TCS Heater In",
+    "SUS Redundant",
+    "Deployment Mechanism",
+    "Payload PCDU CH6",
+    "ACS B Side",
+    "Payload Camera",
+]
+
+PDU_CHANNEL_NAMES = [PDU1_CHANNELS_NAMES, PDU2_CHANNELS_NAMES]
+
+
+class WdtInfo:
+    def __init__(self):
+        self.wdt_reboots_list = []
+        self.time_pings_left_list = []
+
+    def print(self, printer: FsfwTmTcPrinter):
+        wdt_info = "WDT Type | Reboots | Time or Pings left (CSP only)"
+        log_to_both(printer, wdt_info)
+        for idx in range(len(self.wdt_reboots_list)):
+            log_to_both(
+                printer,
+                f"{WDT_LIST[idx].ljust(5)} | "
+                f"{self.wdt_reboots_list[idx]:010} | {self.time_pings_left_list[idx]:010}",
+            )
+
+    def parse(self, wdt_data: bytes, current_idx: int) -> int:
+        priv_idx = 0
+        self.wdt_reboots_list = []
+        self.time_pings_left_list = []
+        for idx in range(5):
+            self.wdt_reboots_list.append(
+                struct.unpack("!I", wdt_data[priv_idx : priv_idx + 4])[0]
+            )
+            priv_idx += 4
+            current_idx += 4
+        for idx in range(3):
+            self.time_pings_left_list.append(
+                struct.unpack("!I", wdt_data[priv_idx : priv_idx + 4])[0]
+            )
+            priv_idx += 4
+            current_idx += 4
+        for idx in range(2):
+            self.time_pings_left_list.append(wdt_data[priv_idx])
+            current_idx += 1
+            priv_idx += 1
+        return current_idx
+
+
+def handle_pdu_data(
+    printer: FsfwTmTcPrinter, pdu_idx: int, set_id: int, hk_data: bytes
+):
+    current_idx = 0
+    priv_idx = pdu_idx - 1
+    if set_id == SetIds.PDU_1_AUX or set_id == SetIds.PDU_2_AUX:
+        fmt_str = "!hhBBBIIH"
+        inc_len = struct.calcsize(fmt_str)
+        (
+            vcc,
+            vbat,
+            conv_enb_0,
+            conv_enb_1,
+            conv_enb_2,
+            boot_cause,
+            uptime,
+            reset_cause,
+        ) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
+        log_to_both(printer, f"VCC {vcc} mV | VBAT {vbat} mV")
+        log_to_both(
+            printer, f"Converter Enables [{conv_enb_0},{conv_enb_1},{conv_enb_2}]"
+        )
+        log_to_both(
+            printer,
+            f"Boot Cause {boot_cause} | Uptime {uptime} | Reset Cause {reset_cause}",
+        )
+        current_idx += inc_len
+        latchup_list = []
+        log_to_both(printer, "Latchups")
+        for idx in range(len(PDU1_CHANNELS_NAMES)):
+            latchup_list.append(
+                struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
+            )
+            content_line = (
+                f"{PDU_CHANNEL_NAMES[priv_idx][idx].ljust(24)} | {latchup_list[idx]}"
+            )
+            log_to_both(printer, content_line)
+            current_idx += 2
+        device_types = []
+        for idx in range(len(PDU1_CHANNELS_NAMES)):
+            device_types.append(hk_data[current_idx])
+            current_idx += 1
+        device_statuses = []
+        for idx in range(len(PDU1_CHANNELS_NAMES)):
+            device_statuses.append(hk_data[current_idx])
+            current_idx += 1
+        wdt = WdtInfo()
+        current_idx = wdt.parse(wdt_data=hk_data[current_idx:], current_idx=current_idx)
+        wdt.print(printer=printer)
+    if set_id == SetIds.PDU_1_CORE or set_id == SetIds.PDU_2_CORE:
+        log_to_both(printer, f"Received PDU HK from PDU {pdu_idx}")
+        current_list = []
+        for idx in range(len(PDU1_CHANNELS_NAMES)):
+            current_list.append(
+                struct.unpack("!h", hk_data[current_idx : current_idx + 2])[0]
+            )
+            current_idx += 2
+        voltage_list = []
+        for idx in range(len(PDU1_CHANNELS_NAMES)):
+            voltage_list.append(
+                struct.unpack("!H", hk_data[current_idx : current_idx + 2])[0]
+            )
+            current_idx += 2
+        output_enb_list = []
+        for idx in range(len(PDU1_CHANNELS_NAMES)):
+            output_enb_list.append(hk_data[current_idx])
+            current_idx += 1
+        header_str = f"{'Name'.ljust(24)} | OutEnb | U [mV] | I [mA]"
+        print(header_str)
+        printer.file_logger.info(header_str)
+        for idx in range(len(PDU1_CHANNELS_NAMES)):
+            out_enb = f"{output_enb_list[idx]}".ljust(6)
+            content_line = (
+                f"{PDU_CHANNEL_NAMES[priv_idx][idx].ljust(24)} | {out_enb} | "
+                f"{voltage_list[idx]:05} | {current_list[idx]:04}"
+            )
+            log_to_both(printer, content_line)
+        fmt_str = "!IBh"
+        inc_len = struct.calcsize(fmt_str)
+        (boot_count, batt_mode, temperature) = struct.unpack(
+            fmt_str, hk_data[current_idx : current_idx + inc_len]
+        )
+        info = (
+            f"Boot Count {boot_count} | Battery Mode {batt_mode} | "
+            f"Temperature {temperature / 10.0}"
+        )
+        log_to_both(printer, info)
+
 
 def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
     if set_id == SetIds.P60_CORE:
@@ -506,21 +659,8 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
             dearm_status,
         ) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
         current_idx += inc_len
-        wdt_reboots_list = []
+        wdt = WdtInfo()
-        for idx in range(5):
+        current_idx = wdt.parse(wdt_data=hk_data[current_idx:], current_idx=current_idx)
-            wdt_reboots_list.append(
-                struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
-            )
-            current_idx += 4
-        time_pings_left_list = []
-        for idx in range(3):
-            time_pings_left_list.append(
-                struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
-            )
-            current_idx += 4
-        for idx in range(2):
-            time_pings_left_list.append(hk_data[current_idx])
-            current_idx += 1
         fmt_str = "!hhbb"
         inc_len = struct.calcsize(fmt_str)
         (
@@ -547,14 +687,7 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
         )
         log_to_both(printer, util_info)
         log_to_both(printer, util_info_2)
-        wdt_info = "WDT Type | Reboots | Time or Pings left (CSP only)"
+        wdt.print(printer)
-        log_to_both(printer, wdt_info)
-        for idx in range(len(wdt_reboots_list)):
-            log_to_both(
-                printer,
-                f"{WDT_LIST[idx].ljust(5)} | "
-                f"{wdt_reboots_list[idx]:010} | {time_pings_left_list[idx]:010}",
-            )
         misc_info = (
             f"Dearm {dearm_status} | ANT6 Depl {ant6_depl} | AR6 Deply {ar6_depl}"
         )