eive-tmtc/eive_tmtc/tmtc/power/pdu1.py

# -*- coding: utf-8 -*-
"""PDU1 is mounted on the X2 slot of the P60 dock
@author J. Meier
@date   17.12.2020
"""
import enum

from eive_tmtc.config.definitions import CustomServiceList
from eive_tmtc.config.object_ids import PDU_1_HANDLER_ID
from eive_tmtc.gomspace.gomspace_common import (
    pack_ping_command,
    TableIds,
    pack_get_param_command,
)
from eive_tmtc.gomspace.gomspace_pdu_definitions import PduHkTable
from eive_tmtc.tmtc.power.common_power import (
    add_gomspace_nodes,
    pack_common_gomspace_cmds,
    req_hk_cmds,
    PowerOpCodes,
    generic_on_cmd,
    generic_off_cmd,
    add_gomspace_cmd_defs,
    create_generic_on_cmd,
    create_generic_off_cmd,
    pack_common_power_cmds,
    add_common_power_defs,
    SetId,
)
from spacepackets.ecss import PusTelecommand

from tmtccmd.config import CmdTreeNode, OpCodeEntry, TmtcDefinitionWrapper
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tmtc import DefaultPusQueueHelper
from tmtccmd.util import ObjectIdU32


class Pdu1InfoBase:
    TCS = "Switch TCS Board"
    SYRLINKS = "Switch Syrlinks (COM)"
    STR = "Switch Startracker"
    MGT = "Switch Magnetorquer"
    SUS_N = "Switch Sun Sensor Board Nominal"
    SCEX = "Switch Solar Cell Experiment"
    PLOC = "Switch Payload On-Board Computer"
    ACS_A = "Switch ACS Board A-Side"


class Pdu1ChIndex(enum.IntEnum):
    TCS = 0
    SYRLINKS = 1
    STR = 2
    MGT = 3
    SUS_N = 4
    SCEX = 5
    PLOC = 6
    ACS_A = 7


class PDU1TestProcedure:
    """
    @brief  Use this class to define the tests to perform for the PDU2.
    @details    Setting all to True will run all tests.
                Setting all to False will only run the tests set to True.
    """

    all = False
    reboot = False
    ping = False
    read_temperature = False
    turn_channel_2_on = False  # Star Tracker connected to this channel (5V)
    turn_channel_2_off = False
    turn_channel_3_on = False  # MTQ connected to this channel (5V)
    turn_channel_3_off = False


