small bugfix

Reviewed-on: #284

CHANGELOG.md

@@ -10,6 +10,14 @@ list yields a list of all related PRs for each release.
 
 # [unreleased]
 
+## Added
+
+- Added version set for STR
+
+## Fixed
+
+- EPS power commands working again.
+
 # [v6.1.1] 2024-03-06
 
 ## Added

eive_tmtc/config/custom_mode_op.py

@@ -3,6 +3,7 @@
 @details    Template configuration file. Copy this folder to the TMTC commander root and adapt
             it to your needs.
 """
+
 import enum
 
 from tmtccmd import CcsdsTmtcBackend

eive_tmtc/config/object_ids.py

@@ -3,6 +3,7 @@
 @details    Template configuration file. Copy this folder to the TMTC commander root and adapt
             it to your needs.
 """
+
 import logging
 import os.path
 from typing import Dict

eive_tmtc/pus_tc/cmd_definitions.py

@@ -1,21 +0,0 @@
-from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry, CoreServiceList
-from tmtccmd.config.tmtc import (
-    call_all_definitions_providers,
-)
-from tmtccmd.config.globals import get_default_tmtc_defs
-
-
-def get_eive_service_op_code_dict() -> TmtcDefinitionWrapper:
-    """Call all registered TMTC definition providers. They were registered using
-    the :py:func:`tmtc_definitions_provider` decorator.
-    """
-    def_wrapper = get_default_tmtc_defs()
-    srv_5 = OpCodeEntry()
-    srv_5.add("0", "Event Test")
-    def_wrapper.add_service(
-        name=CoreServiceList.SERVICE_5.value,
-        info="PUS Service 5 Event",
-        op_code_entry=srv_5,
-    )
-    call_all_definitions_providers(def_wrapper)
-    return def_wrapper

eive_tmtc/pus_tc/cmd_demux.py

@@ -1,5 +1,6 @@
 """Hook function which packs telecommands based on service and operation code string
 """
+
 import logging
 from typing import List, cast
 
@@ -51,6 +52,7 @@ from eive_tmtc.tmtc.com.subsystem import build_com_subsystem_procedure
 from eive_tmtc.tmtc.com.syrlinks_handler import pack_syrlinks_command
 from eive_tmtc.tmtc.core import pack_core_commands
 from eive_tmtc.tmtc.health import build_health_cmds
+from eive_tmtc.tmtc.payload.subsystem import create_payload_subsystem_cmd
 from eive_tmtc.tmtc.payload.ploc_mpsoc import pack_ploc_mpsoc_commands
 from eive_tmtc.tmtc.payload.ploc_supervisor import pack_ploc_supv_commands
 from eive_tmtc.tmtc.payload.plpcdu import pack_pl_pcdu_commands
@@ -116,9 +118,9 @@ def handle_pus_procedure(
 
 def handle_eps_procedure(queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]):
     obj_id_man = get_object_ids()
-    if len(cmd_path_list) == 1:
+    assert len(cmd_path_list) >= 1
-        return pack_power_commands(queue_helper, cmd_path_list[0])
+    if cmd_path_list[0] == "power":
-    assert len(cmd_path_list) >= 2
+        return pack_power_commands(queue_helper, cmd_path_list[1])
     if cmd_path_list[0] == "pwr_ctrl":
         return pack_power_ctrl_command(queue_helper, cmd_path_list[1])
     if cmd_path_list[0] == "p60_dock":
@@ -251,7 +253,8 @@ def handle_payload_procedure(
     queue_helper: DefaultPusQueueHelper, cmd_path_list: List[str]
 ):
     obj_id_man = get_object_ids()
-    assert len(cmd_path_list) >= 2
+    if cmd_path_list[0] == "subsystem":
+        return create_payload_subsystem_cmd(queue_helper, cmd_path_list[1])
     if cmd_path_list[0] == "ploc_mpsoc":
         return pack_ploc_mpsoc_commands(queue_helper, cmd_path_list[1])
     if cmd_path_list[0] == "ploc_supv":
