eive-tmtc/pus_tc/devs/common_power.py

import enum

from config.object_ids import PDU_1_HANDLER_ID, PDU_2_HANDLER_ID
from gomspace.gomspace_common import (
    pack_set_param_command,
    Channel,
    GomspaceOpCodes,
    GsInfo,
    SetIds,
)
from gomspace.gomspace_pdu_definitions import PDU_CONFIG_LIST
from tmtccmd.config import OpCodeEntry
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_3_fsfw_hk import (
    make_sid,
    generate_one_diag_command,
    generate_one_hk_command,
)


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


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 Pdu2InfoBase:
    PL_PCDU_BAT_NOM = "Switch PL PCDU Nominal Battery Channel"
    RW = "Switch Reaction Wheel"
    HEATER = "Switch Heater"
    SUS_R = "Switch Sun Sensor Board Redundant"
    SOLAR_ARRAY_DEPL = "Switch Solar Array Deployment"
    PL_PCDU_BAT_RED = "Switch PL PCDU Redundant Battery Channel"
    ACS_B = "Switch ACS Board B-Side"
    PL_CAM = "Switch Payload Camera"


class PowerInfo:
    INFO_CORE = "Core Information"
    INFO_AUX = "Auxiliary Information"
    INFO_ALL = "All Information"


class Pdu2ChIndex(enum.IntEnum):
    PL_PCDU_BAT_NOM = 1
    RW = 2
    HEATER = 3
    SUS_R = 4
    SOLAR_ARRAY_DEPL = 5
    PL_PCDU_BAT_RED = 6
    ACS_B = 7
    PL_CAM = 8


class PowerOpCodes:
    # PDU 1
    TCS_ON = ["tcs-on"]
    TCS_OFF = ["tcs-off"]
    SYRLINKS_ON = ["syrlinks-on"]
    SYRLINKS_OFF = ["syrlinks-off"]
    STAR_TRACKER_ON = ["str-on"]
    STAR_TRACKER_OFF = ["str-off"]
    MGT_ON = ["mgt-on"]
    MGT_OFF = ["mgt-off"]
    SUS_N_ON = ["sus-nom-on"]
    SUS_N_OFF = ["sus-nom-off"]
    SCEX_ON = ["scex-on"]
    SCEX_OFF = ["scex-off"]
    PLOC_ON = ["ploc-on"]
    PLOC_OFF = ["ploc-off"]
    ACS_A_ON = ["acs-a-on"]
    ACS_A_OFF = ["acs-a-off"]

    # PDU 2
    PL_PCDU_VBAT_NOM_ON = ["plpcdu-vbat-nom-on"]
    PL_PCDU_VBAT_NOM_OFF = ["plpcdu-vbat-nom-off"]
    RW_ON = ["rw-on"]
    RW_OFF = ["rw-off"]
    HEATER_ON = ["heater-on"]
    HEATER_OFF = ["heater-off"]
    SUS_R_ON = ["sus-red-on"]
    SUS_R_OFF = ["sus-red-off"]
    SOLAR_ARRAY_DEPL_ON = ["sa-depl-on"]
    SOLAR_ARRAY_DEPL_OFF = ["sa-depl-off"]
    PL_PCDU_VBAT_RED_ON = ["plpcdu-vbat-red-on"]
    PL_PCDU_VBAT_RED_OFF = ["plpcdu-vbat-red-off"]
    ACS_B_ON = ["acs-b-on"]
    ACS_B_OFF = ["acs-b-off"]
    PL_CAM_ON = ["cam-on"]
    PL_CAM_OFF = ["cam-off"]

    INFO_CORE = ["info"]
    INFO_AUX = ["info-aux"]
    INFO_ALL = ["info-all"]


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


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


def info_on_pdu2(base: str) -> str:
    return "PDU2: " + base + " on"


