tmtccmd update

Reviewed-on: #51

.idea/runConfigurations/PDU1_Commanding.xml

@@ -13,7 +13,7 @@
     <option name="ADD_SOURCE_ROOTS" value="true" />
     <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
     <option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtccli.py" />
-    <option name="PARAMETERS" value="-s pdu1 -l -t 6" />
+    <option name="PARAMETERS" value="-s pdu1 -t 6" />
     <option name="SHOW_COMMAND_LINE" value="false" />
     <option name="EMULATE_TERMINAL" value="true" />
     <option name="MODULE_MODE" value="false" />

.idea/runConfigurations/tmtccli_example.xml

@@ -0,0 +1,24 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="tmtccli-example" type="PythonConfigurationType" factoryName="Python" folderName="Example">
+    <module name="tmtc" />
+    <option name="INTERPRETER_OPTIONS" value="" />
+    <option name="PARENT_ENVS" value="true" />
+    <envs>
+      <env name="PYTHONUNBUFFERED" value="1" />
+    </envs>
+    <option name="SDK_HOME" value="" />
+    <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/tmtccmd/examples" />
+    <option name="IS_MODULE_SDK" value="true" />
+    <option name="ADD_CONTENT_ROOTS" value="true" />
+    <option name="ADD_SOURCE_ROOTS" value="true" />
+    <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
+    <option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtccmd/examples/tmtccli.py" />
+    <option name="PARAMETERS" value="" />
+    <option name="SHOW_COMMAND_LINE" value="false" />
+    <option name="EMULATE_TERMINAL" value="true" />
+    <option name="MODULE_MODE" value="false" />
+    <option name="REDIRECT_INPUT" value="false" />
+    <option name="INPUT_FILE" value="" />
+    <method v="2" />
+  </configuration>
+</component>

.idea/runConfigurations/tmtcgui_example.xml

@@ -0,0 +1,24 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="tmtcgui-example" type="PythonConfigurationType" factoryName="Python" folderName="Example">
+    <module name="tmtc" />
+    <option name="INTERPRETER_OPTIONS" value="" />
+    <option name="PARENT_ENVS" value="true" />
+    <envs>
+      <env name="PYTHONUNBUFFERED" value="1" />
+    </envs>
+    <option name="SDK_HOME" value="" />
+    <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/tmtccmd/examples" />
+    <option name="IS_MODULE_SDK" value="true" />
+    <option name="ADD_CONTENT_ROOTS" value="true" />
+    <option name="ADD_SOURCE_ROOTS" value="true" />
+    <EXTENSION ID="PythonCoverageRunConfigurationExtension" runner="coverage.py" />
+    <option name="SCRIPT_NAME" value="$PROJECT_DIR$/tmtccmd/examples/tmtcgui.py" />
+    <option name="PARAMETERS" value="" />
+    <option name="SHOW_COMMAND_LINE" value="false" />
+    <option name="EMULATE_TERMINAL" value="false" />
+    <option name="MODULE_MODE" value="false" />
+    <option name="REDIRECT_INPUT" value="false" />
+    <option name="INPUT_FILE" value="" />
+    <method v="2" />
+  </configuration>
+</component>

config/globals_config.py

@@ -12,10 +12,7 @@ import argparse
 from config.definitions import CustomServiceList, PUS_APID
 from config.custom_mode_op import CustomModeList
 from tmtccmd.config.definitions import CoreComInterfaces
-from tmtccmd.config.globals import (
-    set_default_globals_pre_args_parsing,
-    set_default_globals_post_args_parsing,
-)
+from tmtccmd.config.globals import set_default_globals_pre_args_parsing
 from tmtccmd.logging import get_console_logger
 
 LOGGER = get_console_logger()
@@ -34,12 +31,3 @@ def set_globals_pre_args_parsing(gui: bool = False):
         tm_apid=PUS_APID,
         com_if_id=CoreComInterfaces.TCPIP_UDP.value,
     )
-
-
-def add_globals_post_args_parsing(args: argparse.Namespace, json_cfg_path: str):
-    set_default_globals_post_args_parsing(
-        args=args,
-        custom_services_list=[CustomServiceList],
-        custom_modes_list=[CustomModeList],
-        json_cfg_path=json_cfg_path,
-    )

config/hook_implementations.py

@@ -21,34 +21,16 @@ from config.retvals import get_retval_dict
 
 
 class EiveHookObject(TmTcHookBase):
+    def __init__(self, json_cfg_path: str):
+        super().__init__(json_cfg_path=json_cfg_path)
+
     def get_service_op_code_dictionary(self) -> ServiceOpCodeDictT:
         from tmtccmd.config.globals import get_default_service_op_code_dict
 
         service_op_code_dict = get_default_service_op_code_dict()
-
         get_eive_service_op_code_dict(service_op_code_dict=service_op_code_dict)
