syrlinks hk handler test

config/tmtcc_definitions.py

@@ -36,6 +36,7 @@ class ServiceList(enum.Enum):
     TMP1075_2 = auto()
     HEATER = auto()
     SA_DEPL = auto()
+    SYRLINKS = auto()
 
 
 class SerialConfig(enum.Enum):

config/tmtcc_globals.py

@@ -162,6 +162,8 @@ def add_globals_post_args_parsing(args: argparse.Namespace):
         service = ServiceList.HEATER
     elif service == "sa_depl":
         service = ServiceList.SA_DEPL
+    elif service == "syrlinks":
+        service = ServiceList.SYRLINKS
     else:
         logger.warning("Service not known! Setting standard service 17")
         service = ServiceList.SERVICE_17

config/tmtcc_object_ids.py

@@ -16,11 +16,13 @@ class ObjectIds(enum.Enum):
     P60DOCK_HANDLER_ID = auto()
     PDU1_HANDLER_ID = auto()
     PDU2_HANDLER_ID = auto()
+    PCDU_HANDLER = auto()
     ACU_HANDLER_ID = auto()
     TMP1075_1_HANDLER_ID = auto()
     TMP1075_2_HANDLER_ID = auto()
     HEATER = auto()
-    SOLAR_ARRAY_DEPLOYMENT_HANDLER = auto()
+    SOLAR_ARRAY_DEPLOYMENT = auto()
+    SYRLINKS_HK_HANDLER = auto()
 
 
 def set_object_ids(object_id_dict: Dict[ObjectIds, bytearray]):
@@ -34,8 +36,10 @@ def set_object_ids(object_id_dict: Dict[ObjectIds, bytearray]):
          o_ids.ACU_HANDLER_ID: bytearray([0x44, 0x00, 0x00, 0x4]),
          o_ids.TMP1075_1_HANDLER_ID: bytearray([0x44, 0x00, 0x00, 0x5]),
          o_ids.TMP1075_2_HANDLER_ID: bytearray([0x44, 0x00, 0x00, 0x6]),
-         o_ids.HEATER: bytearray([0x54, 0x00, 0x00, 0x3]),
-         o_ids.SOLAR_ARRAY_DEPLOYMENT_HANDLER: bytearray([0x44, 0x00, 0x00, 0x8]),
+         o_ids.HEATER: bytearray([0x54, 0x00, 0x00, 0x1]),
+         o_ids.PCDU_HANDLER: bytearray([0x44, 0x00, 0x10, 0x00]),
+         o_ids.SOLAR_ARRAY_DEPLOYMENT: bytearray([0x44, 0x00, 0x10, 0x01]),
+         o_ids.SYRLINKS_HK_HANDLER: bytearray([0x44, 0x00, 0x00, 0x9]),
          o_ids.INVALID: bytearray([0xFF, 0xFF, 0xFF, 0xFF]),
          }
     )

pus_tc/tmtcc_tc_packer_hook.py

@@ -21,6 +21,7 @@ from config.tmtcc_object_ids import ObjectIds
 from pus_tc.tmtcc_tc_tmp1075 import pack_tmp1075_test_into
 from pus_tc.tmtcc_tc_heater import pack_heater_test_into
 from pus_tc.tmtcc_tc_solar_array_deployment import pack_solar_array_deployment_test_into
+from pus_tc.tmtcc_tc_syrlinks_hk_handler import pack_syrlinks_hk_handler_test_into
 
 LOGGER = get_logger()
 
