STR bugfixes

CHANGELOG.md

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

eive_tmtc/tmtc/acs/star_tracker.py

@@ -103,6 +103,7 @@ class OpCodes:
     TAKE_IMAGE = "take_image"
     UPLOAD_IMAGE = "upload_image"
     SET_IMG_PROCESSOR_MODE = "set_img_proc_mode"
+    FW_UPDATE = "fw_update"
 
 
 class Info:
@@ -111,6 +112,7 @@ class Info:
     UPLOAD_IMAGE = "Upload Image"
     TAKE_IMAGE = "Take Image"
     SET_IMG_PROCESSOR_MODE = "Set Image Processor Mode"
+    FW_UPDATE = "Firmware Update"
 
 
 class SetId(enum.IntEnum):
@@ -141,6 +143,7 @@ class FileDefs:
     firmware2_1 = "/home/pi/arcsec/firmware/sagitta-2-1.bin"
     firmware22_1 = "/home/pi/arcsec/firmware/sagitta-22-1.bin"
     firmware_origin = "/home/pi/arcsec/firmware/sagitta-origin.bin"
+    FW_SLOT_Q7S = "/mnt/sd0/startracker/updates/sagitta-update.bin"
 
 
 json_dict = {
@@ -158,13 +161,15 @@ json_dict = {
     ),
 }
 