@@ -263,7 +266,6 @@ def handle_payload_procedure(
             object_id=object_id, q=queue_helper, cmd_str=cmd_path_list[1]
         )
     if cmd_path_list[0] == "pl_pcdu":
-        assert len(cmd_path_list) >= 3
         return pack_pl_pcdu_commands(q=queue_helper, cmd_str=cmd_path_list[1])
     if cmd_path_list[0] == "scex":
         return pack_scex_cmds(

eive_tmtc/pus_tm/hk_handler.py

@@ -1,4 +1,5 @@
 """HK Handling for EIVE OBSW"""
+
 import dataclasses
 import logging
 import base64  # noqa

eive_tmtc/pus_tm/pus_handler.py

@@ -1,5 +1,6 @@
 """Core EIVE TM handler module
 """
+
 import logging
 import sqlite3
 import uuid

eive_tmtc/tmtc/acs/star_tracker.py

@@ -140,9 +140,15 @@ class OpCode:
     SELECT_TARGET_FIRMWARE_BACKUP = "fw_backup"
     SELECT_TARGET_FIRMWARE_MAIN_PERSISTENT = "fw_main_persistent"
     SELECT_TARGET_FIRMWARE_BACKUP_PERSISTENT = "fw_backup_persistent"
+    REQUEST_VERSION = "request_version"
 
 
 class Info:
+    ON_BOOTLOADER = "Switch to Mode On, Submode Bootloder"
+    ON_FIRMWARE = "Switch to Mode On, Submode Firmware"
+    NORMAL = "Switch to Mode Normal"
+    OFF = "Switch to Mode Off"
+    PING = "Send Ping"
     ONE_SHOOT_HK = "One shoot HK Set"
     ENABLE_HK = "Enable Periodic HK"
     DISABLE_HK = "Disable Periodic HK"
@@ -163,6 +169,7 @@ class Info:
     SELECT_TARGET_FIRMWARE_BACKUP_PERSISTENT = (
         "Select backup firmware slot persistently"
     )
+    REQUEST_VERSION = "Request the active Firmware Version"
 
 
 class SetId(enum.IntEnum):
@@ -316,10 +323,6 @@ def pack_star_tracker_commands(  # noqa C901
         q.add_log_cmd("Star tracker: Temperature request")
         data = obyt + struct.pack("!I", StarTrackerActionId.REQ_TEMPERATURE)
         q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
-    if cmd_str == "8":
-        q.add_log_cmd("Star tracker: Request version")
-        data = obyt + struct.pack("!I", StarTrackerActionId.REQ_VERSION)
-        q.add_pus_tc(PusTelecommand(service=8, subservice=128, app_data=data))
     if cmd_str == "9":
         q.add_log_cmd("Star tracker: Request interface")
         data = obyt + struct.pack("!I", StarTrackerActionId.REQ_INTERFACE)
@@ -730,6 +733,13 @@ def pack_star_tracker_commands(  # noqa C901
     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))
+    if cmd_str == OpCode.REQUEST_VERSION:
+        q.add_log_cmd(Info.REQUEST_VERSION)
+        q.add_pus_tc(
+            create_action_cmd(
+                object_id=STAR_TRACKER_ID, action_id=StarTrackerActionId.REQ_VERSION
+            )
+        )
 
 
 def create_update_firmware_target_cmd(
@@ -880,6 +890,8 @@ def handle_str_hk_data(set_id: int, hk_data: bytes, pw: PrintWrapper):
         handle_contrast_set(hk_data, pw)
     elif set_id == SetId.BLOB_STATS:
         handle_blob_stats_set(hk_data, pw)
+    elif set_id == SetId.VERSION:
+        handle_version_set(hk_data, pw)
     else:
         _LOGGER.warning(f"HK parsing for Star Tracker set ID {set_id} unimplemented")
 
@@ -902,6 +914,25 @@ def unpack_time_hk(hk_data: bytes, current_idx: int, pw: PrintWrapper) -> int:
     return current_idx
 
 
+def handle_version_set(hk_data: bytes, pw: PrintWrapper):
+    pw.dlog("Received Version Set")
+    if len(hk_data) != 16:
+        _LOGGER.warning(
+            f"Version dataset HK with length {len(hk_data)} of unexpected size"
+        )
+    current_idx = unpack_time_hk(hk_data, 0, pw)
+    program = struct.unpack("!B", hk_data[current_idx : current_idx + 1])[0]
+    pw.dlog(f"Program: {program}")
+    current_idx += 1
+    major = struct.unpack("!B", hk_data[current_idx : current_idx + 1])[0]
+    pw.dlog(f"Major: {major}")
+    current_idx += 1
+    minor = struct.unpack("!B", hk_data[current_idx : current_idx + 1])[0]
+    pw.dlog(f"Minor: {minor}")
+    current_idx += 1
+    pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=5))
+
+
 def handle_temperature_set(hk_data: bytes, pw: PrintWrapper):
     pw.dlog("Received temperature set")
     if len(hk_data) < 24:
@@ -1208,7 +1239,7 @@ def handle_blob_stats_set(hk_data: bytes, pw: PrintWrapper):
                 i, noise_list[i], threshold_list[i], lvalid_list[i], oflow_list[i]
             )
         )
-    pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=4))
+    pw.dlog(FsfwTmTcPrinter.get_validity_buffer(hk_data[current_idx:], num_vars=6))
 
 
 def handle_star_tracker_action_replies(
@@ -1251,60 +1282,15 @@ def handle_read_secondary_tm_set(pw: PrintWrapper, custom_data: bytes):
 
 
 def create_str_node() -> CmdTreeNode:
+    # Zip the two classes together into a dictionary
+    op_code_strs = [
+        getattr(OpCode, key) for key in dir(OpCode) if not key.startswith("__")
+    ]
+    info_strs = [getattr(Info, key) for key in dir(OpCode) if not key.startswith("__")]
+    combined_dict = dict(zip(op_code_strs, info_strs))
     node = CmdTreeNode(
         "str", "Star Tracker Device", hide_children_which_are_leaves=True
     )
-    node.add_child(CmdTreeNode(OpCode.ON_BOOTLOADER, "Mode On, Submode Bootloader"))
+    for op_code, info in combined_dict.items():
-    node.add_child(CmdTreeNode(OpCode.ON_FIRMWARE, "Mode On, Submode Firmware"))
+        node.add_child(CmdTreeNode(op_code, info))
-    node.add_child(CmdTreeNode(OpCode.NORMAL, "Mode Normal"))
-    node.add_child(CmdTreeNode(OpCode.OFF, "Mode Off"))
-    node.add_child(CmdTreeNode(OpCode.PING, "Star Tracker: Ping"))
-    node.add_child(CmdTreeNode(OpCode.TAKE_IMAGE, "Take Image"))
-    node.add_child(CmdTreeNode(OpCode.UPLOAD_IMAGE, Info.UPLOAD_IMAGE))
-    node.add_child(CmdTreeNode(OpCode.DOWNLOAD_IMAGE, Info.DOWNLOAD_IMAGE))
-    node.add_child(CmdTreeNode(OpCode.ONE_SHOOT_HK, Info.ONE_SHOOT_HK))
-    node.add_child(CmdTreeNode(OpCode.ENABLE_HK, Info.ENABLE_HK))
-    node.add_child(CmdTreeNode(OpCode.DISABLE_HK, Info.DISABLE_HK))
-    node.add_child(
-        CmdTreeNode(OpCode.SET_IMG_PROCESSOR_MODE, Info.SET_IMG_PROCESSOR_MODE)
-    )
-    node.add_child(
-        CmdTreeNode(
-            OpCode.ADD_SECONDARY_TM_TO_NORMAL_MODE,
-            Info.ADD_SECONDARY_TM_TO_NORMAL_MODE,
-        )
-    )
-    node.add_child(
-        CmdTreeNode(OpCode.READ_SECONDARY_TM_SET, Info.READ_SECONDARY_TM_SET)
-    )
-    node.add_child(
-        CmdTreeNode(OpCode.RESET_SECONDARY_TM_SET, Info.RESET_SECONDARY_TM_SET)
-    )
-    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

eive_tmtc/tmtc/payload/subsystem.py

@@ -21,20 +21,27 @@ class ModeId(enum.IntEnum):
 class OpCode(str, enum.Enum):
     OFF = "off"
     REPORT_ALL_MODES = "report_modes"
+    MPSOC_STREAM = "mode_mpsoc_stream"
+    CAM_STREAM = "mode_cam_stream"
+    EARTH_OBSV = "mode_eart_obsv"
 
 
 class Info(str, enum.Enum):
     OFF = "Off Command"
     REPORT_ALL_MODES = "Report all modes"
+    MPSOC_STREAM = "MPSoC Stream Mode"
+    CAM_STREAM = "Camera Stream Mode"
+    EARTH_OBSV = "Earth Observation Mode"
 
 
 HANDLER_LIST: Dict[str, Tuple[int, str]] = {
     OpCode.OFF: (ModeId.OFF, Info.OFF),
+    OpCode.MPSOC_STREAM: (ModeId.MPSOC_STREAM, Info.MPSOC_STREAM),
 }
 
 
-def build_acs_subsystem_cmd(q: DefaultPusQueueHelper, cmd_str: str):
+def create_payload_subsystem_cmd(q: DefaultPusQueueHelper, cmd_str: str):
-    info_prefix = "ACS Subsystem"
+    info_prefix = "Payload Subsystem"
     if cmd_str == OpCode.REPORT_ALL_MODES:
         q.add_log_cmd(f"{info_prefix}: {Info.REPORT_ALL_MODES}")
         q.add_pus_tc(
@@ -45,19 +52,28 @@ def build_acs_subsystem_cmd(q: DefaultPusQueueHelper, cmd_str: str):
             )
         )
     mode_info_tup = HANDLER_LIST.get(cmd_str)