@@ -54,8 +55,11 @@ def pack_service_queue_user(service: ServiceList, op_code: str, service_queue: T
         object_id = get_object_id(ObjectIds.HEATER)
         return pack_heater_test_into(object_id, service_queue)
     if service == ServiceList.SA_DEPL:
-        object_id = get_object_id(ObjectIds.SOLAR_ARRAY_DEPLOYMENT_HANDLER)
+        object_id = get_object_id(ObjectIds.SOLAR_ARRAY_DEPLOYMENT)
         return pack_solar_array_deployment_test_into(object_id, service_queue)
+    if service == ServiceList.SYRLINKS:
+        object_id = get_object_id(ObjectIds.SYRLINKS_HK_HANDLER)
+        return pack_syrlinks_hk_handler_test_into(object_id, service_queue)
     LOGGER.warning("Invalid Service !")
 
 

pus_tc/tmtcc_tc_syrlinks_hk_handler.py

@@ -0,0 +1,31 @@
+# -*- coding: utf-8 -*-
+"""
+@file   tmtcc_tc_syrlinks_hk_handler.py
+@brief  Syrlinks Hk Handler tests
+@author J. Meier
+@date   13.12.2020
+"""
+
+from tmtc_core.core.tmtc_core_definitions import QueueCommands
+from tmtc_core.pus_tc.tmtcc_pus_tc_packer import TcQueueT
+from tmtc_core.pus_tc.tmtcc_pus_tc_base import PusTelecommand
+from tmtc_core.pus_tc.tmtcc_tc_service_3_housekeeping import *
+
+
+class SetIds:
+    RX_REGISTERS_DATASET = 1
+    TX_REGISTERS_DATASET = 2
+
+
+def pack_syrlinks_hk_handler_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
+    tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Get RX Registers"))
+    sid = make_sid(object_id, SetIds.RX_REGISTERS_DATASET)
+    command = generate_one_hk_command(sid, 200)
+    tc_queue.appendleft(command.pack_command_tuple())
+
+    tc_queue.appendleft((QueueCommands.PRINT, "Syrlinks: Get TX Registers"))
+    sid = make_sid(object_id, SetIds.TX_REGISTERS_DATASET)
+    command = generate_one_hk_command(sid, 201)
+    tc_queue.appendleft(command.pack_command_tuple())
+
+    return tc_queue

pus_tm/tmtcc_pus_hk_handling.py

@@ -1,26 +0,0 @@
-"""
-@brief      This file transfers control of housekeeping handling (PUS service 3) to the
-            developer
-@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
-            it to your needs.
-"""
-from typing import Tuple
-
-from tmtc_core.pus_tm.tmtcc_pus_service_3 import Service3Base
-from tmtc_core.utility.tmtcc_logger import get_logger
-
-LOGGER = get_logger()
-
-
-def handle_user_hk_packet(
-        object_id: bytearray, hk_data: bytearray,
-        service3_packet: Service3Base) -> Tuple[list, list, bytearray]:
-    """
-    This function is called when a Service 3 Housekeeping packet is received.
-    @param object_id:
-    @param hk_data:
-    @param service3_packet:
-    @return:
-    """
-    LOGGER.info("Service3TM: Parsing for this SID has not been implemented.")
-    return [], [], bytearray()

pus_tm/tmtcc_pus_hk_handling_hook.py

@@ -0,0 +1,81 @@
+"""
+@brief      This file transfers control of housekeeping handling (PUS service 3) to the
+            developer
+@details    Template configuration file. Copy this folder to the TMTC commander root and adapt
+            it to your needs.
+"""
+import struct
+from typing import Tuple
+
+from config.tmtcc_object_ids import ObjectIds
+from tmtc_core.pus_tm.tmtcc_tm_service3_base import Service3Base
+from tmtc_core.utility.tmtcc_logger import get_logger
+from config.tmtcc_object_ids import ObjectIds
+from pus_tc.tmtcc_tc_syrlinks_hk_handler import SetIds
+
+LOGGER = get_logger()
+
+
+def handle_user_hk_packet(object_id: ObjectIds, set_id: int, hk_data: bytearray,
+                          service3_packet: Service3Base) -> Tuple[list, list, bytearray, int]:
+    """
+    This function is called when a Service 3 Housekeeping packet is received.
+
+    Please note that the object IDs should be compared by value because direct comparison of
+    enumerations does not work in Python. For example use:
+
+        if object_id.value == ObjectIds.TEST_OBJECT.value
+
+    to test equality based on the object ID list.
+
+    @param object_id:
+    @param set_id:
+    @param hk_data:
+    @param service3_packet:
+    @return: Expects a tuple, consisting of two lists, a bytearray and an integer
+    The first list contains the header columns, the second list the list with
+    the corresponding values. The bytearray is the validity buffer, which is usually appended
+    at the end of the housekeeping packet. The last value is the number of parameters.
+    """
+    if object_id == ObjectIds.SYRLINKS_HK_HANDLER:
+        if set_id == SetIds.RX_REGISTERS_DATASET:
+            return handle_syrlinks_rx_registers_dataset(hk_data)
+        elif set_id == SetIds.TX_REGISTERS_DATASET:
+            return handle_syrlinks_tx_registers_dataset(hk_data)
+        else:
+            LOGGER.info("Serive 3 TM: Syrlinks handler reply with unknown set id")
+            return [], [], bytearray(), 0
+    else:
+        LOGGER.info("Service 3 TM: Parsing for this SID has not been implemented.")
+        return [], [], bytearray(), 0
+
+
+def handle_syrlinks_rx_registers_dataset(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
+    hk_header = []
+    hk_content = []
+    validity_buffer = bytearray()
+    hk_header = ["RX Status", "RX Sensitivity", "RX Frequency Shift", "RX IQ Power", "RX AGC Value", "RX Demod Eb",
+                 "RX Demod N0", "RX Datarate"]
+    rx_status = hk_data[0]
+    rx_sensitivity = struct.unpack('!I', hk_data[1:5])
+    rx_frequency_shift = struct.unpack('!I', hk_data[5:9])
+    rx_iq_power = struct.unpack('!H', hk_data[9:11])
+    rx_agc_value = struct.unpack('!H', hk_data[11:13])
+    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]
+    hk_content = [rx_status, rx_sensitivity, rx_frequency_shift, rx_iq_power, rx_agc_value, rx_demod_eb, rx_demod_n0,
+                  rx_data_rate]
+    return hk_header, hk_content, validity_buffer, 8
+
+
+def handle_syrlinks_tx_registers_dataset(hk_data: bytearray) -> Tuple[list, list, bytearray, int]:
+    hk_header = []
+    hk_content = []
+    validity_buffer = bytearray()
+    hk_header = ["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])
+    hk_content = [tx_status,  tx_waveform, tx_agc_value]
+    return hk_header, hk_content, validity_buffer, 3

pus_tm/tmtcc_pus_tm_factory_hook.py

@@ -3,6 +3,7 @@
 @details    Template configuration file. Copy this folder to the TMTC commander root and adapt
             it to your needs.
 """
+from tmtc_core.pus_tm.tmtcc_pus_service_3 import Service3TM
 from tmtc_core.pus_tm.tmtcc_pus_service_8 import Service8TM
 from tmtc_core.pus_tm.tmtcc_pus_tm_base import PusTelemetry
 from tmtc_core.utility.tmtcc_logger import get_logger
@@ -18,6 +19,8 @@ def tm_user_factory_hook(raw_tm_packet: bytearray) -> PusTelemetry:
     service_type = raw_tm_packet[7]
     if service_type == 1:
         return Service1TM(raw_tm_packet)
+    if service_type == 3:
+        return Service3TM(raw_tm_packet)
     if service_type == 5:
         return Service5TM(raw_tm_packet)
     if service_type == 8: