.idea/runConfigurations/PLOC_MPSoC.xml

@@ -1,5 +1,5 @@
 <component name="ProjectRunConfigurationManager">
-  <configuration default="false" name="PLOC Test UDP Local" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
+  <configuration default="false" name="PLOC MPSoC" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
     <module name="tmtc" />
     <option name="INTERPRETER_OPTIONS" value="" />
     <option name="PARENT_ENVS" value="true" />
@@ -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="-m 1 -c 2 -s PLOC -l" />
+    <option name="PARAMETERS" value="-s ploc_mpsoc -l -t 6" />
     <option name="SHOW_COMMAND_LINE" value="false" />
     <option name="EMULATE_TERMINAL" value="true" />
     <option name="MODULE_MODE" value="false" />

.idea/runConfigurations/PLOC_SUPV_Test.xml

@@ -13,9 +13,9 @@
     <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 ploc_supv -l -t 6 --hk" />
+    <option name="PARAMETERS" value="-s ploc_supv -l -t 6" />
     <option name="SHOW_COMMAND_LINE" value="false" />
-    <option name="EMULATE_TERMINAL" 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="" />

.idea/runConfigurations/PLOC_Test_UDP_Flatsat.xml

@@ -1,24 +0,0 @@
-<component name="ProjectRunConfigurationManager">
-  <configuration default="false" name="PLOC Test UDP Flatsat" type="PythonConfigurationType" factoryName="Python" folderName="Devices">
-    <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$" />
-    <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$/tmtccli.py" />
-    <option name="PARAMETERS" value="-m 1 -c 2 -s PLOC -l" />
-    <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>

config/definitions.py

@@ -22,7 +22,8 @@ class CustomServiceList(enum.Enum):
     TMP1075_2 = "tmp1075_2"
     HEATER = "heater"
     IMTQ = "imtq"
-    PLOC = "ploc"
+    PLOC_SUPV = "ploc_supv"
+    PLOC_MPSOC = "ploc_mpsoc"
     PCDU = "pcdu"
     PL_PCDU = "plpcdu"
     SA_DEPLYOMENT = "sa_depl"
@@ -31,7 +32,6 @@ class CustomServiceList(enum.Enum):
     REACTION_WHEEL_3 = "reaction_wheel_3"
     REACTION_WHEEL_4 = "reaction_wheel_4"
     RAD_SENSOR = "rad_sensor"
-    PLOC_SUPV = "ploc_supv"
     PLOC_UPDATER = "ploc_updater"
     GPS_0 = "gps0"
     GPS_1 = "gps1"

config/hook_implementations.py

@@ -124,7 +124,8 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
         add_imtq_cmds,
         add_rw_cmds,
         add_rad_sens_cmds,
-        add_ploc_cmds,
+        add_ploc_mpsoc_cmds,
+        add_ploc_supv_cmds,
         add_system_cmds,
     )
     from pus_tc.devs.gps import GpsOpCodes
@@ -136,7 +137,8 @@ def get_eive_service_op_code_dict(service_op_code_dict: ServiceOpCodeDictT):
     add_imtq_cmds(cmd_dict=service_op_code_dict)
     add_rad_sens_cmds(cmd_dict=service_op_code_dict)
     add_rw_cmds(cmd_dict=service_op_code_dict)