-
         return service_op_code_dict
 
-    def get_json_config_file_path(self) -> str:
-        """The user can specify a path and filename for the JSON configuration file by overriding
-        this function.
-        :return:
-        """
-        return "config/tmtc_config.json"
-
-    def add_globals_pre_args_parsing(self, gui: bool = False):
-        from config.globals_config import set_globals_pre_args_parsing
-
-        set_globals_pre_args_parsing(gui=gui)
-
-    def add_globals_post_args_parsing(self, args: argparse.Namespace):
-        from config.globals_config import add_globals_post_args_parsing
-
-        add_globals_post_args_parsing(
-            args=args, json_cfg_path=self.get_json_config_file_path()
-        )
-
     def assign_communication_interface(
         self, com_if_key: str
     ) -> Union[CommunicationInterface, None]:
@@ -56,7 +38,7 @@ class EiveHookObject(TmTcHookBase):
 
         return create_communication_interface_default(
             com_if_key=com_if_key,
-            json_cfg_path=self.get_json_config_file_path(),
+            json_cfg_path=self.json_cfg_path,
             space_packet_ids=(0x0865,),
         )
 
@@ -328,6 +310,26 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
             "Syrlinks Handler: Read LCL config register",
             {OpCodeDictKeys.TIMEOUT: 2.0},
         ),
+        "15": (
+            "Syrlinks Handler: Set waveform OQPSK",
+            {OpCodeDictKeys.TIMEOUT: 2.0},
+        ),
+        "16": (
+            "Syrlinks Handler: Set waveform BPSK",
+            {OpCodeDictKeys.TIMEOUT: 2.0},
+        ),
+        "17": (
+            "Syrlinks Handler: Set second config",
+            {OpCodeDictKeys.TIMEOUT: 2.0},
+        ),
+        "18": (
+            "Syrlinks Handler: Enable debug output",
+            {OpCodeDictKeys.TIMEOUT: 2.0},
+        ),
+        "19": (
+            "Syrlinks Handler: Disable debug output",
+            {OpCodeDictKeys.TIMEOUT: 2.0},
+        ),
     }
     service_syrlinks_handler_tuple = (
         "Syrlinks Handler",

config/version.py

@@ -1,6 +1,6 @@
 SW_NAME = "eive"
 VERSION_MAJOR = 1
-VERSION_MINOR = 8
+VERSION_MINOR = 9
 VERSION_SUBMINOR = 0
 
-__version__ = "1.8.0"
+__version__ = "1.9.0"

pus_tc/devs/p60dock.py

@@ -7,7 +7,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 config.object_ids import P60_DOCK_HANDLER
 

pus_tc/devs/plpcdu.py

@@ -172,7 +172,7 @@ def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
                 | (1 << NormalSubmodesMask.X8_ON)
                 | (1 << NormalSubmodesMask.TX_ON)
                 | (1 << NormalSubmodesMask.MPA_ON)
-                | (1 << NormalSubmodesMask.MPA_ON)
+                | (1 << NormalSubmodesMask.HPA_ON)
             ),
         )
     if op_code in OpCodes.UPDATE_DRO_TO_X8_WAIT:

pus_tc/devs/syrlinks_hk_handler.py

@@ -20,16 +20,21 @@ class SetIds:
 
 
 class CommandIds:
-    SET_TX_MODE_STANDBY = bytearray([0x0, 0x0, 0x0, 0x3])
-    SET_TX_MODE_MODULATION = bytearray([0x0, 0x0, 0x0, 0x4])
-    SET_TX_MODE_CW = bytearray([0x0, 0x0, 0x0, 0x5])
-    READ_TX_STATUS = bytearray([0x0, 0x0, 0x0, 0x7])
-    READ_TX_WAVEFORM = bytearray([0x0, 0x0, 0x0, 0x8])
-    READ_TX_AGC_VALUE_HIGH_BYTE = bytearray([0x0, 0x0, 0x0, 0x9])
-    READ_TX_AGC_VALUE_LOW_BYTE = bytearray([0x0, 0x0, 0x0, 0x9])
+    READ_RX_STATUS_REGISTERS = 2
+    SET_TX_MODE_STANDBY = 3
+    SET_TX_MODE_MODULATION = 4
+    SET_TX_MODE_CW = 5
+    READ_TX_STATUS = 7
+    READ_TX_WAVEFORM = 8
+    READ_TX_AGC_VALUE_HIGH_BYTE = 9
+    READ_TX_AGC_VALUE_LOW_BYTE = 10
     WRITE_LCL_CONFIG = 11
     READ_LCL_CONFIG_REGISTER = 12
