Merge pull request 'RW commanding fix' (#277) from rw-commanding-fix into main

Reviewed-on: #277

CHANGELOG.md

@@ -10,7 +10,15 @@ list yields a list of all related PRs for each release.
 
 # [unreleased]
 
-# [v6.0.0] 2024-01-31
+## Added
+
+- Added commands to unlock and use STR secondary firmware slot.
+
+## Fixed
+
+- RW commanding
+
+# [v6.0.0] 2024-02-08
 
 ## Changed
 

eive_tmtc/pus_tc/cmd_demux.py

@@ -70,6 +70,7 @@ from eive_tmtc.tmtc.tcs.subsystem import pack_tcs_sys_commands
 from eive_tmtc.tmtc.tcs.tmp1075 import pack_tmp1075_test_into
 from eive_tmtc.tmtc.test import build_test_commands
 from eive_tmtc.tmtc.time import pack_time_management_cmd
+from eive_tmtc.tmtc.tm_store import pack_tm_store_commands
 from eive_tmtc.tmtc.wdt import pack_wdt_commands
 from eive_tmtc.utility.input_helper import InputHelper
 
@@ -197,29 +198,29 @@ def handle_acs_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: Lis
         )
 
     if cmd_path_list[0] == "rws":
-        assert len(cmd_path_list) >= 3
+        assert len(cmd_path_list) >= 2
         if cmd_path_list[1] == "rw_assy":
-            assert len(cmd_path_list) >= 4
+            assert len(cmd_path_list) >= 3
             return pack_rw_ass_cmds(
                 q=queue_helper, object_id=RW_ASSEMBLY, cmd_str=cmd_path_list[2]
             )
         if cmd_path_list[1] == "rw_1":
-            assert len(cmd_path_list) >= 4
+            assert len(cmd_path_list) >= 3
             return create_single_rw_cmd(
                 object_id=RW1_ID, rw_idx=1, q=queue_helper, cmd_str=cmd_path_list[2]
             )
         if cmd_path_list[1] == "rw_2":
-            assert len(cmd_path_list) >= 4
+            assert len(cmd_path_list) >= 3
             return create_single_rw_cmd(
                 object_id=RW2_ID, rw_idx=2, q=queue_helper, cmd_str=cmd_path_list[2]
             )
         if cmd_path_list[1] == "rw_3":
-            assert len(cmd_path_list) >= 4
+            assert len(cmd_path_list) >= 3
             return create_single_rw_cmd(
                 object_id=RW3_ID, rw_idx=3, q=queue_helper, cmd_str=cmd_path_list[2]
             )
         if cmd_path_list[1] == "rw_4":
-            assert len(cmd_path_list) >= 4
+            assert len(cmd_path_list) >= 3
             return create_single_rw_cmd(
                 object_id=RW4_ID, rw_idx=4, q=queue_helper, cmd_str=cmd_path_list[2]
             )
@@ -281,6 +282,8 @@ def handle_obdh_procedure(
         return pack_wdt_commands(queue_helper, cmd_path_list[1])
     if cmd_path_list[0] == "time":
         return pack_time_management_cmd(queue_helper, cmd_path_list[1])
+    if cmd_path_list[0] == "tm_store":
+        return pack_tm_store_commands(queue_helper, cmd_path_list[1])
 
 
 def handle_com_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]):

eive_tmtc/tmtc/acs/reaction_wheels.py

