diff --git a/eive_tmtc/config/definitions.py b/eive_tmtc/config/definitions.py index a46b708..e49abe9 100644 --- a/eive_tmtc/config/definitions.py +++ b/eive_tmtc/config/definitions.py @@ -65,7 +65,8 @@ class CustomServiceList(str, enum.Enum): PL_SS = "pl_subsystem" ACS_BRD_ASS = "acs_brd_ass" SUS_BRD_ASS = "sus_brd_ass" - TCS = "tcs" + TCS_SS = "tcs_subsystem" + TCS_CTRL = "tcs_ctrl" TCS_ASS = "tcs_ass" TIME = "time" PROCEDURE = "proc" diff --git a/eive_tmtc/pus_tc/procedure_packer.py b/eive_tmtc/pus_tc/procedure_packer.py index 97d0577..5107979 100644 --- a/eive_tmtc/pus_tc/procedure_packer.py +++ b/eive_tmtc/pus_tc/procedure_packer.py @@ -6,6 +6,7 @@ from typing import cast from eive_tmtc.tmtc.acs.gyros import handle_gyr_cmd from eive_tmtc.tmtc.acs.mgms import handle_mgm_cmd from eive_tmtc.tmtc.power.power import pack_power_commands +from eive_tmtc.tmtc.tcs.ctrl import pack_tcs_ctrl_commands from eive_tmtc.tmtc.tcs.rtd import pack_rtd_commands from eive_tmtc.tmtc.payload.scex import pack_scex_cmds from eive_tmtc.tmtc.tcs.subsystem import pack_tcs_sys_commands @@ -99,8 +100,10 @@ def handle_default_procedure( # noqa C901: Complexity okay here. if service == CustomServiceList.ACU.value: object_id = cast(ObjectIdU32, obj_id_man.get(ACU_HANDLER_ID)) return pack_acu_commands(object_id=object_id, q=queue_helper, op_code=op_code) - if service == CustomServiceList.TCS.value: + if service == CustomServiceList.TCS_SS.value: return pack_tcs_sys_commands(q=queue_helper, op_code=op_code) + if service == CustomServiceList.TCS_CTRL.value: + return pack_tcs_ctrl_commands(q=queue_helper, op_code=op_code) if service == CustomServiceList.TMP1075.value: menu_dict = { "0": ("TMP1075 TCS Board 0", TMP1075_HANDLER_TCS_BRD_0_ID), diff --git a/eive_tmtc/pus_tm/hk_handling.py b/eive_tmtc/pus_tm/hk_handling.py index fcc7e97..d56c536 100644 --- a/eive_tmtc/pus_tm/hk_handling.py +++ b/eive_tmtc/pus_tm/hk_handling.py @@ -48,10 +48,7 @@ FORWARD_SENSOR_TEMPS = False def handle_hk_packet( - raw_tm: bytes, - obj_id_dict: ObjectIdDictT, - printer: FsfwTmTcPrinter, - hk_level: int + raw_tm: bytes, obj_id_dict: ObjectIdDictT, printer: FsfwTmTcPrinter, hk_level: int ): tm_packet = Service3FsfwTm.unpack(raw_telemetry=raw_tm, custom_hk_handling=False) named_obj_id = obj_id_dict.get(tm_packet.object_id.as_bytes) diff --git a/eive_tmtc/pus_tm/pus_demux.py b/eive_tmtc/pus_tm/pus_demux.py index 66de46f..9bab90f 100644 --- a/eive_tmtc/pus_tm/pus_demux.py +++ b/eive_tmtc/pus_tm/pus_demux.py @@ -29,7 +29,7 @@ def pus_factory_hook( # noqa C901 : Complexity okay here verif_wrapper: VerificationWrapper, printer: FsfwTmTcPrinter, raw_logger: RawTmtcTimedLogWrapper, - hk_level: int + hk_level: int, ): if len(packet) < 8: _LOGGER.warning("Detected packet shorter than 8 bytes!") @@ -49,7 +49,9 @@ def pus_factory_hook( # noqa C901 : Complexity okay here if service == 1: handle_service_1_fsfw_packet(wrapper=verif_wrapper, raw_tm=packet) elif service == 3: - handle_hk_packet(printer=printer, raw_tm=packet, obj_id_dict=obj_id_dict, hk_level=hk_level) + handle_hk_packet( + printer=printer, raw_tm=packet, obj_id_dict=obj_id_dict, hk_level=hk_level + ) elif service == 5: handle_event_packet(raw_tm=packet, pw=pw) elif service == 8: diff --git a/eive_tmtc/tmtc/tcs/ctrl.py b/eive_tmtc/tmtc/tcs/ctrl.py new file mode 100644 index 0000000..17a1131 --- /dev/null +++ b/eive_tmtc/tmtc/tcs/ctrl.py @@ -0,0 +1,92 @@ +from eive_tmtc.config.definitions import CustomServiceList +from eive_tmtc.config.object_ids import TCS_CONTROLLER +from eive_tmtc.tmtc.tcs import CtrlSetId +from eive_tmtc.tmtc.tcs.brd_assy import pack_tcs_ass_cmds +from tmtccmd.config.tmtc import ( + tmtc_definitions_provider, + TmtcDefinitionWrapper, + OpCodeEntry, +) +from tmtccmd.tc import DefaultPusQueueHelper +from tmtccmd.tc.pus_3_fsfw_hk import ( + make_sid, + generate_one_hk_command, + create_request_one_diag_command, + create_enable_periodic_hk_command_with_interval, + create_request_one_hk_command, +) + + +class OpCode: + REQUEST_PRIMARY_TEMP_SET = "temp" + ENABLE_TEMP_SET = "enable_temp_set" + REQUEST_DEVICE_TEMP_SET = "temp_devs" + REQUEST_DEVICE_SUS_SET = "temp_sus" + REQUEST_HEATER_INFO = "heater_info" + REQUEST_TCS_CTRL_INFO = "tcs_ctrl_info" + + +class Info: + ENABLE_TEMP_SET = "Enable Primary Temperature Set" + REQUEST_PRIMARY_TEMP_SET = "Request HK set of primary sensor temperatures" + REQUEST_DEVICE_TEMP_SET = "Request HK set of device sensor temperatures" + REQUEST_DEVICE_SUS_SET = "Request HK set of the SUS temperatures" + REQUEST_HEATER_INFO = "Request heater information" + REQUEST_TCS_CTRL_INFO = "Request TCS controller information" + + +def pack_tcs_ctrl_commands(q: DefaultPusQueueHelper, op_code: str): + if op_code == OpCode.REQUEST_PRIMARY_TEMP_SET: + sensor_set_sid = make_sid(TCS_CONTROLLER, CtrlSetId.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 == 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, CtrlSetId.DEVICE_SENSORS)) + ) + if op_code == 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, CtrlSetId.SUS_TEMP_SENSORS) + ) + ) + if op_code == OpCode.REQUEST_HEATER_INFO: + q.add_log_cmd(Info.REQUEST_HEATER_INFO) + q.add_pus_tc( + create_request_one_diag_command( + make_sid(TCS_CONTROLLER, CtrlSetId.HEATER_INFO) + ) + ) + if op_code == OpCode.REQUEST_TCS_CTRL_INFO: + q.add_log_cmd(Info.REQUEST_TCS_CTRL_INFO) + q.add_pus_tc( + create_request_one_hk_command( + make_sid(TCS_CONTROLLER, CtrlSetId.TCS_CTRL_INFO) + ) + ) + if op_code == OpCode.ENABLE_TEMP_SET: + interval_seconds = float(input("Please specify interval in seconds: ")) + cmds = create_enable_periodic_hk_command_with_interval( + False, make_sid(TCS_CONTROLLER, CtrlSetId.PRIMARY_SENSORS), interval_seconds + ) + for cmd in cmds: + q.add_pus_tc(cmd) + pack_tcs_ass_cmds(q, op_code) + + +@tmtc_definitions_provider +def add_tcs_ctrl_cmds(defs: TmtcDefinitionWrapper): + oce = OpCodeEntry() + oce.add(keys=OpCode.ENABLE_TEMP_SET, info=Info.ENABLE_TEMP_SET) + 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) + oce.add(keys=OpCode.REQUEST_HEATER_INFO, info=Info.REQUEST_HEATER_INFO) + oce.add(keys=OpCode.REQUEST_TCS_CTRL_INFO, info=Info.REQUEST_TCS_CTRL_INFO) + defs.add_service( + name=CustomServiceList.TCS_CTRL, + info="TCS controller", + op_code_entry=oce, + ) diff --git a/eive_tmtc/tmtc/tcs/defs.py b/eive_tmtc/tmtc/tcs/defs.py index 8eb631a..eeaefdb 100644 --- a/eive_tmtc/tmtc/tcs/defs.py +++ b/eive_tmtc/tmtc/tcs/defs.py @@ -12,3 +12,16 @@ class CtrlSetId(enum.IntEnum): class TcsSubmode(enum.IntEnum): DEFAULT = 0 NO_HEATER_CTRL = 1 + + +class Heater(enum.IntEnum): + HEATER_0_PLOC_PROC_BRD = 0 + HEATER_1_PCDU_BRD = 1 + HEATER_2_ACS_BRD = 2 + HEATER_3_OBC_BRD = 3 + HEATER_4_CAMERA = 4 + HEATER_5_STR = 5 + HEATER_6_DRO = 6 + HEATER_7_SYRLINKS = 7 + NUMBER_OF_SWITCHES = 8 + NONE = 0xFF diff --git a/eive_tmtc/tmtc/tcs/heater.py b/eive_tmtc/tmtc/tcs/heater.py index c33ec8c..c6558c0 100644 --- a/eive_tmtc/tmtc/tcs/heater.py +++ b/eive_tmtc/tmtc/tcs/heater.py @@ -7,6 +7,7 @@ import enum from eive_tmtc.config.definitions import CustomServiceList from eive_tmtc.config.object_ids import get_object_ids +from eive_tmtc.tmtc.tcs.defs import Heater from tmtccmd.config import TmtcDefinitionWrapper, OpCodeEntry from tmtccmd.config.tmtc import tmtc_definitions_provider from tmtccmd.tc import DefaultPusQueueHelper @@ -20,18 +21,6 @@ from tmtccmd.pus.s8_fsfw_funccmd import create_action_cmd from spacepackets.ecss.tc import PusTelecommand -class Heater(enum.IntEnum): - HEATER_0_PLOC_PROC_BRD = 0 - HEATER_1_PCDU_BRD = 1 - HEATER_2_ACS_BRD = 2 - HEATER_3_OBC_BRD = 3 - HEATER_4_CAMERA = 4 - HEATER_5_STR = 5 - HEATER_6_DRO = 6 - HEATER_7_SYRLINKS = 7 - NUMBER_OF_SWITCHES = 8 - - HEATER_LOCATION = [ "PLOC Processing Board", "PCDU PDU", diff --git a/eive_tmtc/tmtc/tcs/subsystem.py b/eive_tmtc/tmtc/tcs/subsystem.py index 04e87b8..7b32a9d 100644 --- a/eive_tmtc/tmtc/tcs/subsystem.py +++ b/eive_tmtc/tmtc/tcs/subsystem.py @@ -1,6 +1,5 @@ -from .defs import CtrlSetId from eive_tmtc.config.definitions import CustomServiceList -from eive_tmtc.config.object_ids import TCS_CONTROLLER, TCS_SUBSYSTEM_ID +from eive_tmtc.config.object_ids import TCS_SUBSYSTEM_ID from eive_tmtc.tmtc.common import pack_mode_cmd_with_info from eive_tmtc.tmtc.tcs.brd_assy import pack_tcs_ass_cmds from tmtccmd.config.tmtc import ( @@ -10,74 +9,28 @@ from tmtccmd.config.tmtc import ( ) from tmtccmd.tc import DefaultPusQueueHelper from tmtccmd.tc.pus_200_fsfw_mode import Mode, create_announce_mode_recursive_command -from tmtccmd.tc.pus_3_fsfw_hk import ( - make_sid, - generate_one_hk_command, - create_enable_periodic_hk_command_with_interval, - create_request_one_diag_command, -) -class OpCodeSys: - OFF = ["off"] - NML = ["nml"] - REQUEST_PRIMARY_TEMP_SET = ["temp"] - ENABLE_TEMP_SET = "enable_temp_set" - REQUEST_DEVICE_TEMP_SET = ["temp_devs"] - REQUEST_DEVICE_SUS_SET = ["temp_sus"] - REQUEST_HEATER_INFO = "heater_info" +class OpCode: + OFF = "off" + NML = "nml" ANNOUNCE_MODES = "announce_modes" class InfoSys: OFF = "Switch TCS subsystem OFF" NML = "Switch TCS subsystem NORMAL (nominal)" - ENABLE_TEMP_SET = "Enable Primary Temperature Set" - REQUEST_PRIMARY_TEMP_SET = "Request HK set of primary sensor temperatures" - REQUEST_DEVICE_TEMP_SET = "Request HK set of device sensor temperatures" - REQUEST_DEVICE_SUS_SET = "Request HK set of the SUS temperatures" - REQUEST_HEATER_INFO = "Request heater information" ANNOUNCE_MODES = "Announce Modes recursively" 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, CtrlSetId.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, CtrlSetId.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, CtrlSetId.SUS_TEMP_SENSORS) - ) - ) - if op_code == OpCodeSys.REQUEST_HEATER_INFO: - q.add_log_cmd(InfoSys.REQUEST_HEATER_INFO) - q.add_pus_tc( - create_request_one_diag_command( - make_sid(TCS_CONTROLLER, CtrlSetId.HEATER_INFO) - ) - ) - if op_code in OpCodeSys.OFF: + if op_code == OpCode.OFF: q.add_log_cmd(InfoSys.OFF) pack_mode_cmd_with_info(TCS_SUBSYSTEM_ID, Mode.OFF, 0, q, InfoSys.OFF) - if op_code in OpCodeSys.NML: + if op_code == OpCode.NML: q.add_log_cmd(InfoSys.NML) pack_mode_cmd_with_info(TCS_SUBSYSTEM_ID, Mode.NORMAL, 0, q, InfoSys.OFF) - if op_code == OpCodeSys.ENABLE_TEMP_SET: - interval_seconds = float(input("Please specify interval in seconds: ")) - cmds = create_enable_periodic_hk_command_with_interval( - False, make_sid(TCS_CONTROLLER, CtrlSetId.PRIMARY_SENSORS), interval_seconds - ) - for cmd in cmds: - q.add_pus_tc(cmd) - if op_code == OpCodeSys.ANNOUNCE_MODES: + if op_code == OpCode.ANNOUNCE_MODES: q.add_log_cmd(InfoSys.ANNOUNCE_MODES) q.add_pus_tc(create_announce_mode_recursive_command(TCS_SUBSYSTEM_ID)) pack_tcs_ass_cmds(q, op_code) @@ -86,20 +39,11 @@ def pack_tcs_sys_commands(q: DefaultPusQueueHelper, op_code: str): @tmtc_definitions_provider def add_tcs_subsystem_cmds(defs: TmtcDefinitionWrapper): 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) - oce.add(keys=OpCodeSys.REQUEST_HEATER_INFO, info=InfoSys.REQUEST_HEATER_INFO) - oce.add(keys=OpCodeSys.ANNOUNCE_MODES, info=InfoSys.ANNOUNCE_MODES) - oce.add(keys=OpCodeSys.ENABLE_TEMP_SET, info=InfoSys.ENABLE_TEMP_SET) + oce.add(keys=OpCode.OFF, info=InfoSys.OFF) + oce.add(keys=OpCode.NML, info=InfoSys.NML) + oce.add(keys=OpCode.ANNOUNCE_MODES, info=InfoSys.ANNOUNCE_MODES) defs.add_service( - name=CustomServiceList.TCS, - info="TCS", + name=CustomServiceList.TCS_SS, + info="TCS subsystem", op_code_entry=oce, ) diff --git a/eive_tmtc/tmtc/tcs/tm.py b/eive_tmtc/tmtc/tcs/tm.py index e73a6fa..985969c 100644 --- a/eive_tmtc/tmtc/tcs/tm.py +++ b/eive_tmtc/tmtc/tcs/tm.py @@ -1,7 +1,11 @@ +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 @@ -11,6 +15,45 @@ 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( object_id: ObjectIdU32, pw: PrintWrapper, set_id: int, hk_data: bytes ): @@ -119,5 +162,50 @@ def handle_thermal_controller_hk_data( ) 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}") diff --git a/tmtcc.py b/tmtcc.py index 473aeb3..d6d3645 100755 --- a/tmtcc.py +++ b/tmtcc.py @@ -166,7 +166,7 @@ class PusHandler(SpecificApidHandlerBase): wrapper: VerificationWrapper, printer: FsfwTmTcPrinter, raw_logger: RawTmtcTimedLogWrapper, - hk_level: int + hk_level: int, ): super().__init__(PUS_APID, None) self.printer = printer @@ -177,7 +177,9 @@ class PusHandler(SpecificApidHandlerBase): def handle_tm(self, packet: bytes, _user_args: any): # with open("tc.bin", "wb") as of: # of.write(packet) - pus_factory_hook(packet, self.verif_wrapper, self.printer, self.raw_logger, self.hk_level) + pus_factory_hook( + packet, self.verif_wrapper, self.printer, self.raw_logger, self.hk_level + ) class UnknownApidHandler(GenericApidHandlerBase): @@ -408,7 +410,7 @@ def setup_tmtc_handlers( printer: FsfwTmTcPrinter, raw_logger: RawTmtcTimedLogWrapper, gui: bool, - hk_level: int + hk_level: int, ) -> (CcsdsTmHandler, TcHandler): cfdp_in_ccsds_wrapper = setup_cfdp_handler() verification_wrapper = VerificationWrapper(