########################################################################### # # Copyright (c) 2025-2026 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 drybicart.py # # @author (last) Zoltan Miskolci # @date (last) 04-May-2026 # @author (original) Sameer Poyil # @date (original) 19-Nov-2025 # ############################################################################ # Module imports from logging import Logger from typing import Union # Project imports from leahi_dialin.common.constants import NO_RESET , RESET from leahi_dialin.common.dd_defs import DDBicarbChFillExecStates, DDDryBicartDrainExecStates, DDDryBicartFillExecStates from leahi_dialin.common.disp_defs import BicarbTypes from leahi_dialin.common.generic_defs import DataTypes from leahi_dialin.common.msg_ids 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 class DDDryBicart(AbstractSubSystem): """ DryBicart Dialysate Delivery (DD) Dialin API sub-class for Dry Bicart related commands. """ def __init__(self, can_interface: CanMessenger, logger: Logger): """ @param can_interface: 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.dd_sync_broadcast_ch_id, message_id = MsgIds.MSG_ID_DD_DRY_BICART_DATA.value, function = self._handler_dry_bicart_sync) self.dd_dry_bicart_timestamp = 0.0 #: The timestamp of the last message self.dd_dry_bicart_fill_execution_state = DDDryBicartFillExecStates.NUM_OF_DRY_BICART_FILL_EXEC_STATES self.dd_bicarb_chamber_fill_execution_state = DDBicarbChFillExecStates.NUM_OF_BICARB_CHAMBER_FILL_EXEC_STATES self.dd_dry_bicart_drain_execution_state = DDDryBicartDrainExecStates.NUM_OF_DRY_BICART_DRAIN_EXEC_STATES self.dd_dry_bicart_fill_cycle_counter = 0 self.dd_dry_bicart_max_fill_cycle_count = 0 self.dd_dry_bicart_fill_request = False self.dd_dry_bicart_drain_request = False self.dd_bicarb_chamber_fill_request = False self.dd_dry_bicart_last_fill_time = 0 self.dd_dry_bicart_current_fill_time = 0 self.dry_bicart_type = BicarbTypes.NUM_OF_BICARB_TYPES self.dry_bicart_drain_time_period = 0 # ============================================================ Properties ============================================================ @property def dd_dry_bicart_fill_execution_state(self) -> DDDryBicartFillExecStates: """ The Dry Bicart fill execution state """ return self._dry_bicart_fill_execution_state @dd_dry_bicart_fill_execution_state.setter def dd_dry_bicart_fill_execution_state(self, value: Union[int, DDDryBicartFillExecStates]): if isinstance(value, int): self._dry_bicart_fill_execution_state = DDDryBicartFillExecStates(value) else: self._dry_bicart_fill_execution_state = value @property def dd_dry_bicart_drain_execution_state(self) -> DDDryBicartDrainExecStates: """ The Dry Bicart drain execution state """ return self._dry_bicart_drain_execution_state @dd_dry_bicart_drain_execution_state.setter def dd_dry_bicart_drain_execution_state(self, value: Union[int, DDDryBicartDrainExecStates]): if isinstance(value, int): self._dry_bicart_drain_execution_state = DDDryBicartDrainExecStates(value) else: self._dry_bicart_drain_execution_state = value @property def dd_bicarb_chamber_fill_execution_state(self) -> DDBicarbChFillExecStates: """ The Bicarb chamber F fill execution state """ return self._bicarb_chamber_fill_execution_state @dd_bicarb_chamber_fill_execution_state.setter def dd_bicarb_chamber_fill_execution_state(self, value: Union[int, DDBicarbChFillExecStates]): if isinstance(value, int): self._bicarb_chamber_fill_execution_state = DDBicarbChFillExecStates(value) else: self._bicarb_chamber_fill_execution_state = value @property def dd_dry_bicart_fill_cycle_counter(self) -> int: """ The Dry Bicart fill cycle counter """ return self._dry_bicart_fill_cycle_counter @dd_dry_bicart_fill_cycle_counter.setter def dd_dry_bicart_fill_cycle_counter(self, value): self._dry_bicart_fill_cycle_counter = value @property def dd_dry_bicart_max_fill_cycle_count(self) -> int: """ The Dry Bicart max fill cycle state """ return self._dry_bicart_max_fill_cycle_count @dd_dry_bicart_max_fill_cycle_count.setter def dd_dry_bicart_max_fill_cycle_count(self, value): self._dry_bicart_max_fill_cycle_count = value @property def dd_dry_bicart_fill_request(self) -> bool: """ Is fill requested for Dry Bicart or not """ return self._dry_bicart_fill_request @dd_dry_bicart_fill_request.setter def dd_dry_bicart_fill_request(self, value): self._dry_bicart_fill_request = value @property def dd_dry_bicart_drain_request(self) -> bool: """ Is drain requested for Dry Bicart or not """ return self._dry_bicart_drain_request @dd_dry_bicart_drain_request.setter def dd_dry_bicart_drain_request(self, value): self._dry_bicart_drain_request = value @property def dd_bicarb_chamber_fill_request(self) -> bool: """ Is fill requested for Bicarb Chamber or not """ return self._bicarb_chamber_fill_request @dd_bicarb_chamber_fill_request.setter def dd_bicarb_chamber_fill_request(self, value): self._bicarb_chamber_fill_request = value @property def dd_dry_bicart_last_fill_time(self) -> int: """ The Dry Bicart last fill time """ return self._dry_bicart_last_fill_time @dd_dry_bicart_last_fill_time.setter def dd_dry_bicart_last_fill_time(self, value): self._dry_bicart_last_fill_time = value @property def dd_dry_bicart_current_fill_time(self) -> int: """ The Dry Bicart current fill time """ return self._dry_bicart_current_fill_time @dd_dry_bicart_current_fill_time.setter def dd_dry_bicart_current_fill_time(self, value): self._dry_bicart_current_fill_time = value @property def dry_bicart_type(self) -> BicarbTypes: """ The Dry Bicart concentrate option """ return self._dry_bicart_type @dry_bicart_type.setter def dry_bicart_type(self, value: Union[int, BicarbTypes]): if isinstance(value, int): self._dry_bicart_type = BicarbTypes(value) else: self._dry_bicart_type = value @property def dry_bicart_drain_time_period(self) -> int: """ The Dry Bicart drain time period in sec """ return self._dry_bicart_drain_time_period @dry_bicart_drain_time_period.setter def dry_bicart_drain_time_period(self, value): self._dry_bicart_drain_time_period = value # ============================================================ Handlers ============================================================ @publish(["msg_id_dd_dry_bicart_data", "dd_dry_bicart_fill_execution_state", "dd_bicarb_chamber_fill_execution_state", "dd_dry_bicart_drain_execution_state", "dd_dry_bicart_fill_cycle_counter", "dd_dry_bicart_max_fill_cycle_count", "dd_dry_bicart_fill_request", "dry_bicarb_chamber_fill_request", "dd_dry_bicart_drain_request", "dd_dry_bicart_last_fill_time", "dd_dry_bicart_current_fill_time", "dry_bicart_type", "dry_bicart_drain_time_period", "dd_dry_bicart_timestamp"]) def _handler_dry_bicart_sync(self, message, timestamp=0.0): """ Handles published dry bicart data messages. @param message: published dry bicart data message @return: None """ msg_list = [] msg_list.append((type(self).dd_dry_bicart_fill_execution_state, DataTypes.U32)) msg_list.append((type(self).dd_bicarb_chamber_fill_execution_state, DataTypes.U32)) msg_list.append((type(self).dd_dry_bicart_drain_execution_state, DataTypes.U32)) msg_list.append((type(self).dd_dry_bicart_fill_cycle_counter, DataTypes.U32)) msg_list.append((type(self).dd_dry_bicart_max_fill_cycle_count, DataTypes.U32)) msg_list.append((type(self).dd_dry_bicart_fill_request, DataTypes.U32)) msg_list.append((type(self).dd_bicarb_chamber_fill_request, DataTypes.U32)) msg_list.append((type(self).dd_dry_bicart_drain_request, DataTypes.U32)) msg_list.append((type(self).dd_dry_bicart_last_fill_time, DataTypes.U32)) msg_list.append((type(self).dd_dry_bicart_current_fill_time, DataTypes.U32)) msg_list.append((type(self).dry_bicart_type, DataTypes.U32)) msg_list.append((type(self).dry_bicart_drain_time_period, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) self.dd_dry_bicart_timestamp = timestamp # ============================================================ Overrides and Requests ============================================================ def cmd_dry_bicart_broadcast_interval_override(self, ms: int, reset: int = NO_RESET) -> int: """ Constructs and sends the dry bicart data broadcast 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_dd_ch_id, msg_id=MsgIds.MSG_ID_DD_DRY_BICART_DATA_PUBLISH_INTERVAL_OVERRIDE_REQUEST, module_name='DD Dry Bicart', logger=self.logger, can_interface=self.can_interface) def cmd_bicart_max_fill_cycle_count_override(self, count: int, reset: int = NO_RESET) -> int: """ Constructs and sends the dry bicart max cycle count override command Constraints: Must be logged into DD. @param count: int - count value to override max fill cycle count @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ cycle_count = integer_to_bytearray(count) reset_byte_array = integer_to_bytearray(reset) payload = reset_byte_array + cycle_count return cmd_generic_override( payload=payload, reset=reset, channel_id=CanChannels.dialin_to_dd_ch_id, msg_id=MsgIds.MSG_ID_DD_DRY_BICART_FILL_CYCLE_MAX_OVERRIDE_REQUEST, entity_name='max fill cycle count', override_text=str(count), logger=self.logger, can_interface=self.can_interface) def cmd_dry_bicart_fill_request_override(self, start_stop: int, reset: int = NO_RESET) -> int: """ Constructs and sends the dry bicart fill request override command Constraints: Must be logged into DD. @param start_stop: int - value to start or stop dry bicart fill ( start = 1, stop = 0 @return: 1 if successful, zero otherwise """ request = integer_to_bytearray(start_stop) reset_byte_array = integer_to_bytearray(reset) payload = reset_byte_array + request return cmd_generic_override( payload=payload, reset=reset, channel_id=CanChannels.dialin_to_dd_ch_id, msg_id=MsgIds.MSG_ID_DD_DRY_BICART_FILL_REQUEST_OVERRIDE_REQUEST, entity_name='Bicart Fill Request', override_text=str(start_stop), logger=self.logger, can_interface=self.can_interface) def cmd_dry_bicart_bicarb_chamber_fill_request_override(self, start_stop: int, reset: int = NO_RESET) -> int: """ Constructs and sends the dry bicart bicarbonate chamber F fill request override command Constraints: Must be logged into DD. @param start_stop: int - value to start or stop dry bicart fill ( start = 1, stop = 0 @return: 1 if successful, zero otherwise """ request = integer_to_bytearray(start_stop) reset_byte_array = integer_to_bytearray(reset) payload = reset_byte_array + request return cmd_generic_override( payload=payload, reset=NO_RESET, channel_id=CanChannels.dialin_to_dd_ch_id, msg_id=MsgIds.MSG_ID_DD_BICARB_CHAMBER_FILL_REQUEST_OVERRIDE_REQUEST, entity_name='Bicarb Fill Request', override_text=str(start_stop), logger=self.logger, can_interface=self.can_interface) def cmd_dry_bicart_drain_request_override(self, start_stop: int, reset: int = NO_RESET) -> int: """ Constructs and sends the dry bicart drain request override command Constraints: Must be logged into DD. @param start_stop: int - value to start or stop dry bicart drain ( start = 1, stop = 0 @return: 1 if successful, zero otherwise """ request = integer_to_bytearray(start_stop) reset_byte_array = integer_to_bytearray(reset) payload = reset_byte_array + request return cmd_generic_override( payload=payload, reset=reset, channel_id=CanChannels.dialin_to_dd_ch_id, msg_id=MsgIds.MSG_ID_DD_BICART_DRAIN_REQUEST_OVERRIDE_REQUEST, entity_name='Bicart Drain Request', override_text=str(start_stop), logger=self.logger, can_interface=self.can_interface) def cmd_dry_bicart_cartridge_size_override(self, small_large: int, reset: int = NO_RESET) -> int: """ Constructs and sends the dry bicart cartridge size override command Constraints: Must be logged into DD. @param small_large: int - value for small or large dry bicart ( small = 0, large = 1 @return: 1 if successful, zero otherwise """ cartridge_size = integer_to_bytearray(small_large) reset_byte_array = integer_to_bytearray(reset) payload = reset_byte_array + cartridge_size return cmd_generic_override( payload=payload, reset=reset, channel_id=CanChannels.dialin_to_dd_ch_id, msg_id=MsgIds.MSG_ID_DD_BICART_CARTRIDGE_SELECT_OVERRIDE_REQUEST, entity_name='Bicart Cartridge Size', override_text=str(small_large), logger=self.logger, can_interface=self.can_interface) def cmd_dry_bicart_depressurise_override(self, depressurise: int, reset: int = NO_RESET) -> int: """ Constructs and sends the dry bicart depressurise override command Constraints: Must be logged into DD. @param depressurise: int - value for depressurise ( depressurise = 1 @return: 1 if successful, zero otherwise """ depressurise = integer_to_bytearray(depressurise) reset_byte_array = integer_to_bytearray(reset) payload = reset_byte_array + depressurise return cmd_generic_override( payload=payload, reset=reset, channel_id=CanChannels.dialin_to_dd_ch_id, msg_id=MsgIds.MSG_ID_DD_BICART_DEPRESSURISE_REQUEST_OVERRIDE_REQUEST, entity_name='Bicart Depressurise', override_text=str(depressurise), logger=self.logger, can_interface=self.can_interface)