@@ -6,12 +6,11 @@
 """
 import enum
 import struct
-from typing import List
+from typing import List, Tuple
 
 from eive_tmtc.pus_tm.defs import PrintWrapper
 from eive_tmtc.config.object_ids import RW1_ID, RW2_ID, RW3_ID, RW4_ID
-from tmtccmd.config import CmdTreeNode, TmtcDefinitionWrapper, OpCodeEntry
-from tmtccmd.config.tmtc import tmtc_definitions_provider
+from tmtccmd.config import CmdTreeNode
 from tmtccmd.tmtc import DefaultPusQueueHelper
 from tmtccmd.pus.tc.s3_fsfw_hk import (
     generate_one_hk_command,
@@ -23,7 +22,6 @@ from tmtccmd.pus.tc.s3_fsfw_hk import (
 from tmtccmd.pus.s8_fsfw_action import create_action_cmd
 from spacepackets.ecss.tc import PusTelecommand
 from tmtccmd.pus.s200_fsfw_mode import pack_mode_data, Mode, Subservice
-from eive_tmtc.config.definitions import CustomServiceList
 from tmtccmd.util import ObjectIdU32
 from tmtccmd.fsfw.tmtc_printer import FsfwTmTcPrinter
 
@@ -126,59 +124,14 @@ def create_reaction_wheel_assembly_node() -> CmdTreeNode:
     node = CmdTreeNode(
         "rw_assy", "Reaction Wheels Assembly", hide_children_which_are_leaves=True
     )
-    node.add_child(CmdTreeNode(InfoAss.ON, OpCodesAss.ON))
-    node.add_child(CmdTreeNode(InfoAss.NML, OpCodesAss.NML))
-    node.add_child(CmdTreeNode(InfoAss.OFF, OpCodesAss.OFF))
-    node.add_child(CmdTreeNode(InfoAss.ALL_SPEED_UP, OpCodesAss.ALL_SPEED_UP))
-    node.add_child(CmdTreeNode(InfoAss.ALL_SPEED_OFF, OpCodesAss.ALL_SPEED_OFF))
+    node.add_child(CmdTreeNode(OpCodesAss.ON, InfoAss.ON))
+    node.add_child(CmdTreeNode(OpCodesAss.NML, InfoAss.NML))
+    node.add_child(CmdTreeNode(OpCodesAss.OFF, InfoAss.OFF))
+    node.add_child(CmdTreeNode(OpCodesAss.ALL_SPEED_UP, InfoAss.ALL_SPEED_UP))
+    node.add_child(CmdTreeNode(OpCodesAss.ALL_SPEED_OFF, InfoAss.ALL_SPEED_OFF))
     return node
 
 
-@tmtc_definitions_provider
-def add_rw_cmds(defs: TmtcDefinitionWrapper):
-    oce = OpCodeEntry()
-    oce.add(info=InfoDev.SPEED, keys=OpCodesDev.SPEED)
-    oce.add(info=InfoDev.ON, keys=OpCodesDev.ON)
-    oce.add(info=InfoDev.OFF, keys=OpCodesDev.OFF)
-    oce.add(info=InfoDev.NML, keys=OpCodesDev.NML)
-    oce.add(info=InfoDev.REQ_TM, keys=OpCodesDev.REQ_TM)
-    oce.add(info=InfoDev.GET_STATUS, keys=OpCodesDev.GET_STATUS)
-    oce.add(info=InfoDev.GET_TM, keys=OpCodesDev.GET_TM)
-    oce.add(info=InfoDev.ENABLE_STATUS_HK, keys=OpCodesDev.ENABLE_STATUS_HK)
-    oce.add(info=InfoDev.DISABLE_STATUS_HK, keys=OpCodesDev.DISABLE_STATUS_HK)
-    defs.add_service(
-        name=CustomServiceList.REACTION_WHEEL_1.value,
-        info="Reaction Wheel 1",
-        op_code_entry=oce,
-    )
-    defs.add_service(
-        name=CustomServiceList.REACTION_WHEEL_2.value,
-        info="Reaction Wheel 2",
-        op_code_entry=oce,
-    )
-    defs.add_service(
-        name=CustomServiceList.REACTION_WHEEL_3.value,
-        info="Reaction Wheel 3",
-        op_code_entry=oce,
-    )
-    defs.add_service(
-        name=CustomServiceList.REACTION_WHEEL_4.value,
-        info="Reaction Wheel 4",
-        op_code_entry=oce,
-    )
-    oce = OpCodeEntry()
-    oce.add(info=InfoAss.ON, keys=OpCodesAss.ON)
-    oce.add(info=InfoAss.NML, keys=OpCodesAss.NML)
-    oce.add(info=InfoAss.OFF, keys=OpCodesAss.OFF)
-    oce.add(info=InfoAss.ALL_SPEED_UP, keys=OpCodesAss.ALL_SPEED_UP)
-    oce.add(info=InfoAss.ALL_SPEED_OFF, keys=OpCodesAss.ALL_SPEED_OFF)
-    defs.add_service(
-        name=CustomServiceList.RW_ASSEMBLY.value,
-        info="Reaction Wheel Assembly",
-        op_code_entry=oce,
-    )
-
-
 def create_single_rw_cmd(  # noqa C901: Complexity is okay here.
     object_id: bytes, rw_idx: int, q: DefaultPusQueueHelper, cmd_str: str
 ):
@@ -272,7 +225,7 @@ def pack_rw_ass_cmds(q: DefaultPusQueueHelper, object_id: bytes, cmd_str: str):
         )
 
 
-def prompt_speed_ramp_time() -> (int, int):
+def prompt_speed_ramp_time() -> Tuple[int, int]:
     speed = int(
         input("Specify speed [0.1 RPM, 0 or range [-65000, -1000] and [1000, 65000]: ")
     )

eive_tmtc/tmtc/acs/star_tracker.py

@@ -10,12 +10,10 @@ import enum
 import logging
 import struct
 
-from eive_tmtc.config.definitions import CustomServiceList
 from eive_tmtc.pus_tm.defs import PrintWrapper
 from eive_tmtc.utility.input_helper import InputHelper
 from spacepackets.ecss.tc import PusTelecommand
-from tmtccmd.config import CmdTreeNode, TmtcDefinitionWrapper, OpCodeEntry
-from tmtccmd.config.tmtc import tmtc_definitions_provider
+from tmtccmd.config import CmdTreeNode
 from tmtccmd.pus.tc.s3_fsfw_hk import (
     create_request_one_diag_command,
     create_request_one_hk_command,
@@ -23,6 +21,7 @@ from tmtccmd.pus.tc.s3_fsfw_hk import (
     disable_periodic_hk_command,
     make_sid,
 )
+from tmtccmd.pus.s20_fsfw_param import create_load_param_cmd, create_scalar_u8_parameter
 from tmtccmd.pus.s8_fsfw_action import create_action_cmd
 from tmtccmd.tmtc import DefaultPusQueueHelper
 from tmtccmd.pus.s200_fsfw_mode import pack_mode_data, Mode
@@ -33,6 +32,16 @@ from eive_tmtc.config.object_ids import STR_ASSEMBLY, STAR_TRACKER_ID
 _LOGGER = logging.getLogger(__name__)
 
 
+class FirmwareTarget(enum.IntEnum):
+    MAIN = 1
+    BACKUP = 10
+
+
+class ParamId(enum.IntEnum):
+    FIRMWARE_TARGET = 1
+    FIRMWARE_TARGET_PERSISTENT = 2
+
+
 class StarTrackerActionId(enum.IntEnum):
     PING = 0
     BOOT = 1
@@ -95,7 +104,10 @@ class StarTrackerActionId(enum.IntEnum):
     LOGLEVEL = 81
     LOG_SUBSCRIPTION = 82
     DEBUG_CAMERA = 83
+    # Legacy variable.
     FIRMWARE_UPDATE = 84
+    FIRMWARE_UPDATE_MAIN = 84
+    FIRMWARE_UPDATE_BACKUP = 101
     SET_TIME_FROM_SYS_TIME = 87
     ADD_SECONDARY_TM_TO_NORMAL_MODE = 95
     RESET_SECONDARY_TM_SET = 96
@@ -119,9 +131,14 @@ class OpCode:
     UPLOAD_IMAGE = "upload_image"
     DOWNLOAD_IMAGE = "download_image"
     SET_IMG_PROCESSOR_MODE = "set_img_proc_mode"
-    FW_UPDATE = "fw_update"
+    FW_UPDATE_MAIN = "fw_update_main"
     SET_TIME_FROM_SYS_TIME = "set_time"
     RELOAD_JSON_CFG_FILE = "reload_json_cfg"
+    FW_UPDATE_BACKUP = "fw_update_backup"
+    SELECT_TARGET_FIRMWARE_MAIN = "fw_main"
+    SELECT_TARGET_FIRMWARE_BACKUP = "fw_backup"
+    SELECT_TARGET_FIRMWARE_MAIN_PERSISTENT = "fw_main_persistent"
+    SELECT_TARGET_FIRMWARE_BACKUP_PERSISTENT = "fw_backup_persistent"
 
 
 class Info:
@@ -135,9 +152,16 @@ class Info:
     DOWNLOAD_IMAGE = "Download Optical Image"
     TAKE_IMAGE = "Take Image"
     SET_IMG_PROCESSOR_MODE = "Set Image Processor Mode"
-    FW_UPDATE = "Firmware Update"
+    FW_UPDATE_MAIN = "Update Main Firmware Slot"
+    FW_UPDATE_BACKUP = "Update Backup Firmware Slot"
     SET_TIME_FROM_SYS_TIME = "Set time from system time"
     RELOAD_JSON_CFG_FILE = "Reload JSON configuration file. Reboot still required."
+    SELECT_TARGET_FIRMWARE_MAIN = "Select main firmware slot"
+    SELECT_TARGET_FIRMWARE_BACKUP = "Select backup firmware slot"
+    SELECT_TARGET_FIRMWARE_MAIN_PERSISTENT = "Select main firmware slot persistently"
+    SELECT_TARGET_FIRMWARE_BACKUP_PERSISTENT = (
+        "Select backup firmware slot persistently"
+    )
 
 
 class SetId(enum.IntEnum):
@@ -642,12 +666,21 @@ def pack_star_tracker_commands(  # noqa C901
             + bytearray(json_file, "utf-8")
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
-    if cmd_str == OpCode.FW_UPDATE:
-        q.add_log_cmd(Info.FW_UPDATE)
+    if cmd_str == OpCode.FW_UPDATE_MAIN:
+        q.add_log_cmd(Info.FW_UPDATE_MAIN)
         firmware = get_firmware()
         data = (
             obyt
-            + struct.pack("!I", StarTrackerActionId.FIRMWARE_UPDATE)
+            + struct.pack("!I", StarTrackerActionId.FIRMWARE_UPDATE_MAIN)
+            + firmware.encode()
+        )
+        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
+    if cmd_str == OpCode.FW_UPDATE_BACKUP:
+        q.add_log_cmd(Info.FW_UPDATE_BACKUP)
+        firmware = get_firmware()
+        data = (
+            obyt
+            + struct.pack("!I", StarTrackerActionId.FIRMWARE_UPDATE_BACKUP)
             + firmware.encode()
         )
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
@@ -683,6 +716,35 @@ def pack_star_tracker_commands(  # noqa C901
         q.add_pus_tc(
             create_action_cmd(STAR_TRACKER_ID, StarTrackerActionId.RELOAD_JSON_CFG_FILE)
         )
+    if cmd_str == OpCode.SELECT_TARGET_FIRMWARE_MAIN:
+        q.add_log_cmd(Info.SELECT_TARGET_FIRMWARE_MAIN)
+        q.add_pus_tc(create_update_firmware_target_cmd(False, FirmwareTarget.MAIN))
+    if cmd_str == OpCode.SELECT_TARGET_FIRMWARE_BACKUP:
+        q.add_log_cmd(Info.SELECT_TARGET_FIRMWARE_BACKUP)
+        q.add_pus_tc(create_update_firmware_target_cmd(False, FirmwareTarget.BACKUP))
+    if cmd_str == OpCode.SELECT_TARGET_FIRMWARE_MAIN_PERSISTENT:
+        q.add_log_cmd(Info.SELECT_TARGET_FIRMWARE_BACKUP)
+        q.add_pus_tc(create_update_firmware_target_cmd(True, FirmwareTarget.MAIN))
+    if cmd_str == OpCode.SELECT_TARGET_FIRMWARE_BACKUP_PERSISTENT:
+        q.add_log_cmd(Info.SELECT_TARGET_FIRMWARE_BACKUP)
+        q.add_pus_tc(create_update_firmware_target_cmd(True, FirmwareTarget.BACKUP))
+
+
+def create_update_firmware_target_cmd(
+    persistent: bool, fw_target: FirmwareTarget
+) -> PusTelecommand:
+    if persistent:
+        param_id = ParamId.FIRMWARE_TARGET_PERSISTENT
+    else:
+        param_id = ParamId.FIRMWARE_TARGET
+    return create_load_param_cmd(
+        create_scalar_u8_parameter(
+            STAR_TRACKER_ID,
+            0,
+            param_id,
+            fw_target,
+        )
+    )
 
 
 def request_dataset(q: DefaultPusQueueHelper, req_type: DataSetRequest):
@@ -1185,34 +1247,31 @@ def create_str_node() -> CmdTreeNode:
     node.add_child(
         CmdTreeNode(OpCode.RESET_SECONDARY_TM_SET, Info.RESET_SECONDARY_TM_SET)
     )
-    node.add_child(CmdTreeNode(OpCode.FW_UPDATE, Info.FW_UPDATE))
+    node.add_child(CmdTreeNode(OpCode.FW_UPDATE_MAIN, Info.FW_UPDATE_MAIN))
+    node.add_child(CmdTreeNode(OpCode.FW_UPDATE_BACKUP, Info.FW_UPDATE_BACKUP))
+    node.add_child(
+        CmdTreeNode(
+            OpCode.SELECT_TARGET_FIRMWARE_MAIN, Info.SELECT_TARGET_FIRMWARE_MAIN
+        )
+    )
+    node.add_child(
+        CmdTreeNode(
+            OpCode.SELECT_TARGET_FIRMWARE_BACKUP, Info.SELECT_TARGET_FIRMWARE_BACKUP
+        )
+    )
+    node.add_child(
+        CmdTreeNode(
+            OpCode.SELECT_TARGET_FIRMWARE_MAIN_PERSISTENT,
+            Info.SELECT_TARGET_FIRMWARE_MAIN_PERSISTENT,
+        )
+    )
+    node.add_child(
+        CmdTreeNode(
+            OpCode.SELECT_TARGET_FIRMWARE_BACKUP_PERSISTENT,
+            Info.SELECT_TARGET_FIRMWARE_BACKUP_PERSISTENT,
+        )
+    )
     node.add_child(
         CmdTreeNode(OpCode.SET_TIME_FROM_SYS_TIME, Info.SET_TIME_FROM_SYS_TIME)
     )
     return node
-
-
-@tmtc_definitions_provider
-def add_str_cmds(defs: TmtcDefinitionWrapper):
-    oce = OpCodeEntry()
-    oce.add(OpCode.ON_BOOTLOADER, "Mode On, Submode Bootloader")
-    oce.add(OpCode.ON_FIRMWARE, "Mode On, Submode Firmware")
-    oce.add(OpCode.NORMAL, "Mode Normal")
-    oce.add(OpCode.OFF, "Mode Off")
-    oce.add(OpCode.PING, "Star Tracker: Ping")
-    oce.add(OpCode.TAKE_IMAGE, "Take Image")
-    oce.add(OpCode.UPLOAD_IMAGE, Info.UPLOAD_IMAGE)
-    oce.add(OpCode.DOWNLOAD_IMAGE, Info.DOWNLOAD_IMAGE)
-    oce.add(OpCode.ONE_SHOOT_HK, Info.ONE_SHOOT_HK)
-    oce.add(OpCode.ENABLE_HK, Info.ENABLE_HK)
-    oce.add(OpCode.DISABLE_HK, Info.DISABLE_HK)
-    oce.add(OpCode.SET_IMG_PROCESSOR_MODE, Info.SET_IMG_PROCESSOR_MODE)
-    oce.add(
-        OpCode.ADD_SECONDARY_TM_TO_NORMAL_MODE, Info.ADD_SECONDARY_TM_TO_NORMAL_MODE
-    )
-    oce.add(OpCode.READ_SECONDARY_TM_SET, Info.READ_SECONDARY_TM_SET)
-    oce.add(OpCode.RESET_SECONDARY_TM_SET, Info.RESET_SECONDARY_TM_SET)
-    oce.add(OpCode.FW_UPDATE, Info.FW_UPDATE)
-    oce.add(OpCode.SET_TIME_FROM_SYS_TIME, Info.SET_TIME_FROM_SYS_TIME)
-    oce.add(OpCode.RELOAD_JSON_CFG_FILE, Info.RELOAD_JSON_CFG_FILE)
-    defs.add_service(CustomServiceList.STAR_TRACKER.value, "Star Tracker", oce)

eive_tmtc/tmtc/tm_store.py

@@ -107,6 +107,7 @@ def pack_tm_store_commands(q: DefaultPusQueueHelper, cmd_path: str):
 def create_persistent_tm_store_node() -> CmdTreeNode:
     node = CmdTreeNode("tm_store", "Persistent TM Store")
     node.add_child(CmdTreeNode(OpCode.DELETE_UP_TO, Info.DELETE_UP_TO))
+    node.add_child(CmdTreeNode(OpCode.DELETE_BY_TIME_RANGE, Info.DELETE_BY_TIME_RANGE))
     node.add_child(
         CmdTreeNode(OpCode.RETRIEVAL_BY_TIME_RANGE, Info.RETRIEVAL_BY_TIME_RANGE)
     )

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 name = "eive-tmtc"
 description = "TMTC Commander EIVE"
 readme = "README.md"
-version = "5.13.0"
+version = "6.0.0"
 requires-python = ">=3.10"
 license = {text = "Apache-2.0"}
 authors = [