def info_off_pdu2(base: str) -> str:
    return "PDU2: " + base + " off"


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_pdu2_common_defs(oce: OpCodeEntry):
    oce.add(keys=PowerOpCodes.ACS_B_ON, info=info_on_pdu2(Pdu2InfoBase.ACS_B))
    oce.add(keys=PowerOpCodes.ACS_B_OFF, info=info_off_pdu2(Pdu2InfoBase.ACS_B))
    oce.add(keys=PowerOpCodes.SUS_R_ON, info=info_on_pdu2(Pdu2InfoBase.SUS_R))
    oce.add(keys=PowerOpCodes.SUS_R_OFF, info=info_off_pdu2(Pdu2InfoBase.SUS_R))
    oce.add(keys=PowerOpCodes.RW_ON, info=info_on_pdu2(Pdu2InfoBase.RW))
    oce.add(keys=PowerOpCodes.RW_OFF, info=info_off_pdu2(Pdu2InfoBase.RW))
    oce.add(
        keys=PowerOpCodes.PL_PCDU_VBAT_NOM_ON,
        info=info_on_pdu2(Pdu2InfoBase.PL_PCDU_BAT_NOM),
    )
    oce.add(
        keys=PowerOpCodes.PL_PCDU_VBAT_NOM_OFF,
        info=info_off_pdu2(Pdu2InfoBase.PL_PCDU_BAT_NOM),
    )
    oce.add(
        keys=PowerOpCodes.PL_PCDU_VBAT_RED_ON,
        info=info_on_pdu2(Pdu2InfoBase.PL_PCDU_BAT_RED),
    )
    oce.add(
        keys=PowerOpCodes.PL_PCDU_VBAT_RED_OFF,
        info=info_off_pdu2(Pdu2InfoBase.PL_PCDU_BAT_RED),
    )
    oce.add(keys=PowerOpCodes.HEATER_ON, info=info_on_pdu2(Pdu2InfoBase.HEATER))
    oce.add(keys=PowerOpCodes.HEATER_OFF, info=info_off_pdu2(Pdu2InfoBase.HEATER))
    oce.add(
        keys=PowerOpCodes.SOLAR_ARRAY_DEPL_ON,
        info=info_on_pdu2(Pdu2InfoBase.SOLAR_ARRAY_DEPL),
    )
    oce.add(
        keys=PowerOpCodes.SOLAR_ARRAY_DEPL_OFF,
        info=info_off_pdu2(Pdu2InfoBase.SOLAR_ARRAY_DEPL),
    )
    oce.add(keys=PowerOpCodes.PL_CAM_ON, info=info_on_pdu2(Pdu2InfoBase.PL_CAM))
    oce.add(keys=PowerOpCodes.PL_CAM_OFF, info=info_off_pdu2(Pdu2InfoBase.PL_CAM))


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


def pdu2_cmds(q: DefaultPusQueueHelper, op_code: str):
    if op_code in PowerOpCodes.PL_PCDU_VBAT_NOM_ON:
        pl_pcdu_bat_nom_on_cmd(q)
    elif op_code in PowerOpCodes.PL_PCDU_VBAT_NOM_OFF:
        pl_pcdu_bat_nom_off_cmd(q)
    elif op_code in PowerOpCodes.RW_ON:
        reaction_wheel_on_cmd(q)
    elif op_code in PowerOpCodes.RW_OFF:
        reaction_wheel_off_cmd(q)
    elif op_code in PowerOpCodes.HEATER_ON:
        heater_on_cmd(q)
    elif op_code in PowerOpCodes.HEATER_OFF:
        heater_off_cmd(q)
    elif op_code in PowerOpCodes.SUS_R_ON:
        sus_red_on_cmd(q)
    elif op_code in PowerOpCodes.SUS_R_OFF:
        sus_red_off_cmd(q)
    elif op_code in PowerOpCodes.SOLAR_ARRAY_DEPL_ON:
        solar_array_deployment_on_cmd(q)
    elif op_code in PowerOpCodes.SOLAR_ARRAY_DEPL_OFF:
        solar_array_deployment_off_cmd(q)
    elif op_code in PowerOpCodes.PL_PCDU_VBAT_RED_ON:
        pl_pcdu_bat_red_on_cmd(q)
    elif op_code in PowerOpCodes.PL_PCDU_VBAT_RED_OFF:
        pl_pcdu_bat_nom_off_cmd(q)
    elif op_code in PowerOpCodes.ACS_B_ON:
        acs_board_b_side_on_cmd(q)
    elif op_code in PowerOpCodes.ACS_B_OFF:
        acs_board_b_side_off_cmd(q)
    elif op_code in PowerOpCodes.PL_CAM_ON:
        payload_camera_on_cmd(q)
    elif op_code in PowerOpCodes.PL_CAM_OFF:
        payload_camera_off_cmd(q)


def pdu1_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
    if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
        q.add_log_cmd(f"PDU1: {GsInfo.REQUEST_CORE_HK_ONCE}")
        hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1_CORE)
        q.add_pus_tc(generate_one_diag_command(sid=hk_sid))
    if op_code in GomspaceOpCodes.REQUEST_AUX_HK_ONCE:
        q.add_log_cmd(f"PDU1: {GsInfo.REQUEST_AUX_HK_ONCE}")
        hk_sid = make_sid(object_id=PDU_1_HANDLER_ID, set_id=SetIds.PDU_1_AUX)
        q.add_pus_tc(generate_one_hk_command(sid=hk_sid))


def pdu2_req_hk_cmds(q: DefaultPusQueueHelper, op_code: str):
    if op_code in GomspaceOpCodes.REQUEST_CORE_HK_ONCE:
        q.add_log_cmd(f"PDU2: {GsInfo.REQUEST_CORE_HK_ONCE}")
        hk_sid = make_sid(object_id=PDU_2_HANDLER_ID, set_id=SetIds.PDU_2_CORE)
        q.add_pus_tc(generate_one_diag_command(sid=hk_sid))
    if op_code in GomspaceOpCodes.REQUEST_AUX_HK_ONCE:
        q.add_log_cmd(f"PDU2: {GsInfo.REQUEST_AUX_HK_ONCE}")
        hk_sid = make_sid(object_id=PDU_2_HANDLER_ID, set_id=SetIds.PDU_2_AUX)
        q.add_pus_tc(generate_one_hk_command(sid=hk_sid))


