ploc tc mem write and tc mem read test

config/definitions.py

@@ -15,3 +15,4 @@ class CustomServiceList(enum.Enum):
     TMP1075_1 = "tmp1075_1"
     TMP1075_2 = "tmp1075_2"
     HEATER = "heater"
+    PLOC = "ploc"

config/object_ids.py

@@ -18,6 +18,7 @@ TMP_1075_2_HANDLER_ID = bytearray([0x44, 0x00, 0x00, 0x6])
 HEATER_ID = bytearray([0x54, 0x00, 0x00, 0x1])
 PCDU_HANDLER_ID = bytearray([0x44, 0x00, 0x10, 0x00])
 SOLAR_ARRAY_DEPLOYMENT_ID = bytearray([0x44, 0x00, 0x10, 0x01])
+PLOC_ID = bytearray([0x44, 0x00, 0x00, 0x15])
 
 
 class ObjIdIds(enum.IntEnum):
@@ -32,6 +33,7 @@ class ObjIdIds(enum.IntEnum):
     TMP1075_2_HANDLER_ID = 8
     HEATER_ID = 9
     SOLAR_ARRAY_DEPLOYMENT_ID = 10
+    PLOC_ID = 11
 
 
 def set_object_ids() -> Dict[int, bytearray]:
@@ -47,5 +49,6 @@ def set_object_ids() -> Dict[int, bytearray]:
          ObjIdIds.HEATER_ID: HEATER_ID,
          ObjIdIds.PCDU_HANDLER_ID: PCDU_HANDLER_ID,
          ObjIdIds.SOLAR_ARRAY_DEPLOYMENT_ID: SOLAR_ARRAY_DEPLOYMENT_ID,
+         ObjIdIds.PLOC_ID: PLOC_ID,
     })
     return object_id_dict

config/version.py

@@ -1,4 +1,4 @@
 SW_NAME = "eive"
 VERSION_MAJOR = 1
-VERSION_MINOR = 2
+VERSION_MINOR = 3
 VERSION_SUBMINOR = 0

pus_tc/ploc.py

