eive-tmtc/eive_tmtc/tmtc/health.py

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,
    TmtcDefinitionWrapper,
    OpCodeEntry,
)
from tmtccmd.tc import service_provider
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.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)
        )
    else:
        raise ValueError(f"unknown op code {o} for service {CustomServiceList.HEALTH}")


@tmtc_definitions_provider
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)