eive-tmtc/pus_tc/devs/plpcdu.py

import enum
from typing import Optional

from tmtccmd.config import QueueCommands
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.service_200_mode import pack_mode_data, Modes, Subservices
from tmtccmd.tc.service_20_parameter import (
    pack_scalar_double_param_app_data,
    pack_fsfw_load_param_cmd,
    pack_boolean_parameter_app_data
)
from tmtccmd.utility.logger import get_console_logger
from spacepackets.ecss.tc import PusTelecommand
from config.object_ids import PL_PCDU_ID

LOGGER = get_console_logger()


class OpCodes:
    SWITCH_ON = ["0", "on"]
    SWITCH_ADC_NORMAL = ["1", "adc-normal"]
    SWITCH_ALL_NORMAL = ["2", "all-normal"]
    SWITCH_OFF = ["3", "off"]
    UPDATE_DRO_TO_X8_WAIT = ["6", "dro-to-x8-wait"]
    UPDATE_X8_TO_TX_WAIT_TIME = ["7", "x8-to-tx-wait"]
    UPDATE_TX_TO_MPA_WAIT_TIME = ["8", "tx-to-mpa-wait"]
    UPDATE_MPA_TO_HPA_WAIT_TIME = ["9", "mpa-to-hpa-wait"]

    INJECT_SSR_TO_DRO_FAILURE = ["10", "inject-ssr-dro-fault"]
    INJECT_DRO_TO_X8_FAILURE = ["11", "inject-dro-x8-fault"]
    INJECT_X8_TO_TX_FAILURE = ["12", "inject-x8-tx-fault"]
    INJECT_TX_TO_MPA_FAILURE = ["13", "inject-tx-mpa-fault"]
    INJECT_MPA_TO_HPA_FAILURE = ["14", "inject-mpa-hpa-fault"]
    INJECT_ALL_ON_FAILURE = ["15", "inject-all-on-fault"]


class Submodes(enum.IntEnum):
    ADC_ON = 0
    ALL_ON = 1


class ParamIds(enum.IntEnum):
    NEG_V_LOWER_BOUND = 0
    NEG_V_UPPER_BOUND = 1

    DRO_U_LOWER_BOUND = 2
    DRO_U_UPPER_BOUND = 3
    DRO_I_UPPER_BOUND = 4

    X8_U_LOWER_BOUND = 5
    X8_U_UPPER_BOUND = 6
    X8_I_UPPER_BOUND = 7

    TX_U_LOWER_BOUND = 8
    TX_U_UPPER_BOUND = 9
    TX_I_UPPER_BOUND = 10

    MPA_U_LOWER_BOUND = 11
    MPA_U_UPPER_BOUND = 12
    MPA_I_UPPER_BOUND = 13

    HPA_U_LOWER_BOUND = 14
    HPA_U_UPPER_BOUND = 15
    HPA_I_UPPER_BOUND = 16

    SSR_TO_DRO_WAIT_TIME = 17
    DRO_TO_X8_WAIT_TIME = 18
    X8_TO_TX_WAIT_TIME = 19
    TX_TO_MPA_WAIT_TIME = 20
    MPA_TO_HPA_WAIT_TIME = 21

    INJECT_SSR_TO_DRO_FAILURE = 30
    INJECT_DRO_TO_X8_FAILURE = 31
    INJECT_X8_TO_TX_FAILURE = 32
    INJECT_TX_TO_MPA_FAILURE = 33
    INJECT_MPA_TO_HPA_FAILURE = 34
    INJECT_ALL_ON_FAILURE = 35