def pack_pdu1_commands(object_id: ObjectIdU32, q: DefaultPusQueueHelper, cmd_str: str):
    q.add_log_cmd("Commanding PDU1")
    objb = object_id.as_bytes
    pdu1_switch_cmds(q, cmd_str)
    pdu1_req_hk_cmds(q, cmd_str)
    pack_common_power_cmds("PDU1", object_id, q, cmd_str)
    pack_common_gomspace_cmds("PDU1", object_id, q, cmd_str)
    if PDU1TestProcedure.all or PDU1TestProcedure.ping:
        q.add_log_cmd("PDU1: Ping Test")
        ping_data = bytearray([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
        q.add_pus_tc(pack_ping_command(object_id, ping_data))
    if PDU1TestProcedure.all or PDU1TestProcedure.read_temperature:
        q.add_log_cmd("PDU1: Testing temperature reading")
        q.add_pus_tc(
            pack_get_param_command(
                objb,
                TableIds.HK,
                PduHkTable.temperature.parameter_address,
                PduHkTable.temperature.parameter_size,
            )
        )


def pdu1_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
    req_hk_cmds("PDU1", q, op_code, PDU_1_HANDLER_ID, (SetId.CORE, SetId.AUX))


def info_on_pdu1(base: str) -> str:
    return "PDU1: " + base + " on"


def info_off_pdu1(base: str) -> str:
    return "PDU1: " + base + " off"


def pdu1_switch_cmds(  # noqa C901: Complexity is okay here.
    q: DefaultPusQueueHelper, cmd_str: str
):  # noqa C901: Complexity okay here
    if cmd_str in PowerOpCodes.TCS_ON:
        tcs_on_cmd(q)
    elif cmd_str in PowerOpCodes.TCS_OFF:
        tcs_off_cmd(q)
    elif cmd_str in PowerOpCodes.SYRLINKS_ON:
        syrlinks_on_cmd(q)
    elif cmd_str in PowerOpCodes.SYRLINKS_OFF:
        syrlinks_off_cmd(q)
    elif cmd_str in PowerOpCodes.STAR_TRACKER_ON:
        startracker_on_cmd(q)
    elif cmd_str in PowerOpCodes.STAR_TRACKER_OFF:
        startracker_off_cmd(q)
    elif cmd_str in PowerOpCodes.MGT_ON:
        mgt_on_cmd(q)
    elif cmd_str in PowerOpCodes.MGT_OFF:
        mgt_off_cmd(q)
    elif cmd_str in PowerOpCodes.SUS_N_ON:
        sun_sensor_nominal_on_cmd(q)
    elif cmd_str in PowerOpCodes.SUS_N_OFF:
        sun_sensor_nominal_off_cmd(q)
    elif cmd_str in PowerOpCodes.SCEX_ON:
        solar_cell_experiment_on_cmd(q)
    elif cmd_str in PowerOpCodes.SCEX_OFF:
        solar_cell_experiment_off_cmd(q)
    elif cmd_str in PowerOpCodes.PLOC_ON:
        ploc_on_cmd(q)
    elif cmd_str in PowerOpCodes.PLOC_OFF:
        ploc_off_cmd(q)
    elif cmd_str in PowerOpCodes.ACS_A_ON:
        acs_board_a_on_cmd(q)
    elif cmd_str in PowerOpCodes.ACS_A_OFF:
        acs_board_a_off_cmd(q)


def add_pdu1_common_defs(oce: OpCodeEntry):
    oce.add(keys=PowerOpCodes.TCS_ON, info=info_on_pdu1(Pdu1InfoBase.TCS))
    oce.add(keys=PowerOpCodes.TCS_OFF, info=info_off_pdu1(Pdu1InfoBase.TCS))
    oce.add(keys=PowerOpCodes.STAR_TRACKER_ON, info=info_on_pdu1(Pdu1InfoBase.STR))
    oce.add(keys=PowerOpCodes.STAR_TRACKER_OFF, info=info_off_pdu1(Pdu1InfoBase.STR))
    oce.add(keys=PowerOpCodes.SUS_N_ON, info=info_on_pdu1(Pdu1InfoBase.SUS_N))
    oce.add(keys=PowerOpCodes.SUS_N_OFF, info=info_off_pdu1(Pdu1InfoBase.SUS_N))
    oce.add(keys=PowerOpCodes.ACS_A_ON, info=info_on_pdu1(Pdu1InfoBase.ACS_A))
    oce.add(keys=PowerOpCodes.ACS_A_OFF, info=info_off_pdu1(Pdu1InfoBase.ACS_A))
    oce.add(keys=PowerOpCodes.SYRLINKS_ON, info=info_on_pdu1(Pdu1InfoBase.SYRLINKS))
    oce.add(keys=PowerOpCodes.SYRLINKS_OFF, info=info_off_pdu1(Pdu1InfoBase.SYRLINKS))
    oce.add(keys=PowerOpCodes.MGT_ON, info=info_on_pdu1(Pdu1InfoBase.MGT))
    oce.add(keys=PowerOpCodes.MGT_OFF, info=info_off_pdu1(Pdu1InfoBase.MGT))
    oce.add(keys=PowerOpCodes.PLOC_ON, info=info_on_pdu1(Pdu1InfoBase.PLOC))
    oce.add(keys=PowerOpCodes.PLOC_OFF, info=info_off_pdu1(Pdu1InfoBase.PLOC))
    oce.add(keys=PowerOpCodes.SCEX_ON, info=info_on_pdu1(Pdu1InfoBase.SCEX))
    oce.add(keys=PowerOpCodes.SCEX_OFF, info=info_off_pdu1(Pdu1InfoBase.SCEX))


def add_pdu1_subnodes(node: CmdTreeNode):
    node.add_child(CmdTreeNode(PowerOpCodes.TCS_ON[0], info_on_pdu1(Pdu1InfoBase.TCS)))
    node.add_child(
        CmdTreeNode(PowerOpCodes.TCS_OFF[0], info_off_pdu1(Pdu1InfoBase.TCS))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.STAR_TRACKER_ON[0], info_on_pdu1(Pdu1InfoBase.STR))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.STAR_TRACKER_OFF[0], info_off_pdu1(Pdu1InfoBase.STR))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.SUS_N_ON[0], info_on_pdu1(Pdu1InfoBase.SUS_N))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.SUS_N_OFF[0], info_off_pdu1(Pdu1InfoBase.SUS_N))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.ACS_A_ON[0], info_on_pdu1(Pdu1InfoBase.ACS_A))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.ACS_A_OFF[0], info_off_pdu1(Pdu1InfoBase.ACS_A))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.SYRLINKS_ON[0], info_on_pdu1(Pdu1InfoBase.SYRLINKS))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.SYRLINKS_OFF[0], info_off_pdu1(Pdu1InfoBase.SYRLINKS))
    )
    node.add_child(CmdTreeNode(PowerOpCodes.MGT_ON[0], info_on_pdu1(Pdu1InfoBase.MGT)))
    node.add_child(
        CmdTreeNode(PowerOpCodes.MGT_OFF[0], info_off_pdu1(Pdu1InfoBase.MGT))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.PLOC_ON[0], info_on_pdu1(Pdu1InfoBase.PLOC))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.PLOC_OFF[0], info_off_pdu1(Pdu1InfoBase.PLOC))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.SCEX_ON[0], info_on_pdu1(Pdu1InfoBase.SCEX))
    )
    node.add_child(
        CmdTreeNode(PowerOpCodes.SCEX_OFF[0], info_off_pdu1(Pdu1InfoBase.SCEX))
    )


