add better assy names, group legacy object IDs

Reviewed-on: #156

CHANGELOG.md

@@ -10,6 +10,23 @@ list yields a list of all related PRs for each release.
 
 # [unreleased]
 
+# [v2.17.1] 2023-03-07
+
+## Added
+
+- `MGT_ASSEMBLY` object ID
+
+# [v2.17.0] 2023-03-07
+
+## Added
+
+- Health set and health announce commands
+- Camera Switcher, Syrlinks Assembly, IMTQ assembly and Star Tracker assembly object IDs added
+
+## Changed
+
+- Breaking typo fix in TMP1075 `SetId`
+
 # [v2.16.4] 2023-03-04
 
 ## Added

eive_tmtc/__init__.py

@@ -1,12 +1,12 @@
-__version__ = "2.16.4"
+__version__ = "2.17.1"
 
 import logging
 from pathlib import Path
 
 SW_NAME = "eive-tmtc"
 VERSION_MAJOR = 2
-VERSION_MINOR = 16
-VERSION_REVISION = 4
+VERSION_MINOR = 17
+VERSION_REVISION = 1
 
 EIVE_TMTC_ROOT = Path(__file__).parent
 PACKAGE_ROOT = EIVE_TMTC_ROOT.parent

eive_tmtc/config/object_ids.py

@@ -132,10 +132,21 @@ PL_SUBSYSTEM_ID = bytes([0x73, 0x01, 0x00, 0x02])
 TCS_SUBSYSTEM_ID = bytes([0x73, 0x01, 0x00, 0x03])
 COM_SUBSYSTEM_ID = bytes([0x73, 0x01, 0x00, 0x04])
 
+# Legacy names, kept for backwards compatibility
 ACS_BOARD_ASS_ID = bytes([0x73, 0x00, 0x00, 0x01])
-SUS_BOARD_ASS_ID = bytes([0x73, 0x00, 0x00, 0x02])
 TCS_BOARD_ASS_ID = bytes([0x73, 0x00, 0x00, 0x03])
+SUS_BOARD_ASS_ID = bytes([0x73, 0x00, 0x00, 0x02])
+IMTQ_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x08])
+
+ACS_BOARD_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x01])
+SUS_BOARD_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x02])
+TCS_BOARD_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x03])
 RW_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x04])
+CAM_SWITCHER = bytes([0x73, 0x00, 0x00, 0x06])
+SYRLINKS_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x07])
+# To be removed soon..
+MGT_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x08])
+STR_ASSEMBLY = bytes([0x73, 0x00, 0x00, 0x09])
 
 # Controllers
 TCS_CONTROLLER = bytes([0x43, 0x40, 0x00, 0x01])

eive_tmtc/tmtc/health.py

@@ -1,4 +1,7 @@
+import struct
+
 from eive_tmtc.config.definitions import CustomServiceList
