bump tmtc to v3.1.0

CHANGELOG.md

@@ -10,6 +10,12 @@ list yields a list of all related PRs for each release.
 
 # [unreleased]
 
+# [v3.1.0] 2023-04-16
+
+## Added
+
+- New core controller file system helper commands.
+
 # [v3.0.0] 2023-04-14
 
 ## Fixed

eive_tmtc/__init__.py

@@ -1,11 +1,11 @@
-__version__ = "3.0.0"
+__version__ = "3.1.0"
 
 import logging
 from pathlib import Path
 
 SW_NAME = "eive-tmtc"
 VERSION_MAJOR = 3
-VERSION_MINOR = 0
+VERSION_MINOR = 1
 VERSION_REVISION = 0
 
 EIVE_TMTC_ROOT = Path(__file__).parent

eive_tmtc/pus_tm/action_reply_handler.py

@@ -2,6 +2,7 @@ import struct
 from eive_tmtc.config.object_ids import *
 from eive_tmtc.tmtc.acs.imtq import ImtqActionId
 from eive_tmtc.pus_tm.defs import PrintWrapper
+from eive_tmtc.tmtc.core import handle_core_ctrl_action_replies
 from eive_tmtc.tmtc.payload.ploc_mpsoc import PlocReplyIds
 from eive_tmtc.tmtc.payload.ploc_supervisor import SupvActionId
 from eive_tmtc.tmtc.acs.star_tracker import StarTrackerActionId
@@ -37,6 +38,8 @@ def handle_action_reply(
         return handle_ploc_replies(action_id, printer, custom_data)
     elif object_id.as_bytes == PLOC_SUPV_ID:
         return handle_supervisor_replies(action_id, printer, custom_data)
+    elif object_id.as_bytes == CORE_CONTROLLER_ID:
+        return handle_core_ctrl_action_replies(action_id, printer, custom_data)
     elif object_id.as_bytes == STAR_TRACKER_ID:
         return handle_startracker_replies(action_id, printer, custom_data)
     elif object_id.as_bytes in [

eive_tmtc/tmtc/core.py

@@ -1,6 +1,8 @@
 import enum
 import logging
+import os
 import struct
+from pathlib import Path
 
 from eive_tmtc.pus_tm.defs import PrintWrapper
 
@@ -23,7 +25,6 @@ _LOGGER = logging.getLogger(__name__)
 
 
 class ActionId(enum.IntEnum):
-    LIST_DIR_INTO_FILE = 0
     ANNOUNCE_VERSION = 1
     ANNOUNCE_CURRENT_IMAGE = 2
     ANNOUNCE_BOOT_COUNTS = 3
@@ -42,6 +43,12 @@ class ActionId(enum.IntEnum):
     EXECUTE_SHELL_CMD_BLOCKING = 40
     EXECUTE_SHELL_CMD_NON_BLOCKING = 41
     SYSTEMCTL_CMD_EXECUTOR = 42
+    LIST_DIR_INTO_FILE = 50
+    LIST_DIR_DUMP_DIRECTLY = 51
+    CP_HELPER = 52
+    MV_HELPER = 53
+    RM_HELPER = 54
+    MKDIR_HELPER = 55
 
 
 class ParamId(enum.IntEnum):
@@ -59,6 +66,12 @@ class OpCode:
     EXECUTE_SHELL_CMD_BLOCKING = "exec_cmd_blocking"
     EXECUTE_SHELL_CMD_NON_BLOCKING = "exec_cmd_non_blocking"
     SYSTEMCTL_CMD_EXECUTOR = "systemctl_cmd"
+    LIST_DIR_INTO_FILE = "list_dir_into_file"
+    LIST_DIR_DUMP_DIRECTLY = "list_dir_dump_directly"
+    CP_HELPER = "cp_helper"
+    MV_HELPER = "mv_helper"
+    RM_HELPER = "rm_helper"
+    MKDIR_HELPER = "mkdir_helper"
     SET_PREF_SD = "set_pref_sd"
     REBOOT_XSC = ["reboot_xsc"]
     XSC_REBOOT_SELF = ["reboot_self"]
@@ -100,6 +113,12 @@ class Info:
     SWITCH_TO_SD_0 = "Switch to SD card 0"
     SWITCH_TO_SD_1 = "Switch to SD card 1"
     SWITCH_TO_BOTH_SD_CARDS = "Switch to both SD cards with specified active card"
+    LIST_DIR_INTO_FILE = "List directory, dump output into file"
+    LIST_DIR_DUMP_DIRECTLY = "List directory, dump content directly"
+    CP_HELPER = "Filesystem Copy Helper"
+    MV_HELPER = "Filesystem Move Helper"
+    RM_HELPER = "Filesystem Removal Helper"
+    MKDIR_HELPER = "Filesystem Directory Creation Helper"
 
 
 class Chip(enum.IntEnum):
@@ -184,6 +203,12 @@ def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
     oce.add(keys=OpCode.SWITCH_TO_SD_0, info=Info.SWITCH_TO_SD_0)
     oce.add(keys=OpCode.SWITCH_TO_SD_1, info=Info.SWITCH_TO_SD_1)
     oce.add(keys=OpCode.SWITCH_TO_BOTH_SD_CARDS, info=Info.SWITCH_TO_BOTH_SD_CARDS)
+    oce.add(keys=OpCode.LIST_DIR_INTO_FILE, info=Info.LIST_DIR_INTO_FILE)
+    oce.add(keys=OpCode.LIST_DIR_DUMP_DIRECTLY, info=Info.LIST_DIR_DUMP_DIRECTLY)
+    oce.add(keys=OpCode.MV_HELPER, info=Info.MV_HELPER)
+    oce.add(keys=OpCode.CP_HELPER, info=Info.CP_HELPER)
+    oce.add(keys=OpCode.RM_HELPER, info=Info.RM_HELPER)
+    oce.add(keys=OpCode.MKDIR_HELPER, info=Info.MKDIR_HELPER)
     defs.add_service(CustomServiceList.CORE.value, "Core Controller", oce)
 
 
@@ -369,12 +394,97 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
                 ).pack()
             )
         )