def create_pdu1_node() -> CmdTreeNode:
    node = CmdTreeNode(
        "pdu1", "P60 PCDU PDU1 device", hide_children_which_are_leaves=True
    )
    add_gomspace_nodes(node)
    add_pdu1_subnodes(node)
    node.add_child(
        CmdTreeNode(
            PowerOpCodes.PRINT_SWITCH_V_I[0], "PDU1: Print Switches, Voltages, Currents"
        )
    )
    return node


@tmtc_definitions_provider
def add_pdu1_cmds(defs: TmtcDefinitionWrapper):
    oce = OpCodeEntry()
    add_pdu1_common_defs(oce)
    add_common_power_defs(oce)
    add_gomspace_cmd_defs(oce)
    oce.add(
        keys=PowerOpCodes.PRINT_SWITCH_V_I,
        info="PDU1: Print Switches, Voltages, Currents",
    )

    defs.add_service(
        name=CustomServiceList.PDU1.value,
        info="PDU1 Device",
        op_code_entry=oce,
    )


PDU1_DICT = {
    Pdu1ChIndex.TCS: Pdu1InfoBase.TCS,
    Pdu1ChIndex.STR: Pdu1InfoBase.STR,
    Pdu1ChIndex.SYRLINKS: Pdu1InfoBase.SYRLINKS,
    Pdu1ChIndex.MGT: Pdu1InfoBase.MGT,
    Pdu1ChIndex.SCEX: Pdu1InfoBase.SCEX,
    Pdu1ChIndex.PLOC: Pdu1InfoBase.PLOC,
    Pdu1ChIndex.ACS_A: Pdu1InfoBase.ACS_A,
    Pdu1ChIndex.SUS_N: Pdu1InfoBase.SUS_N,
}


def pdu1_on_cmd(idx: Pdu1ChIndex, q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, PDU1_DICT[idx], idx)


def pdu1_off_cmd(idx: Pdu1ChIndex, q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, PDU1_DICT[idx], idx)


def tcs_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.TCS, q)


def create_tcs_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.TCS)


def tcs_off_cmd(q: DefaultPusQueueHelper):
    pdu1_off_cmd(Pdu1ChIndex.TCS, q)


def create_tcs_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.TCS)


def syrlinks_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.SYRLINKS, q)


def create_syrlinks_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.SYRLINKS)


def syrlinks_off_cmd(q: DefaultPusQueueHelper):
    pdu1_off_cmd(Pdu1ChIndex.SYRLINKS, q)


def create_syrlinks_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.SYRLINKS)


def startracker_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.STR, q)


def create_startracker_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.STR)


def startracker_off_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.STR, q)


def create_startracker_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.STR)


def mgt_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.MGT, q)


def create_mgt_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.MGT)


def mgt_off_cmd(q: DefaultPusQueueHelper):
    pdu1_off_cmd(Pdu1ChIndex.MGT, q)


def create_mgt_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.MGT)


def sun_sensor_nominal_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.SUS_N, q)


def create_sun_sensor_nominal_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.SUS_N)


def sun_sensor_nominal_off_cmd(q: DefaultPusQueueHelper):
    pdu1_off_cmd(Pdu1ChIndex.SUS_N, q)


def create_sun_sensor_nominal_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.SUS_N)


def solar_cell_experiment_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.SCEX, q)


def create_solar_cell_experiment_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.SCEX)


def solar_cell_experiment_off_cmd(q: DefaultPusQueueHelper):
    pdu1_off_cmd(Pdu1ChIndex.SCEX, q)


def create_solar_cell_experiment_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.SCEX)


def ploc_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.PLOC, q)


def create_ploc_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.PLOC)


def ploc_off_cmd(q: DefaultPusQueueHelper):
    pdu1_off_cmd(Pdu1ChIndex.PLOC, q)


def create_ploc_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.PLOC)


def acs_board_a_on_cmd(q: DefaultPusQueueHelper):
    pdu1_on_cmd(Pdu1ChIndex.ACS_A, q)


def create_acs_board_a_on_cmd() -> PusTelecommand:
    return create_generic_on_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.ACS_A)


def acs_board_a_off_cmd(q: DefaultPusQueueHelper):
    pdu1_off_cmd(Pdu1ChIndex.ACS_A, q)


def create_acs_board_a_off_cmd() -> PusTelecommand:
    return create_generic_off_cmd(PDU_1_HANDLER_ID, Pdu1ChIndex.ACS_A)