From aa52ed79ea8643931e9d465ad1b126658a678fc6 Mon Sep 17 00:00:00 2001 From: "Jakob.Meier" Date: Sun, 11 Apr 2021 11:08:52 +0200 Subject: [PATCH] plocHandler test wip --- config/definitions.py | 1 + config/object_ids.py | 3 ++ pus_tc/ploc.py | 64 ++++++++++++++++++++++++++++++++++++++++ pus_tc/tc_packer_hook.py | 4 +++ pus_tc/tmp1075.py | 2 +- pus_tm/service_8_hook.py | 15 ++++++++++ 6 files changed, 88 insertions(+), 1 deletion(-) create mode 100644 pus_tc/ploc.py diff --git a/config/definitions.py b/config/definitions.py index f48c235..bf9e7d1 100644 --- a/config/definitions.py +++ b/config/definitions.py @@ -15,3 +15,4 @@ class CustomServiceList(enum.Enum): TMP1075_1 = "tmp1075_1" TMP1075_2 = "tmp1075_2" HEATER = "heater" + PLOC = "ploc" diff --git a/config/object_ids.py b/config/object_ids.py index bffa66e..4b3c428 100644 --- a/config/object_ids.py +++ b/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 diff --git a/pus_tc/ploc.py b/pus_tc/ploc.py new file mode 100644 index 0000000..abde171 --- /dev/null +++ b/pus_tc/ploc.py @@ -0,0 +1,64 @@ +# -*- 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.pus_tc.base 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, 0x7, 0x14]) + tc_mem_read = bytearray([0x0, 0x0, 0x7, 0x15]) + + +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")) + command = generateWriteMemCommand(object_id) + 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_write: + 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 \ No newline at end of file diff --git a/pus_tc/tc_packer_hook.py b/pus_tc/tc_packer_hook.py index 576ecf4..49e8bf9 100644 --- a/pus_tc/tc_packer_hook.py +++ b/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) + return pack_ploc_test_into(object_id, service_queue) LOGGER.warning("Invalid Service !") diff --git a/pus_tc/tmp1075.py b/pus_tc/tmp1075.py index 9dba368..068bd08 100644 --- a/pus_tc/tmp1075.py +++ b/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 diff --git a/pus_tm/service_8_hook.py b/pus_tm/service_8_hook.py index 12a8617..8933b40 100644 --- a/pus_tm/service_8_hook.py +++ b/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 PlocActionIds def user_analyze_service_8_data( @@ -25,7 +27,20 @@ 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(custom_data) else: header_list = [] content_list = [] return header_list, content_list + + +def handle_ploc_replies(custom_data: bytearray) -> Tuple[list, list]: + header_list = [] + content_list = [] + action_id = struct.unpack('!I', custom_data[0:4]) + if action_id == PlocActionIds.tc_mem_read: + header_list = ['PLOC Read Memory Data'] + memory_data = struct.unpack('!I', custom_data[4:8]) + content_list = [memory_data] + return header_list, content_list