eive-tmtc/pus_tc/system/proc.py

from tmtccmd.config import QueueCommands
from tmtccmd.tc.definitions import TcQueueT
from tmtccmd.tc.pus_3_fsfw_hk import *
"""
from config.object_ids import (
    BPX_HANDLER_ID,
    P60_DOCK_HANDLER,
    PDU_1_HANDLER_ID,
    PDU_2_HANDLER_ID,
    ACU_HANDLER_ID,
    CORE_CONTROLLER_ID,
)
"""
import config.object_ids as oids
from pus_tc.devs.bpx_batt import BpxSetIds
from pus_tc.system.core import SetIds as CoreSetIds
from gomspace.gomspace_common import SetIds as GsSetIds
from pus_tc.devs.rad_sensor import CommandIds as RadSetIds
from pus_tc.system.tcs import pack_tcs_sys_commands
from pus_tc.system.controllers import pack_controller_commands


class OpCodes:
    HEATER = ["0", "heater"]
    BAT_FT = ["bat-ft"]
    CORE_FT = ["core-ft"]
    PCDU_FT = ["pcdu-ft"]
    RAD_SEN_FT = ["rad-sen-ft"]
    TCS_FT = ["tcs-ft-on"]


class KeyAndInfo:
    HEATER = ["Heater", "heater procedure"]
    BAT_FT = ["BPX Battery", "battery functional test"]
    CORE_FT = ["OBC", "OBC functional test"]
    PCDU_FT = ["PCDU", "PCDU functional test"]
    RAD_SEN_FT = ["Radiation Sensor", "Radiation Sensor functional test"]
    TCS_FT = ["TCS Act.", "TCS functional test activation"]


KAI = KeyAndInfo

PROC_INFO_DICT = {
    KAI.BAT_FT[0]: [OpCodes.BAT_FT, KAI.BAT_FT[1], 120.0, 10.0],
    KAI.CORE_FT[0]: [OpCodes.CORE_FT, KAI.CORE_FT[1], 120.0, 10.0],
    KAI.PCDU_FT[0]: [OpCodes.PCDU_FT, KAI.PCDU_FT[1], 120.0, 10.0],
    KAI.RAD_SEN_FT[0]: [OpCodes.RAD_SEN_FT, KAI.RAD_SEN_FT[1], 120.0, 10.0],
    KAI.TCS_FT[0]: [OpCodes.TCS_FT, KAI.TCS_FT[1], 1800.0, 10.0],


}


def generic_print(tc_queue: TcQueueT, info: dict):
    tc_queue.appendleft(
        (QueueCommands.PRINT, f"Executing {info[1]} Procedure (OpCodes: {info[0]})")
    )


def pack_generic_hk_listening_cmds(
    tc_queue: TcQueueT, proc_key: str, sid: bytes, diag: bool
):
    info = PROC_INFO_DICT[proc_key]
    generic_print(tc_queue=tc_queue, info=info)
    listen_to_hk_for_x_seconds(
        diag=diag,
        tc_queue=tc_queue,
        device=proc_key,
        sid=sid,
        interval_seconds=info[3],
        collection_time=info[2],
    )


def pack_proc_commands(tc_queue: TcQueueT, op_code: str):
    if op_code in OpCodes.BAT_FT:
        key = KAI.BAT_FT[0]
        sid = make_sid(oids.BPX_HANDLER_ID, BpxSetIds.GET_HK_SET)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )

    if op_code in OpCodes.CORE_FT:
        key = KAI.CORE_FT[0]
        sid = make_sid(oids.CORE_CONTROLLER_ID, CoreSetIds.HK)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )

# PCDU

    if op_code in OpCodes.PCDU_FT:
        key = KAI.PCDU_FT[0]

        sid = make_sid(oids.P60_DOCK_HANDLER, GsSetIds.P60_CORE)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )
        sid = make_sid(oids.PDU_1_HANDLER_ID, GsSetIds.PDU_1_CORE)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )
        sid = make_sid(oids.PDU_2_HANDLER_ID, GsSetIds.PDU_2_CORE)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )
        sid = make_sid(oids.ACU_HANDLER_ID, GsSetIds.ACU)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )

# AUX und 2er
        sid = make_sid(oids.P60_DOCK_HANDLER, GsSetIds.P60_AUX)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )
        sid = make_sid(oids.PDU_1_HANDLER_ID, GsSetIds.PDU_1_AUX)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )
        sid = make_sid(oids.PDU_2_HANDLER_ID, GsSetIds.PDU_2_AUX)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )

    if op_code in OpCodes.RAD_SEN_FT:
        key = KAI.CORE_FT[0]
        sid = make_sid(oids.RAD_SENSOR_ID, RadSetIds.READ_CONVERSIONS)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )

    if op_code in OpCodes.TCS_FT:
        pack_tcs_sys_commands(tc_queue=tc_queue, op_code="tcs-normal")
        pack_controller_commands(tc_queue=tc_queue, op_code="thermal_controller")

        key = KAI.CORE_FT[0]
# Ids for TCS Board missing. No HK generation?
        sid = make_sid(oids.TCS_BOARD_ASS_ID, RadSetIds.READ_CONVERSIONS)
        pack_generic_hk_listening_cmds(
            tc_queue=tc_queue, proc_key=key, sid=sid, diag=False
        )

        pack_tcs_sys_commands(tc_queue=tc_queue, op_code="tcs-off")
        pack_controller_commands(tc_queue=tc_queue, op_code="thermal_controller")


def listen_to_hk_for_x_seconds(
    tc_queue: TcQueueT,
    diag: bool,
    device: str,
    sid: bytes,
    interval_seconds: float,
    collection_time: float,
):

    tc_queue.appendleft((QueueCommands.PRINT, f"Enabling periodic HK for {device}"))
    cmd_tuple = enable_periodic_hk_command_with_interval(
        diag=diag, sid=sid, interval_seconds=interval_seconds, ssc=0
    )
    for cmd in cmd_tuple:
        tc_queue.appendleft(cmd.pack_command_tuple())

    tc_queue.appendleft((QueueCommands.WAIT, collection_time))

    tc_queue.appendleft((QueueCommands.PRINT, f"Disabling periodic HK for {device}"))
    tc_queue.appendleft(
        disable_periodic_hk_command(diag=diag, sid=sid, ssc=0).pack_command_tuple()
    )