this is more pythonic

CHANGELOG.md

@@ -10,6 +10,13 @@ list yields a list of all related PRs for each release.
 
 # [unreleased]
 
+## Added
+
+- Added version announce and current image announce commands.
+- Added persistent TM store object IDs
+- GPS enable and disable HK commands
+- All mode commands for the COM subsystem.
+
 # [v2.12.0] 2023-02-06
 
 ## Changed

eive_tmtc/config/definitions.py

@@ -73,3 +73,4 @@ class CustomServiceList(str, enum.Enum):
     TVTTESTPROCEDURE = "tvtestproc"
     CONTROLLERS = "controllers"
     SCEX = "scex"
+    TM_STORE = "tm_store"

eive_tmtc/config/events.csv

@@ -244,3 +244,5 @@ Event ID (dec); Event ID (hex); Name; Severity; Description; File Path
 14002;0x36b2;REBOOT_MECHANISM_TRIGGERED;MEDIUM;The reboot mechanism was triggered. P1: First 16 bits: Last Chip, Last 16 bits: Last Copy, P2: Each byte is the respective reboot count for the slots;bsp_q7s/core/CoreController.h
 14003;0x36b3;REBOOT_HW;MEDIUM;;bsp_q7s/core/CoreController.h
 14004;0x36b4;NO_SD_CARD_ACTIVE;HIGH;No SD card was active. Core controller will attempt to re-initialize a SD card.;bsp_q7s/core/CoreController.h
+14005;0x36b5;VERSION_INFO;INFO;P1: Byte 0: Major, Byte 1: Minor, Byte 2: Patch, Byte 3: Has Git Hash P2: First four letters of Git SHA is the last byte of P1 is set.;bsp_q7s/core/CoreController.h
+14006;0x36b6;CURRENT_IMAGE_INFO;INFO;P1: Current Chip, P2: Current Copy;bsp_q7s/core/CoreController.h

eive_tmtc/config/object_ids.py

@@ -141,6 +141,11 @@ TCS_CONTROLLER = bytes([0x43, 0x40, 0x00, 0x01])
 ACS_CONTROLLER = bytes([0x43, 0x00, 0x00, 0x02])
 CORE_CONTROLLER_ID = bytes([0x43, 0x00, 0x00, 0x03])
 
+MISC_TM_STORE = bytes([0x73, 0x02, 0x00, 0x01])
+OK_TM_STORE = bytes([0x73, 0x02, 0x00, 0x02])
+NOT_OK_TM_STORE = bytes([0x73, 0x02, 0x00, 0x03])
+HK_TM_STORE = bytes([0x73, 0x02, 0x00, 0x04])
+CFDP_TM_STORE = bytes([0x73, 0x03, 0x00, 0x00])
 
 ObjectIdDict = Dict[bytes, ObjectIdU32]
 

eive_tmtc/pus_tm/event_handler.py

@@ -1,5 +1,7 @@
 import logging
 import datetime
+import struct
+import sys
 
 from eive_tmtc.config.events import get_event_dict
 from eive_tmtc.config.object_ids import get_object_ids
@@ -74,6 +76,19 @@ def handle_event_packet(raw_tm: bytes, printer: FsfwTmTcPrinter):
                 mode_name = "Raw"
             pw.dlog(f"Mode Number {event_def.param1}, Mode Name {mode_name}")
             pw.dlog(f"Submode: {event_def.param2}")
+    elif info.name == "VERSION_INFO":
+        ver_major = (event_def.param1 >> 24) & 0xFF
+        ver_minor = (event_def.param1 >> 16) & 0xFF
+        ver_rev = (event_def.param1 >> 8) & 0xFF
+        has_git_sha = bool(event_def.param1 & 0xFF)
+        git_sha = ""
+        if has_git_sha:
+            p2_as_bytes = event_def.param2.to_bytes(4, sys.byteorder)
+            git_sha = p2_as_bytes.decode("ascii")
+        version_string = f"v{ver_major}.{ver_minor}.{ver_rev}"
+        pw.dlog(f"Version {version_string}")
+        if has_git_sha:
+            pw.dlog(f"Git SHA first four letters: {git_sha}")
     elif info.name == "CLOCK_SET":
         old_time = event_def.param1
         new_time = event_def.param2

eive_tmtc/tmtc/acs/acs_ctrl.py

@@ -528,40 +528,40 @@ def handle_mgm_data_processed(pw: PrintWrapper, hk_data: bytes):
     current_idx = 0
     fmt_str = "!fff"
     inc_len = struct.calcsize(fmt_str)
-    mgm_0 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
+    mgm_0 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     mgm_0_str = [f"{val:8.3f}" for val in mgm_0]
     pw.dlog(f"MGM 0 Vec: {mgm_0_str}")
     current_idx += inc_len
