eive-tmtc/eive_tmtc/tmtc/tcs.py

import enum
import pprint
import struct

from eive_tmtc.pus_tm.defs import PrintWrapper

from eive_tmtc.config.definitions import CustomServiceList
from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry
from tmtccmd.config.tmtc import tmtc_definitions_provider
from tmtccmd.tc import DefaultPusQueueHelper
from tmtccmd.tc.pus_200_fsfw_modes import Mode
from tmtccmd.tc.pus_3_fsfw_hk import (
    make_sid,
    generate_one_hk_command,
)

from eive_tmtc.tmtc.common import pack_mode_cmd_with_info
from eive_tmtc.config.object_ids import (
    TCS_BOARD_ASS_ID,
    TCS_CONTROLLER,
    TCS_SUBSYSTEM_ID,
)
from tmtccmd.util import ObjectIdU32
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter


class OpCodeAssy:
    TCS_BOARD_ASS_NORMAL = ["nml"]
    TCS_BOARD_ASS_OFF = ["off"]


class OpCodeSys:
    OFF = ["off"]
    NML = ["nml"]
    REQUEST_PRIMARY_TEMP_SET = ["temp"]
    REQUEST_DEVICE_TEMP_SET = ["temp_devs"]
    REQUEST_DEVICE_SUS_SET = ["temp_sus"]


class InfoSys:
    OFF = "Switch TCS subsystem OFF"
    NML = "Switch TCS subsystem NORMAL (nominal)"
    REQUEST_PRIMARY_TEMP_SET = "Request HK set of primary sensor temperatures"
    REQUEST_DEVICE_TEMP_SET = (
        "Request HK set of device sensor (not dedicated sensors) temperatures"
    )
    REQUEST_DEVICE_SUS_SET = "Request HK set of the SUS temperatures"


class InfoAssy:
    TCS_BOARD_ASS_NORMAL = "Switching TCS board assembly on"
    TCS_BOARD_ASS_OFF = "Switching TCS board assembly off"


class SetId(enum.IntEnum):
    PRIMARY_SENSORS = 0
    DEVICE_SENSORS = 1
    SUS_TEMP_SENSORS = 2


@tmtc_definitions_provider
def add_tcs_cmds(defs: TmtcDefinitionWrapper):
    oce = OpCodeEntry()
    oce.add(
        keys=OpCodeAssy.TCS_BOARD_ASS_NORMAL,
        info=InfoAssy.TCS_BOARD_ASS_NORMAL,
    )
    oce.add(
        keys=OpCodeAssy.TCS_BOARD_ASS_OFF,
        info=InfoAssy.TCS_BOARD_ASS_OFF,
    )
    defs.add_service(
        name=CustomServiceList.TCS_ASS.value,
        info="TCS Board Assembly",
        op_code_entry=oce,
    )
    oce = OpCodeEntry()
    oce.add(keys=OpCodeSys.OFF, info=InfoSys.OFF)
    oce.add(keys=OpCodeSys.NML, info=InfoSys.NML)
    oce.add(
        keys=OpCodeSys.REQUEST_PRIMARY_TEMP_SET, info=InfoSys.REQUEST_PRIMARY_TEMP_SET
    )
    oce.add(
        keys=OpCodeSys.REQUEST_DEVICE_TEMP_SET, info=InfoSys.REQUEST_DEVICE_TEMP_SET
    )
    oce.add(keys=OpCodeSys.REQUEST_DEVICE_SUS_SET, info=InfoSys.REQUEST_DEVICE_SUS_SET)
    defs.add_service(
        name=CustomServiceList.TCS,
        info="TCS Board",
        op_code_entry=oce,
    )


