diff --git a/config/definitions.py b/config/definitions.py index 249fbe5..6e0c9f8 100644 --- a/config/definitions.py +++ b/config/definitions.py @@ -16,3 +16,4 @@ class CustomServiceList(enum.Enum): TMP1075_2 = "tmp1075_2" HEATER = "heater" IMTQ = "imtq" + PLOC = "ploc" diff --git a/config/object_ids.py b/config/object_ids.py index 92f076c..11d1d28 100644 --- a/config/object_ids.py +++ b/config/object_ids.py @@ -19,6 +19,7 @@ HEATER_ID = bytearray([0x54, 0x00, 0x00, 0x1]) PCDU_HANDLER_ID = bytearray([0x44, 0x00, 0x10, 0x00]) SOLAR_ARRAY_DEPLOYMENT_ID = bytearray([0x44, 0x00, 0x10, 0x01]) IMTQ_HANDLER_ID = bytearray([0x44, 0x00, 0x00, 0x14]) +PLOC_ID = bytearray([0x44, 0x00, 0x00, 0x15]) class ObjIdIds(enum.IntEnum): @@ -34,6 +35,7 @@ class ObjIdIds(enum.IntEnum): HEATER_ID = 9 SOLAR_ARRAY_DEPLOYMENT_ID = 10 IMTQ_HANDLER_ID = 11 + PLOC_ID = 12 def set_object_ids() -> Dict[int, bytearray]: @@ -50,5 +52,6 @@ def set_object_ids() -> Dict[int, bytearray]: ObjIdIds.PCDU_HANDLER_ID: PCDU_HANDLER_ID, ObjIdIds.SOLAR_ARRAY_DEPLOYMENT_ID: SOLAR_ARRAY_DEPLOYMENT_ID, ObjIdIds.IMTQ_HANDLER_ID: IMTQ_HANDLER_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..b7955ac --- /dev/null +++ b/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 diff --git a/pus_tc/tc_packer_hook.py b/pus_tc/tc_packer_hook.py index 113daea..aafd9de 100644 --- a/pus_tc/tc_packer_hook.py +++ b/pus_tc/tc_packer_hook.py @@ -21,6 +21,7 @@ from pus_tc.pdu1 import pack_pdu1_test_into from pus_tc.acu import pack_acu_test_into from pus_tc.imtq import pack_imtq_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 @@ -60,6 +61,9 @@ def pack_service_queue_user(service: Union[str, int], op_code: str, service_queu if service == CustomServiceList.IMTQ.value: object_id = get_object_id(ObjIdIds.IMTQ_HANDLER_ID) return pack_imtq_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 !") diff --git a/pus_tc/tmp1075.py b/pus_tc/tmp1075.py index 7b667cb..3057e86 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 7db95db..d661055 100644 --- a/pus_tm/service_8_hook.py +++ b/pus_tm/service_8_hook.py @@ -2,6 +2,7 @@ import struct from typing import Tuple from config.object_ids import ObjIdIds from pus_tc.imtq import ImtqActionIds +from pus_tc.ploc import PlocReplyIds def user_analyze_service_8_data( @@ -29,6 +30,8 @@ def user_analyze_service_8_data( content_list = [data_string] elif object_id == ObjIdIds.IMTQ_HANDLER_ID: return handle_imtq_replies(action_id, custom_data) + elif object_id == ObjIdIds.PLOC_ID: + return handle_ploc_replies(action_id, custom_data) else: header_list = [] content_list = [] @@ -44,4 +47,12 @@ def handle_imtq_replies(action_id: int, custom_data: bytearray) -> Tuple[list, l y_dipole = struct.unpack('!H', custom_data[2:4]) z_dipole = struct.unpack('!H', custom_data[4:6]) content_list = [x_dipole[0], y_dipole[0], z_dipole[0]] + + +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