-    mgm_1 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
+    mgm_1 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     mgm_1_str = [f"{val:8.3f}" for val in mgm_1]
     pw.dlog(f"MGM 1 Vec: {mgm_1_str}")
     current_idx += inc_len
-    mgm_2 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
+    mgm_2 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     mgm_2_str = [f"{val:8.3f}" for val in mgm_2]
     pw.dlog(f"MGM 2 Vec: {mgm_2_str}")
     current_idx += inc_len
-    mgm_3 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
+    mgm_3 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     mgm_3_str = [f"{val:8.3f}" for val in mgm_3]
     pw.dlog(f"MGM 3 Vec: {mgm_3_str}")
     current_idx += inc_len
-    mgm_4 = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
+    mgm_4 = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     mgm_4_str = [f"{val:8.3f}" for val in mgm_4]
     pw.dlog(f"MGM 4 Vec: {mgm_4_str}")
     current_idx += inc_len
     fmt_str = "!ddd"
     inc_len = struct.calcsize(fmt_str)
-    mgm_vec_tot = struct.unpack(fmt_str, hk_data[current_idx: current_idx + inc_len])
+    mgm_vec_tot = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     mgm_vec_tot = [f"{val:8.3f}" for val in mgm_vec_tot]
     current_idx += inc_len
     pw.dlog(f"MGM Total Vec: {mgm_vec_tot}")
     mgm_vec_tot_deriv = struct.unpack(
-        fmt_str, hk_data[current_idx: current_idx + inc_len]
+        fmt_str, hk_data[current_idx : current_idx + inc_len]
     )
     mgm_vec_tot_deriv = [f"{val:8.3f}" for val in mgm_vec_tot_deriv]
     pw.dlog(f"MGM Total Vec Deriv: {mgm_vec_tot_deriv}")
     current_idx += inc_len
     mag_igrf_model = struct.unpack(
-        fmt_str, hk_data[current_idx: current_idx + inc_len]
+        fmt_str, hk_data[current_idx : current_idx + inc_len]
     )
     mag_igrf_model = [f"{val:8.3f}" for val in mag_igrf_model]
     pw.dlog(f"MAG IGRF Model: {mag_igrf_model}")

eive_tmtc/tmtc/acs/gps.py

@@ -1,3 +1,4 @@
+import datetime
 import logging
 import struct
 
@@ -6,7 +7,12 @@ from eive_tmtc.pus_tm.defs import PrintWrapper
 from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
 from tmtccmd.config.tmtc import tmtc_definitions_provider
 from tmtccmd.tc import DefaultPusQueueHelper
-from tmtccmd.tc.pus_3_fsfw_hk import make_sid, generate_one_hk_command
+from tmtccmd.tc.pus_3_fsfw_hk import (
+    make_sid,
+    create_request_one_hk_command,
+    create_enable_periodic_hk_command_with_interval,
+    create_disable_periodic_hk_command,
+)
 from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
 
 _LOGGER = logging.getLogger(__name__)
@@ -14,11 +20,15 @@ _LOGGER = logging.getLogger(__name__)
 
 class OpCode:
     REQ_OS_HK = ["hk"]
+    ENABLE_HK = ["enable_hk"]
+    DISABLE_HK = ["disable_hk"]
     RESET_GNSS = ["reset"]
 
 
 class Info:
     REQ_OS_HK = "Request One-Shot HK"
+    ENABLE_HK = "Enable HK"
+    DISABLE_HK = "Disable HK"
     RESET_GNSS = "Reset GNSS using reset pin"
 
 
@@ -31,6 +41,8 @@ def add_gps_cmds(defs: TmtcDefinitionWrapper):
     oce = OpCodeEntry()
     oce.add(keys=OpCode.RESET_GNSS, info=Info.RESET_GNSS)
     oce.add(keys=OpCode.REQ_OS_HK, info=Info.REQ_OS_HK)