def pack_tcs_sys_commands(q: DefaultPusQueueHelper, op_code: str):
    if op_code in OpCodeSys.REQUEST_PRIMARY_TEMP_SET:
        sensor_set_sid = make_sid(TCS_CONTROLLER, SetId.PRIMARY_SENSORS)
        q.add_log_cmd(InfoSys.REQUEST_PRIMARY_TEMP_SET)
        q.add_pus_tc(generate_one_hk_command(sensor_set_sid))
    if op_code in OpCodeSys.REQUEST_DEVICE_TEMP_SET:
        q.add_log_cmd(InfoSys.REQUEST_DEVICE_TEMP_SET)
        q.add_pus_tc(
            generate_one_hk_command(make_sid(TCS_CONTROLLER, SetId.DEVICE_SENSORS))
        )
    if op_code in OpCodeSys.REQUEST_DEVICE_SUS_SET:
        q.add_log_cmd(InfoSys.REQUEST_DEVICE_SUS_SET)
        q.add_pus_tc(
            generate_one_hk_command(make_sid(TCS_CONTROLLER, SetId.SUS_TEMP_SENSORS))
        )
    if op_code in OpCodeSys.OFF:
        q.add_log_cmd(InfoSys.OFF)
        q.add_pus_tc(
            pack_mode_cmd_with_info(TCS_SUBSYSTEM_ID, Mode.OFF, 0, q, InfoSys.OFF)
        )
    if op_code in OpCodeSys.NML:
        q.add_log_cmd(InfoSys.NML)
        q.add_pus_tc(
            pack_mode_cmd_with_info(TCS_SUBSYSTEM_ID, Mode.NORMAL, 0, q, InfoSys.OFF)
        )
    pack_tcs_ass_cmds(q, op_code)


def pack_tcs_ass_cmds(q: DefaultPusQueueHelper, op_code: str):
    if op_code in OpCodeAssy.TCS_BOARD_ASS_NORMAL:
        pack_mode_cmd_with_info(
            object_id=TCS_BOARD_ASS_ID,
            mode=Mode.NORMAL,
            submode=0,
            q=q,
            info=InfoAssy.TCS_BOARD_ASS_NORMAL,
        )
    if op_code in OpCodeAssy.TCS_BOARD_ASS_OFF:
        pack_mode_cmd_with_info(
            object_id=TCS_BOARD_ASS_ID,
            mode=Mode.OFF,
            submode=0,
            q=q,
            info=InfoAssy.TCS_BOARD_ASS_OFF,
        )


