Index: leahi_dialin/td/modules/treatment.py =================================================================== diff -u --- leahi_dialin/td/modules/treatment.py (revision 0) +++ leahi_dialin/td/modules/treatment.py (revision 0c1da102269169d5b63f91e146488b20eaa883a7) @@ -0,0 +1,212 @@ +########################################################################### +# +# 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 Index: leahi_dialin/td/treatment_delivery.py =================================================================== diff -u -rbd45f04f6e4f66ec2b121f496bc8db20ca00a6b8 -r0c1da102269169d5b63f91e146488b20eaa883a7 --- leahi_dialin/td/treatment_delivery.py (.../treatment_delivery.py) (revision bd45f04f6e4f66ec2b121f496bc8db20ca00a6b8) +++ leahi_dialin/td/treatment_delivery.py (.../treatment_delivery.py) (revision 0c1da102269169d5b63f91e146488b20eaa883a7) @@ -23,6 +23,7 @@ from .modules.buttons import TDButtons from .modules.pressure_sensors import TDPressureSensors from .modules.switches import TDSwitches +from .modules.treatment import TDTreatment from .modules.valves import TDValves from .modules.voltages import TDVoltages @@ -122,6 +123,7 @@ self.buttons = TDButtons(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.valves = TDValves(self.can_interface, self.logger) self.voltages = TDVoltages(self.can_interface, self.logger) @@ -267,7 +269,7 @@ payload = integer_to_bytearray(new_mode) message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_td_ch_id, - message_id=MsgIds.MSG_ID_TD_SET_OP_MODE_REQUEST.value, + message_id=MsgIds.MSG_ID_TD_OP_MODE_OVERRIDE_REQUEST.value, payload=payload) self.logger.debug("Requesting TD mode change to " + str(new_mode)) @@ -318,5 +320,48 @@ self.logger.debug("Sending an UI version request to the TD.") self.can_interface.send(message, 0) + 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 + + +