-    add_ploc_cmds(cmd_dict=service_op_code_dict)
+    add_ploc_mpsoc_cmds(cmd_dict=service_op_code_dict)
+    add_ploc_supv_cmds(cmd_dict=service_op_code_dict)
     add_system_cmds(cmd_dict=service_op_code_dict)
 
     op_code_dict = {

pus_tc/cmd_definitions.py

@@ -268,7 +268,17 @@ def add_pcdu_cmds(cmd_dict: ServiceOpCodeDictT):
     add_op_code_entry(
         op_code_dict=op_code_dict,
         keys=Pdu1OpCodes.MGT_OFF.value,
-        info="PDU1: Turn MGT on",
+        info="PDU1: Turn MGT off",
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=Pdu1OpCodes.PLOC_ON.value,
+        info="PDU1: Turn PLOC on",
+    )
+    add_op_code_entry(
+        op_code_dict=op_code_dict,
+        keys=Pdu1OpCodes.PLOC_OFF.value,
+        info="PDU1: Turn PLOC off",
     )
     add_op_code_entry(
         op_code_dict=op_code_dict,
@@ -446,7 +456,23 @@ def add_rad_sens_cmds(cmd_dict: ServiceOpCodeDictT):
     cmd_dict[CustomServiceList.RAD_SENSOR.value] = service_rad_sensor_tuple
 
 
-def add_ploc_cmds(cmd_dict: ServiceOpCodeDictT):
+def add_ploc_mpsoc_cmds(cmd_dict: ServiceOpCodeDictT):
+    op_code_dict_srv_ploc_mpsoc = {
+        "0": ("Ploc MPSoC: Memory write", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "1": ("Ploc MPSoC: Memory read", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "2": ("Ploc MPSoC: Flash write", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "3": ("Ploc MPSoC: Flash delete", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "4": ("Ploc MPSoC: Replay start", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "5": ("Ploc MPSoC: Replay stop", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "6": ("Ploc MPSoC: Downlink pwr on", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "7": ("Ploc MPSoC: Downlink pwr off", {OpCodeDictKeys.TIMEOUT: 2.0}),
+        "8": ("Ploc MPSoC: Replay write sequence", {OpCodeDictKeys.TIMEOUT: 2.0}),
+    }
+    service_ploc_mpsoc_tuple = ("Ploc MPSoC", op_code_dict_srv_ploc_mpsoc)
+    cmd_dict[CustomServiceList.PLOC_MPSOC.value] = service_ploc_mpsoc_tuple
+
+
+def add_ploc_supv_cmds(cmd_dict: ServiceOpCodeDictT):
     op_code_dict_ploc_mem_dumper = {
         "0": ("PLOC Memory Dumper: MRAM dump", {OpCodeDictKeys.TIMEOUT: 2.0}),
     }

pus_tc/devs/pdu1.py

@@ -27,6 +27,8 @@ class Pdu1OpCodes(enum.Enum):
     # Solar Cell Experiment
     SCEX_ON = "12"
     SCEX_OFF = "13"
+    PLOC_ON = "14"
+    PLOC_OFF = "15"
 
     TESTS = "32"
 
@@ -190,6 +192,24 @@ def pack_pdu1_commands(object_id: bytearray, tc_queue: TcQueueT, op_code: str):
             Channel.off,
         )
         tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == Pdu1OpCodes.PLOC_ON.value:
+        tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Turn PLOC on"))
+        command = pack_set_param_command(
+            object_id,
+            PDUConfigTable.out_en_6.parameter_address,
+            PDUConfigTable.out_en_6.parameter_size,
+            Channel.on,
+        )
+        tc_queue.appendleft(command.pack_command_tuple())
+    if op_code == Pdu1OpCodes.PLOC_OFF.value:
+        tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Turn PLOC off"))
+        command = pack_set_param_command(
+            object_id,
+            PDUConfigTable.out_en_6.parameter_address,
+            PDUConfigTable.out_en_6.parameter_size,
+            Channel.off,
+        )
+        tc_queue.appendleft(command.pack_command_tuple())
     if op_code == GomspaceOpCodes.REQUEST_HK_ONCE.value:
         tc_queue.appendleft((QueueCommands.PRINT, "PDU1: Requesting HK Table Once"))
         hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1)

pus_tc/devs/ploc_mpsoc.py

@@ -7,63 +7,112 @@
 @date   06.03.2021
 """
 import struct
+import enum
 
 from tmtccmd.config.definitions import QueueCommands
-
+from tmtccmd.utility.logger import get_console_logger
 from tmtccmd.tc.packer import TcQueueT
 from spacepackets.ecss.tc import PusTelecommand
+from utility.input_helper import InputHelper
+
+LOGGER = get_console_logger()
+
+MANUAL_INPUT = "3"
+
+flash_write_file_dict = {
+    "1": ["q7s test file", "/mnt/sd0/ploc-mpsoc/flashwrite.bin"],
+    "2": ["te0720-1cfa test file", "/mnt/sd0/ploc-mpsoc/flashwrite.bin"],
+    MANUAL_INPUT: ["manual input", ""],
+}
+
+mpsoc_file_dict = {
+    MANUAL_INPUT: ["manual input", ""],
+}
 
 
-class PlocTestProcedure:
+class CommandIds(enum.IntEnum):
-    """
+    TC_MEM_WRITE = 1
-    @brief  Use this class to define the tests to perform for the PLOC.
+    TC_MEM_READ = 2
-    @details    Setting all to True will run all tests.
+    FLASH_WRITE = 9
-                Setting all to False will only run the tests set to True.
+    TC_FLASH_DELETE = 10
-    """
+    TC_REPLAY_START = 11
-
+    TC_REPLAY_STOP = 12
-    all = False
+    TC_REPLAY_WRITE_SEQUENCE = 13
-    test_tc_mem_write = False
+    TC_DOWNLINK_PWR_ON = 14
-    test_tc_mem_read = True
+    TC_DOWNLINK_PWR_OFF = 15
-
-
-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_mpsoc_test_into(object_id: bytearray, tc_queue: TcQueueT) -> TcQueueT:
+def pack_ploc_mpsoc_commands(
+        object_id: bytearray, tc_queue: TcQueueT, op_code: str
+) -> TcQueueT:
     tc_queue.appendleft(
-        (QueueCommands.PRINT, "Testing PLOC MPSoC with object id: 0x" + object_id.hex())
+        (
+            QueueCommands.PRINT,
+            "Generate command for PLOC MPSoC with object id: 0x" + object_id.hex(),
+        )
     )
 
-    if PlocTestProcedure.all or PlocTestProcedure.test_tc_mem_write:
+    if op_code == "0":
-        tc_queue.appendleft((QueueCommands.PRINT, "PLOC: TC Mem Write Test"))
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: TC mem write test"))
-        memory_address = int(input("PLOC Tc Mem Write: Type memory address: 0x"), 16)
+        memory_address = int(input("PLOC MPSoC: Tc Mem Write: Type memory address: 0x"), 16)
-        memory_data = int(input("PLOC Tc Mem Write: Type memory data: 0x"), 16)
+        memory_data = int(input("PLOC MPSoC: Tc Mem Write: Type memory data: 0x"), 16)
+        # TODO: implement variable length mem write command
+        mem_len = 1  # 1 32-bit word
         command = generate_write_mem_command(
-            object_id, struct.pack("!I", memory_address), memory_data
+            object_id, memory_address, memory_data, mem_len
         )
+        command = PusTelecommand(service=8, subservice=128, ssc=20, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "1":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: TC mem read test"))
+        command = prepare_mem_read_command(object_id)
+        command = PusTelecommand(service=8, subservice=128, ssc=21, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "2":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Flash write"))
+        command = prepare_flash_write_cmd(object_id)
+        command = PusTelecommand(service=8, subservice=128, ssc=22, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "3":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Flash delete"))
+        command = prepare_flash_delete_cmd(object_id)
         command = PusTelecommand(service=8, subservice=128, ssc=23, app_data=command)
         tc_queue.appendleft(command.pack_command_tuple())
-
+    elif op_code == "4":
-    if PlocTestProcedure.all or PlocTestProcedure.test_tc_mem_read:
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Replay start"))
-        tc_queue.appendleft((QueueCommands.PRINT, "PLOC: TC Mem Read Test"))
+        command = prepare_replay_start_cmd(object_id)
-        memory_address = int(input("PLOC Tc Mem Read: 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())
+    elif op_code == "5":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Replay stop"))
+        command = object_id + struct.pack('!I', CommandIds.TC_REPLAY_STOP)
+        command = PusTelecommand(service=8, subservice=128, ssc=25, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "6":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Downlink pwr on"))
+        command = prepare_downlink_pwr_on_cmd(object_id)
+        command = PusTelecommand(service=8, subservice=128, ssc=26, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "7":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Downlink pwr off"))
+        command = object_id + struct.pack('!I', CommandIds.TC_DOWNLINK_PWR_OFF)
+        command = PusTelecommand(service=8, subservice=128, ssc=26, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
+    elif op_code == "8":
+        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Replay write sequence"))
+        command = prepare_replay_write_sequence_cmd(object_id)
+        command = PusTelecommand(service=8, subservice=128, ssc=27, app_data=command)
+        tc_queue.appendleft(command.pack_command_tuple())
 
     return tc_queue
 
 
 def generate_write_mem_command(
-    object_id: bytearray, memory_address: bytes, memory_data: int
+        object_id: bytearray, memory_address: int, memory_data: int, mem_len: 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
@@ -71,9 +120,76 @@ def generate_write_mem_command(
     @param memory_data    The data to write to the memory address specified by the bytearray memory_address.
     """
     command = (
-        object_id
+            object_id
-        + PlocActionIds.tc_mem_write
+            + struct.pack('!I', CommandIds.TC_MEM_WRITE)
-        + memory_address
+            + struct.pack('!I', memory_address)
-        + struct.pack("!I", memory_data)
+            + struct.pack('!H', mem_len)
+            + struct.pack("!I", memory_data)
     )
     return command
+
+
+def prepare_mem_read_command(object_id: bytearray) -> bytearray:
+    memory_address = int(input("PLOC MPSoC Tc Mem Read: Type memory address: 0x"), 16)
+    num_words = int(input("PLOC MPSoC specify number of words (32-bit) to read: "))
+    command = (
+            object_id + struct.pack('!I', CommandIds.TC_MEM_READ) + struct.pack("!I", memory_address) + struct.pack(
+        '!H', num_words)
+    )
+    return command
+
+
+def prepare_flash_write_cmd(object_id: bytearray) -> bytearray:
+    file = get_flash_write_file()
+    command = object_id + struct.pack('!I', CommandIds.FLASH_WRITE) + bytearray(file, 'utf-8')
+    return command
+
+
+def prepare_flash_delete_cmd(object_id: bytearray) -> bytearray:
+    file = get_mpsoc_file()
+    command = object_id + struct.pack('!I', CommandIds.TC_FLASH_DELETE) + bytearray(file, 'utf-8')
+    return command
+
+
+def prepare_replay_start_cmd(object_id: bytearray) -> bytearray:
+    replay = int(input("Specify replay mode (0 - repeated, 1 - once): "))
+    command = object_id + struct.pack('!I', CommandIds.TC_REPLAY_START) + struct.pack('!B', replay)
+    return command
+
+
+def prepare_downlink_pwr_on_cmd(object_id: bytearray) -> bytearray:
+    mode = int(input("Specify JESD mode (0 - 5): "))
+    lane_rate = int(input("Specify lane rate (0 - 9): "))
+    command = object_id + struct.pack('!I', CommandIds.TC_DOWNLINK_PWR_ON) + struct.pack('!B', mode) \
+              + struct.pack('!B', lane_rate)
+    return command
+
+
+def prepare_replay_write_sequence_cmd(object_id: bytearray) -> bytearray:
+    use_decoding = int(input("Use decoding  (set to 1): "))
+    file = get_mpsoc_file()
+    command = object_id + struct.pack('!I', CommandIds.TC_REPLAY_WRITE_SEQUENCE) + struct.pack('!B', use_decoding) + \
+              bytearray(file, 'utf-8')
+    return command
+
+
+def get_flash_write_file() -> str:
+    LOGGER.info("Specify flash file")
+    input_helper = InputHelper(flash_write_file_dict)
+    key = input_helper.get_key()
+    if key == MANUAL_INPUT:
+        file = input("Ploc MPSoC: Specify absolute name of flash file: ")
+    else:
+        file = flash_write_file_dict[key][1]
+    return file
+
+
+def get_mpsoc_file() -> str:
+    LOGGER.info("Specify mpsoc file")
+    input_helper = InputHelper(mpsoc_file_dict)
+    key = input_helper.get_key()
+    if key == MANUAL_INPUT:
+        file = input("Ploc MPSoC: Specify absolute name file: ")
+    else:
+        file = mpsoc_file_dict[key][1]
+    return file

pus_tc/devs/ploc_supervisor.py

@@ -75,7 +75,7 @@ class SupvHkIds:
     BOOT_STATUS_REPORT = 53
 
 
-def pack_ploc_supv_test_into(
+def pack_ploc_supv_commands(
     object_id: bytearray, tc_queue: TcQueueT, op_code: str
 ) -> TcQueueT:
     tc_queue.appendleft(

pus_tc/devs/rad_sensor.py

@@ -5,7 +5,6 @@
 @author J. Meier
 @date   01.07.2021
 """
-import struct
 from tmtccmd.config.definitions import QueueCommands
 
 from tmtccmd.tc.packer import TcQueueT

pus_tc/devs/star_tracker.py

@@ -15,6 +15,7 @@ from tmtccmd.tc.service_200_mode import pack_mode_data, Modes
 from tmtccmd.utility.logger import get_console_logger
 from utility.input_helper import InputHelper
 
+
 LOGGER = get_console_logger()
 
 

pus_tc/tc_packer_hook.py

@@ -19,8 +19,8 @@ from pus_tc.devs.acu import pack_acu_test_into
 from pus_tc.devs.solar_array_deployment import pack_solar_array_deployment_test_into
 from pus_tc.devs.imtq import pack_imtq_test_into
 from pus_tc.devs.tmp1075 import pack_tmp1075_test_into
-from pus_tc.devs.ploc_mpsoc import pack_ploc_mpsoc_test_into
+from pus_tc.devs.ploc_mpsoc import pack_ploc_mpsoc_commands
-from pus_tc.devs.ploc_supervisor import pack_ploc_supv_test_into
+from pus_tc.devs.ploc_supervisor import pack_ploc_supv_commands
 from pus_tc.devs.heater import pack_heater_test_into
 from pus_tc.devs.reaction_wheels import pack_single_rw_test_into
 from pus_tc.devs.rad_sensor import pack_rad_sensor_test_into
@@ -118,9 +118,9 @@ def pack_service_queue_user(
         return pack_imtq_test_into(
             object_id=object_id, tc_queue=service_queue, op_code=op_code
         )
-    if service == CustomServiceList.PLOC.value:
+    if service == CustomServiceList.PLOC_MPSOC.value:
         object_id = PLOC_MPSOC_ID
-        return pack_ploc_mpsoc_test_into(object_id=object_id, tc_queue=service_queue)
+        return pack_ploc_mpsoc_commands(object_id=object_id, tc_queue=service_queue, op_code=op_code)
     if service == CustomServiceList.REACTION_WHEEL_1.value:
         object_id = RW1_ID
         return pack_single_rw_test_into(
@@ -148,7 +148,7 @@ def pack_service_queue_user(
         )
     if service == CustomServiceList.PLOC_SUPV.value:
         object_id = PLOC_SUPV_ID
-        return pack_ploc_supv_test_into(
+        return pack_ploc_supv_commands(
             object_id=object_id, tc_queue=service_queue, op_code=op_code
         )
     if service == CustomServiceList.PLOC_UPDATER.value:

pus_tm/service_8_hook.py

@@ -69,7 +69,8 @@ def handle_ploc_replies(action_id: int, custom_data: bytearray) -> DataReplyUnpa
             "PLOC Mem Len",
             "PLOC Read Memory Data",
         ]
-        reply.content_list = [custom_data[:4], custom_data[4:6], custom_data[6:10]]
+        reply.content_list = ["0x" + custom_data[:4].hex(), struct.unpack('!H', custom_data[4:6])[0],
+                              "0x" + custom_data[6:10].hex()]
     return reply