Merge branch 'main' into kranz_enums

eive_tmtc/tmtc/__init__.py

@@ -4,3 +4,4 @@ from .test import add_test_defs
 from .time import add_time_cmds
 from .health import add_health_cmd_defs
 from .system import add_system_cmd_defs
+from .tm_store import add_persistent_tm_store_cmd_defs

eive_tmtc/tmtc/tm_store.py

@@ -1,8 +1,161 @@
+import datetime
+import enum
+import logging
+import math
+import struct
+
+from eive_tmtc.config.object_ids import (
+    HK_TM_STORE,
+    MISC_TM_STORE,
+    OK_TM_STORE,
+    NOT_OK_TM_STORE,
+    CFDP_TM_STORE,
+    get_object_ids,
+)
 from eive_tmtc.config.definitions import CustomServiceList
+from tmtccmd.config import TmtcDefinitionWrapper
+from tmtccmd.config.tmtc import tmtc_definitions_provider, OpCodeEntry
 from tmtccmd.tc import service_provider
 from tmtccmd.tc.decorator import ServiceProviderParams
+from dateutil.parser import parse
+
+from spacepackets.ecss import PusService  # noqa
+from spacepackets.ecss.tc import PusTelecommand
+from spacepackets.ecss.pus_15_tm_storage import Subservice
+from tmtccmd.util import ObjectIdU32
+
+
+class OpCode:
+    RETRIEVAL_BY_TIME_RANGE = "retrieval_time_range"
+    DELETE_UP_TO = "delete_up_to"
+
+
+class Info:
+    RETRIEVAL_BY_TIME_RANGE = "Dump Telemetry Packets by time range"
+    DELETE_UP_TO = "Delete Telemetry Packets up to time"
+
+
+_LOGGER = logging.getLogger(__name__)
 
 
 @service_provider(CustomServiceList.TM_STORE)
 def pack_tm_store_commands(p: ServiceProviderParams):
-    pass
+    q = p.queue_helper
+    o = p.op_code
+    if o == OpCode.DELETE_UP_TO:
+        obj_id, store_string = store_select_prompt()
+        app_data = bytearray(obj_id.as_bytes)
+        delete_up_to_time = time_prompt()
+        end_stamp = int(math.floor(delete_up_to_time.timestamp()))
+        app_data.extend(struct.pack("!I", end_stamp))
+        q.add_log_cmd(Info.DELETE_UP_TO)
+        q.add_log_cmd(f"Selected Store: {obj_id}")
+        q.add_log_cmd(
+            f"Deletion up to time "
+            f"{datetime.datetime.fromtimestamp(end_stamp, tz=datetime.timezone.utc)}"
+        )
+        q.add_pus_tc(
+            PusTelecommand(
+                service=15, subservice=Subservice.DELETE_UP_TO, app_data=app_data
+            )
+        )
+    elif o == OpCode.RETRIEVAL_BY_TIME_RANGE:
+        q.add_log_cmd(Info.RETRIEVAL_BY_TIME_RANGE)
+        pass
+
+
+@tmtc_definitions_provider
+def add_persistent_tm_store_cmd_defs(defs: TmtcDefinitionWrapper):
+    oce = OpCodeEntry()
+    oce.add(keys=OpCode.DELETE_UP_TO, info=Info.DELETE_UP_TO)
+    oce.add(keys=OpCode.RETRIEVAL_BY_TIME_RANGE, info=Info.RETRIEVAL_BY_TIME_RANGE)
+    defs.add_service(
+        CustomServiceList.TM_STORE, "Persistent TM Store", op_code_entry=oce
+    )
+
+
+class TmStoreSelect(enum.IntEnum):
+    OK_TM_STORE = 0
+    NOT_OK_TM_STORE = 1
+    MISC_TM_STORE = 2
+    HK_TM_STORE = 3
+    CFDP_TM_STORE = 4
+
+
+STORE_DICT = {
+    TmStoreSelect.OK_TM_STORE: (OK_TM_STORE, "OK Store (Verification)"),
+    TmStoreSelect.NOT_OK_TM_STORE: (NOT_OK_TM_STORE, "NOT OK Store (Events, Verification Failures..)"),
+    TmStoreSelect.MISC_TM_STORE: (MISC_TM_STORE, "Miscellaneous Store"),
+    TmStoreSelect.HK_TM_STORE: (HK_TM_STORE, "HK TM Store"),
+    TmStoreSelect.CFDP_TM_STORE: (CFDP_TM_STORE, "CFDP TM Store"),
+}
+
+
+TIME_INPUT_DICT = {
+    1: "Full manual input with dateutil.parser.parse",
+    2: "Offset from now in seconds",
+}
+
+
+def time_prompt() -> datetime.datetime:
+    print("Available time input types: ")
+    for k, v in TIME_INPUT_DICT.items():
+        print(f"  {k}: {v}")
+    while True:
+        time_input_key = int(input("Please specify a time input type by key: "))
+        if time_input_key not in TIME_INPUT_DICT.keys():
+            _LOGGER.warning("Invalid key, try again")
+            continue
+        break
+    if time_input_key == 1:
+        return time_prompt_fully_manually()
+    elif time_input_key == 2:
+        return time_prompt_offset_from_now()
+
+
+def time_prompt_fully_manually() -> datetime.datetime:
+    # TODO: Add support for offset from now in seconds
+    time = parse(
+        input(
+            "Please enter the time in any format supported by dateutil.parser.parse\n"
+            "Recommended format: UTC ISO format YYYY-MM-DDThh:mm:ssZ: "
+        )
+    )
+    print(f"Parsed timestamp: {time}")
+    return time
+
+
+def time_prompt_offset_from_now() -> datetime.datetime:
+    seconds_offset = math.floor(
+        float(
+            input(
+                "Please enter the time as a offset from now in seconds. Negative offset is allowed: "
+            )
+        )
+    )
+    time_now_with_offset = datetime.datetime.now(
+        tz=datetime.timezone.utc
+    ) + datetime.timedelta(seconds=seconds_offset)
+    print(f"Absolute resulting time: {time_now_with_offset}")
+    return time_now_with_offset
+
+
+def store_select_prompt() -> (ObjectIdU32, str):
+    obj_id_dict = get_object_ids()
+    print("Available TM stores:")
+    idx_to_obj_id = dict()
+    for k, v in STORE_DICT.items():
+        print(f"  {k}: {v[1]}")
+    while True:
+        target_index = int(
+            input("Please enter the target store for the TM store transaction: ")
+        )
+        desc_and_obj_id = STORE_DICT.get(TmStoreSelect(target_index))
+        if desc_and_obj_id is None:
+            _LOGGER.warning("Invalid index. Try again")
+            continue
+        break
+    obj_id_raw = desc_and_obj_id[0]
+    obj_id = obj_id_dict.get(obj_id_raw)
+    print(f"Selected store: {obj_id} ({desc_and_obj_id[1]})")
+    return obj_id_dict.get(obj_id_raw), desc_and_obj_id[1]