@@ -0,0 +1,67 @@
+# -*- coding: utf-8 -*-
+"""
+@file   ploc.py
+@brief  TMP1075 tests
+@author J. Meier
+@date   06.01.2021
+"""
+import struct
+
+from tmtccmd.core.definitions import QueueCommands
+
+from tmtccmd.pus_tc.packer import TcQueueT
+from tmtccmd.ecss.tc import PusTelecommand
+
+
+class PlocTestProcedure:
+    """
+    @brief  Use this class to define the tests to perform for the PLOC.
+    @details    Setting all to True will run all tests.
+                Setting all to False will only run the tests set to True.
+    """
+    all = False
+    test_tc_mem_write = False
+    test_tc_mem_read = True
+
+
+class PlocActionIds:
+    tc_mem_write = bytearray([0x0, 0x0, 0x0, 0x1])
+    tc_mem_read = bytearray([0x0, 0x0, 0x0, 0x2])
+
+
+class PlocReplyIds:
+    tm_mem_read_report = 6
+
+
+def pack_ploc_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
+    tc_queue.appendleft(
+        (QueueCommands.PRINT,
+         "Testing PLOC Handler with object id: 0x" + object_id.hex())
+    )
+
+    if PlocTestProcedure.all or PlocTestProcedure.test_tc_mem_write:
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC: TC Mem Write Test"))
+        memory_address = int(input("Type memory address: 0x"), 16)
+        memory_data = int(input("Type memory data: 0x"), 16)
+        command = generateWriteMemCommand(object_id, struct.pack('!I', memory_address), memory_data)
+        command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+
+    if PlocTestProcedure.all or PlocTestProcedure.test_tc_mem_read:
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC: TC Mem Read Test"))
+        memory_address = int(input("Type memory address: 0x"), 16)
+        command = object_id + PlocActionIds.tc_mem_read + struct.pack('!I', memory_address)
+        command = PusTelecommand(service=8, subservice=128, ssc=24, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+
+    return tc_queue
+
+
+def generateWriteMemCommand(object_id: bytearray, memory_address: bytearray, memory_data: int) -> bytearray:
+    """ This function generates the command to write to a memory address within the PLOC
+    @param object_id    The object id of the PlocHandler
+    @param memory_address   The PLOC memory address where to write to.
+    @param memory_data    The data to write to the memory address specified by the bytearray memory_address.
+    """
+    command = object_id + PlocActionIds.tc_mem_write + memory_address + struct.pack('!I', memory_data)
+    return command

pus_tc/tc_packer_hook.py

@@ -20,6 +20,7 @@ from pus_tc.pdu2 import pack_pdu2_test_into
 from pus_tc.pdu1 import pack_pdu1_test_into
 from pus_tc.acu import pack_acu_test_into
 from pus_tc.tmp1075 import pack_tmp1075_test_into
+from pus_tc.ploc import pack_ploc_test_into
 from pus_tc.heater import pack_heater_test_into
 from config.definitions import CustomServiceList
 from config.object_ids import ObjIdIds
@@ -56,6 +57,9 @@ def pack_service_queue_user(service: Union[str, int], op_code: str, service_queu
     if service == CustomServiceList.HEATER.value:
         object_id = get_object_id(ObjIdIds.HEATER)
         return pack_heater_test_into(object_id, service_queue)
+    if service == CustomServiceList.PLOC.value:
+        object_id = get_object_id(ObjIdIds.PLOC_ID)
+        return pack_ploc_test_into(object_id, service_queue)
     LOGGER.warning("Invalid Service !")
 
 

pus_tc/tmp1075.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 """
-@file   tmtcc_tc_tmp1075.py
+@file   tmp1075.py
 @brief  TMP1075 tests
 @author J. Meier
 @date   06.01.2021

pus_tc/tmtcc_tc_solar_array_deployment.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+"""
+@file   tmtcc_tc_solar_array_deployment.py
+@brief  The test function in this file simply returns a command which triggers the solar array deployment.
+@author J. Meier
+@date   15.02.2021
+"""
+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
+
+
+class ActionIds:
+    DEPLOY_SOLAR_ARRAYS = bytearray([0x0, 0x0, 0x0, 0x5])
+
+
+def pack_solar_array_deployment_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
+    tc_queue.appendleft((QueueCommands.PRINT, "Testing S/A Deployment"))
+
+    command = object_id + ActionIds.DEPLOY_SOLAR_ARRAYS
+    command = PusTelecommand(service=8, subservice=128, ssc=200, app_data=command)
+    tc_queue.appendleft(command.pack_command_tuple())

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/factory_hook.py

@@ -8,6 +8,7 @@ from tmtccmd.ecss.tm import PusTelemetry
 from tmtccmd.utility.tmtcc_logger import get_logger
 
 from tmtccmd.pus_tm.service_1_verification import Service1TM
+from tmtccmd.pus_tm.service_3_base import Service3Base
 from tmtccmd.pus_tm.service_5_event import Service5TM
 from tmtccmd.pus_tm.service_17_test import Service17TM
 
@@ -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 Service3Base(raw_tm_packet)
     if service_type == 5:
         return Service5TM(raw_tm_packet)
     if service_type == 8:

pus_tm/service_8_hook.py

@@ -1,5 +1,7 @@
+import struct
 from typing import Tuple
 from config.object_ids import ObjIdIds
+from pus_tc.ploc import PlocReplyIds
 
 
 def user_analyze_service_8_data(
@@ -25,7 +27,18 @@ def user_analyze_service_8_data(
         data_string = data_string.rstrip(',')
         data_string = data_string.rstrip()
         content_list = [data_string]
+    elif object_id == ObjIdIds.PLOC_ID:
+        return handle_ploc_replies(action_id, custom_data)
     else:
         header_list = []
         content_list = []
     return header_list, content_list
+
+
+def handle_ploc_replies(action_id: int, custom_data: bytearray) -> Tuple[list, list]:
+    header_list = []
+    content_list = []
+    if action_id == PlocReplyIds.tm_mem_read_report:
+        header_list = ['PLOC Memory Address', 'PLOC Mem Len', 'PLOC Read Memory Data']
+        content_list = [custom_data[:4], custom_data[4:6], custom_data[6:10]]
+    return header_list, content_list

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