add to exposed commands

CHANGELOG.md

@@ -10,6 +10,10 @@ list yields a list of all related PRs for each release.
 
 # [unreleased]
 
+## Added
+
+- Added core commands to execute `systemctl` commands and to execute arbitrary Linux commands.
+
 # [v2.22.1] 2023-04-12
 
 ## Added

eive_tmtc/tmtc/core.py

@@ -39,6 +39,9 @@ class ActionId(enum.IntEnum):
     SWITCH_TO_BOTH_SD_CARDS = 18
     XSC_REBOOT = 32
     FULL_REBOOT = 34
+    EXECUTE_SHELL_CMD_BLOCKING = 40
+    EXECUTE_SHELL_CMD_NON_BLOCKING = 41
+    SYSTEMCTL_CMD_EXECUTOR = 42
 
 
 class ParamId(enum.IntEnum):
@@ -53,6 +56,9 @@ class OpCode:
     ANNOUNCE_VERSION = "announce_version"
     ANNOUNCE_CURRENT_IMAGE = "announce_current_image"
     ANNOUNCE_BOOT_COUNTS = "announce_boot_counts"
+    EXECUTE_SHELL_CMD_BLOCKING = "exec_cmd_blocking"
+    EXECUTE_SHELL_CMD_NON_BLOCKING = "exec_cmd_non_blocking"
+    SYSTEMCTL_CMD_EXECUTOR = "systemctl_cmd"
     SET_PREF_SD = "set_pref_sd"
     REBOOT_XSC = ["reboot_xsc"]
     XSC_REBOOT_SELF = ["reboot_self"]
@@ -82,6 +88,9 @@ class Info:
     ANNOUNCE_VERSION = "Announce version"
     ANNOUNCE_CURRENT_IMAGE = "Announce current image"
     ANNOUNCE_BOOT_COUNTS = "Announce boot counts"
+    SYSTEMCTL_CMD_EXECUTOR = "Perform systemctl command"
+    EXECUTE_SHELL_CMD_BLOCKING = "Execute shell command blocking"
+    EXECUTE_SHELL_CMD_NON_BLOCKING = "Execute shell command non-blocking"
     SET_PREF_SD = "Set preferred SD card"
     REBOOT_XSC = "XSC reboot with prompt"
     REBOOT_FULL = "Full regular reboot"
@@ -105,6 +114,12 @@ class Copy(enum.IntEnum):
     NONE = 2
 
 
+class SystemctlCmd(enum.IntEnum):
+    START = 0
+    STOP = 1
+    RESTART = 2
+
+
 @tmtc_definitions_provider
 def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
     oce = OpCodeEntry()
@@ -123,6 +138,9 @@ def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
     oce.add(keys=OpCode.OBSW_UPDATE_FROM_TMP, info=Info.OBSW_UPDATE_FROM_TMP)
     oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_0, info=Info.OBSW_UPDATE_FROM_SD_0)
     oce.add(keys=OpCode.OBSW_UPDATE_FROM_SD_1, info=Info.OBSW_UPDATE_FROM_SD_1)
+    oce.add(keys=OpCode.SYSTEMCTL_CMD_EXECUTOR, info=Info.SYSTEMCTL_CMD_EXECUTOR)
+    oce.add(keys=OpCode.EXECUTE_SHELL_CMD_BLOCKING, info=Info.EXECUTE_SHELL_CMD_BLOCKING)
+    oce.add(keys=OpCode.EXECUTE_SHELL_CMD_NON_BLOCKING, info=Info.EXECUTE_SHELL_CMD_NON_BLOCKING)
     oce.add(
         keys=OpCode.GET_HK,
         info="Request housekeeping set",
@@ -195,6 +213,33 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
         )
     elif op_code in OpCode.XSC_REBOOT_SELF:
         add_xsc_reboot_cmd(q=q, reboot_self=True)
+    elif op_code == OpCode.SYSTEMCTL_CMD_EXECUTOR:
+        print("systemctl command types: ")
+        for entry in SystemctlCmd:
+            print(f"{entry}: {entry.name}")
+        systemctl_cmd = SystemctlCmd(int(input("Specify systemctl command type by key: ")))
+        unit_name = input("Specify unit name: ")
+        cmd_data = bytearray([systemctl_cmd])
+        cmd_data.extend(unit_name.encode())
+        q.add_pus_tc(create_action_cmd(
+            object_id=CORE_CONTROLLER_ID,
+            action_id=ActionId.SYSTEMCTL_CMD_EXECUTOR,
+            user_data=cmd_data
+        ))
+    elif op_code == OpCode.EXECUTE_SHELL_CMD_BLOCKING:
+        custom_cmd = input("Please specify command to execute: ")
+        q.add_pus_tc(create_action_cmd(
+            object_id=CORE_CONTROLLER_ID,
+            action_id=ActionId.EXECUTE_SHELL_CMD_BLOCKING,
+            user_data=custom_cmd.encode()
+        ))
+    elif op_code == OpCode.EXECUTE_SHELL_CMD_NON_BLOCKING:
+        custom_cmd = input("Please specify command to execute: ")
+        q.add_pus_tc(create_action_cmd(
+            object_id=CORE_CONTROLLER_ID,
+            action_id=ActionId.EXECUTE_SHELL_CMD_NON_BLOCKING,
+            user_data=custom_cmd.encode()
+        ))
     elif op_code in OpCode.XSC_REBOOT_0_0:
         add_xsc_reboot_cmd(
             q=q, reboot_self=False, chip=Chip.CHIP_0, copy=Copy.COPY_0_NOM