########################################################################### # # 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 rinse_pump.py # # @author (last) Zoltan Miskolci # @date (last) 07-Jan-2026 # @author (original) Jonny Paguio # @date (original) 13-Oct-2025 # ############################################################################ # Module imports import struct from logging import Logger # Project imports from leahi_dialin.common.constants import NO_RESET from leahi_dialin.common import td_enum_repository from leahi_dialin.common.generic_defs import DataTypes from leahi_dialin.common.msg_defs 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 TDHemoDiaFiltration(AbstractSubSystem): """ HDF module HemoDiaFiltration (TD) Dialin API sub-class for HDF 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_HEMODIAFILTRATION_DATA.value, function = self._handler_hdf_sync) self.td_hdf_data = { td_enum_repository.TDHemodiafiltrationAttributes.HDF_EXEC_STATE: 0, td_enum_repository.TDHemodiafiltrationAttributes.HDF_REQUESTED_RATE: 0.0, td_enum_repository.TDHemodiafiltrationAttributes.HDF_TARGET_TMP: 0.0, td_enum_repository.TDHemodiafiltrationAttributes.HDF_CURRENT_RATE: 0.0, td_enum_repository.TDHemodiafiltrationAttributes.HDF_CURRENT_TMP: 0.0, td_enum_repository.TDHemodiafiltrationAttributes.HDF_REQUESTED_VOLUME: 0.0, td_enum_repository.TDHemodiafiltrationAttributes.HDF_TOTAL_VOLUME: 0.0, td_enum_repository.TDHemodiafiltrationAttributes.HDF_REQUESTED: 0, } self.td_hdf_timestamp = 0 #: The timestamp of the latest message @publish(["msg_id_td_hemodiafiltration_data", "td_hdf", "td_hdf_timestamp"]) def _handler_hdf_sync(self, message, timestamp=0.0): """ Handles published hdf data messages. @param message: published hdf data message @return: None """ hdf_list = [] hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_EXEC_STATE, DataTypes.U32)) hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_REQUESTED_RATE, DataTypes.F32)) hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_TARGET_TMP, DataTypes.F32)) hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_CURRENT_RATE, DataTypes.F32)) hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_CURRENT_TMP, DataTypes.F32)) hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_REQUESTED_VOLUME, DataTypes.F32)) hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_TOTAL_VOLUME, DataTypes.F32)) hdf_list.append((td_enum_repository.TDHemodiafiltrationAttributes.HDF_REQUESTED, DataTypes.U32)) self.process_into_dict(dict_to_update = self.td_hdf_data, decoder_list = hdf_list, message = message) self.td_hdf_timestamp = timestamp def cmd_hdf_data_publish_interval_override(self, ms: int, reset: int = NO_RESET) -> int: """ Constructs and sends the substitution pump data publish interval override command Constraints: Must be logged into DD. Given interval must be non-zero and a multiple of the DD 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_TD_HDF_PUBLISH_INTERVAL_OVERRIDE_REQUEST, module_name='TD HDF', logger=self.logger, can_interface=self.can_interface) def cmd_hdf_set_start_stop(self, command: int, target_tmp: float, volume: float, speed: float) -> int: """ Constructs and sends the substitution pump start stop command @param command: int - value to command the substitution pump @param target_tmp: float - target transmembrane pressure that pump will control to @param volume: float - target total volume to deliver. @param speed: float - ml/min to set the speed to. Used for open loop control only and will take precendence over closed control if non-zero. @return: 1 if successful, zero otherwise """ cmd = integer_to_bytearray(command) tgt = float_to_bytearray(target_tmp) vol = float_to_bytearray(volume) spd = float_to_bytearray(speed) payload = cmd + tgt + vol + spd return cmd_generic_override( payload=payload, reset=NO_RESET, channel_id=CanChannels.dialin_to_td_ch_id, msg_id=MsgIds.MSG_ID_TD_HDF_START_STOP_OVERRIDE_REQUEST, entity_name=f'TD HDF start stop request', override_text=str(spd), logger=self.logger, can_interface=self.can_interface) def cmd_hdf_target_tmp_override_request(self, tmp: int) -> int: """ Constructs and sends the HDF target TMP override. Constraints: Must be logged into TD. @param tmp: float @return: 1 if successful, zero otherwise """ trans = float_to_bytearray(tmp) payload = trans return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_BLOOD_PUMP_SET_SPEED_REQUEST, entity_name = f'TD Target Tmp', override_text = f'{str(trans)}', logger = self.logger, can_interface = self.can_interface) def cmd_hdf_current_rate_request(self, rpm: int) -> int: """ Constructs and sends the hdf current command Constraints: Must be logged into TD. @param rpm: integer - speed in ml/min. @return: 1 if successful, zero otherwise """ rpm = integer_to_bytearray(rpm) payload = rpm return cmd_generic_override( payload = payload, reset = NO_RESET, channel_id = CanChannels.dialin_to_td_ch_id, msg_id = MsgIds.MSG_ID_TD_BLOOD_PUMP_SET_SPEED_REQUEST, entity_name = f'TD rate', override_text = f'{str(rpm)}', logger = self.logger, can_interface = self.can_interface)