Index: leahi_dialin/td/treatment_delivery.py =================================================================== diff -u -r9357937d8acaa33c4a5a18799349fbf4aa432ac1 -rfc671af6d1a18c900812e68c42ca14da3280e4f1 --- leahi_dialin/td/treatment_delivery.py (.../treatment_delivery.py) (revision 9357937d8acaa33c4a5a18799349fbf4aa432ac1) +++ leahi_dialin/td/treatment_delivery.py (.../treatment_delivery.py) (revision fc671af6d1a18c900812e68c42ca14da3280e4f1) @@ -7,25 +7,29 @@ # # @file treatment_delivery.py # -# @author (last) Dara Navaei -# @date (last) 26-Feb-2024 +# @author (last) Zoltan Miskolci +# @date (last) 08-Jan-2026 # @author (original) Peter Lucia # @date (original) 02-Apr-2020 # ############################################################################ + +# Module imports import struct +from time import sleep +# Project imports from .modules.air_pump import TDAirPump from .modules.air_trap import TDAirTrap from .modules.alarms import TDAlarms from .modules.blood_flow import TDBloodFlow from .modules.bubble_detector import TDBubbleDetector from .modules.buttons import TDButtons -from .modules.constants import NO_RESET, RESET from .modules.ejector import TDEjector from .modules.events import TDEvents from .modules.pressure_sensors import TDPressureSensors from .modules.switches import TDSwitches +from .modules.temperature_sensors import TDTemperatureSensors from .modules.treatment import TDTreatment from .modules.td_test_configs import TDTestConfig from .modules.valves import TDValves @@ -34,13 +38,13 @@ from .proxies.dd_proxy import DDProxy from .proxies.ui_proxy import UIProxy +from ..common.constants import NO_RESET from ..common.msg_defs import MsgIds, MsgFieldPositions, MsgFieldPositionsFWVersions -from ..common.td_defs import TDOpModes +from ..common import td_enum_repository +from ..common.override_templates import cmd_generic_broadcast_interval_override, cmd_generic_override from ..protocols.CAN import DenaliMessage, DenaliCanMessenger, DenaliChannels from ..utils.base import AbstractSubSystem, publish, LogManager -from ..utils.checks import check_broadcast_interval_override_ms -from ..utils.conversions import integer_to_bytearray, unsigned_short_to_bytearray, bytearray_to_integer, \ - bytearray_to_byte +from ..utils.conversions import integer_to_bytearray, bytearray_to_byte class TD(AbstractSubSystem): @@ -99,43 +103,48 @@ self.can_interface.register_receiving_publication_function(DenaliChannels.ui_to_td_ch_id, self.msg_id_ui_version_info_response, self._handler_ui_version_response_sync) + # Dialin will send a login message during construction. This is for the leahi subsystems to start + # publishing CAN data when there is no UI connected as the UI typically does this job. + self.cmd_log_in_to_td() # create properties - self.td_op_mode_timestamp = 0.0 - self.td_debug_events_timestamp = 0.0 - self.ui_version_info_response_timestamp = 0.0 - self.td_version_response_timestamp = 0.0 - self.td_operation_mode = TDOpModes.MODE_INIT.value - self.td_operation_sub_mode = 0 - self.td_logged_in = False + self.td_op_mode_timestamp = 0.0 #: The timestamp of the latest operation mode message + self.td_debug_events_timestamp = 0.0 #: The timestamp of the latest events message + self.ui_version_info_response_timestamp = 0.0 #: The timestamp of the latest UI version info message + self.td_version_response_timestamp = 0.0 #: The timestamp of the latest TD version info message + self.td_operation_mode = td_enum_repository.TDOpModes.MODE_INIT.value #: The Operation Mode's value + self.td_operation_sub_mode = 0 #: The Operation Sub-Mode's value + self.td_logged_in = False #: The value showing if the user is logged in or not self.td_set_logged_in_status(False) - self.ui_version = None - self.td_version = None - self.td_fpga_version = None - self.td_debug_events = [''] * self._TD_DEBUG_EVENT_LIST_COUNT - self.td_debug_event_index = 0 - self.td_last_debug_event = '' + self.ui_version = None #: The UI's version value + self.td_version = None #: The TD's version value + self.td_fpga_version = None #: The TD's FPGA version value + self.td_debug_events = [''] * self._TD_DEBUG_EVENT_LIST_COUNT #: The Debug Event's list + self.td_debug_event_index = 0 #: The index of the last Event + self.td_last_debug_event = '' #: The name of the last Event # Create command groups - self.air_pump = TDAirPump(self.can_interface, self.logger) - self.air_trap = TDAirTrap(self.can_interface, self.logger) - self.alarms = TDAlarms(self.can_interface, self.logger) - self.blood_flow = TDBloodFlow(self.can_interface, self.logger) - self.bubbles = TDBubbleDetector(self.can_interface, self.logger) - self.buttons = TDButtons(self.can_interface, self.logger) - self.ejector = TDEjector(self.can_interface, self.logger) - self.events = TDEvents(self.can_interface, self.logger) - self.pressure_sensors = TDPressureSensors(self.can_interface, self.logger) - self.switches = TDSwitches(self.can_interface, self.logger) - self.treatment = TDTreatment(self.can_interface, self.logger) - self.test_configs = TDTestConfig(self.can_interface, self.logger) - self.valves = TDValves(self.can_interface, self.logger) - self.voltages = TDVoltages(self.can_interface, self.logger) + self.air_pump = TDAirPump(self.can_interface, self.logger) #: The Air Pump module + self.air_trap = TDAirTrap(self.can_interface, self.logger) #: The Air Trap module + self.alarms = TDAlarms(self.can_interface, self.logger) #: The Alarms module + self.blood_flow = TDBloodFlow(self.can_interface, self.logger) #: The Blood Flow module + self.bubbles = TDBubbleDetector(self.can_interface, self.logger) #: The Bubble Detector module + self.buttons = TDButtons(self.can_interface, self.logger) #: The Buttons module + self.ejector = TDEjector(self.can_interface, self.logger) #: The Ejector module + self.events = TDEvents(self.can_interface, self.logger) #: The Events module + self.pressure_sensors = TDPressureSensors(self.can_interface, self.logger) #: The Pressure Sensors module + self.switches = TDSwitches(self.can_interface, self.logger) #: The Switches module + self.temperature_sensors = TDTemperatureSensors(self.can_interface, self.logger) #: The Temperature Sensprs module + self.treatment = TDTreatment(self.can_interface, self.logger) #: The Treatment module + self.test_configs = TDTestConfig(self.can_interface, self.logger) #: The Test Configs module + self.valves = TDValves(self.can_interface, self.logger) #: The Valves module + self.voltages = TDVoltages(self.can_interface, self.logger) #: The Voltages module - self.dd_proxy = DDProxy(self.can_interface, self.logger) - self.ui_proxy = UIProxy(self.can_interface, self.logger) + self.dd_proxy = DDProxy(self.can_interface, self.logger) #: The DD Proxy module (imitates commands sent by DD) + self.ui_proxy = UIProxy(self.can_interface, self.logger) #: The UI Proxy module (imitates commands sent by UI) + @publish(["msg_id_td_debug_event", "td_debug_events_timestamp","td_debug_events"]) def _handler_td_debug_event_sync(self, message, timestamp = 0.0): @@ -158,15 +167,17 @@ if self.td_debug_event_index == self._TD_DEBUG_EVENT_LIST_COUNT: self.td_debug_event_index = 0 - @publish(["td_logged_in"]) + def td_set_logged_in_status(self, logged_in: bool = False): """ Callback for td logged in status change. - @param logged_in boolean logged in status for TD - @return: none + + @param logged_in: Logged in status for TD + @return: None """ self.td_logged_in = logged_in + @publish(["msg_id_td_op_mode_data", "td_op_mode_timestamp","td_operation_mode", "td_operation_sub_mode"]) def _handler_td_op_mode_sync(self, message, timestamp = 0.0): """ @@ -185,6 +196,7 @@ self.td_operation_sub_mode = smode[0] self.td_op_mode_timestamp = timestamp + @publish(["msg_id_td_version_response", "td_version, td_fpga_version"]) def _handler_td_version_response_sync(self,message, timestamp = 0.0): """ @@ -223,6 +235,7 @@ self.td_version_response_timestamp = timestamp + @publish(["msg_id_ui_version_info_response", "ui_version", "ui_version_info_response_timestamp"]) def _handler_ui_version_response_sync(self,message, timestamp = 0.0): """ @@ -258,6 +271,28 @@ self.ui_version = None self.logger.debug("Failed to retrieve UI Version.") + + def cmd_op_mode_broadcast_interval_override(self, ms: int, reset: int = NO_RESET) -> int: + """ + Constructs and sends the measured op mode broadcast interval override command + Constraints: + Must be logged into TD. + Given interval must be non-zero and a multiple of the TD general task interval (50 ms). + + @param ms: integer - interval (in ms) to override with + @param reset: integer - 1 to reset a previous override, 0 to override + @return: 1 if successful, zero otherwise + """ + return cmd_generic_broadcast_interval_override( + ms = ms, + reset = reset, + channel_id = DenaliChannels.dialin_to_td_ch_id, + msg_id = MsgIds.MSG_ID_TD_OP_MODE_PUBLISH_INTERVAL_OVERRIDE_REQUEST, + module_name = 'TD Operation Mode', + logger = self.logger, + can_interface = self.can_interface) + + def cmd_log_in_to_td(self, resend: bool = False) -> int: """ Constructs and sends a login command via CAN bus. Login required before \n @@ -266,7 +301,6 @@ @param resend: (bool) if False (default), try to login once. Otherwise, tries to login indefinitely @return: 1 if logged in, 0 if log in failed """ - message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, message_id=MsgIds.MSG_ID_TD_TESTER_LOGIN_REQUEST.value, payload=list(map(int, map(ord, self.TD_LOGIN_PASSWORD)))) @@ -289,6 +323,7 @@ self.logger.debug("Login Timeout!!!!") return False + def cmd_td_set_operation_mode(self, new_mode: int = 0) -> int: """ Constructs and sends a set operation mode request command via CAN bus. @@ -308,31 +343,20 @@ TD_OP_MODE_POST_TREATMENT = 7 @return: 1 if successful, zero otherwise - """ - payload = integer_to_bytearray(new_mode) - message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, - message_id=MsgIds.MSG_ID_TD_OP_MODE_OVERRIDE_REQUEST.value, - payload=payload) + return cmd_generic_override( + payload = payload, + reset = NO_RESET, + channel_id = DenaliChannels.dialin_to_td_ch_id, + msg_id = MsgIds.MSG_ID_TD_OP_MODE_OVERRIDE_REQUEST, + entity_name = 'TD Operation Mode', + override_text = td_enum_repository.TDOpModes(new_mode).name, + logger = self.logger, + can_interface = self.can_interface) - self.logger.debug("Requesting TD mode change to " + str(new_mode)) - # Send message - received_message = self.can_interface.send(message) - - if received_message is not None: - if received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] == 1: - self.logger.debug("Success: Mode change accepted") - else: - self.logger.debug("Failure: Mode change rejected.") - return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] - else: - self.logger.debug("TD mode change request Timeout!!!!") - return False - - def cmd_td_software_reset_request(self) -> None: """ Constructs and sends an TD software reset request via CAN bus. @@ -341,30 +365,59 @@ @return: None """ + resp = cmd_generic_override( + payload = None, + reset = NO_RESET, + channel_id = DenaliChannels.dialin_to_td_ch_id, + msg_id = MsgIds.MSG_ID_TD_SOFTWARE_RESET_REQUEST, + entity_name = 'TD Software Reset', + override_text = '', + logger = self.logger, + can_interface = self.can_interface) + + sleep(1) # Wait for reboot to happen + resp_login = self.cmd_log_in_to_td() + return resp and resp_login - message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, - message_id=MsgIds.MSG_ID_TD_SOFTWARE_RESET_REQUEST.value) - self.logger.debug("requesting TD software reset") + def cmd_td_safety_shutdown_override(self, active: int, reset: int = NO_RESET) -> int: + """ + Constructs and sends an TD safety shutdown override command via CAN bus. + Constraints: + Must be logged into TD. - # Send message - self.can_interface.send(message, 0) - self.logger.debug("Sent request to TD to reset...") - self.td_set_logged_in_status(False) + @param active: int - True to activate safety shutdown, False to deactivate + @param reset: integer - 1 to reset a previous override, 0 to override + @return: 1 if successful, zero otherwise + """ + rst = integer_to_bytearray(reset) + saf = integer_to_bytearray(active) + payload = rst + saf + return cmd_generic_override( + payload = payload, + reset = NO_RESET, + channel_id = DenaliChannels.dialin_to_td_ch_id, + msg_id = MsgIds.MSG_ID_TD_SAFETY_SHUTDOWN_OVERRIDE_REQUEST, + entity_name = 'TD Safety Shutdown', + override_text = str(active), + logger = self.logger, + can_interface = self.can_interface) + + def cmd_request_ui_version(self) -> None: """ Constructs and sends a ui version request to the TD. @return: none """ - message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, message_id=MsgIds.MSG_ID_TD_UI_VERSION_INFO_REQUEST.value) self.logger.debug("Sending an UI version request to the TD.") self.can_interface.send(message, 0) + def cmd_training_test_override(self, value: int, reset: int = NO_RESET) -> int: """ Constructs and sends a training test override request to the TD. @@ -373,109 +426,16 @@ @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ - rst = integer_to_bytearray(reset) val = integer_to_bytearray(value) payload = rst + val - message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, - message_id=MsgIds.MSG_ID_TD_TRAINING_TEST_OVERRIDE_REQUEST.value, - payload=payload) - - self.logger.debug("Sending a training test override request to the TD.") - received_message = self.can_interface.send(message) - - # If there is content... - if received_message is not None: - # self.logger.debug(received_message) - if reset == RESET: - str_res = "reset back to normal: " - else: - str_res = str(val) + " value: " - self.logger.debug("Training test value overridden to " + str_res + - str(received_message['message'][DenaliMessage.PAYLOAD_START_INDEX])) - # response payload is OK or not OK - return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] - else: - self.logger.debug("Timeout!!!!") - return False - - def cmd_op_mode_broadcast_interval_override(self, ms: int, reset: int = NO_RESET) -> int: - """ - Constructs and sends the measured op mode broadcast interval override command - Constraints: - Must be logged into TD. - Given interval must be non-zero and a multiple of the TD general task interval (50 ms). - - @param ms: integer - interval (in ms) to override with - @param reset: integer - 1 to reset a previous override, 0 to override - @return: 1 if successful, zero otherwise - """ - - if not check_broadcast_interval_override_ms(ms): - return False - - rst = integer_to_bytearray(reset) - mis = integer_to_bytearray(ms) - payload = rst + mis - - message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, - message_id=MsgIds.MSG_ID_TD_OP_MODE_PUBLISH_INTERVAL_OVERRIDE_REQUEST.value, - payload=payload) - - self.logger.debug("override op mode broadcast interval") - - # Send message - received_message = self.can_interface.send(message) - - # If there is content... - if received_message is not None: - # self.logger.debug(received_message) - if reset == RESET: - str_res = "reset back to normal: " - else: - str_res = str(ms) + " ms: " - self.logger.debug("Blood flow broadcast interval overridden to " + str_res + - str(received_message['message'][DenaliMessage.PAYLOAD_START_INDEX])) - # response payload is OK or not OK - return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] - else: - self.logger.debug("Timeout!!!!") - return False - - def cmd_td_safety_shutdown_override(self, active: int, reset: int = NO_RESET) -> int: - """ - Constructs and sends an TD safety shutdown override command via CAN bus. - Constraints: - Must be logged into TD. - - @param active: int - True to activate safety shutdown, False to deactivate - @param reset: integer - 1 to reset a previous override, 0 to override - @return: 1 if successful, zero otherwise - - """ - - rst = integer_to_bytearray(reset) - saf = integer_to_bytearray(active) - payload = rst + saf - - message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, - message_id=MsgIds.MSG_ID_TD_SAFETY_SHUTDOWN_OVERRIDE_REQUEST.value, - payload=payload) - - self.logger.debug("overriding TD safety shutdown") - - # Send message - received_message = self.can_interface.send(message) - - if received_message is not None: - if received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] == 1: - self.logger.debug("Safety shutdown signal overridden") - else: - self.logger.debug("Safety shutdown signal override failed.") - return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] - else: - self.logger.debug("Timeout!!!!") - return False - - + return cmd_generic_override( + payload = payload, + reset = reset, + channel_id = DenaliChannels.dialin_to_td_ch_id, + msg_id = MsgIds.MSG_ID_TD_TRAINING_TEST_OVERRIDE_REQUEST, + entity_name = 'TD Training Test Override', + override_text = str(value), + logger = self.logger, + can_interface = self.can_interface)