+    assert mode_info_tup is not None
     if mode_info_tup is None:
         return
     pack_mode_cmd_with_info(
         object_id=PL_SUBSYSTEM_ID,
         info=f"{info_prefix}: {mode_info_tup[1]}",
-        submode=0,
         mode=mode_info_tup[0],
+        submode=0,
         q=q,
     )
 
 
 def create_payload_subsystem_node() -> CmdTreeNode:
     payload_node = CmdTreeNode("payload", "Payload Subsystem")
-    payload_node.add_child(CmdTreeNode(OpCode.OFF, Info.OFF))
+    subsystem_node = CmdTreeNode("subsystem", "Subsystem Commands")
-    payload_node.add_child(CmdTreeNode(OpCode.REPORT_ALL_MODES, Info.REPORT_ALL_MODES))
+    subsystem_node.add_child(CmdTreeNode(OpCode.OFF, Info.OFF))
+    subsystem_node.add_child(CmdTreeNode(OpCode.MPSOC_STREAM, Info.MPSOC_STREAM))
+    subsystem_node.add_child(CmdTreeNode(OpCode.EARTH_OBSV, Info.EARTH_OBSV))
+    subsystem_node.add_child(CmdTreeNode(OpCode.CAM_STREAM, Info.CAM_STREAM))
+
+    subsystem_node.add_child(
+        CmdTreeNode(OpCode.REPORT_ALL_MODES, Info.REPORT_ALL_MODES)
+    )
+    payload_node.add_child(subsystem_node)
     return payload_node

eive_tmtc/tmtc/tcs/heater.py

@@ -59,7 +59,7 @@ CTN = CmdTreeNode
 
 
 def create_heater_node() -> CmdTreeNode:
-    node = CmdTreeNode("heater", "Heater Device", hide_children_which_are_leaves=True)
+    node = CmdTreeNode("heaters", "Heater Device", hide_children_which_are_leaves=True)
     node.add_child(CTN(OpCode.HEATER_CMD, Info.HEATER_CMD))
     node.add_child(CTN(OpCode.HEATER_HEALTHY_CMD, Info.HEATER_HEALTHY_CMD))
     node.add_child(CTN(OpCode.HEATER_EXT_CTRL, Info.HEATER_EXT_CTRL))