-firmware_dict = {
+FW_DICT = {
+    "0": ("Firmware Update Q7S", FileDefs.FW_SLOT_Q7S),
     "1": ("Firmware Major = 2, Minor = 1", FileDefs.firmware2_1),
     "2": ("Firmware Major = 22, Minor = 1", FileDefs.firmware22_1),
     "3": ("Firmware Origin", FileDefs.firmware_origin),
 }
 
-upload_image_dict = {
+UPLOAD_IMAGE_DICT = {
+    "0": ("custom path", "Custom Path"),
     "1": ("q7s gemma", "/mnt/sd0/startracker/gemma.bin"),
     "2": ("egse gemma", "/home/pi/arcsec/star-images/gemma.bin"),
     "3": ("q7s polaris", "/mnt/sd0/startracker/polaris.bin"),
@@ -630,13 +635,13 @@ def pack_star_tracker_commands(
             + bytearray(json_file, "utf-8")
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
-    if op_code == "69":
-        q.add_log_cmd("Star tracker: Firmware update")
+    if op_code == OpCodes.FW_UPDATE:
+        q.add_log_cmd(Info.FW_UPDATE)
         firmware = get_firmware()
         data = (
             obyt
             + struct.pack("!I", StarTrackerActionId.FIRMWARE_UPDATE)
-            + bytearray(firmware, "utf-8")
+            + firmware.encode()
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
 
@@ -681,17 +686,24 @@ def get_config_file() -> str:
 
 def get_firmware() -> str:
     _LOGGER.info("Specify firmware file")
-    input_helper = InputHelper(firmware_dict)
-    key = input_helper.get_key()
-    firmware = firmware_dict[key][1]
+    bin_select = int(input("Use hardcoded paths (0) or specify path manually (1) ?: "))
+    if bin_select == 0:
+        input_helper = InputHelper(FW_DICT)
+        key = input_helper.get_key()
+        firmware = FW_DICT[key][1]
+    else:
+        firmware = input("Specify absolute path of the firmware update file: ")
     return firmware
 
 
 def get_upload_image() -> str:
     _LOGGER.info("Specify image to upload")
-    input_helper = InputHelper(upload_image_dict)
+    input_helper = InputHelper(UPLOAD_IMAGE_DICT)
     key = input_helper.get_key()
-    image = upload_image_dict[key][1]
+    if UPLOAD_IMAGE_DICT[key][0] == "custom path":
+        image = input("Please specify custom absolute path: ")
+    else:
+        image = UPLOAD_IMAGE_DICT[key][1]
     return image
 
 
@@ -797,8 +809,11 @@ def handle_solution_set(hk_data: bytes, pw: PrintWrapper):
     )
     pw.dlog(f"Number of close stars in LISA solution: {lisa_number_close_stars}")
     current_idx += fmt_len
-    is_trusworthy = hk_data[current_idx]
-    pw.dlog(f"Trustworthy solution: {is_trusworthy}")
+    str_mode = hk_data[current_idx]
+    pw.dlog(f"STR mode: {str_mode}")
+    current_idx += 1
+    is_trustworthy = hk_data[current_idx]
+    pw.dlog(f"Trustworthy solution: {is_trustworthy}")
     current_idx += 1
     stable_count = struct.unpack("!I", hk_data[current_idx : current_idx + 4])[0]
     pw.dlog(f"Stable count: {stable_count}")
@@ -870,7 +885,7 @@ def add_str_cmds(defs: TmtcDefinitionWrapper):
     oce.add("66", "Star Tracker: Set log level parameters")
     oce.add("67", "Star Tracker: Set log subscription parameters")
     oce.add("68", "Star Tracker: Set debug camera parameters")
-    oce.add("69", "Star Tracker: Firmware update")
+    oce.add(OpCodes.FW_UPDATE, Info.FW_UPDATE)
     oce.add("70", "Star Tracker: Disable timestamp generation")
     oce.add("71", "Star Tracker: Enable timestamp generation")
     defs.add_service(CustomServiceList.STAR_TRACKER.value, "Star Tracker", oce)

eive_tmtc/tmtc/core.py

@@ -11,6 +11,10 @@ from tmtccmd.config import TmtcDefinitionWrapper
 from tmtccmd.tc import DefaultPusQueueHelper
 from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd
 from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_hk_command
+from tmtccmd.tc.pus_20_fsfw_param import (
+    create_scalar_u8_parameter,
+    create_load_param_cmd,
+)
 from tmtccmd.config.tmtc import OpCodeEntry, tmtc_definitions_provider
 from eive_tmtc.config.object_ids import CORE_CONTROLLER_ID
 from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
@@ -35,6 +39,13 @@ 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):
+    PREF_SD = 0
 
 
 class SetId(enum.IntEnum):
@@ -45,6 +56,10 @@ 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"]
     XSC_REBOOT_0_0 = ["reboot_00"]
@@ -73,6 +88,10 @@ 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"
     OBSW_UPDATE_FROM_SD_0 = "Update OBSW from SD Card 0"
@@ -95,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()
@@ -109,9 +134,13 @@ def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
     oce.add(keys=OpCode.XSC_REBOOT_0_1, info="Reboot 0 1")
     oce.add(keys=OpCode.XSC_REBOOT_1_0, info="Reboot 1 0")
     oce.add(keys=OpCode.XSC_REBOOT_1_1, info="Reboot 1 1")
+    oce.add(keys=OpCode.SET_PREF_SD, info=Info.SET_PREF_SD)
     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",
@@ -184,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
@@ -283,6 +339,23 @@ def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
         q.add_log_cmd("Requesting housekeeping set")
         sid = make_sid(object_id=CORE_CONTROLLER_ID, set_id=SetId.HK)
         q.add_pus_tc(generate_one_hk_command(sid))
+    elif op_code in OpCode.SET_PREF_SD:
+        q.add_log_cmd("Set preferred SD card")
+        pref_sd = int(
+            input("Specify which SD card to set as the preferred one (0/1): ")
+        )
+        if pref_sd not in [0, 1]:
+            raise ValueError("Only 0 or 1 allowed for preferred SD card")
+        q.add_pus_tc(
+            create_load_param_cmd(
+                create_scalar_u8_parameter(
+                    object_id=CORE_CONTROLLER_ID,
+                    domain_id=0,
+                    unique_id=ParamId.PREF_SD,
+                    parameter=pref_sd,
+                ).pack()
+            )
+        )
     else:
         _LOGGER.warning(
             f"Unknown operation code {op_code} for core controller commands"