+    oce.add(keys=OpCode.ENABLE_HK, info=Info.ENABLE_HK)
+    oce.add(keys=OpCode.DISABLE_HK, info=Info.DISABLE_HK)
     defs.add_service(
         name=CustomServiceList.GPS_CTRL.value,
         info="GPS/GNSS Controller",
@@ -39,14 +51,26 @@ def add_gps_cmds(defs: TmtcDefinitionWrapper):
 
 
 def pack_gps_command(object_id: bytes, q: DefaultPusQueueHelper, op_code: str):
+    sid = make_sid(object_id=object_id, set_id=SetId.HK)
     if op_code in OpCode.RESET_GNSS:
         # TODO: This needs to be re-implemented
         _LOGGER.warning("Reset pin handling needs to be re-implemented")
-    if op_code in OpCode.REQ_OS_HK:
-        q.add_log_cmd(f"GMSS: {Info.REQ_OS_HK}")
-        q.add_pus_tc(
-            generate_one_hk_command(sid=make_sid(object_id=object_id, set_id=SetId.HK))
+    if op_code in OpCode.ENABLE_HK:
+        interval = float(input("Please specify interval in floating point seconds: "))
+        if interval <= 0:
+            raise ValueError("invalid interval")
+        q.add_log_cmd(f"GPS: {Info.ENABLE_HK}")
+        cmds = create_enable_periodic_hk_command_with_interval(
+            diag=False, sid=sid, interval_seconds=interval
         )
+        for cmd in cmds:
+            q.add_pus_tc(cmd)
+    if op_code in OpCode.DISABLE_HK:
+        q.add_log_cmd(f"gps: {Info.DISABLE_HK}")
+        q.add_pus_tc(create_disable_periodic_hk_command(diag=False, sid=sid))
+    if op_code in OpCode.REQ_OS_HK:
+        q.add_log_cmd(f"GPS: {Info.REQ_OS_HK}")
+        q.add_pus_tc(create_request_one_hk_command(sid=sid))
 
 
 def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes):
@@ -72,7 +96,12 @@ def handle_gps_data(printer: FsfwTmTcPrinter, hk_data: bytes):
         unix_seconds,
     ) = struct.unpack(fmt_str, hk_data[current_idx : current_idx + inc_len])
     current_idx += inc_len
-    date_string = f"{day}.{month}.{year} {hours}:{minutes}:{seconds}"
+    if year == 0:
+        date_string = "No date string, year is 0"
+    else:
+        date_string = datetime.datetime(
+            year=year, month=month, day=day, hour=hours, minute=minutes, second=seconds
+        )
     pw.dlog(f"Lat: {lat} deg")
     pw.dlog(f"Long: {long} deg")
     pw.dlog(f"Altitude: {alt} m | Speed: {speed} m/s")

eive_tmtc/tmtc/com/subsystem.py

@@ -10,7 +10,12 @@ from tmtccmd.config.tmtc import (
 )
 from tmtccmd.tc import service_provider
 from tmtccmd.tc.decorator import ServiceProviderParams
