213 lines
7.7 KiB
Python
213 lines
7.7 KiB
Python
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
|
|
from tmtccmd.util import ObjectIdU32
|
|
from tmtccmd.util.tmtc_printer import FsfwTmTcPrinter
|
|
|
|
|
|
class OpCode:
|
|
TCS_BOARD_ASS_NORMAL = ["0", "normal"]
|
|
TCS_BOARD_ASS_OFF = ["1", "off"]
|
|
REQUEST_PRIMARY_TEMP_SET = ["temp"]
|
|
REQUEST_DEVICE_TEMP_SET = ["temp_devs"]
|
|
REQUEST_DEVICE_SUS_SET = ["temp_sus"]
|
|
|
|
|
|
class Info:
|
|
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"
|
|
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=OpCode.TCS_BOARD_ASS_NORMAL,
|
|
info=Info.TCS_BOARD_ASS_NORMAL,
|
|
)
|
|
oce.add(
|
|
keys=OpCode.TCS_BOARD_ASS_OFF,
|
|
info=Info.TCS_BOARD_ASS_OFF,
|
|
)
|
|
oce.add(keys=OpCode.REQUEST_PRIMARY_TEMP_SET, info=Info.REQUEST_PRIMARY_TEMP_SET)
|
|
oce.add(keys=OpCode.REQUEST_DEVICE_TEMP_SET, info=Info.REQUEST_DEVICE_TEMP_SET)
|
|
oce.add(keys=OpCode.REQUEST_DEVICE_SUS_SET, info=Info.REQUEST_DEVICE_SUS_SET)
|
|
defs.add_service(
|
|
name=CustomServiceList.TCS.value,
|
|
info="TCS Board Assembly",
|
|
op_code_entry=oce,
|
|
)
|
|
|
|
|
|
def pack_tcs_sys_commands(q: DefaultPusQueueHelper, op_code: str):
|
|
if op_code in OpCode.REQUEST_PRIMARY_TEMP_SET:
|
|
sensor_set_sid = make_sid(TCS_CONTROLLER, SetId.PRIMARY_SENSORS)
|
|
q.add_log_cmd(Info.REQUEST_PRIMARY_TEMP_SET)
|
|
q.add_pus_tc(generate_one_hk_command(sensor_set_sid))
|
|
if op_code in OpCode.REQUEST_DEVICE_TEMP_SET:
|
|
q.add_log_cmd(Info.REQUEST_DEVICE_TEMP_SET)
|
|
q.add_pus_tc(
|
|
generate_one_hk_command(make_sid(TCS_CONTROLLER, SetId.DEVICE_SENSORS))
|
|
)
|
|
if op_code in OpCode.REQUEST_DEVICE_SUS_SET:
|
|
q.add_log_cmd(Info.REQUEST_DEVICE_SUS_SET)
|
|
q.add_pus_tc(
|
|
generate_one_hk_command(make_sid(TCS_CONTROLLER, SetId.SUS_TEMP_SENSORS))
|
|
)
|
|
pack_tcs_ass_cmds(q, op_code)
|
|
|
|
|
|
def pack_tcs_ass_cmds(q: DefaultPusQueueHelper, op_code: str):
|
|
if op_code in OpCode.TCS_BOARD_ASS_NORMAL:
|
|
pack_mode_cmd_with_info(
|
|
object_id=TCS_BOARD_ASS_ID,
|
|
mode=Mode.NORMAL,
|
|
submode=0,
|
|
q=q,
|
|
info=Info.TCS_BOARD_ASS_NORMAL,
|
|
)
|
|
if op_code in OpCode.TCS_BOARD_ASS_OFF:
|
|
pack_mode_cmd_with_info(
|
|
object_id=TCS_BOARD_ASS_ID,
|
|
mode=Mode.OFF,
|
|
submode=0,
|
|
q=q,
|
|
info=Info.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)
|
|
# tcp_server_sensor_temperatures.report_parsed_hk_data(
|
|
# object_id, set_id, 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)
|
|
# print(parsed_data)
|
|
# tcp_server_device_temperatures.report_parsed_hk_data(
|
|
# object_id, set_id, parsed_data
|
|
# )
|
|
elif set_id == SetId.SUS_TEMP_SENSORS:
|
|
pass
|
|
# 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],
|
|
# }
|
|
# print(parsed_data)
|
|
# TODO: Forward data to space simulator
|