eive_tmtc/pus_tm/action_reply_handler.py

@@ -24,7 +24,7 @@ def handle_action_reply(
         raw_telemetry=raw_tm, time_reader=CdsShortTimestamp.empty()
     )
     object_id = obj_id_dict.get(tm_packet.source_object_id_as_bytes)
-    pw = PrintWrapper(printer)
+    pw = PrintWrapper(printer.file_logger)
     custom_data = tm_packet.custom_data
     action_id = tm_packet.action_id
     generic_print_str = printer.generic_action_packet_tm_print(
@@ -32,15 +32,15 @@ def handle_action_reply(
     )
     pw.dlog(generic_print_str)
     if object_id.as_bytes == IMTQ_HANDLER_ID:
-        return handle_imtq_replies(action_id, printer, custom_data)
+        return handle_imtq_replies(action_id, pw, custom_data)
     elif object_id.as_bytes == PLOC_MPSOC_ID:
-        return handle_mpsoc_data_reply(action_id, printer, custom_data)
+        return handle_mpsoc_data_reply(action_id, pw, custom_data)
     elif object_id.as_bytes == PLOC_SUPV_ID:
-        return handle_supervisor_replies(action_id, printer, custom_data)
+        return handle_supervisor_replies(action_id, pw, custom_data)
     elif object_id.as_bytes == CORE_CONTROLLER_ID:
-        return handle_core_ctrl_action_replies(action_id, printer, custom_data)
+        return handle_core_ctrl_action_replies(action_id, pw, custom_data)
     elif object_id.as_bytes == STAR_TRACKER_ID:
-        return handle_startracker_replies(action_id, printer, custom_data)
+        return handle_startracker_replies(action_id, pw, custom_data)
     elif object_id.as_bytes in [
         ACU_HANDLER_ID,
         PDU_1_HANDLER_ID,
@@ -53,9 +53,7 @@ def handle_action_reply(
         pw.dlog(f"Raw Data: {tm_packet.custom_data.hex(sep=',')}")
 
 
-def handle_imtq_replies(
+def handle_imtq_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray):
-    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
-):
     if action_id == struct.unpack("!I", ImtqActionId.get_commanded_dipole)[0]:
         header_list = [
             "Commanded X-Dipole",
@@ -64,33 +62,25 @@ def handle_imtq_replies(
         ]
         [x_dipole, y_dipole, z_dipole] = struct.unpack("!HHH", custom_data[0:6])
         content_list = [x_dipole, y_dipole, z_dipole]
-        print(header_list)
+        pw.dlog(f"{header_list}")
-        print(content_list)
+        pw.dlog(f"{content_list}")
-        printer.file_logger.info(header_list)
-        printer.file_logger.info(content_list)
 
 
-def handle_supervisor_replies(
+def handle_supervisor_replies(action_id: int, pw: PrintWrapper, custom_data: bytearray):
-    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
-):
     if action_id == SupvActionId.DUMP_MRAM:
         header_list = ["MRAM Dump"]
         content_list = [custom_data[: len(custom_data)]]
-        print(header_list)
+        pw.dlog(f"{header_list}")
-        print(content_list)
+        pw.dlog(f"{content_list}")
-        printer.file_logger.info(header_list)
-        printer.file_logger.info(content_list)
     elif action_id == SupvActionId.READ_GPIO:
         header_list = ["GPIO state"]
         content_list = [struct.unpack("!H", custom_data[:2])[0]]
-        print(header_list)
+        pw.dlog(f"{header_list}")
-        print(content_list)
+        pw.dlog(f"{content_list}")
-        printer.file_logger.info(header_list)
-        printer.file_logger.info(content_list)
 
 
 def handle_startracker_replies(
-    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
+    action_id: int, pw: PrintWrapper, custom_data: bytearray
 ):
     if action_id == StarTrackerActionId.CHECKSUM:
         if len(custom_data) != 5:
@@ -102,7 +92,5 @@ def handle_startracker_replies(
         print(custom_data[4])
         checksum_valid_flag = custom_data[4] >> 8
         content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
-        print(header_list)
+        pw.dlog(f"{header_list}")
-        print(content_list)
+        pw.dlog(f"{content_list}")
-        printer.file_logger.info(header_list)
-        printer.file_logger.info(content_list)

eive_tmtc/pus_tm/factory_hook.py

@@ -41,14 +41,14 @@ def pus_factory_hook(
         return
     service = tm_packet.service
     obj_id_dict = get_object_ids()
-    pw = PrintWrapper(printer)
+    pw = PrintWrapper(printer.file_logger)
     dedicated_handler = True
     if service == 1:
         handle_service_1_fsfw_packet(wrapper=verif_wrapper, raw_tm=packet)
     elif service == 3:
         handle_hk_packet(printer=printer, raw_tm=packet, obj_id_dict=obj_id_dict)
     elif service == 5:
-        handle_event_packet(raw_tm=packet, printer=printer)
+        handle_event_packet(raw_tm=packet, pw=pw)
     elif service == 8:
         handle_action_reply(raw_tm=packet, printer=printer, obj_id_dict=obj_id_dict)
     elif service == 17:

eive_tmtc/pus_tm/hk_handling.py

@@ -108,9 +108,9 @@ def handle_regular_hk_print(
     elif objb == obj_ids.PCDU_HANDLER_ID:
         return handle_pcdu_hk(pw=pw, set_id=set_id, hk_data=hk_data)
     elif objb == obj_ids.BPX_HANDLER_ID:
-        return handle_bpx_hk_data(hk_data=hk_data, set_id=set_id, printer=printer)
+        return handle_bpx_hk_data(hk_data=hk_data, set_id=set_id, pw=pw)
     elif objb == obj_ids.CORE_CONTROLLER_ID:
-        return handle_core_hk_data(printer=printer, hk_data=hk_data, set_id=set_id)
+        return handle_core_hk_data(pw=pw, hk_data=hk_data, set_id=set_id)
     elif objb == obj_ids.PDU_1_HANDLER_ID:
         return handle_pdu_data(pw=pw, pdu_idx=1, set_id=set_id, hk_data=hk_data)
     elif objb == obj_ids.PDU_2_HANDLER_ID:

eive_tmtc/tmtc/core.py

@@ -581,9 +581,8 @@ def create_xsc_reboot_cmds(
     )
 
 
-def handle_core_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
+def handle_core_hk_data(pw: PrintWrapper, set_id: int, hk_data: bytes):
     if set_id == SetId.HK:
-        pw = PrintWrapper(printer)
         fmt_str = "!fff"
         inc_len = struct.calcsize(fmt_str)
         (temperature, ps_voltage, pl_voltage) = struct.unpack(
@@ -594,13 +593,14 @@ def handle_core_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
             f"PL Voltage [mV] {pl_voltage}"
         )
         pw.dlog(printout)
-        printer.get_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=3)
+        FsfwTmTcPrinter.get_validity_buffer(
+            validity_buffer=hk_data[inc_len:], num_vars=3
+        )
 
 
 def handle_core_ctrl_action_replies(
-    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytes
+    action_id: int, pw: PrintWrapper, custom_data: bytes
 ):
-    pw = PrintWrapper(printer)
     if action_id == ActionId.LIST_DIR_DUMP_DIRECTLY:
         if len(custom_data) < 4:
             _LOGGER.warning("Data unexpectedly small")

eive_tmtc/tmtc/power/bpx_batt.py

@@ -126,8 +126,7 @@ HEADER_LIST = [
 ]
 
 
-def handle_bpx_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
+def handle_bpx_hk_data(pw: PrintWrapper, set_id: int, hk_data: bytes):
-    pw = PrintWrapper(printer)
     if set_id == BpxSetId.GET_HK_SET:
         fmt_str = "!HHHHhhhhIB"
         inc_len = struct.calcsize(fmt_str)
@@ -158,7 +157,9 @@ def handle_bpx_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
         validity_buffer = hk_data[inc_len:]
         pw.dlog(str(HEADER_LIST))
         pw.dlog(str(content_list))
-        printer.get_validity_buffer(validity_buffer=validity_buffer, num_vars=10)
+        FsfwTmTcPrinter.get_validity_buffer(
+            validity_buffer=validity_buffer, num_vars=10
+        )
     elif set_id == BpxSetId.GET_CFG_SET:
         battheat_mode = hk_data[0]
         battheat_low = struct.unpack("!b", hk_data[1:2])[0]
@@ -172,4 +173,6 @@ def handle_bpx_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
         validity_buffer = hk_data[3:]
         pw.dlog(str(header_list))
         pw.dlog(str(content_list))
-        printer.get_validity_buffer(validity_buffer=validity_buffer, num_vars=10)
+        FsfwTmTcPrinter.get_validity_buffer(
+            validity_buffer=validity_buffer, num_vars=10
+        )