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}")