-from tmtccmd.tc.pus_200_fsfw_mode import create_mode_command
+from tmtccmd.tc.pus_200_fsfw_mode import (
+    create_mode_command,
+    create_read_mode_command,
+    create_announce_mode_command,
+    create_announce_mode_recursive_command,
+)
 from tmtccmd.tc.pus_20_fsfw_param import (
     create_load_param_cmd,
     pack_scalar_u8_parameter_app_data,
@@ -37,6 +42,9 @@ class OpCode:
     TX_AND_RX_CARRIER_WAVE = "rx_and_tx_carrier_wave"
     UPDATE_DEFAULT_DATARATE_LOW = "update_default_rate_low"
     UPDATE_DEFAULT_DATARATE_HIGH = "update_default_rate_high"
+    READ_MODE = "read_mode"
+    ANNOUNCE_MODE = "announce_mode"
+    ANNOUNCE_MODE_RECURSIVE = "announce_mode_recursive"
 
 
 class Info:
@@ -47,6 +55,9 @@ class Info:
     TX_AND_RX_CARRIER_WAVE = "Syrlinks with TX carrier wave"
     UPDATE_DEFAULT_DATARATE_LOW = "Configure default low datarate (BPSK modulation)"
     UPDATE_DEFAULT_DATARATE_HIGH = "Configure default high datarate (0QPSK modulation)"
+    READ_MODE = "Read Mode"
+    ANNOUNCE_MODE = "Announce Mode"
+    ANNOUNCE_MODE_RECURSIVE = "Announce mode recursively"
 
 
 @service_provider(CustomServiceList.COM_SS)
@@ -101,6 +112,15 @@ def build_com_subsystem_cmd(p: ServiceProviderParams):
         q.add_pus_tc(
             create_mode_command(COM_SUBSYSTEM_ID, Submode.RX_AND_TX_CARRIER_WAVE, 0)
         )
+    elif o == OpCode.READ_MODE:
+        q.add_log_cmd(Info.READ_MODE)
+        q.add_pus_tc(create_read_mode_command(COM_SUBSYSTEM_ID))
+    elif o == OpCode.ANNOUNCE_MODE:
+        q.add_log_cmd(Info.ANNOUNCE_MODE)
+        q.add_pus_tc(create_announce_mode_command(COM_SUBSYSTEM_ID))
+    elif o == OpCode.ANNOUNCE_MODE_RECURSIVE:
+        q.add_log_cmd(Info.ANNOUNCE_MODE_RECURSIVE)
+        q.add_pus_tc(create_announce_mode_recursive_command(COM_SUBSYSTEM_ID))
 
 
 @tmtc_definitions_provider
@@ -112,4 +132,7 @@ def add_com_subsystem_cmds(defs: TmtcDefinitionWrapper):
     oce.add(OpCode.TX_AND_RX_DEF_RATE, Info.TX_AND_RX_DEF_DATARATE)
     oce.add(OpCode.UPDATE_DEFAULT_DATARATE_LOW, Info.UPDATE_DEFAULT_DATARATE_LOW)
     oce.add(OpCode.UPDATE_DEFAULT_DATARATE_HIGH, Info.UPDATE_DEFAULT_DATARATE_HIGH)
+    oce.add(OpCode.READ_MODE, Info.READ_MODE)
+    oce.add(OpCode.ANNOUNCE_MODE, Info.ANNOUNCE_MODE)
+    oce.add(OpCode.ANNOUNCE_MODE_RECURSIVE, Info.ANNOUNCE_MODE_RECURSIVE)
     defs.add_service(CustomServiceList.COM_SS, "COM Subsystem", oce)

eive_tmtc/tmtc/core.py

@@ -20,6 +20,8 @@ _LOGGER = logging.getLogger(__name__)
 
 class ActionId(enum.IntEnum):
     LIST_DIR_INTO_FILE = 0
+    ANNOUNCE_VERSION = 1
+    ANNOUNCE_CURRENT_IMAGE = 2
     SWITCH_REBOOT_FILE_HANDLING = 5
     RESET_REBOOT_COUNTER = 6
     SWITCH_IMG_LOCK = 7
@@ -39,6 +41,8 @@ class SetId(enum.IntEnum):
 
 
 class OpCode:
+    ANNOUNCE_VERSION = "announce_version"
+    ANNOUNCE_CURRENT_IMAGE = "announce_current_image"
     REBOOT_XSC = ["0", "reboot_xsc"]
     XSC_REBOOT_SELF = ["1", "reboot_self"]
     XSC_REBOOT_0_0 = ["2", "reboot_00"]
@@ -64,6 +68,8 @@ class OpCode:
 
 
 class Info:
+    ANNOUNCE_VERSION = "Announce version"
+    ANNOUNCE_CURRENT_IMAGE = "Announce current image"
     REBOOT_XSC = "XSC reboot with prompt"
     REBOOT_FULL = "Full regular reboot"
     OBSW_UPDATE_FROM_SD_0 = "Update OBSW from SD Card 0"
@@ -89,6 +95,8 @@ class Copy(enum.IntEnum):
 @tmtc_definitions_provider
 def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
     oce = OpCodeEntry()
+    oce.add(keys=OpCode.ANNOUNCE_VERSION, info=Info.ANNOUNCE_VERSION)
+    oce.add(keys=OpCode.ANNOUNCE_CURRENT_IMAGE, info=Info.ANNOUNCE_CURRENT_IMAGE)
     oce.add(keys=OpCode.REBOOT_XSC, info=Info.REBOOT_XSC)
     oce.add(keys=OpCode.REBOOT_XSC, info=Info.REBOOT_XSC)
     oce.add(keys=OpCode.REBOOT_FULL, info=Info.REBOOT_FULL)
@@ -142,6 +150,14 @@ def add_core_controller_definitions(defs: TmtcDefinitionWrapper):
 
 
 def pack_core_commands(q: DefaultPusQueueHelper, op_code: str):
+    if op_code == OpCode.ANNOUNCE_VERSION:
+        q.add_log_cmd(f"{Info.ANNOUNCE_VERSION}")
+        q.add_pus_tc(create_action_cmd(CORE_CONTROLLER_ID, ActionId.ANNOUNCE_VERSION))
+    if op_code == OpCode.ANNOUNCE_CURRENT_IMAGE:
+        q.add_log_cmd(f"{Info.ANNOUNCE_CURRENT_IMAGE}")
+        q.add_pus_tc(
+            create_action_cmd(CORE_CONTROLLER_ID, ActionId.ANNOUNCE_CURRENT_IMAGE)
+        )
     if op_code in OpCode.REBOOT_XSC:
         reboot_self, chip_select, copy_select = determine_reboot_params()
         add_xsc_reboot_cmd(

eive_tmtc/tmtc/tm_store.py

@@ -0,0 +1,8 @@
+from eive_tmtc.config.definitions import CustomServiceList
+from tmtccmd.tc import service_provider
+from tmtccmd.tc.decorator import ServiceProviderParams
+
+
+@service_provider(CustomServiceList.TM_STORE)
+def pack_tm_store_commands(p: ServiceProviderParams):
+    pass