########################################################################### # # Copyright (c) 2021-2025 Diality Inc. - All Rights Reserved. # # THIS CODE MAY NOT BE COPIED OR REPRODUCED IN ANY FORM, IN PART OR IN # WHOLE, WITHOUT THE EXPLICIT PERMISSION OF THE COPYRIGHT OWNER. # # @file syringe_pump.py # # @author (last) Micahel Garthwaite # @date (last) 17-Aug-2023 # @author (original) Sean Nash # @date (original) 12-Mar-2021 # ############################################################################ # Module imports from logging import Logger # Project imports from leahi_dialin.common.constants import RESET, NO_RESET from leahi_dialin.common.generic_defs import DataTypes from leahi_dialin.common.td_defs import TDTreaHeparinStates, TDTreaSyringePumpStates, SyringePumpOperations from leahi_dialin.common.msg_ids import MsgIds from leahi_dialin.common.override_templates import cmd_generic_broadcast_interval_override, cmd_generic_override from leahi_dialin.protocols.CAN import CanMessenger, CanChannels from leahi_dialin.utils.abstract_classes import AbstractSubSystem from leahi_dialin.utils.base import publish from leahi_dialin.utils.conversions import integer_to_bytearray, float_to_bytearray class TDSyringePump(AbstractSubSystem): """ TDSyringePump Treatment Delivery (TD) Dialin API sub-class for syringe pump related commands. """ def __init__(self, can_interface: CanMessenger, logger: Logger): """ @param can_interface: Denali Can Messenger object """ super().__init__() self.can_interface = can_interface self.logger = logger if self.can_interface is not None: self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_DATA.value, function = self._handler_syringe_pump_data) self.syringe_pump_state = TDTreaSyringePumpStates.SYRINGE_PUMP_INIT_STATE.value self.heparin_state = TDTreaHeparinStates.HEPARIN_STATE_OFF.value self.syringe_pump_set_rate_ml_hr = 0.0 self.syringe_pump_meas_rate_ml_hr = 0.0 self.syringe_pump_position = 0 self.syringe_pump_volume_ml = 0.0 self.syringe_pump_safety_volume_ml = 0.0 self.syringe_pump_home_v = 0.0 self.syringe_pump_switch_v = 0.0 self.syringe_pump_force_v = 0.0 self.syringe_pump_status = 0 self.syringe_pump_encoder_status = 0 self.syringe_pump_adc_dac_status = 0 self.syringe_pump_adc_read_counter = 0 self.td_syringe_pump_timestamp = 0.0 #: The timestamp of the syringe pump message # ============================================================ Properties ============================================================ @property def syringe_pump_state(self) -> int: """ The current Srynge Pump State """ return self._syringe_pump_state @syringe_pump_state.setter def syringe_pump_state(self, value): self._syringe_pump_state = value @property def heparin_state(self) -> int: """ The current Heparin State """ return self._heparin_state @heparin_state.setter def heparin_state(self, value): self._heparin_state = value @property def syringe_pump_set_rate_ml_hr(self) -> float: """ The current Set Syringe Pump rate """ return self._syringe_pump_set_rate_ml_hr @syringe_pump_set_rate_ml_hr.setter def syringe_pump_set_rate_ml_hr(self, value): self._syringe_pump_set_rate_ml_hr = value @property def syringe_pump_meas_rate_ml_hr(self) -> float: """ The current Measured Syringe Pump rate """ return self._syringe_pump_meas_rate_ml_hr @syringe_pump_meas_rate_ml_hr.setter def syringe_pump_meas_rate_ml_hr(self, value): self._syringe_pump_meas_rate_ml_hr = value @property def syringe_pump_position(self) -> int: """ The current Syringe Pump position """ return self._syringe_pump_position @syringe_pump_position.setter def syringe_pump_position(self, value): self._syringe_pump_position = value @property def syringe_pump_volume_ml(self) -> float: """ The current Syringe Pump volume delivered in ml """ return self._syringe_pump_volume_ml @syringe_pump_volume_ml.setter def syringe_pump_volume_ml(self, value): self._syringe_pump_volume_ml = value @property def syringe_pump_safety_volume_ml(self) -> float: """ The current Syringe Pump safety volume reading """ return self._syringe_pump_safety_volume_ml @syringe_pump_safety_volume_ml.setter def syringe_pump_safety_volume_ml(self, value): self._syringe_pump_safety_volume_ml = value @property def syringe_pump_home_v(self) -> float: """ The current Syringe Pump home voltage reading """ return self._syringe_pump_home_v @syringe_pump_home_v.setter def syringe_pump_home_v(self, value): self._syringe_pump_home_v = value @property def syringe_pump_switch_v(self) -> float: """ The current Syringe Pump switch voltage reading """ return self._syringe_pump_switch_v @syringe_pump_switch_v.setter def syringe_pump_switch_v(self, value): self._syringe_pump_switch_v = value @property def syringe_pump_force_v(self) -> float: """ The current Syringe Pump force voltage reading """ return self._syringe_pump_force_v @syringe_pump_force_v.setter def syringe_pump_force_v(self, value): self._syringe_pump_force_v = value @property def syringe_pump_status(self) -> int: """ The current Syringe Pump status """ return self._syringe_pump_status @syringe_pump_status.setter def syringe_pump_status(self, value): self._syringe_pump_status = value @property def syringe_pump_encoder_status(self) -> int: """ The current Syringe Pump encoder status """ return self._syringe_pump_encoder_status @syringe_pump_encoder_status.setter def syringe_pump_encoder_status(self, value): self._syringe_pump_encoder_status = value @property def syringe_pump_adc_dac_status(self) -> int: """ The current Syringe Pump ADC & DAC status """ return self._syringe_pump_adc_dac_status @syringe_pump_adc_dac_status.setter def syringe_pump_adc_dac_status(self, value): self._syringe_pump_adc_dac_status = value @property def syringe_pump_adc_read_counter(self) -> int: """ The current Syringe Pump ADC read counter """ return self._syringe_pump_adc_read_counter @syringe_pump_adc_read_counter.setter def syringe_pump_adc_read_counter(self, value): self._syringe_pump_adc_read_counter = value # ============================================================ Handlers ============================================================ @publish(["td_syringe_pump_timestamp", "syringe_pump_state", "syringe_pump_set_rate_ml_hr", "syringe_pump_meas_rate_ml_hr", "syringe_pump_position", "syringe_pump_volume_ml", "syringe_pump_home_v", "syringe_pump_switch_v", "syringe_pump_force_v", "heparin_state", "syringe_pump_safety_volume_ml", "syringe_pump_status", "syringe_pump_encoder_status", "syringe_pump_adc_dac_status", "syringe_pump_adc_read_counter"]) def _handler_syringe_pump_data(self, message, timestamp=0.0): """ Handles published syringe pump data messages. Syringe pump data are captured for reference. @param message: published syringe pump data message @return: None """ msg_list = [] msg_list.append((type(self).syringe_pump_state, DataTypes.U32)) msg_list.append((type(self).heparin_state, DataTypes.U32)) msg_list.append((type(self).syringe_pump_set_rate_ml_hr, DataTypes.F32)) msg_list.append((type(self).syringe_pump_meas_rate_ml_hr, DataTypes.F32)) msg_list.append((type(self).syringe_pump_position, DataTypes.U32)) msg_list.append((type(self).syringe_pump_volume_ml, DataTypes.F32)) msg_list.append((type(self).syringe_pump_home_v, DataTypes.F32)) msg_list.append((type(self).syringe_pump_switch_v, DataTypes.F32)) msg_list.append((type(self).syringe_pump_force_v, DataTypes.F32)) msg_list.append((type(self).syringe_pump_safety_volume_ml, DataTypes.F32)) msg_list.append((type(self).syringe_pump_status, DataTypes.U08)) msg_list.append((type(self).syringe_pump_encoder_status, DataTypes.U08)) msg_list.append((type(self).syringe_pump_adc_dac_status, DataTypes.U08)) msg_list.append((type(self).syringe_pump_adc_read_counter, DataTypes.U08)) self.process_into_vars(decoder_list = msg_list, message = message) self.td_syringe_pump_timestamp = timestamp # ============================================================ Overrides and Requests ============================================================ def cmd_syringe_pump_operation(self, operation: int = SyringePumpOperations.SYRINGE_PUMP_OP_STOP.value, rate: float = 0.0, volume: float = 0.0) -> int: """ Constructs and sends the syringe pump operation command Constraints: Must be logged into TD. Syringe pump must be in appropriate state for the given operation. Given rate/volume (when applicable) must be within valid range for the given operation. The following Treatment parameters should be set before executing operations 3,4 & 5: - Treatment Duration - Heparin Pre-Stop Time - Rate if volume is being sent - Volume if rate is being sent Failure to set these treatment parameters may result in faulting the TD device. @param operation: unsigned int - ID of operation being requested @param rate: float - target rate for given operation (if applicable) @param volume: float - target volume for given operation (if applicable) @return: 1 if successful, zero otherwise Syringe pump operation IDs: SYRINGE_PUMP_OP_STOP = 0 SYRINGE_PUMP_OP_RETRACT = 1 SYRINGE_PUMP_OP_SEEK = 2 SYRINGE_PUMP_OP_PRIME = 3 SYRINGE_PUMP_OP_BOLUS = 4 SYRINGE_PUMP_OP_CONTINUOUS = 5 """ op = integer_to_bytearray(operation) rat = float_to_bytearray(rate) vol = float_to_bytearray(volume) payload = op + rat + vol return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_OPERATION_REQUEST, entity_name = f'TD Syringe Pump Operation', override_text = f'{operation} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_data_broadcast_interval_override(self, ms: int = 1000, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump data 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 = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_HD_SYRINGE_PUMP_PUBLISH_INTERVAL_OVERRIDE, module_name = 'TD Syringe Pump', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_meas_rate_override(self, rate: float = 0.0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump measured rate override command Constraints: Must be logged into TD. @param rate: float - rate (in mL/hr) to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) rat = float_to_bytearray(rate) payload = rst + rat return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_RATE_OVERRIDE_REQUEST, entity_name = f'TD syringe pump measured rate', override_text = f'{rate} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_meas_force_override(self, volts: float = 0.0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump measured force override command Constraints: Must be logged into TD. @param volts: float - volts to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) vol = float_to_bytearray(volts) payload = rst + vol return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_FORCE_OVERRIDE_REQUEST, entity_name = f'TD syringe pump measured force', override_text = f'{volts} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_meas_home_override(self, volts: float = 0.0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump measured home override command Constraints: Must be logged into TD. @param volts: float - volts to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) vol = float_to_bytearray(volts) payload = rst + vol return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_HOME_OVERRIDE_REQUEST, entity_name = f'TD syringe pump measured home', override_text = f'{volts} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_meas_position_override(self, position: int = 0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump measured position override command Constraints: Must be logged into TD. @param position: integer - position (in encoder counts) to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) pos = integer_to_bytearray(position) payload = rst + pos return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_POSITION_OVERRIDE_REQUEST, entity_name = f'TD syringe pump measured position', override_text = f'{position} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_meas_volume_override(self, volume: float = 0.0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump measured volume override command Constraints: Must be logged into TD. @param volume: float - volume (in mL) to override with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) vol = float_to_bytearray(volume) payload = rst + vol return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_VOLUME_OVERRIDE_REQUEST, entity_name = f'TD syringe pump volume delivered', override_text = f'{volume} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_status_override(self, status: int = 0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump status override command Constraints: Must be logged into TD. @param status: integer - status (0..255) @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) sts = integer_to_bytearray(status & 0x000000FF) payload = rst + sts return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_STATUS_OVERRIDE_REQUEST, entity_name = f'TD syringe pump status', override_text = f'{status} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_encoder_status_override(self, status: int = 0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump encoder status override command Constraints: Must be logged into TD. @param status: integer - encoder status (0..255) @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) sts = integer_to_bytearray(status & 0x000000FF) payload = rst + sts return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_SYRINGE_PUMP_ENCODER_STATUS_OVERRIDE_REQUEST, entity_name = f'TD syringe pump encoder status', override_text = f'{status} ', logger = self.logger, can_interface = self.can_interface) def cmd_syringe_pump_adc_dac_status_override(self, status: int = 0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump ADC & DAC status override command Constraints: Must be logged into TD. @param status: integer - ADC and DAC status (0..255) @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) sts = integer_to_bytearray(status & 0x000000FF) payload = rst + sts return cmd_generic_override( payload=payload, reset=NO_RESET, channel_id=CanChannels.dialin_to_td_ch_id, msg_id=MsgIds.MSG_ID_TD_SYRINGE_PUMP_ADC_DAC_STATUS_OVERRIDE_REQUEST, entity_name=f'TD syringe pump ADC & DAC status', override_text=f'{status} ', logger=self.logger, can_interface=self.can_interface) def cmd_syringe_pump_adc_read_counter_override(self, counter: int = 0, reset: int = NO_RESET) -> int: """ Constructs and sends the syringe pump ADC read counter override command Constraints: Must be logged into TD. @param counter: integer - status (0..255) @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) ctr = integer_to_bytearray(counter & 0x000000FF) payload = rst + ctr return cmd_generic_override( payload=payload, reset=NO_RESET, channel_id=CanChannels.dialin_to_td_ch_id, msg_id=MsgIds.MSG_ID_TD_SYRINGE_PUMP_ADC_READ_COUNTER_OVERRIDE_REQUEST, entity_name=f'TD syringe pump ADC read counter', override_text=f'{counter} ', logger=self.logger, can_interface=self.can_interface) def cmd_heparin_target_rate_override(self, rate: float, reset: int = NO_RESET) -> int: """ Constructs and sends the heparin bolus target rate value override command Constraints: Must be logged into TD. @param rate: (float) the heparin bolus target rate to be set in mL/hour @param reset: (int) 1 to reset a previous override, 0 to override @return 1 if successful, zero otherwise """ reset_value = integer_to_bytearray(reset) vlu = float_to_bytearray(rate) # TD expects the rate in mL/hour payload = reset_value + vlu return cmd_generic_override( payload=payload, reset=NO_RESET, channel_id=CanChannels.dialin_to_td_ch_id, msg_id=MsgIds.MSG_ID_TD_HEPARIN_BOLUS_TARGET_RATE_OVERRIDE_REQUEST, entity_name=f'TD heparin bolus target rate value', override_text=f'{rate} ', logger=self.logger, can_interface=self.can_interface) def cmd_calibrate_force_sensor(self) -> int: """ Constructs and sends the set syringe pump DAC vRef. The value the DAC is set to is within TD Calibration Record. Constraints: Must be logged into TD. TD NVM heparin calibration record must be up to date (min: 0.0, max: 3.3) Syringe Pump must be in off state @return: 1 if successful, zero otherwise """ return cmd_generic_override( payload=None, reset=NO_RESET, channel_id=CanChannels.dialin_to_td_ch_id, msg_id=MsgIds.MSG_ID_TD_SYRINGE_PUMP_FORCE_SENSOR_CALIBRATION_REQUEST, entity_name=f'TD syringe pump calibrate force sensor', override_text=f' ', logger=self.logger, can_interface=self.can_interface)