restructure TCS module a bit
This commit is contained in:
212
eive_tmtc/tmtc/tcs.py
Normal file
212
eive_tmtc/tmtc/tcs.py
Normal file
@ -0,0 +1,212 @@
|
||||
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
|
||||
Reference in New Issue
Block a user