########################################################################### # # 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 ui_proxy.py # # @author (last) Zoltan Miskolci # @date (last) 05-May-2026 # @author (original) Peter Lucia # @date (original) 02-Apr-2020 # ############################################################################ # Module imports from logging import Logger from typing import List, Union # Project imports from leahi_dialin.common.constants import ACCEPTED from leahi_dialin.common.generic_defs import DataTypes from leahi_dialin.common.msg_ids import MsgIds from leahi_dialin.common.msg_defs import RequestRejectReasons from leahi_dialin.common.override_templates import cmd_generic_override from leahi_dialin.protocols.CAN import CanMessenger, CanChannels from leahi_dialin.utils.abstract_classes import AbstractSubSystem from leahi_dialin.utils.conversions import integer_to_bytearray, unsigned_byte_to_bytearray, float_to_bytearray, \ short_to_bytearray class UIProxy(AbstractSubSystem): """ Treatment Delivery (TD) Dialin API sub-class for UI proxy ( injection ) related commands. """ SW_COMPATIBILITY_REV = 0 def __init__(self, can_interface: CanMessenger, logger: Logger): """ DDProxy constructor @param can_interface: the CAN interface 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_to_ui_ch_id, message_id = MsgIds.MSG_ID_TD_RESP_TREATMENT_PARAMS_TO_VALIDATE.value, function = self._handler_tx_parameters_val_resp_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_to_ui_ch_id, message_id = MsgIds.MSG_ID_TD_RESP_ULTRAFILTRATION_VOLUME_TO_VALIDATE.value, function = self._handler_uf_val_resp_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_to_ui_ch_id, message_id = MsgIds.MSG_ID_TD_RESP_INITIATE_TREATMENT_WORKFLOW.value, function = self._handler_initiate_treatment_resp_sync) self.can_interface.register_receiving_publication_function(channel_id = CanChannels.td_to_ui_ch_id, message_id = MsgIds.MSG_ID_TD_UF_PAUSE_RESUME_RESPONSE.value, function = self._handler_uf_pause_resp_sync) # MSG_ID_TD_RESP_TREATMENT_PARAMS_TO_VALIDATE self.tx_param_ack = False self.tx_param_reject_reason = { 'treatment_modality': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'hdf_dilution': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'blood_flow_rate': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'dialysate_flow_rate': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'tx_duration': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'heparin_duration': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'heparin_type': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'dry_bicarb_cart_size': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'sodium': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'bicarb_concentrate': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'dialyzer_type': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'fluid_bolus_volume': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'bp_interval': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'rinseback_volume': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'hepatitis_b_status': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'acid_concentrate': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'subst_fluid_volume': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'heparin_bolus_volume': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'heparin_delivery_rate': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'dialysate_temperature': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'uf_pre_weight': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'uf_est_target_weight': RequestRejectReasons.REQUEST_REJECT_REASON_NONE, 'uf_volume': RequestRejectReasons.REQUEST_REJECT_REASON_NONE } self.tx_params_validate_timestamp = 0 #: The timestamp of the latest treatment parameters validate response message # MSG_ID_TD_RESP_ULTRAFILTRATION_VOLUME_TO_VALIDATE self.uf_volume_ack = False self.uf_volume_reject_reason = RequestRejectReasons.REQUEST_REJECT_REASON_NONE self.uf_volume_ml = 0.0 self.uf_volume_timestamp = 0 #: The timestamp of the latest ultrafiltration validate response message # MSG_ID_TD_RESP_INITIATE_TREATMENT_WORKFLOW self.initiate_tx_ack = False self.initiate_tx_reject_reason = RequestRejectReasons.REQUEST_REJECT_REASON_NONE self.initiate_tx_timestamp = 0 #: The timestamp of the latest initiate treatment reponse message # MSG_ID_TD_UF_PAUSE_RESUME_RESPONSE self.uf_pause_resume_ack = False self.uf_pause_resume_reject_reason = RequestRejectReasons.REQUEST_REJECT_REASON_NONE self.uf_pause_resume_timestamp = 0 #: The timestamp of the latest pause ultrafiltration response message # ============================================================ Properties ============================================================ @property def tx_param_ack(self) -> bool: """ The Treatment Parameter Validate message's Ack data """ return self._tx_param_ack @tx_param_ack.setter def tx_param_ack(self, value: bool): self._tx_param_ack = value @property def tx_param_reject_reason(self) -> dict: """ The Treatment Parameter Validate Reject Reasons data """ return self._tx_param_reject_reason @tx_param_reject_reason.setter def tx_param_reject_reason(self, value: Union[int, RequestRejectReasons]): if isinstance(value, int): self._tx_param_reject_reason = RequestRejectReasons(value) else: self._tx_param_reject_reason = value @property def uf_volume_ack(self) -> bool: """ The Ultrafiltration Validate message's Ack data """ return self._uf_volume_ack @uf_volume_ack.setter def uf_volume_ack(self, value: bool): self._uf_volume_ack = value @property def uf_volume_reject_reason(self) -> RequestRejectReasons: """ The Ultrafiltration Validate Reject Reasons data """ return self._uf_volume_reject_reason @uf_volume_reject_reason.setter def uf_volume_reject_reason(self, value: Union[int, RequestRejectReasons]): if isinstance(value, int): self._uf_volume_reject_reason = RequestRejectReasons(value) else: self._uf_volume_reject_reason = value @property def uf_volume_ml(self) -> float: """ The Ultrafiltration Volume (mL) data """ return self._uf_volume_ml @uf_volume_ml.setter def uf_volume_ml(self, value: float): self._uf_volume_ml = value @property def initiate_tx_ack(self) -> bool: """ The Initiate Treatment message's Ack data """ return self._initiate_tx_ack @initiate_tx_ack.setter def initiate_tx_ack(self, value: bool): self._initiate_tx_ack = value @property def initiate_tx_reject_reason(self) -> RequestRejectReasons: """ The Initiate Treatment Reject Reasons data """ return self._initiate_tx_reject_reason @initiate_tx_reject_reason.setter def initiate_tx_reject_reason(self, value: Union[int, RequestRejectReasons]): if isinstance(value, int): self._initiate_tx_reject_reason = RequestRejectReasons(value) else: self._initiate_tx_reject_reason = value @property def uf_pause_resume_ack(self) -> bool: """ The Pause Ultrafiltration message's Ack data """ return self._uf_pause_resume_ack @uf_pause_resume_ack.setter def uf_pause_resume_ack(self, value: bool): self._uf_pause_resume_ack = value @property def uf_pause_resume_reject_reason(self) -> RequestRejectReasons: """ The Pause Ultrafiltration Reject Reasons data """ return self._uf_pause_resume_reject_reason @uf_pause_resume_reject_reason.setter def uf_pause_resume_reject_reason(self, value: Union[int, RequestRejectReasons]): if isinstance(value, int): self._uf_pause_resume_reject_reason = RequestRejectReasons(value) else: self._uf_pause_resume_reject_reason = value # ============================================================ Handlers ============================================================ def _handler_tx_parameters_val_resp_sync(self, message: dict, timestamp=0.0) -> None: """ Handles treatment parameters validation response from the TD. Treatment parameters acceptance and rejection are captured. :param message: the published TD treatment parameters validation response message. :param timestamp: timestamp in epoch time of the received message :return: none """ msg_list = [] msg_list.append((type(self).tx_param_ack, DataTypes.BOOL)) msg_list.append((self.tx_param_reject_reason, 'treatment_modality', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'hdf_dilution', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'blood_flow_rate', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'dialysate_flow_rate', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'tx_duration', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'heparin_duration', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'heparin_type', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'dry_bicarb_cart_size', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'sodium', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'bicarb_concentrate', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'dialyzer_type', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'fluid_bolus_volume', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'bp_interval', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'rinseback_volume', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'hepatitis_b_status', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'acid_concentrate', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'subst_fluid_volume', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'heparin_bolus_volume', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'heparin_delivery_rate', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'dialysate_temperature', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'uf_pre_weight', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'uf_est_target_weight', DataTypes.U32)) msg_list.append((self.tx_param_reject_reason, 'uf_volume', DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) self.tx_params_validate_timestamp = timestamp def _handler_uf_val_resp_sync(self, message: dict, timestamp=0.0) -> None: """ Handles ultrafiltration volume validation response from the TD. Ultrafiltration volume acceptance and rejection are captured. :param message: the published TD ultrafiltration volume validation response message. :param timestamp: timestamp in epoch time of the received message :return: none """ msg_list = [] msg_list.append((type(self).uf_pause_resume_ack, DataTypes.BOOL)) msg_list.append((type(self).uf_pause_resume_reject_reason, DataTypes.U32)) msg_list.append((type(self).uf_volume_ml, DataTypes.F32)) self.process_into_vars(decoder_list = msg_list, message = message) self.uf_volume_timestamp = timestamp def _handler_initiate_treatment_resp_sync(self, message: dict, timestamp=0.0) -> None: """ Handles inititate treatment response from the TD. Treatment acceptance and rejection are captured. :param message: the published TD inititate treatment response message. :param timestamp: timestamp in epoch time of the received message :return: none """ msg_list = [] msg_list.append((type(self).initiate_tx_ack, DataTypes.BOOL)) msg_list.append((type(self).initiate_tx_reject_reason, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) self.initiate_tx_timestamp = timestamp def _handler_uf_pause_resp_sync(self, message: dict, timestamp=0.0) -> None: """ Handles ultrafiltration pause response from the TD. Ultrafiltration pause acceptance and rejection are captured. :param message: the published TD inititate treatment response message. :param timestamp: timestamp in epoch time of the received message :return: none """ msg_list = [] msg_list.append((type(self).uf_pause_resume_ack, DataTypes.BOOL)) msg_list.append((type(self).uf_pause_resume_reject_reason, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, bmessage = message) self.uf_pause_resume_timestamp = timestamp # ============================================================ Overrides and Requests ============================================================ def cmd_send_ui_treatment_param_data(self, final_confirmation: int, treatment_modality: int, hdf_dilution: int, blood_flow_rate: int, dialysate_flow_rate: int, treatment_duration: int, heparin_duration: int, heparin_type: int, dry_bicarb_cart_size: int, sodium: int, bicarb_concentrate: int, dialyzer_type: int, fluid_bolus_volume: int, bp_interval: int, rinseback_volume: int, hepatitis_b_status: int, acid_concentrate: int, subst_fluid_volume: float, heparin_bolus_volume: float, heparin_delivery_rate: float, dialysate_temp: float, uf_pre_weight: float, uf_est_target_weight: float, uf_volume: float) -> None: """ Constructs and sends a UI set treatment parameters message. :param final_confirmation: (int) 0 - if only validating parameters, 1 - if validating and moving the state forward :param treatment_modality: (int) type of treatment: 0-HD, 1-HDF :param hdf_dilution: (int) HDF dilution option (TDTreatmentHDFDilution) :param blood_flow_rate: (int) blood flow rate (in mL/min) :param dialysate_flow_rate: (int) dialysate flow rate (in mL/min) :param treatment_duration: (int) treatment duration (in min) :param heparin_duration: (int) heparin delivery duration (in min) :param heparin_type: (int) heparin Type (HeparinTypes) :param dry_bicarb_cart_size: (int) dry bicarb cart size option :param sodium: (int) sodium (in mEq/L) :param bicarb_concentrate: (int) bicarbonate concentrate type (BicarbTypes) :param dialyzer_type: (int) dialyzer type (DialyzerTypes) :param fluid_bolus_volume: (int) fluid bolus volume (in mL) :param bp_interval: (int) blood pressure measurement interval (in min) :param rinseback_volume: (int) rinse back volume (in mL) :param hepatitis_b_status: (int) Hepatitis B status option (TDTreatmentHepatitisB) :param acid_concentrate: (int) acid concentrate type (AcidTypes) :param subst_fluid_volume: (float) substitution fluid volume (in L) :param heparin_bolus_volume: (float) heparin bolus volume (in mL) :param heparin_delivery_rate: (float) heparin delivery rate (in mL/hr) :param dialysate_temp: (float) dialysate temperature (in deg C) :param uf_pre_weight: (float) patient pre weight prior to treatment (in Kilogram) :param uf_est_target_weight: (float) patient estimated target weight after the treatment (in Kilogram) :param uf_volume: (float) ultrafiltration volume (in L) :return: None """ payload = integer_to_bytearray(final_confirmation) payload += integer_to_bytearray(treatment_modality) payload += integer_to_bytearray(hdf_dilution) payload += integer_to_bytearray(blood_flow_rate) payload += integer_to_bytearray(dialysate_flow_rate) payload += integer_to_bytearray(treatment_duration) payload += integer_to_bytearray(heparin_duration) payload += integer_to_bytearray(heparin_type) payload += integer_to_bytearray(dry_bicarb_cart_size) payload += integer_to_bytearray(sodium) payload += integer_to_bytearray(bicarb_concentrate) payload += integer_to_bytearray(dialyzer_type) payload += integer_to_bytearray(fluid_bolus_volume) payload += integer_to_bytearray(bp_interval) payload += integer_to_bytearray(rinseback_volume) payload += integer_to_bytearray(hepatitis_b_status) payload += integer_to_bytearray(acid_concentrate) payload += float_to_bytearray(subst_fluid_volume) payload += float_to_bytearray(heparin_bolus_volume) payload += float_to_bytearray(heparin_delivery_rate) payload += float_to_bytearray(dialysate_temp) payload += float_to_bytearray(uf_pre_weight) payload += float_to_bytearray(uf_est_target_weight) payload += float_to_bytearray(uf_volume) cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_UI_TREATMENT_PARAMS_TO_VALIDATE, entity_name = 'Sneding UI Treatment Parameter to TD', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_version_request(self) -> None: """ Constructs and sends a UI version request broadcast. Can be used to retrieve all sub system version responses. @return: none """ major = unsigned_byte_to_bytearray(0) minor = unsigned_byte_to_bytearray(0) micro = unsigned_byte_to_bytearray(0) build = short_to_bytearray(0) compatibility = integer_to_bytearray(self.SW_COMPATIBILITY_REV) payload = major + minor + micro + build + compatibility cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.ui_sync_broadcast_ch_id, msg_id = MsgIds.MSG_ID_FW_VERSIONS_REQUEST, entity_name = 'UI Version Request', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_initiate_treatment(self, accepted:int = ACCEPTED ) -> None: """ Constructs and sends a UI initiate treatment message. :param accepted: (int) the acceptance value :return: none """ acc = integer_to_bytearray(accepted) payload = acc cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_UI_INITIATE_TREATMENT_WORKFLOW, entity_name = 'UI Initiate Treatment Workflow', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_confirmed_treatment_parameters (self, accepted: int = ACCEPTED ) -> None: """ Constructs and sends a UI confirmation of treatment parameters message. :param accepted: (int) the acceptance value :return: none """ acc = integer_to_bytearray(accepted) payload = acc cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_UI_TREATMENT_PARAMS_CONFIRMED, entity_name = 'UI Confirm Treatment Parameters', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_pause_resume_uf(self, accepted:int = ACCEPTED, rejection_reason:int = RequestRejectReasons.REQUEST_REJECT_REASON_NONE.value ) -> None: """ Constructs and sends a UI pause/resume ultrafiltration message. :param accepted: (int) the acceptance value :param rejection_reason: (int) the rejection reason value :return: none """ acc = integer_to_bytearray(accepted) rjr = integer_to_bytearray(rejection_reason) payload = acc + rjr cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_UI_UF_PAUSE_RESUME_REQUEST, entity_name = 'UI Pause or Resume UF Request', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_validate_uf_volume (self, uf_volume: float = 0.0 ) -> None: """ Constructs and sends a UI validate ultrafiltration message. :param uf_volume: (float) the uf volume between 0.0 - 8.0 :return: none """ ufv = float_to_bytearray(uf_volume) payload = ufv cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_UI_ULTRAFILTRATION_VOLUME_TO_VALIDATE, entity_name = 'Validate UI Ultrafiltration Volume', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_alarm_list_request (self) -> None: """ Constructs and sends a UI request for the active alarm list. :return: none """ cmd_generic_override(payload = None, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_UI_ACTIVE_ALARMS_LIST_REQUEST, entity_name = 'UI Active Alarm List Request', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_silence_alarm(self, silence: int = 1) -> None: """ Constructs and sends a UI request to silence the alarm sound. :param silence: (int) 1 to silence, 0 to de-silence :return: none """ sil = float_to_bytearray(silence) payload = sil cmd_generic_override(payload = payload, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_USER_ALARM_SILENCE_REQUEST, entity_name = 'UI Alarm Silence Request', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface) def cmd_send_ui_blood_pressure_request(self) -> None: """ Constructs and sends a UI request to measure blood pressure. :return: none """ cmd_generic_override(payload = None, reset = None, channel_id = CanChannels.ui_to_td_ch_id, msg_id = MsgIds.MSG_ID_UI_BLOOD_PRESSURE_REQUEST, entity_name = 'UI Blood Pressure Request', override_text = 'N/A', logger = self.logger, can_interface = self.can_interface)