new commands for PL PCDU mode commanding

2022-03-30 12:08:50 +02:00
parent e37430423e
commit b5a9dac8f6
4 changed files with 173 additions and 80 deletions
--- a/pus_tc/devs/ploc_mpsoc.py
+++ b/pus_tc/devs/ploc_mpsoc.py
@ -58,7 +58,7 @@ class PlocReplyIds:


 def pack_ploc_mpsoc_commands(
-        object_id: bytearray, tc_queue: TcQueueT, op_code: str
+    object_id: bytearray, tc_queue: TcQueueT, op_code: str
 ) -> TcQueueT:
    tc_queue.appendleft(
        (
@ -69,7 +69,9 @@ def pack_ploc_mpsoc_commands(

    if op_code == "0":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: TC mem write test"))
-        memory_address = int(input("PLOC MPSoC: 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 MPSoC: Tc Mem Write: Type memory data: 0x"), 16)
        # TODO: implement variable length mem write command
        mem_len = 1  # 1 32-bit word
@ -100,7 +102,7 @@ def pack_ploc_mpsoc_commands(
        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 = 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":
@ -110,7 +112,7 @@ def pack_ploc_mpsoc_commands(
        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 = 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":
@ -119,20 +121,26 @@ def pack_ploc_mpsoc_commands(
        command = PusTelecommand(service=8, subservice=128, ssc=27, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "9":
-        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Reset OBSW sequence count"))
-        command = object_id + struct.pack('!I', CommandIds.OBSW_RESET_SEQ_COUNT)
+        tc_queue.appendleft(
+            (QueueCommands.PRINT, "PLOC MPSoC: Reset OBSW sequence count")
+        )
+        command = object_id + struct.pack("!I", CommandIds.OBSW_RESET_SEQ_COUNT)
        command = PusTelecommand(service=8, subservice=128, ssc=28, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "10":
        num_words = 1
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Read DEADBEEF address"))
-        command = object_id + struct.pack('!I', CommandIds.TC_MEM_READ) + struct.pack("!I", MemAddresses.DEADBEEF) + \
-                  struct.pack('!H', num_words)
+        command = (
+            object_id
+            + struct.pack("!I", CommandIds.TC_MEM_READ)
+            + struct.pack("!I", MemAddresses.DEADBEEF)
+            + struct.pack("!H", num_words)
+        )
        command = PusTelecommand(service=8, subservice=128, ssc=29, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())
    elif op_code == "11":
        tc_queue.appendleft((QueueCommands.PRINT, "PLOC MPSoC: Tc mode replay"))
-        command = object_id + struct.pack('!I', CommandIds.TC_MODE_REPLAY)
+        command = object_id + struct.pack("!I", CommandIds.TC_MODE_REPLAY)
        command = PusTelecommand(service=8, subservice=128, ssc=30, app_data=command)
        tc_queue.appendleft(command.pack_command_tuple())

@ -140,7 +148,7 @@ def pack_ploc_mpsoc_commands(


 def generate_write_mem_command(
-        object_id: bytearray, memory_address: int, memory_data: int, mem_len: 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
@ -148,11 +156,11 @@ 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
-            + struct.pack('!I', CommandIds.TC_MEM_WRITE)
-            + struct.pack('!I', memory_address)
-            + struct.pack('!H', mem_len)
-            + struct.pack("!I", memory_data)
+        object_id
+        + struct.pack("!I", CommandIds.TC_MEM_WRITE)
+        + struct.pack("!I", memory_address)
+        + struct.pack("!H", mem_len)
+        + struct.pack("!I", memory_data)
    )
    return command

@ -161,8 +169,10 @@ 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)
+        object_id
+        + struct.pack("!I", CommandIds.TC_MEM_READ)
+        + struct.pack("!I", memory_address)
+        + struct.pack("!H", num_words)
    )
    return command

@ -170,28 +180,44 @@ def prepare_mem_read_command(object_id: bytearray) -> bytearray:
 def prepare_flash_write_cmd(object_id: bytearray) -> bytearray:
    obcFile = get_obc_file()
    mpsocFile = get_mpsoc_file()
-    command = object_id + struct.pack('!I', CommandIds.FLASH_WRITE) + bytearray(obcFile, 'utf-8') + bytearray(mpsocFile,
-                                                                                                              'utf-8')
+    command = (
+        object_id
+        + struct.pack("!I", CommandIds.FLASH_WRITE)
+        + bytearray(obcFile, "utf-8")
+        + bytearray(mpsocFile, "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')
+    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 - once, 1 - repeated): "))
-    command = object_id + struct.pack('!I', CommandIds.TC_REPLAY_START) + struct.pack('!B', replay)
+    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)
+    command = (
+        object_id
+        + struct.pack("!I", CommandIds.TC_DOWNLINK_PWR_ON)
+        + struct.pack("!B", mode)
+        + struct.pack("!B", lane_rate)
+    )
    return command


@ -199,8 +225,12 @@ def prepare_replay_write_sequence_cmd(object_id: bytearray) -> bytearray:
    null_terminator = 0
    use_decoding = int(input("Use decoding  (set to 1): "))
    file = get_sequence_file()
-    command = object_id + struct.pack('!I', CommandIds.TC_REPLAY_WRITE_SEQUENCE) + struct.pack('!B', use_decoding) + \
-              bytearray(file, 'utf-8')
+    command = (
+        object_id
+        + struct.pack("!I", CommandIds.TC_REPLAY_WRITE_SEQUENCE)
+        + struct.pack("!B", use_decoding)
+        + bytearray(file, "utf-8")
+    )
    return command