+    elif op_code == OpCode.CP_HELPER:
+        cp_recursive = int(input("Copy recursively (0/1) ?: "))
+        if cp_recursive not in [0, 1]:
+            raise ValueError("Invalid value, only 0 or 1 allowed")
+        user_data = bytearray([cp_recursive])
+        user_data.extend(packet_source_dest_path("Copy"))
+        q.add_log_cmd(Info.CP_HELPER)
+        q.add_pus_tc(
+            create_action_cmd(CORE_CONTROLLER_ID, ActionId.CP_HELPER, user_data)
+        )
+    elif op_code == OpCode.MV_HELPER:
+        user_data = packet_source_dest_path("Move")
+        q.add_log_cmd(Info.MV_HELPER)
+        q.add_pus_tc(
+            create_action_cmd(CORE_CONTROLLER_ID, ActionId.MV_HELPER, user_data)
+        )
+    elif op_code == OpCode.RM_HELPER:
+        rm_recursive = int(input("Remove with recursive (-r) option (0/1) ?: "))
+        if rm_recursive not in [0, 1]:
+            raise ValueError("Invalid value, only 0 or 1 allowed")
+        rm_force = int(input("Remove with force (-f) option (0/1) ?: "))
+        if rm_force not in [0, 1]:
+            raise ValueError("Invalid value, only 0 or 1 allowed")
+        user_data = bytearray([rm_recursive, rm_force])
+        removed_file_or_dir = input("Specify absolute path to be removed: ")
+        user_data.extend(removed_file_or_dir.encode())
+        user_data.append(0)
+        q.add_log_cmd(Info.RM_HELPER)
+        q.add_pus_tc(
+            create_action_cmd(CORE_CONTROLLER_ID, ActionId.RM_HELPER, user_data)
+        )
+    elif op_code == OpCode.LIST_DIR_INTO_FILE:
+        q.add_log_cmd(Info.LIST_DIR_INTO_FILE)
+        user_data = list_directory_base_user_data()
+        dest_file_path = input("Destination file path: ")
+        user_data.extend(dest_file_path.encode())
+        user_data.append(0)
+        q.add_pus_tc(
+            create_action_cmd(
+                CORE_CONTROLLER_ID, ActionId.LIST_DIR_INTO_FILE, user_data
+            )
+        )
+    elif op_code == OpCode.LIST_DIR_DUMP_DIRECTLY:
+        q.add_log_cmd(Info.LIST_DIR_DUMP_DIRECTLY)
+        user_data = list_directory_base_user_data()
+        q.add_pus_tc(
+            create_action_cmd(
+                CORE_CONTROLLER_ID, ActionId.LIST_DIR_DUMP_DIRECTLY, user_data
+            )
+        )
+    elif op_code == OpCode.MKDIR_HELPER:
+        q.add_log_cmd(Info.MKDIR_HELPER)
+        user_data = input("Specify absolute path of newly created directory: ")
+        user_data = bytearray(user_data.encode())
+        user_data.append(0)
+        q.add_pus_tc(
+            create_action_cmd(CORE_CONTROLLER_ID, ActionId.MKDIR_HELPER, user_data)
+        )
     else:
         _LOGGER.warning(
             f"Unknown operation code {op_code} for core controller commands"
         )
 
 
