eive-tmtc/eive_tmtc/tmtc/tm_store.py

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):
    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]