def pack_power_commands(q: DefaultPusQueueHelper, op_code: str):
    pdu1_cmds(q, op_code)
    pdu2_cmds(q, op_code)
    if op_code in PowerOpCodes.INFO_CORE:
        pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
        pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
        q.add_wait_seconds(3.0)
    elif op_code in PowerOpCodes.INFO_AUX:
        pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
        pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
        q.add_wait_seconds(3.0)
    elif op_code in PowerOpCodes.INFO_ALL:
        pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
        pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_CORE_HK_ONCE[0])
        pdu1_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
        pdu2_req_hk_cmds(q, GomspaceOpCodes.REQUEST_AUX_HK_ONCE[0])
        q.add_wait_seconds(3.0)


def tcs_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.TCS, Pdu1ChIndex.TCS)


def tcs_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.TCS, Pdu1ChIndex.TCS)


def syrlinks_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SYRLINKS, Pdu1ChIndex.SYRLINKS)


def syrlinks_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SYRLINKS, Pdu1ChIndex.SYRLINKS)


def startracker_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.STR, Pdu1ChIndex.STR)


def startracker_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.STR, Pdu1ChIndex.STR)


def mgt_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.MGT, Pdu1ChIndex.MGT)


def mgt_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.MGT, Pdu1ChIndex.MGT)


def sun_sensor_nominal_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SUS_N, Pdu1ChIndex.SUS_N)


def sun_sensor_nominal_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SUS_N, Pdu1ChIndex.SUS_N)


def solar_cell_experiment_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SCEX, Pdu1ChIndex.SCEX)


def solar_cell_experiment_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.SCEX, Pdu1ChIndex.SCEX)


def ploc_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.PLOC, Pdu1ChIndex.PLOC)


def ploc_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.PLOC, Pdu1ChIndex.PLOC)


def acs_board_a_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.ACS_A, Pdu1ChIndex.ACS_A)


def acs_board_a_off_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_1_HANDLER_ID, q, Pdu1InfoBase.ACS_A, Pdu1ChIndex.ACS_A)


def pl_pcdu_bat_nom_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(
        PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_NOM, Pdu2ChIndex.PL_PCDU_BAT_NOM
    )


def pl_pcdu_bat_nom_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(
        PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_NOM, Pdu2ChIndex.PL_PCDU_BAT_NOM
    )


def reaction_wheel_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.RW, Pdu2ChIndex.RW)


def reaction_wheel_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.RW, Pdu2ChIndex.RW)


def heater_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.HEATER, Pdu2ChIndex.HEATER)


def heater_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.HEATER, Pdu2ChIndex.HEATER)


def sus_red_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.SUS_R, Pdu2ChIndex.SUS_R)


def sus_red_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.SUS_R, Pdu2ChIndex.SUS_R)


def solar_array_deployment_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(
        PDU_2_HANDLER_ID, q, Pdu2InfoBase.SOLAR_ARRAY_DEPL, Pdu2ChIndex.SOLAR_ARRAY_DEPL
    )


def solar_array_deployment_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(
        PDU_2_HANDLER_ID, q, Pdu2InfoBase.SOLAR_ARRAY_DEPL, Pdu2ChIndex.SOLAR_ARRAY_DEPL
    )


def pl_pcdu_bat_red_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(
        PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_RED, Pdu2ChIndex.PL_PCDU_BAT_RED
    )


def pl_pcdu_bat_red_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(
        PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_PCDU_BAT_RED, Pdu2ChIndex.PL_PCDU_BAT_RED
    )


def acs_board_b_side_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.ACS_B, Pdu2ChIndex.ACS_B)


def acs_board_b_side_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.ACS_B, Pdu2ChIndex.ACS_B)


def payload_camera_on_cmd(q: DefaultPusQueueHelper):
    generic_on_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_CAM, Pdu2ChIndex.PL_CAM)


def payload_camera_off_cmd(q: DefaultPusQueueHelper):
    generic_off_cmd(PDU_2_HANDLER_ID, q, Pdu2InfoBase.PL_CAM, Pdu2ChIndex.PL_CAM)


def generic_on_cmd(
    object_id: bytes, q: DefaultPusQueueHelper, info_str: str, out_idx: int
):
    q.add_log_cmd(info_str + " on")
    q.add_pus_tc(
        pack_set_param_command(
            object_id,
            PDU_CONFIG_LIST[out_idx].parameter_address,
            PDU_CONFIG_LIST[out_idx].parameter_size,
            Channel.on,
        )
    )


def generic_off_cmd(
    object_id: bytes, q: DefaultPusQueueHelper, info_str: str, out_idx: int
):
    q.add_log_cmd(info_str + " off")
    q.add_pus_tc(
        pack_set_param_command(
            object_id,
            PDU_CONFIG_LIST[out_idx].parameter_address,
            PDU_CONFIG_LIST[out_idx].parameter_size,
            Channel.off,
        )
    )