+def list_directory_base_user_data() -> bytearray:
+    all_opt = int(input("Use all (-a) option (0/1) ?: "))
+    if all_opt not in [0, 1]:
+        raise ValueError("Invalid value, only 0 or 1 allowed")
+    recursive_opt = int(input("Use recursive (-R) option (0/1) ?: "))
+    if recursive_opt not in [0, 1]:
+        raise ValueError("Invalid value, only 0 or 1 allowed")
+    compression_opt = int(input("Compress target file (0/1) ?: "))
+    if compression_opt not in [0, 1]:
+        raise ValueError("Invalid value, only 0 or 1 allowed")
+    listing_path = input("Specify listing path (absolute path): ")
+    user_data = bytearray([all_opt, recursive_opt, compression_opt])
+    user_data.extend(listing_path.encode())
+    user_data.append(0)
+    return user_data
+
+
+def packet_source_dest_path(context: str) -> bytes:
+    source = input(f"Specify {context} source file: ")
+    dest = input(f"Specify {context} destination file: ")
+    raw_src_dest = bytearray(source.encode())
+    raw_src_dest.append(0)
+    raw_src_dest.extend(dest.encode())
+    raw_src_dest.append(0)
+    return raw_src_dest
+
+
 def reset_specific_boot_counter(q: DefaultPusQueueHelper, chip: int, copy: int):
     q.add_log_cmd(f"Resetting boot counter {chip} {copy}")
     q.add_pus_tc(
@@ -483,3 +593,49 @@ def handle_core_hk_data(printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes):
         )
         pw.dlog(printout)
         printer.print_validity_buffer(validity_buffer=hk_data[inc_len:], num_vars=3)
+
+
+def handle_core_ctrl_action_replies(
+    action_id: int, printer: FsfwTmTcPrinter, custom_data: bytes
+):
+    pw = PrintWrapper(printer)
+    if action_id == ActionId.LIST_DIR_DUMP_DIRECTLY:
+        if len(custom_data) < 4:
+            _LOGGER.warning("Data unexpectedly small")
+            return
+        seq_idx = struct.unpack("!I", custom_data[0:4])[0]
+        total_chunks = struct.unpack("!I", custom_data[4:8])[0]
+        compressed = custom_data[8]
+        ls_cmd = custom_data[9:].split(b"\x00")[0].decode()
+        # Include length of NULL termination
+        file_data_offset = 9 + len(ls_cmd) + 1
+        pw.dlog(
+            f"Received directory listing dump for ls command {ls_cmd}. "
+            f"Chunk {seq_idx + 1}/{total_chunks}"
+        )
+
+        def remove_if_exists_and_new(seq_idx_: int, path_: Path):
+            if seq_idx_ == 0 and path_.exists():
+                os.remove(path_)
+
+        if compressed:
+            path = Path("dir_listing.txt.gz")
+            remove_if_exists_and_new(seq_idx, path)
+            pw.dlog(
+                f"Compression option: {compressed}. Dumping file into dir_listing.txt.gz"
+            )
+            with open(path, "ab") as listing_file:
+                listing_file.write(custom_data[file_data_offset:])
+        else:
+            path = Path("dir_listing.txt")
+            remove_if_exists_and_new(seq_idx, path)
+            pw.dlog(
+                f"Compression option: {compressed}. Dumping file into dir_listing.txt"
+            )
+            with open(path, "a") as listing_file:
+                listing_file_str = custom_data[file_data_offset:].decode()
+                listing_file.write(listing_file_str)
+            if seq_idx + 1 == total_chunks:
+                pw.dlog("Full directory listing: ")
+                with open("dir_listing.txt", "r") as listing_file:
+                    print(listing_file.read())