def pack_pl_pcdu_commands(tc_queue: TcQueueT, op_code: str):
    if op_code in OpCodes.SWITCH_ON:
        tc_queue.appendleft((QueueCommands.PRINT, "Switching PL PCDU on"))
        mode_data = pack_mode_data(
            object_id=PL_PCDU_ID, mode=Modes.ON, submode=0
        )
        mode_cmd = PusTelecommand(
            service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data
        )
        tc_queue.appendleft(mode_cmd.pack_command_tuple())
    if op_code in OpCodes.SWITCH_OFF:
        tc_queue.appendleft((QueueCommands.PRINT, "Switching PL PCDU off"))
        mode_data = pack_mode_data(
            object_id=PL_PCDU_ID, mode=Modes.OFF, submode=0
        )
        mode_cmd = PusTelecommand(
            service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data
        )
        tc_queue.appendleft(mode_cmd.pack_command_tuple())
    if op_code in OpCodes.SWITCH_ADC_NORMAL:
        tc_queue.appendleft((QueueCommands.PRINT, "Switching PL PCDU ADC module normal, submode ADC ON"))
        mode_data = pack_mode_data(
            object_id=PL_PCDU_ID, mode=Modes.NORMAL, submode=Submodes.ADC_ON
        )
        mode_cmd = PusTelecommand(
            service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data
        )
        tc_queue.appendleft(mode_cmd.pack_command_tuple())
    if op_code in OpCodes.SWITCH_ALL_NORMAL:
        tc_queue.appendleft((QueueCommands.PRINT, "Switching all PL PCDU modules normal, submode ALL ON"))
        mode_data = pack_mode_data(
            object_id=PL_PCDU_ID, mode=Modes.NORMAL, submode=Submodes.ALL_ON
        )
        mode_cmd = PusTelecommand(
            service=200, subservice=Subservices.SWITCH_MODE, app_data=mode_data
        )
        tc_queue.appendleft(mode_cmd.pack_command_tuple())
    if op_code in OpCodes.UPDATE_DRO_TO_X8_WAIT:
        pack_wait_time_cmd(
            tc_queue=tc_queue,
            param_id=ParamIds.DRO_TO_X8_WAIT_TIME,
            print_str="DRO to X8",
        )
    if op_code in OpCodes.UPDATE_X8_TO_TX_WAIT_TIME:
        pack_wait_time_cmd(
            tc_queue=tc_queue,
            param_id=ParamIds.X8_TO_TX_WAIT_TIME,
            print_str="X8 to TX",
        )
    if op_code in OpCodes.UPDATE_TX_TO_MPA_WAIT_TIME:
        pack_wait_time_cmd(
            tc_queue=tc_queue,
            param_id=ParamIds.TX_TO_MPA_WAIT_TIME,
            print_str="TX to MPA",
        )
    if op_code in OpCodes.UPDATE_MPA_TO_HPA_WAIT_TIME:
        pack_wait_time_cmd(
            tc_queue=tc_queue,
            param_id=ParamIds.MPA_TO_HPA_WAIT_TIME,
            print_str="MPA to HPA",
        )
    if op_code in OpCodes.INJECT_ALL_ON_FAILURE:
        pack_failure_injection_cmd(
            tc_queue=tc_queue,
            param_id=ParamIds.INJECT_ALL_ON_FAILURE,
            print_str="All On"
        )


def request_wait_time() -> Optional[float]:
    while True:
        wait_time = input("Please enter DRO to X8 wait time in seconds, x to cancel: ")
        if wait_time.lower() == "x":
            return None
        try:
            wait_time = float(wait_time)
        except ValueError:
            LOGGER.warning("Invalid input")
            continue
        if wait_time <= 0:
            LOGGER.warning("Invalid input")
        else:
            return wait_time


def pack_wait_time_cmd(tc_queue: TcQueueT, param_id: int, print_str: str):
    wait_time = request_wait_time()
    tc_queue.appendleft(
        (QueueCommands.PRINT, f"Updating {print_str} wait time to {wait_time}")
    )
    if wait_time is None:
        return
    param_data = pack_scalar_double_param_app_data(
        object_id=PL_PCDU_ID,
        domain_id=0,
        unique_id=param_id,
        parameter=wait_time,
    )
    cmd = pack_fsfw_load_param_cmd(ssc=0, app_data=param_data)
    tc_queue.appendleft(cmd.pack_command_tuple())


def pack_failure_injection_cmd(tc_queue: TcQueueT, param_id: int, print_str: str):
    tc_queue.appendleft((QueueCommands.PRINT, f"Inserting {print_str} error"))
    param_data = pack_boolean_parameter_app_data(
        object_id=PL_PCDU_ID,
        domain_id=0,
        unique_id=param_id,
        parameter=True
    )
    cmd = pack_fsfw_load_param_cmd(ssc=0, app_data=param_data)
    tc_queue.appendleft(cmd.pack_command_tuple())