def handle_thermal_controller_hk_data(
    object_id: ObjectIdU32, printer: FsfwTmTcPrinter, set_id: int, hk_data: bytes
):
    # need a better solutuon for this is this is used again..
    """
    if TCP_TEMP_SENS_SERVER is None:
        TCP_TEMP_SENS_SERVER = TmTcpServer("localhost", 7305)
    if TCP_TEMP_DEV_SERVER:
        TCP_TEMP_DEV_SERVER = TmTcpServer("localhost", 7306)
    """
    if set_id == SetId.PRIMARY_SENSORS:
        pw = PrintWrapper(printer)
        pw.dlog("Received sensor temperature data")

        # get all the floats
        tm_data = struct.unpack("!fffffffffffffffffffff", hk_data[: 21 * 4])
        parsed_data = {
            "SENSOR_PLOC_HEATSPREADER": tm_data[0],
            "SENSOR_PLOC_MISSIONBOARD": tm_data[1],
            "SENSOR_4K_CAMERA": tm_data[2],
            "SENSOR_DAC_HEATSPREADER": tm_data[3],
            "SENSOR_STARTRACKER": tm_data[4],
            "SENSOR_RW1": tm_data[5],
            "SENSOR_DRO": tm_data[6],
            "SENSOR_SCEX": tm_data[7],
            "SENSOR_X8": tm_data[8],
            "SENSOR_HPA": tm_data[9],
            "SENSOR_TX_MODUL": tm_data[10],
            "SENSOR_MPA": tm_data[11],
            "SENSOR_ACU": tm_data[12],
            "SENSOR_PLPCDU_HEATSPREADER": tm_data[13],
            "SENSOR_TCS_BOARD": tm_data[14],
            "SENSOR_MAGNETTORQUER": tm_data[15],
            "TMP1075 TCS 0": tm_data[16],
            "TMP1075 TCS 1": tm_data[17],
            "TMP1075 PL PCDU 0": tm_data[18],
            "TMP1075 PL PCDU 1": tm_data[19],
            "TMP1075 IF BOARD": tm_data[20],
        }
        printer.file_logger.info(str(parsed_data))
        pp = pprint.PrettyPrinter(depth=4)
        pp.pprint(parsed_data)
    elif set_id == SetId.DEVICE_SENSORS:
        pw = PrintWrapper(printer)
        pw.dlog("Received device temperature data")
        fmt_str = "!fhhhhiiiifffhffffffffffffff"
        tm_data = struct.unpack(fmt_str, hk_data[:98])
        parsed_data = {
            "Q7S_TEMPERATURE": tm_data[0],
            "BATTERY_TEMPERATURE_1": tm_data[1],
            "BATTERY_TEMPERATURE_2": tm_data[2],
            "BATTERY_TEMPERATURE_3": tm_data[3],
            "BATTERY_TEMPERATURE_4": tm_data[4],
            "RW_1_TEMPERATURE": tm_data[5],
            "RW_2_TEMPERATURE": tm_data[6],
            "RW_3_TEMPERATURE": tm_data[7],
            "RW_4_TEMPERATURE": tm_data[8],
            "STARTRACKER_TEMPERATURE": tm_data[9],
            "SYRLINKS_POWER_AMPLIFIER_TEMPERATURE": tm_data[10],
            "SYRLINKS_BASEBAND_BOARD_TEMPERATURE": tm_data[11],
            "MGT_TEMPERATURE": tm_data[12],
            "ACU_TEMPERATURE_1": tm_data[13],
            "ACU_TEMPERATURE_2": tm_data[14],
            "ACU_TEMPERATURE_3": tm_data[15],
            "PDU1_TEMPERATURE": tm_data[16],
            "PDU2_TEMPERATURE": tm_data[17],
            "P60DOCK_TEMPERATURE_1": tm_data[18],
            "P60DOCK_TEMPERATURE_2": tm_data[19],
            "GYRO_0_TEMPERATURE": tm_data[20],
            "GYRO_1_TEMPERATURE": tm_data[21],
            "GYRO_2_TEMPERATURE": tm_data[22],
            "GYRO_3_TEMPERATURE": tm_data[23],
            "MGM_0_TEMPERATURE": tm_data[24],
            "MGM_1_TEMPERATURE": tm_data[25],
            "ADC_PL_PCDU_TEMPERATURE": tm_data[26],
        }
        printer.file_logger.info(str(parsed_data))
        pp = pprint.PrettyPrinter(depth=4)
        pp.pprint(parsed_data)
    elif set_id == SetId.SUS_TEMP_SENSORS:
        pw = PrintWrapper(printer)
        pw.dlog("Received SUS temperature data")
        fmt_str = "!ffffffffffffffffff"
        tm_data = struct.unpack(fmt_str, hk_data[: 4 * 18])
        parsed_data = {
            "SUS_0": tm_data[0],
            "SUS_1": tm_data[1],
            "SUS_2": tm_data[2],
            "SUS_3": tm_data[3],
            "SUS_4": tm_data[4],
            "SUS_5": tm_data[5],
            "SUS_6": tm_data[6],
            "SUS_7": tm_data[7],
            "SUS_8": tm_data[8],
            "SUS_9": tm_data[9],
            "SUS_10": tm_data[10],
            "SUS_11": tm_data[11],
        }
        printer.file_logger.info(str(parsed_data))
        pp = pprint.PrettyPrinter(depth=4)
        pp.pprint(parsed_data)