########################################################################### # # 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 levels.py # # @author (last) Zoltan Miskolci # @date (last) 04-May-2026 # @author (original) Dara Navaei # @date (original) 29-May-2020 # ############################################################################ # Module imports from logging import Logger # Project imports from leahi_dialin.common.constants import NO_RESET from leahi_dialin.common import dd_enum_repository 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 DDLevels(AbstractSubSystem): """ Dialysate Delivery (DD) Dialin API sub-class for levels 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_LEVEL_DATA.value, function = self._handler_levels_sync) self.dd_levels_timestamp = 0 #: The timestamp of the latest message self.dd_level_sensors = { dd_enum_repository.DDLevelSensorNames.D46_LEVEL.name: 0, dd_enum_repository.DDLevelSensorNames.D63_LEVEL.name: 0, dd_enum_repository.DDLevelSensorNames.D98_LEVEL.name: 0, dd_enum_repository.DDLevelSensorNames.D6_LEVEL.name: 0, } self.dd_bicarb_level = 0 # ============================================================ Properties ============================================================ @property def dd_level_sensors(self) -> dict: """ The Level Sensors data """ return self._level_sensors @dd_level_sensors.setter def dd_level_sensors(self, value): self._level_sensors = value @property def dd_bicarb_level(self) -> int: """ The Bicarb level """ return self._bicarb_level @dd_bicarb_level.setter def dd_bicarb_level(self, value): self._bicarb_level = value # ============================================================ Handlers ============================================================ @publish(["msg_id_dd_level_data", "dd_level_sensors", "dd_bicarb_level", "dd_levels_timestamp"]) def _handler_levels_sync(self, message, timestamp=0.0): """ Handles published levels message @param message: published levels data message @returns none """ msg_list =[] msg_list.append((self.dd_level_sensors, dd_enum_repository.DDLevelSensorNames.D46_LEVEL.name, DataTypes.U32)) msg_list.append((self.dd_level_sensors, dd_enum_repository.DDLevelSensorNames.D63_LEVEL.name, DataTypes.U32)) msg_list.append((self.dd_level_sensors, dd_enum_repository.DDLevelSensorNames.D98_LEVEL.name, DataTypes.U32)) msg_list.append((self.dd_level_sensors, dd_enum_repository.DDLevelSensorNames.D6_LEVEL.name, DataTypes.U32)) msg_list.append((type(self).dd_bicarb_level, DataTypes.U32)) self.process_into_vars(decoder_list = msg_list, message = message) self.dd_levels_timestamp = timestamp # ============================================================ Overrides and Requests ============================================================ def cmd_levels_broadcast_interval_override(self, ms: int, reset: int = NO_RESET) -> int: """ Constructs and sends dd levels broadcast time interval override. 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: (int) Publish time interval in ms @param reset: (int) 1 to reset a previous override, 0 to override @returns 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_LEVELS_PUBLISH_INTERVAL_OVERRIDE_REQUEST, module_name = 'DD Levels', logger = self.logger, can_interface = self.can_interface) def cmd_level_status_override(self, level_sensor: int, status: int, reset: int = NO_RESET) -> int: """ Constructs and sends level status cycle override command Must be logged into DD. @param level_sensor the level sensor to override @param status the status value to override @param reset: (int) 1 to reset a previous override, 0 to override @returns 1 if successful, zero otherwise """ reset_value = integer_to_bytearray(reset) lvl_sen = integer_to_bytearray(level_sensor) sts = integer_to_bytearray(status) payload = reset_value + sts + lvl_sen sensor_name = dd_enum_repository.DDLevelSensorNames(level_sensor).name.split('_')[0] return cmd_generic_override( payload = payload, reset = reset, channel_id = CanChannels.dialin_to_dd_ch_id, msg_id = MsgIds.MSG_ID_DD_LEVELS_STATUS_OVERRIDE_REQUEST, entity_name = f'DD {sensor_name} Level Status', override_text = str(status), logger = self.logger, can_interface = self.can_interface) def cmd_floater_level_status_override(self, level_sensor: int, status: int, reset: int = NO_RESET) -> int: """ Constructs and sends level status override command. Must be logged into DD. @param level_sensor: integer - level sensor id to override @param status: integer - the status value to override @param reset: (int) 1 to reset a previous override, 0 to override @returns 1 if successful, zero otherwise """ reset_value = integer_to_bytearray(reset) lvl_sen = integer_to_bytearray(level_sensor) sts = integer_to_bytearray(status) print("sts", sts) payload = reset_value + sts + lvl_sen return cmd_generic_override( payload = payload, reset = reset, channel_id = CanChannels.dialin_to_fp_ch_id, msg_id = MsgIds.MSG_ID_DD_FLOATER_LEVEL_OVERRIDE_REQUEST, entity_name = f'DD Floater Levels status', override_text = f'{str(status)}', logger = self.logger, can_interface = self.can_interface)