-    READ_RX_STATUS_REGISTERS = 2
+    SET_WAVEFORM_OQPSK = 17
+    SET_WAVEFORM_BPSK = 18
+    SET_SECOND_CONFIG = 19
+    ENABLE_DEBUG = 20
+    DISABLE_DEBUG = 21
 
 
 def pack_syrlinks_command(
@@ -121,3 +126,28 @@ def pack_syrlinks_command(
         command = object_id + struct.pack("!I", CommandIds.READ_LCL_CONFIG_REGISTER)
         command = PusTelecommand(service=8, subservice=128, ssc=19, app_data=command)
         tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "15":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Set waveform OQPSK"))
+        command = object_id + struct.pack("!I", CommandIds.SET_WAVEFORM_OQPSK)
+        command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "16":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Set waveform BPSK"))
+        command = object_id + struct.pack("!I", CommandIds.SET_WAVEFORM_BPSK)
+        command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "17":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Set second config"))
+        command = object_id + struct.pack("!I", CommandIds.SET_SECOND_CONFIG)
+        command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "18":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Enable debug printout"))
+        command = object_id + struct.pack("!I", CommandIds.ENABLE_DEBUG)
+        command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == "19":
+        tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Disable debug printout"))
+        command = object_id + struct.pack("!I", CommandIds.DISABLE_DEBUG)
+        command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())

pus_tc/tc_packer_hook.py

@@ -1,15 +1,19 @@
 """Hook function which packs telecommands based on service and operation code string
 """
+import logging
 import os
 from collections import deque
 from typing import Union
 
+from spacepackets.ecss import PusTelecommand
+from tmtccmd.com_if.com_interface_base import CommunicationInterface
 from tmtccmd.config.definitions import CoreServiceList
 from tmtccmd.logging import get_console_logger
 from tmtccmd.logging.pus import log_raw_pus_tc
 from tmtccmd.tc.definitions import TcQueueT
 from tmtccmd.tc.service_5_event import pack_generic_service5_test_into
 from tmtccmd.pus.service_17_test import pack_service_17_ping_command
+from tmtccmd.logging import get_current_time_string
 
 from pus_tc.service_200_mode import pack_service200_test_into
 from pus_tc.devs.p60dock import pack_p60dock_test_into
@@ -69,10 +73,19 @@ from config.object_ids import (
 LOGGER = get_console_logger()
 
 
-def pre_tc_send_cb(packet: bytes, user_args: any):
-    service = packet[7]
-    subservice = packet[8]
-    log_raw_pus_tc(packet=packet, srv_subservice=(service, subservice))
+def pre_tc_send_cb(
+    packet: bytes,
+    com_if: CommunicationInterface,
+    pus_info: Union[PusTelecommand, any],
+    file_logger: logging.Logger,
+):
+    log_raw_pus_tc(
+        packet=packet, srv_subservice=(pus_info.service, pus_info.subservice)
+    )
+    tc_info_string = f"Sent {pus_info}"
+    LOGGER.info(tc_info_string)
+    file_logger.info(f"{get_current_time_string(True)}: {tc_info_string}")
+    com_if.send(data=packet)
 
 
 def pack_service_queue_user(

pus_tm/action_reply_handler.py

@@ -6,99 +6,102 @@ from pus_tc.devs.ploc_supervisor import SupvActionIds
 from pus_tc.devs.star_tracker import StarTrackerActionIds
 from tmtccmd.logging import get_console_logger
 from tmtccmd.config.definitions import DataReplyUnpacked
+from tmtccmd.tm import Service8FsfwTm
+from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
 
 LOGGER = get_console_logger()
 
 
-def user_analyze_service_8_data(
-    object_id: bytes, action_id: int, custom_data: bytearray
-) -> DataReplyUnpacked:
+def handle_action_reply(
+    raw_tm: bytes, printer: FsfwTmTcPrinter, obj_id_dict: ObjectIdDictT
+):
+    """Core Action reply handler
+    :return:
     """
-    This function is called by the TMTC core if a Service 8 data reply (subservice 130)
-    is received. The user can return a tuple of two lists, where the first list
-    is a list of header strings to print and the second list is a list of values to print.
-    The TMTC core will take care of printing both lists and logging them.
-
-    @param object_id:
-    @param action_id:
-    @param custom_data:
-    @return:
-    """
-    if object_id == PDU_2_HANDLER_ID:
-        reply = DataReplyUnpacked()
-        reply.header_list = ["PDU2 Service 8 Reply"]
-        data_string = str()
-        for index in range(len(custom_data)):
-            data_string += str(hex(custom_data[index])) + " , "
-        data_string = data_string.rstrip()
-        data_string = data_string.rstrip(",")
-        data_string = data_string.rstrip()
-        reply.content_list = [data_string]
-        return reply
-    elif object_id == IMTQ_HANDLER_ID:
-        return handle_imtq_replies(action_id, custom_data)
+    tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm)
+    printer.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
+    object_id = obj_id_dict.get(tm_packet.source_object_id)
+    custom_data = tm_packet.custom_data
+    action_id = tm_packet.action_id
+    generic_print_str = printer.generic_action_packet_tm_print(
+        packet=tm_packet, obj_id=object_id
+    )
+    print(generic_print_str)
+    printer.file_logger.info(generic_print_str)
+    if object_id == IMTQ_HANDLER_ID:
+        return handle_imtq_replies(action_id, printer, custom_data)
     elif object_id == PLOC_MPSOC_ID:
-        return handle_ploc_replies(action_id, custom_data)
+        return handle_ploc_replies(action_id, printer, custom_data)
     elif object_id == PLOC_SUPV_ID:
-        return handle_supervisor_replies(action_id, custom_data)
+        return handle_supervisor_replies(action_id, printer, custom_data)
     elif object_id == STAR_TRACKER_ID:
-        return handle_startracker_replies(action_id, custom_data)
-    return DataReplyUnpacked()
+        return handle_startracker_replies(action_id, printer, custom_data)
 
 
-def handle_imtq_replies(action_id: int, custom_data: bytearray) -> DataReplyUnpacked:
-    reply = DataReplyUnpacked()
+def handle_imtq_replies(
+    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
+):
     if action_id == struct.unpack("!I", ImtqActionIds.get_commanded_dipole)[0]:
-        reply.header_list = [
+        header_list = [
             "Commanded X-Dipole",
             "Commanded Y-Dipole",
             "Commanded Z-Dipole",
         ]
-        x_dipole = struct.unpack("!H", custom_data[:2])
-        y_dipole = struct.unpack("!H", custom_data[2:4])
-        z_dipole = struct.unpack("!H", custom_data[4:6])
-        reply.content_list = [x_dipole[0], y_dipole[0], z_dipole[0]]
-    return reply
+        [x_dipole, y_dipole, z_dipole] = struct.unpack("!HHH", custom_data[0:6])
+        content_list = [x_dipole, y_dipole, z_dipole]
+        print(header_list)
+        print(content_list)
+        printer.file_logger.info(header_list)
+        printer.file_logger.info(content_list)
 
 
-def handle_ploc_replies(action_id: int, custom_data: bytearray) -> DataReplyUnpacked:
-    reply = DataReplyUnpacked()
+def handle_ploc_replies(
+    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
+):
     if action_id == PlocReplyIds.tm_mem_read_report:
-        reply.header_list = [
+        header_list = [
             "PLOC Memory Address",
             "PLOC Mem Len",
             "PLOC Read Memory Data",
         ]
-        reply.content_list = [
+        content_list = [
             "0x" + custom_data[:4].hex(),
             struct.unpack("!H", custom_data[4:6])[0],
             "0x" + custom_data[6:10].hex(),
         ]
-    return reply
+        print(header_list)
+        print(content_list)
+        printer.file_logger.info(header_list)
+        printer.file_logger.info(content_list)
 
 
 def handle_supervisor_replies(
-    action_id: int, custom_data: bytearray
-) -> DataReplyUnpacked:
+    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
+):
     reply = DataReplyUnpacked()
     if action_id == SupvActionIds.DUMP_MRAM:
-        reply.header_list = ["MRAM Dump"]
-        reply.content_list = [custom_data[: len(custom_data)]]
-    return reply
+        header_list = ["MRAM Dump"]
+        content_list = [custom_data[: len(custom_data)]]
+        print(header_list)
+        print(content_list)
+        printer.file_logger.info(header_list)
+        printer.file_logger.info(content_list)
 
 
 def handle_startracker_replies(
-    action_id: int, custom_data: bytearray
-) -> DataReplyUnpacked:
-    reply = DataReplyUnpacked()
+    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytearray
+):
     if action_id == StarTrackerActionIds.CHECKSUM:
         if len(custom_data) != 5:
             LOGGER.warning(
                 "Star tracker reply has invalid length {0}".format(len(custom_data))
             )
-            return reply
-        reply.header_list = ["Checksum", "Checksum valid"]
+            return
+        header_list = ["Checksum", "Checksum valid"]
         print(custom_data[4])
         checksum_valid_flag = custom_data[4] >> 8
-        reply.content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
-    return reply
+        content_list = ["0x" + custom_data[:4].hex(), checksum_valid_flag]
+        print(header_list)
+        print(content_list)
+        printer.file_logger.info(header_list)
+        printer.file_logger.info(content_list)

pus_tm/event_handler.py

@@ -26,9 +26,7 @@ def get_event_dict() -> EventDictT:
 
 
 def handle_event_packet(
-        raw_tm: bytes,
-        printer: FsfwTmTcPrinter,
-        file_logger: logging.Logger
+    raw_tm: bytes, printer: FsfwTmTcPrinter, file_logger: logging.Logger
 ) -> str:
     tm = Service5Tm.unpack(raw_telemetry=raw_tm)
     printer.handle_long_tm_print(packet_if=tm, info_if=tm)

pus_tm/factory_hook.py

@@ -1,9 +1,5 @@
+"""Core EIVE TM handler module
 """
-@brief      This file transfers control of TM parsing to the user
-@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
-            it to your needs.
-"""
-from tmtccmd.tm.service_8_fsfw_functional_cmd import Service8FsfwTm
 from spacepackets.ecss.tm import PusTelemetry
 from tmtccmd.logging import get_console_logger
 from tmtccmd.logging.pus import (
@@ -14,17 +10,17 @@ from tmtccmd.logging.pus import (
 )
 
 from tmtccmd.pus.service_17_test import Service17TMExtended
-from tmtccmd.tm.service_3_fsfw_housekeeping import Service3FsfwTm
 from tmtccmd.tm.service_20_fsfw_parameters import Service20FsfwTm
-from tmtccmd.tm.service_5_event import Service5Tm
 from tmtccmd.tm.service_200_fsfw_mode import Service200FsfwTm
 from tmtccmd.utility.tmtc_printer import PrintFormats, FsfwTmTcPrinter
 
 from config.definitions import PUS_APID
 from config.object_ids import get_object_ids
+
 from .event_handler import handle_event_packet
 from .verification_handler import handle_service_1_packet
 from .hk_handling import handle_hk_packet
+from .action_reply_handler import handle_action_reply
 
 LOGGER = get_console_logger()
 
@@ -50,19 +46,16 @@ def pus_factory_hook(raw_tm_packet: bytes):
             handle_service_1_packet(printer=FSFW_PRINTER, raw_tm=raw_tm_packet)
         elif service_type == 3:
             handle_hk_packet(
-                printer=FSFW_PRINTER,
-                raw_tm=raw_tm_packet,
-                obj_id_dict=obj_id_dict
+                printer=FSFW_PRINTER, raw_tm=raw_tm_packet, obj_id_dict=obj_id_dict
             )
         elif service_type == 5:
             handle_event_packet(
-                raw_tm=raw_tm_packet,
-                printer=FSFW_PRINTER,
-                file_logger=file_logger
+                raw_tm=raw_tm_packet, printer=FSFW_PRINTER, file_logger=file_logger
             )
         elif service_type == 8:
-            tm_packet = Service8FsfwTm.unpack(raw_telemetry=raw_tm_packet)
-            FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)
+            handle_action_reply(
+                raw_tm=raw_tm_packet, printer=FSFW_PRINTER, obj_id_dict=obj_id_dict
+            )
         elif service_type == 17:
             tm_packet = Service17TMExtended.unpack(raw_telemetry=raw_tm_packet)
             FSFW_PRINTER.handle_long_tm_print(packet_if=tm_packet, info_if=tm_packet)

pus_tm/hk_handling.py

@@ -5,7 +5,11 @@ import datetime
 
 from tmtccmd.utility.tmtc_printer import FsfwTmTcPrinter
 from tmtccmd.config.definitions import HkReplyUnpacked
-from tmtccmd.tm.service_3_fsfw_housekeeping import Service3Base, HkContentType, Service3FsfwTm
+from tmtccmd.tm.service_3_fsfw_housekeeping import (
+    Service3Base,
+    HkContentType,
+    Service3FsfwTm,
+)
 from tmtccmd.logging import get_console_logger
 from pus_tc.devs.bpx_batt import BpxSetIds
 from pus_tc.devs.syrlinks_hk_handler import SetIds
@@ -19,6 +23,7 @@ from config.object_ids import (
     BPX_HANDLER_ID,
     CORE_CONTROLLER_ID,
     P60_DOCK_HANDLER,
+    PL_PCDU_ID
 )
 
 LOGGER = get_console_logger()
@@ -29,15 +34,13 @@ def handle_hk_packet(
     obj_id_dict: ObjectIdDictT,
     printer: FsfwTmTcPrinter,
 ):
-    tm_packet = Service3FsfwTm.unpack(
-        raw_telemetry=raw_tm, custom_hk_handling=False
-    )
+    tm_packet = Service3FsfwTm.unpack(raw_telemetry=raw_tm, custom_hk_handling=False)
     named_obj_id = obj_id_dict.get(tm_packet.object_id.as_bytes)
     if named_obj_id is None:
         named_obj_id = tm_packet.object_id
     if tm_packet.subservice == 25 or tm_packet.subservice == 26:
         hk_data = tm_packet.tm_data[8:]
-        printer.generic_hk_print(
+        printer.generic_hk_tm_print(
             content_type=HkContentType.HK,
             object_id=named_obj_id,
             set_id=tm_packet.set_id,
@@ -64,36 +67,36 @@ def handle_regular_hk_print(
     """This function is called when a Service 3 Housekeeping packet is received."""
     if object_id == SYRLINKS_HANDLER_ID:
         if set_id == SetIds.RX_REGISTERS_DATASET:
-            return handle_syrlinks_rx_registers_dataset(hk_data)
+            return handle_syrlinks_rx_registers_dataset(printer, hk_data)
         elif set_id == SetIds.TX_REGISTERS_DATASET:
-            return handle_syrlinks_tx_registers_dataset(hk_data)
+            return handle_syrlinks_tx_registers_dataset(printer, hk_data)
         else:
-            LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id")
+            LOGGER.info("Service 3 TM: Syrlinks handler reply with unknown set id")
     elif object_id == IMTQ_HANDLER_ID:
         if (set_id >= ImtqSetIds.POSITIVE_X_TEST) and (
             set_id <= ImtqSetIds.NEGATIVE_Z_TEST
         ):
-            return handle_self_test_data(hk_data)
+            return handle_self_test_data(printer, hk_data)
         else:
-            LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id")
+            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:
-        return handle_gps_data(hk_data=hk_data)
+        handle_gps_data(printer=printer, hk_data=hk_data)
     elif object_id == BPX_HANDLER_ID:
-        return handle_bpx_hk_data(hk_data=hk_data, set_id=set_id)
+        handle_bpx_hk_data(hk_data=hk_data, set_id=set_id, printer=printer)
     elif object_id == CORE_CONTROLLER_ID:
         return handle_core_hk_data(printer=printer, hk_data=hk_data)
     elif object_id == P60_DOCK_HANDLER:
-        return handle_p60_hk_data(printer=printer, hk_data=hk_data)
+        handle_p60_hk_data(printer=printer, hk_data=hk_data)
+    elif object_id == 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.")
         return HkReplyUnpacked()
 
 
-def handle_syrlinks_rx_registers_dataset(
-    hk_data: bytes,
-) -> HkReplyUnpacked:
+def handle_syrlinks_rx_registers_dataset(printer: FsfwTmTcPrinter, hk_data: bytes):
     reply = HkReplyUnpacked()
-    reply.header_list = [
+    header_list = [
         "RX Status",
         "RX Sensitivity",
         "RX Frequency Shift",
@@ -111,7 +114,7 @@ def handle_syrlinks_rx_registers_dataset(
     rx_demod_eb = struct.unpack("!I", hk_data[13:17])
     rx_demod_n0 = struct.unpack("!I", hk_data[17:21])
     rx_data_rate = hk_data[21]
-    reply.content_list = [
+    content_list = [
         rx_status,
         rx_sensitivity,
         rx_frequency_shift,
@@ -121,28 +124,30 @@ def handle_syrlinks_rx_registers_dataset(
         rx_demod_n0,
         rx_data_rate,
     ]
-    reply.validity_buffer = hk_data[22:]
-    reply.num_of_vars = 8
-    return reply
+    validity_buffer = hk_data[22:]
+    log_to_both(printer, str(header_list))
+    log_to_both(printer, str(content_list))
+    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=8)
 
 
 def handle_syrlinks_tx_registers_dataset(
+    printer: FsfwTmTcPrinter,
     hk_data: bytes,
-) -> HkReplyUnpacked:
+):
     reply = HkReplyUnpacked()
-    reply.header_list = ["TX Status", "TX Waveform", "TX AGC value"]
+    header_list = ["TX Status", "TX Waveform", "TX AGC value"]
     tx_status = hk_data[0]
     tx_waveform = hk_data[1]
     tx_agc_value = struct.unpack("!H", hk_data[2:4])
-    reply.content_list = [tx_status, tx_waveform, tx_agc_value]
-    reply.validity_buffer = hk_data[4:]
-    reply.num_of_vars = 3
-    return reply
+    content_list = [tx_status, tx_waveform, tx_agc_value]
+    validity_buffer = hk_data[4:]
+    log_to_both(printer, str(header_list))
+    log_to_both(printer, str(content_list))
+    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=3)
 
 
-def handle_self_test_data(hk_data: bytes) -> HkReplyUnpacked:
-    reply = HkReplyUnpacked()
-    reply.hk_header = [
+def handle_self_test_data(printer: FsfwTmTcPrinter, hk_data: bytes):
+    header_list = [
         "Init Err",
         "Init Raw Mag X [nT]",
         "Init Raw Mag Y [nT]",
@@ -228,8 +233,8 @@ def handle_self_test_data(hk_data: bytes) -> HkReplyUnpacked:
     fina_coil_y_temperature = struct.unpack("!H", hk_data[125:127])[0]
     fina_coil_z_temperature = struct.unpack("!H", hk_data[127:129])[0]
 
-    reply.validity_buffer = hk_data[129:]
-    reply.content_list = [
+    validity_buffer = hk_data[129:]
+    content_list = [
         init_err,
         init_raw_mag_x,
         init_raw_mag_y,
@@ -270,17 +275,17 @@ def handle_self_test_data(hk_data: bytes) -> HkReplyUnpacked:
         fina_coil_y_temperature,
         fina_coil_z_temperature,
     ]
-    reply.num_of_vars = len(reply.hk_header)
-    return reply
+    num_of_vars = len(header_list)
+    log_to_both(printer, str(header_list))
+    log_to_both(printer, str(content_list))
+    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=num_of_vars)
 
 
-def handle_gps_data(hk_data: bytes) -> HkReplyUnpacked:
+def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes):
     LOGGER.info(f"Received GPS data, HK data length {len(hk_data)}")
     reply = HkReplyUnpacked()
     var_index = 0
-    header_array = []
-    content_array = []
-    reply.header_list = [
+    header_list = [
         "Latitude",
         "Longitude",
         "Altitude",
@@ -302,7 +307,7 @@ def handle_gps_data(hk_data: bytes) -> HkReplyUnpacked:
     seconds = hk_data[32]
     date_string = f"{day}.{month}.{year} {hours}:{minutes}:{seconds}"
     unix_seconds = struct.unpack("!I", hk_data[33:37])[0]
-    content_array = [
+    content_list = [
         latitude,
         longitude,
         altitude,
@@ -324,27 +329,29 @@ def handle_gps_data(hk_data: bytes) -> HkReplyUnpacked:
             f"{datetime.datetime.now()}, {latitude}, {longitude}, {altitude}, "
             f"{fix_mode}, {sat_in_use}, {date_string}, {unix_seconds}\n"
         )
-    reply.header_list = header_array
-    reply.content_list = content_array
-    reply.validity_buffer = hk_data[37:39]
-    return reply
+    validity_buffer = hk_data[37:39]
+    log_to_both(printer, str(header_list))
+    log_to_both(printer, str(content_list))
+    printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=10)
 
 
-def handle_bpx_hk_data(hk_data: bytes, set_id: int) -> HkReplyUnpacked:
-    LOGGER.info(f"Received BPX data, HK data length {len(hk_data)}")
-    reply = HkReplyUnpacked()
+def handle_bpx_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
     if set_id == BpxSetIds.GET_HK_SET:
-        charge_current = struct.unpack("!H", hk_data[0:2])[0]
-        discharge_current = struct.unpack("!H", hk_data[2:4])[0]
-        heater_current = struct.unpack("!H", hk_data[4:6])[0]
-        batt_voltage = struct.unpack("!H", hk_data[6:8])[0]
-        batt_temp_1 = struct.unpack("!h", hk_data[8:10])[0]
-        batt_temp_2 = struct.unpack("!h", hk_data[10:12])[0]
-        batt_temp_3 = struct.unpack("!h", hk_data[12:14])[0]
-        batt_temp_4 = struct.unpack("!h", hk_data[14:16])[0]
-        reboot_cntr = struct.unpack("!I", hk_data[16:20])[0]
-        boot_cause = hk_data[20]
-        reply.header_list = [
+        fmt_str = "!HHHHhhhhIB"
+        inc_len = struct.calcsize(fmt_str)
+        (
+            charge_current,
+            discharge_current,
+            heater_current,
+            batt_voltage,
+            batt_temp_1,
+            batt_temp_2,
+            batt_temp_3,
+            batt_temp_4,
+            reboot_cntr,
+            boot_cause,
+        ) = struct.unpack(fmt_str, hk_data[0:inc_len])
+        header_list = [
             "Charge Current",
             "Discharge Current",
             "Heater Current",
@@ -356,7 +363,7 @@ def handle_bpx_hk_data(hk_data: bytes, set_id: int) -> HkReplyUnpacked:
             "Reboot Counter",
             "Boot Cause",
         ]
-        reply.content_list = [
+        content_list = [
             charge_current,
             discharge_current,
             heater_current,
@@ -368,33 +375,37 @@ def handle_bpx_hk_data(hk_data: bytes, set_id: int) -> HkReplyUnpacked:
             reboot_cntr,
             boot_cause,
         ]
-        reply.validity_buffer = hk_data[21:]
+        validity_buffer = hk_data[inc_len:]
+        log_to_both(printer, str(header_list))
+        log_to_both(printer, str(content_list))
+        printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=10)
     elif set_id == BpxSetIds.GET_CFG_SET:
         battheat_mode = hk_data[0]
         battheat_low = struct.unpack("!b", hk_data[1:2])[0]
         battheat_high = struct.unpack("!b", hk_data[2:3])[0]
-        reply.header_list = [
+        header_list = [
             "Battery Heater Mode",
             "Battery Heater Low Limit",
             "Battery Heater High Limit",
         ]
-        reply.content_list = [battheat_mode, battheat_low, battheat_high]
-        reply.validity_buffer = hk_data[3:]
-    return reply
+        content_list = [battheat_mode, battheat_low, battheat_high]
+        validity_buffer = hk_data[3:]
+        log_to_both(printer, str(header_list))
+        log_to_both(printer, str(content_list))
+        printer.print_validity_buffer(validity_buffer=validity_buffer, num_vars=10)
 
 
 def handle_core_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes):
 
     fmt_str = "!fffH"
     inc_len = struct.calcsize(fmt_str)
-    (
-        temperature,
-        ps_voltage,
-        pl_voltage,
-        tx_agc_value
-    ) = struct.unpack(fmt_str, hk_data[0:0 + inc_len])
-    printout = f"Chip Temperature [°C] {temperature} | PS Voltage [mV] {ps_voltage} | " \
-               f"PL Voltage [mV] {pl_voltage} | TX AGC {tx_agc_value}"
+    (temperature, ps_voltage, pl_voltage, tx_agc_value) = struct.unpack(
+        fmt_str, hk_data[0 : 0 + inc_len]
+    )
+    printout = (
+        f"Chip Temperature [°C] {temperature} | PS Voltage [mV] {ps_voltage} | "
+        f"PL Voltage [mV] {pl_voltage} | TX AGC {tx_agc_value}"
+    )
     log_to_both(printer, printout)
     printer.print_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=4)
 
@@ -486,8 +497,8 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes):
         batt_current,
         batt_voltage,
         batt_temp_0,
-        batt_temp_1
-    ) = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
+        batt_temp_1,
+    ) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     current_idx += inc_len
     device_types = []
     device_statuses = []
@@ -501,11 +512,15 @@ def handle_p60_hk_data(printer: FsfwTmTcPrinter, hk_data: bytes):
     current_idx += 1
     wdt_reboots_list = []
     for idx in range(5):
-        wdt_reboots_list.append(struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0])
+        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])
+        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])

requirements.txt

@@ -1 +1 @@
-tmtccmd>=1.13.0
+tmtccmd>=2.0.1

tmtccli.py

@@ -31,8 +31,15 @@ import traceback
 
 try:
     import tmtccmd.runner as tmtccmd
+    from tmtccmd.config import default_json_path, SetupArgs
+    from tmtccmd.config.args import (
+        create_default_args_parser,
+        add_default_tmtccmd_args,
+        parse_default_input_arguments,
+    )
     from tmtccmd.ccsds.handler import CcsdsTmHandler, ApidHandler
     from tmtccmd.logging import init_console_logger
+    from tmtccmd.logging.pus import create_tmtc_logger
 except ImportError as error:
     run_tmtc_commander = None
     initialize_tmtc_commander = None
@@ -61,21 +68,26 @@ from pus_tm.factory_hook import ccsds_tm_handler
 def main():
     print(f"-- eive tmtc version {__version__} --")
     print(f"-- spacepackets version {spacepackets.__version__} --")
-    hook_obj = EiveHookObject()
-    tmtccmd.init_tmtccmd(hook_object=hook_obj)
+    tmtccmd.init_printout(False)
+    tmtc_file_logger = create_tmtc_logger()
+    hook_obj = EiveHookObject(json_cfg_path=default_json_path())
+    arg_parser = create_default_args_parser()
+    add_default_tmtccmd_args(arg_parser)
+    args = parse_default_input_arguments(arg_parser, hook_obj)
+    setup_args = SetupArgs(
+        hook_obj=hook_obj, use_gui=False, apid=PUS_APID, cli_args=args
+    )
+    apid_handler = ApidHandler(cb=ccsds_tm_handler, queue_len=50, user_args=None)
     ccsds_handler = CcsdsTmHandler()
-    init_console_logger()
-    pus_handler = ApidHandler(cb=ccsds_tm_handler, queue_len=50, user_args=None)
-    ccsds_handler.add_tm_handler(apid=PUS_APID, handler=pus_handler)
+    ccsds_handler.add_tm_handler(apid=PUS_APID, handler=apid_handler)
+    tmtccmd.setup(setup_args=setup_args)
     tmtccmd.add_ccsds_handler(ccsds_handler)
-    tmtccmd.setup_tmtccmd(use_gui=False, reduced_printout=False)
-    tmtc_backend = tmtccmd.get_default_tmtc_backend(
-        hook_obj=hook_obj,
-        json_cfg_path=hook_obj.get_json_config_file_path(),
+    tmtc_backend = tmtccmd.create_default_tmtc_backend(
+        setup_args=setup_args,
         tm_handler=ccsds_handler,
     )
-    tmtc_backend.set_pre_send_cb(callable=pre_tc_send_cb, user_args=None)
-    tmtccmd.run_tmtccmd(False, tmtc_backend=tmtc_backend, run_setup=False)
+    tmtc_backend.usr_send_wrapper = (pre_tc_send_cb, tmtc_file_logger)
+    tmtccmd.run(tmtc_backend=tmtc_backend)
 
 
 if __name__ == "__main__":