########################################################################### # # Copyright (c) 2020-2024 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 td_proxy.py # # @author (last) Micahel Garthwaite # @date (last) 18-Aug-2023 # @author (original) Peter Lucia # @date (original) 02-Apr-2020 # ############################################################################ import struct from logging import Logger from leahi_dialin.common.msg_defs import MsgIds, MsgFieldPositions from leahi_dialin.protocols.CAN import DenaliMessage, DenaliCanMessenger, DenaliChannels from leahi_dialin.utils.base import AbstractSubSystem, publish from leahi_dialin.utils.conversions import integer_to_bytearray, byte_to_bytearray, float_to_bytearray class TDProxy(AbstractSubSystem): """ Dialysate Delivery (DD) Dialin API sub-class for TD proxy ( injection ) related commands. """ def __init__(self, can_interface: DenaliCanMessenger, logger: Logger): """ TDProxy constructor @param can_interface: the Denali CAN interface object """ super().__init__() self.can_interface = can_interface self.logger = logger if self.can_interface is not None: channel_id = DenaliChannels.td_to_dd_ch_id msg_id = MsgIds.MSG_ID_DD_GEN_DIALYSATE_REQUEST_DATA.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_dialysate_delivery_request_response) self.dialysate_delivery_request_start = 0 self.dialysate_delivery_request_dial_rate = 0.0 self.dialysate_delivery_request_uf_rate = 0.0 self.dialysate_delivery_request_dial_temp = 0.0 self.dialysate_delivery_request_bypass = 0 self.dialysate_delivery_request_acid = 0 self.dialysate_delivery_request_bicarb = 0 self.dd_td_to_dd_request_response_timestamp = 0.0 @publish(["dd_td_to_dd_request_response_timestamp", "dialysate_delivery_request_start","dialysate_delivery_request_dial_rate", "dialysate_delivery_request_uf_rate","dialysate_delivery_request_dial_temp", "dialysate_delivery_request_bypass","dialysate_delivery_request_acid", "dialysate_delivery_request_bicarb", ]) def _handler_dialysate_delivery_request_response(self, message, timestamp=0.0): """ Handles published dialysate delivery request response data messages. @param message: published dialysate delivery request response data message @return: None """ self.dialysate_delivery_request_start = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.dialysate_delivery_request_dial_rate = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.dialysate_delivery_request_uf_rate = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.dialysate_delivery_request_dial_temp = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_4:MsgFieldPositions.END_POS_FIELD_4]))[0] self.dialysate_delivery_request_bypass = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_5:MsgFieldPositions.END_POS_FIELD_5]))[0] self.dialysate_delivery_request_acid = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_6:MsgFieldPositions.END_POS_FIELD_6]))[0] self.dialysate_delivery_request_bicarb = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_7:MsgFieldPositions.END_POS_FIELD_7]))[0] self.dd_td_to_dd_request_response_timestamp = timestamp def cmd_td_send_dd_gen_dialysate_request(self, start: bool = 0, dial_rate: float = 0, uf_rate: float = 0, dial_temp: float = 0, bypass: int = 0, acid: int = 0, bicarb: int = 0): """ Constructs and sends a TD gen dialysate request to the DD. @return: none """ stt = integer_to_bytearray(start) dial = float_to_bytearray(dial_rate) uf = float_to_bytearray(uf_rate) temp = float_to_bytearray(dial_temp) byp = integer_to_bytearray(bypass) acd = integer_to_bytearray(acid) bic = integer_to_bytearray(bicarb) payload = stt + dial + uf + temp + byp + acd + bic message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_dd_ch_id, message_id=MsgIds.MSG_ID_DD_GEN_DIALYSATE_REQUEST_DATA.value, payload=payload) self.logger.debug("Sending TD gen dialysate data request to DD.") self.can_interface.send(message, 0) def cmd_td_send_dd_start_pre_gen_request(self, start: bool = 0, dialysate_rate: float = 0.0, dialysate_temp: float = 0.0, acid_type: int = 0, bicarb_type: int = 0 ): """ Constructs and sends a TD start pre gen dialysate request to the DD. :param start: start and stop boolean :param dialysate_rate: dialysate rate in ml/min :param dialysate_temp: dialysate temp in c :param acid_type: acid type :param bicarb_type: bicarb type :return: """ stt = integer_to_bytearray(start) dial = float_to_bytearray(dialysate_rate) temp = float_to_bytearray(dialysate_temp) acd = integer_to_bytearray(acid_type) bic = integer_to_bytearray(bicarb_type) payload = stt + dial + temp + acd + bic message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_dd_ch_id, message_id=MsgIds.MSG_ID_DD_PRE_GEN_DIALYSATE_REQUEST_DATA.value, payload=payload) self.logger.debug("Sending TD start pre gen request to DD.") self.can_interface.send(message, 0) def cmd_td_override_treatment_parameter(self, paramID: int, value, reset: int = 0): """ Constructs and sends a treatment parameter override request. @param paramID : integer - Treatment parameter ID @param value : float or integer - Override value @param reset : integer 1 to reset override, 0 to apply override @returns None """ rst = integer_to_bytearray(reset) pid = integer_to_bytearray(paramID) # Dialysate rate, UF rate and dialysate temperature are f32 float_params = [ 0, 1, 2 ] # Acid type and bicarbonate type are u32 uint_params = [ 3, 4 ] if paramID in float_params: val = float_to_bytearray(float(value)) elif paramID in uint_params: val = integer_to_bytearray(int(value)) else: val = integer_to_bytearray(int(value)) payload = rst + val + pid message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_dd_ch_id, message_id=MsgIds.MSG_ID_DD_TREATMENT_PARAMS_OVERRIDE_REQUEST.value, payload=payload) self.logger.debug("Sending DD treatment parameter override: id={}, value={}, reset={}".format( paramID, value, reset)) self.can_interface.send(message, 0)