eive-tmtc/eive_tmtc/tmtc/tcs/tm.py

import dataclasses
import datetime
import enum
import logging
import struct

from eive_tmtc.pus_tm.defs import PrintWrapper
from eive_tmtc.tmtc.tcs.defs import Heater
from tmtccmd.fsfw import validity_buffer_list
from tmtccmd.util import ObjectIdU32
from .defs import CtrlSetId
from .heater import HEATER_LOCATION


_LOGGER = logging.getLogger(__name__)


class ThermalComponent(enum.IntEnum):
    NONE = 0
    ACS_BOARD = 1
    MGT = 2
    RW = 3
    STR = 4
    IF_BOARD = 5
    TCS_BOARD = 6
    OBC = 7
    LEGACY_OBCIF_BOARD = 8
    SBAND_TRANSCEIVER = 9
    PCDUP60_BOARD = 10
    PCDUACU = 11
    PCDUPDU = 12
    PLPCDU_BOARD = 13
    PLOCMISSION_BOARD = 14
    PLOCPROCESSING_BOARD = 15
    DAC = 16
    CAMERA = 17
    DRO = 18
    X8 = 19
    HPA = 20
    TX = 21
    MPA = 22
    SCEX_BOARD = 23
    NUM_ENTRIES = 24


@dataclasses.dataclass
class TcsCtrlComponentInfo:
    component: ThermalComponent
    # Heater on or off?
    state: bool
    used_sensor_idx: int
    used_heater: Heater
    start_time: datetime.datetime
    end_time: datetime.datetime


def handle_thermal_controller_hk_data(  # noqa C901: complexity is okay.
    object_id: ObjectIdU32, pw: PrintWrapper, 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 == CtrlSetId.PRIMARY_SENSORS:
        pw.dlog("Received sensor temperature data")

        # get all the floats
        fmt_str = "!fffffffffffffffffffff"
        fmt_len = struct.calcsize(fmt_str)
        tm_data = struct.unpack(fmt_str, hk_data[:fmt_len])
        valid_list = validity_buffer_list(hk_data[fmt_len:], 21)
        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],
        }
        for idx, (k, v) in enumerate(parsed_data.items()):
            print(f"{str(k).ljust(30)}: Valid: {valid_list[idx]}, Value: {v}")
    elif set_id == CtrlSetId.DEVICE_SENSORS:
        pw.dlog("Received device temperature data")
        fmt_str = "!fhhhhiiiifffhffffffffffffff"
        fmt_len = struct.calcsize(fmt_str)
        tm_data = struct.unpack(fmt_str, hk_data[:98])
        valid_list = validity_buffer_list(hk_data[fmt_len:], 25)
        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_TEMPERATURES": (tm_data[13], tm_data[14], 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_LIS3_TEMPERATURE": tm_data[24],
            "MGM_2_LIS3_TEMPERATURE": tm_data[25],
            "ADC_PL_PCDU_TEMPERATURE": tm_data[26],
        }
        for idx, (k, v) in enumerate(parsed_data.items()):
            print(f"{str(k).ljust(30)}: Valid: {valid_list[idx]}, Value: {v}")
    elif set_id == CtrlSetId.SUS_TEMP_SENSORS:
        pw.dlog("Received SUS temperature data")
        fmt_str = "!ffffffffffff"
        fmt_len = struct.calcsize(fmt_str)
        tm_data = struct.unpack(fmt_str, hk_data[: 12 * 4])
        valid_list = validity_buffer_list(hk_data[fmt_len:], 12)
        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],
        }
        for idx, (k, v) in enumerate(parsed_data.items()):
            print(f"{str(k).ljust(30)}: Valid: {valid_list[idx]}, Value: {v}")
    elif set_id == CtrlSetId.HEATER_INFO:
        print("Heater Switch States")
        for i in range(8):
            print(
                f"{HEATER_LOCATION[i].ljust(25)}: {'On' if hk_data[i] == 1 else 'Off'}"
            )
        current_draw = struct.unpack("!H", hk_data[8:10])[0]
        print(f"Heater Power Channel Current Draw: {current_draw} mA")
    elif set_id == CtrlSetId.TCS_CTRL_INFO:
        pw.dlog("Received TCS CTRL information set")
        current_idx = 0
        heater_states = hk_data[0 : ThermalComponent.NUM_ENTRIES]
        current_idx += ThermalComponent.NUM_ENTRIES
        used_sensor_idx = hk_data[
            current_idx : current_idx + ThermalComponent.NUM_ENTRIES
        ]
        current_idx += ThermalComponent.NUM_ENTRIES
        used_heater_idx = hk_data[
            current_idx : current_idx + ThermalComponent.NUM_ENTRIES
        ]
        current_idx += ThermalComponent.NUM_ENTRIES
        start_and_end_time_fmt_str = "!IIIIIIIIIIIIIIIIIIIIIIII"
        data_len = struct.calcsize(start_and_end_time_fmt_str)
        start_times = struct.unpack(
            start_and_end_time_fmt_str, hk_data[current_idx : current_idx + data_len]
        )
        current_idx += data_len
        end_times = struct.unpack(
            start_and_end_time_fmt_str, hk_data[current_idx : current_idx + data_len]
        )
        current_idx += data_len
        component_list = []
        for i in range(ThermalComponent.NUM_ENTRIES):
            info = TcsCtrlComponentInfo(
                component=ThermalComponent(i),
                state=bool(heater_states[i]),
                used_sensor_idx=used_sensor_idx[i],
                used_heater=Heater(used_heater_idx[i]),
                start_time=datetime.datetime.fromtimestamp(
                    start_times[i], datetime.timezone.utc
                ),
                end_time=datetime.datetime.fromtimestamp(
                    end_times[i], datetime.timezone.utc
                ),
            )
            component_str = f"{info.component!r}".ljust(46)
            state_str = "ON" if info.state else "OFF"
            pw.dlog(
                f"{component_str}: {state_str.ljust(4)} | Sensor Index "
                f"{info.used_sensor_idx} | {info.used_heater!r} | Start "
                f"{info.start_time} | End {info.end_time}"
            )
            component_list.append(info)
    else:
        _LOGGER.warning(f"Unimplemented set ID {set_id}")