+from eive_tmtc.tmtc.obj_prompt import prompt_object
 from spacepackets.ecss import PusTelecommand
 from tmtccmd.config.tmtc import (
     tmtc_definitions_provider,
@@ -6,33 +9,58 @@ from tmtccmd.config.tmtc import (
     OpCodeEntry,
 )
 from tmtccmd.tc import service_provider
-from tmtccmd.pus.s201_fsfw_health import Subservice
+from tmtccmd.pus.s201_fsfw_health import Subservice, FsfwHealth
 from tmtccmd.tc.decorator import ServiceProviderParams
 
 
 class OpCode:
+    SET_HEALTH = "set_health"
     ANNOUNCE_HEALTH_ALL = "read_health_all"
     ANNOUNCE_HEALTH = "read_health"
 
 
 class Info:
+    SET_HEALTH = "Set health of specific object"
     ANNOUNCE_HEALTH_ALL = "Read all health states"
     ANNOUNCE_HEALTH = "Read health state of one object"
 
 
+def prompt_health() -> FsfwHealth:
+    for item in FsfwHealth:
+        print(f"{item}: {item.name}")
+    health_idx = int(input("Please enter health by index: "))
+    return FsfwHealth(health_idx)
+
+
 @service_provider(CustomServiceList.HEALTH)
 def pack_test_command(p: ServiceProviderParams):
     o = p.op_code
     q = p.queue_helper
-    if o == OpCode.ANNOUNCE_HEALTH:
-        raise NotImplementedError()
+    if o == OpCode.SET_HEALTH:
+        app_data = bytearray(prompt_object())
+        health = prompt_health()
+        app_data.append(health)
+        q.add_log_cmd(Info.SET_HEALTH)
+        q.add_pus_tc(
+            PusTelecommand(
+                service=201, subservice=Subservice.TC_SET_HEALTH, app_data=app_data
+            )
+        )
+    elif o == OpCode.ANNOUNCE_HEALTH:
+        app_data = bytearray(prompt_object())
+        q.add_log_cmd(Info.ANNOUNCE_HEALTH)
+        q.add_pus_tc(
+            PusTelecommand(
+                service=201, subservice=Subservice.TC_ANNOUNCE_HEALTH, app_data=app_data
+            )
+        )
     elif o == OpCode.ANNOUNCE_HEALTH_ALL:
         q.add_log_cmd(Info.ANNOUNCE_HEALTH_ALL)
         q.add_pus_tc(
             PusTelecommand(service=201, subservice=Subservice.TC_ANNOUNCE_HEALTH_ALL)
         )
-        return
-    raise ValueError(f"unknown op code {o} for service {CustomServiceList.HEALTH}")
+    else:
+        raise ValueError(f"unknown op code {o} for service {CustomServiceList.HEALTH}")
 
 
 @tmtc_definitions_provider
@@ -40,4 +68,5 @@ def add_health_cmd_defs(defs: TmtcDefinitionWrapper):
     oce = OpCodeEntry()
     oce.add(OpCode.ANNOUNCE_HEALTH_ALL, Info.ANNOUNCE_HEALTH_ALL)
     oce.add(OpCode.ANNOUNCE_HEALTH, Info.ANNOUNCE_HEALTH)
+    oce.add(OpCode.SET_HEALTH, Info.SET_HEALTH)
     defs.add_service(CustomServiceList.HEALTH, info="Health Service", op_code_entry=oce)

eive_tmtc/tmtc/obj_prompt.py

@@ -0,0 +1,43 @@
+from eive_tmtc.config.object_ids import (
+    ACS_SUBSYSTEM_ID,
+    ACS_CONTROLLER,
+    IMTQ_HANDLER_ID,
+    GYRO_0_ADIS_HANDLER_ID,
+    GYRO_1_L3G_HANDLER_ID,
+    ACS_BOARD_ASS_ID,
+    RW_ASSEMBLY,
+    SUS_BOARD_ASS_ID,
+)
+
+SUBSYSTEM_DICT = {
+    0: "acs",
+    1: "tcs",
+    2: "com",
+}
+
+ACS_OBJ_DICT = {
+    0: ("Subsystem", ACS_SUBSYSTEM_ID),
+    1: ("SUS Assembly", SUS_BOARD_ASS_ID),
+    2: ("ACS Board Assembly", ACS_BOARD_ASS_ID),
+    3: ("RW Assembly", RW_ASSEMBLY),
+    4: ("iMTQ MGT", IMTQ_HANDLER_ID),
+    5: ("GYR 0 ADIS", GYRO_0_ADIS_HANDLER_ID),
+    6: ("GYR 1 L3G", GYRO_1_L3G_HANDLER_ID)
+}
+
+
+def prompt_object() -> bytes:
+    for k, v in SUBSYSTEM_DICT.items():
+        print(f"{k}: {v}")
+    subsystem_key = int(input("Please specify target subsystem by key: "))
+    subsystem = SUBSYSTEM_DICT[subsystem_key]
+    if subsystem is None:
+        raise ValueError("invalid key")
+    if subsystem == "acs":
+        for k, v in ACS_OBJ_DICT.items():
+            print(f"{k}: {v[0]}")
+        obj_key = int(input("Please specify target object by key: "))
+        acs_obj = ACS_OBJ_DICT[obj_key]
+        if acs_obj is None:
+            raise ValueError("invalid key")
+        return acs_obj[1]

eive_tmtc/tmtc/tcs/tmp1075.py

@@ -40,7 +40,7 @@ class Tmp1075ActionId(enum.IntEnum):
 
 
 class SetId:
-    TMEPERATURE = 1
+    TEMPERATURE = 1
 
 
 def pack_tmp1075_test_into(
@@ -64,7 +64,7 @@ def pack_tmp1075_test_into(
         q.add_pus_tc(PusTelecommand(service=200, subservice=1, app_data=mode_data))
     if op_code == OpCode.HK:
         q.add_log_cmd("TMP1075: Request One-Shot HK")
-        q.add_pus_tc(create_request_one_hk_command(make_sid(obyt, SetId.TMEPERATURE)))
+        q.add_pus_tc(create_request_one_hk_command(make_sid(obyt, SetId.TEMPERATURE)))
     return q