########################################################################### # # Copyright (c) 2020-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 treatment.py # # @author (last) Michael Garthwaite # @date (last) 22-Apr-2025 # @author (original) Michael Garthwaite # @date (original) 22-Apr-2025 # ############################################################################ import struct from logging import Logger from .constants import RESET, NO_RESET from leahi_dialin.common.msg_defs import MsgIds, MsgFieldPositions from leahi_dialin.protocols.CAN import DenaliMessage, DenaliChannels from leahi_dialin.utils.base import AbstractSubSystem, publish from leahi_dialin.utils.checks import check_broadcast_interval_override_ms from leahi_dialin.utils.conversions import integer_to_bytearray, float_to_bytearray class TDTreatment(AbstractSubSystem): """ Treatment Delivery (TD) Dialin API sub-class for treatment related commands. """ def __init__(self, can_interface, logger: Logger): """ TDTreatment constructor """ super().__init__() self.can_interface = can_interface self.logger = logger if self.can_interface is not None: channel_id = DenaliChannels.td_sync_broadcast_ch_id msg_id = MsgIds.MSG_ID_TD_TREATMENT_PARAM_RANGES.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_treatment_param_ranges_sync) msg_id = MsgIds.MSG_ID_TD_SALINE_BOLUS_DATA.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_saline_bolus_sync) msg_id = MsgIds.MSG_ID_TD_ULTRAFILTRATION_DATA.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_uf_sync) msg_id = MsgIds.MSG_ID_TD_TREATMENT_TIME_DATA.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_treatment_time_sync) msg_id = MsgIds.MSG_ID_TD_TREATMENT_STATE_DATA.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_treatment_state_sync) # Treatment param ranges data self.min_tx_time = 0 self.max_tx_time = 0 self.min_uf_volume = 0.0 self.max_uf_volume = 0.0 self.min_dial_rate = 0 self.max_dial_rate = 0 self.tx_params_timestamp = 0 # Saline Bolus data self.tgt_saline_volume = 0 # in mL self.cum_saline_volume = 0.0 # in mL self.bol_saline_volume = 0.0 # in mL self.saline_bolus_state = 0 self.saline_bolus_timestamp = 0 # UF data self.set_uf_volume = 0.0 # in L self.tgt_uf_rate = 0.0 # in L/hr self.uf_volume_delivered = 0.0 # in L self.uf_state = 0 self.uf_timestamp = 0 # Treatment Time Data self.tx_time_prescribed = 0 self.tx_time_elapsed = 0 self.tx_time_remaining = 0 self.tx_time_timestamp = 0 # Treatment State Data self.tx_sub_mode = 0 self.blood_prime_state = 0 self.dialysis_state = 0 self.isolated_uf_state = 0 self.tx_stop_state = 0 self.rinseback_state = 0 self.tx_recirc_state = 0 self.tx_end_state = 0 self.tx_state_timestamp = 0 @publish(["min_tx_time","max_tx_time","min_uf_volume","max_uf_volume", "min_dial_rate","max_dial_rate","tx_params_timestamp"]) def _handler_treatment_param_ranges_sync(self, message, timestamp=0.0): """ Handles published treatment parameter range data messages. @param message: published treatment parameter range data message @return: none """ self.min_tx_time = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.max_tx_time = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.min_uf_volume = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.max_uf_volume = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_4:MsgFieldPositions.END_POS_FIELD_4]))[0] self.min_dial_rate = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_5:MsgFieldPositions.END_POS_FIELD_5]))[0] self.max_dial_rate = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_6:MsgFieldPositions.END_POS_FIELD_6]))[0] self.tx_params_timestamp = timestamp @publish(["tgt_saline_volume","cum_saline_volume","bol_saline_volume", "saline_bolus_state","saline_bolus_timestamp"]) def _handler_saline_bolus_sync(self, message, timestamp=0.0): """ Handles published saline bolus data messages. @param message: published saline bolus data message @return: none """ self.tgt_saline_volume = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.cum_saline_volume = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.bol_saline_volume = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.saline_bolus_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_4:MsgFieldPositions.END_POS_FIELD_4]))[0] self.saline_bolus_timestamp = timestamp @publish(["set_uf_volume","tgt_uf_rate","uf_volume_delivered", "uf_state","uf_timestamp"]) def _handler_uf_sync(self, message, timestamp=0.0): """ Handles published ultrafiltration data messages. @param message: published ultrafiltration data message @return: none """ self.set_uf_volume = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.tgt_uf_rate = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.uf_volume_delivered = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.uf_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_4:MsgFieldPositions.END_POS_FIELD_4]))[0] self.uf_timestamp = timestamp @publish(["tx_time_prescribed","tx_time_elapsed","tx_time_remaining", "tx_time_timestamp"]) def _handler_treatment_time_sync(self, message, timestamp=0.0): """ Handles published treatment time data messages. @param message: published treatment time data message @return: none """ self.tx_time_prescribed = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.tx_time_elapsed = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.tx_time_remaining = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.tx_time_timestamp = timestamp @publish(["tx_sub_mode","blood_prime_state","dialysis_state","isolated_uf_state", "tx_stop_state","rinseback_state","tx_recirc_state","tx_end_state","tx_state_timestamp"]) def _handler_treatment_state_sync(self, message, timestamp=0.0): """ Handles published treatment state data messages. @param message: published treatment state data message @return: none """ self.tx_sub_mode = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.blood_prime_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.dialysis_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.isolated_uf_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_4:MsgFieldPositions.END_POS_FIELD_4]))[0] self.tx_stop_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_5:MsgFieldPositions.END_POS_FIELD_5]))[0] self.rinseback_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_6:MsgFieldPositions.END_POS_FIELD_6]))[0] self.tx_recirc_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_7:MsgFieldPositions.END_POS_FIELD_7]))[0] self.tx_end_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_8:MsgFieldPositions.END_POS_FIELD_8]))[0] self.tx_state_timestamp = timestamp