########################################################################### # # Copyright (c) 2022-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 flow_sensors.py # # @author (last) Micahel Garthwaite # @date (last) 07-Aug-2023 # @author (original) Dara Navaei # @date (original) 19-Sep-2022 # ############################################################################ import struct from enum import unique from logging import Logger from .constants import RESET, NO_RESET from ..common.msg_defs import MsgIds, MsgFieldPositions from ..protocols.CAN import DenaliMessage, DenaliChannels from ..utils.base import AbstractSubSystem, publish from ..utils.base import DialinEnum from ..utils.checks import check_broadcast_interval_override_ms from ..utils.conversions import integer_to_bytearray, float_to_bytearray @unique class FlowSensorsNames(DialinEnum): RO_FLOW_SENSOR = 0 DIALYSATE_FLOW_SENSOR = 1 NUM_OF_FLOW_SENSOR = 2 class FlowSensors(AbstractSubSystem): """ Dialysate Generation (DG) interface for dialysate flow sensor related commands. """ def __init__(self, can_interface, logger: Logger): """ Initialize CAN interface and logger @param can_interface: Denali CAN Messenger object @param: logger: (Logger) object """ super().__init__() self.can_interface = can_interface self.logger = logger # The flow rate of the sensors are L/min self.measured_ro_flow_LPM = 0.0 self.measured_dialysate_flow_LPM = 0.0 self.measured_ro_flow_with_cp_LPM = 0.0 self.measured_raw_ro_flow_LPM = 0.0 self.measured_raw_dialysate_flow_LPM = 0.0 self.dg_flow_sensors_timestamp = 0.0 if self.can_interface is not None: channel_id = DenaliChannels.dg_sync_broadcast_ch_id self.can_interface.register_receiving_publication_function(channel_id, MsgIds.MSG_ID_DG_FLOW_SENSORS_DATA.value, self._handler_flow_sensors_sync) def get_measured_dialysate_flow_rate(self): """ Gets the measured flow value from DG's dialysate flow rate @return: The dialysate flow rate L/min """ return self.measured_dialysate_flow_LPM def get_measured_ro_flow(self): """ Gets the measured flow value from DG's ro pump @return: The RO flow rate L/min """ return self.measured_ro_flow_LPM def get_measured_ro_flow_with_cp(self): """ Gets the measured flow value from DG's ro pump with concentrate pumps @return: The RO flow rate with concentrate pumps flow rate L/min """ return self.measured_ro_flow_with_cp_LPM def get_measured_raw_ro_flow(self): """ Gets the measured raw (uncalibrated) flow value from DG's ro pump (FMP) @return: The raw RO flow rate in L/min """ return self.measured_raw_ro_flow_LPM def get_measured_raw_dialysate_flow(self): """ Gets the measured raw (uncalibrated) flow value from DG's dialydate flow sensor (FMD) @return: The raw dialysate flow rate in L/min """ return self.measured_raw_dialysate_flow_LPM @publish(["dg_flow_sensors_timestamp","measured_ro_flow_LPM", "measured_dialysate_flow_LPM", "measured_ro_flow_with_cp_LPM", "measured_raw_ro_flow_LPM", "measured_raw_dialysate_flow_LPM"]) def _handler_flow_sensors_sync(self, message, timestamp=0.0): """ Handles incoming DG flow sensor broadcast messages and records latest reported flow rates. @return: none """ self.measured_ro_flow_LPM = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.measured_ro_flow_with_cp_LPM = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.measured_dialysate_flow_LPM = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.measured_raw_ro_flow_LPM = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_4:MsgFieldPositions.END_POS_FIELD_4]))[0] self.measured_raw_dialysate_flow_LPM = struct.unpack('f', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_5:MsgFieldPositions.END_POS_FIELD_5]))[0] self.dg_flow_sensors_timestamp = timestamp def cmd_measured_flow_sensor_value_lpm_override(self, sensor_id: int, rate: float, raw: int, reset: int = NO_RESET) -> int: """ Constructs and sends the flow sensor value override command Constraints: Must be logged into DG. @param rate: (float) the sensor flow rate to be set in L/min @param raw: (int) 1 to override measured raw value, 0 to override measured value. @param sensor_id: (int) the sensor id to be overriden @param reset: (int) 1 to reset a previous override, 0 to override @return 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) index = integer_to_bytearray(sensor_id) value = float_to_bytearray(rate) r = integer_to_bytearray(raw) payload = rst + value + index + r message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_dg_ch_id, message_id=MsgIds.MSG_ID_MEASURED_FLOW_SENSORS_OVERRIDE.value, payload=payload) self.logger.debug("Overriding " + FlowSensorsNames(sensor_id).name + " flow sensor measured value to " + str(rate) + " L/min") # Send message received_message = self.can_interface.send(message) # If there is content... if received_message is not None: # response payload is OK or not OK return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] else: self.logger.debug("Flow sensor value override Timeout!!!") return False def cmd_flow_sensor_data_broadcast_interval_override(self, ms: int, reset: int = NO_RESET) -> int: """ Constructs and sends the flow sensor data publish interval command. Constraints: Must be logged into DG. Given interval must be non-zero and a multiple of the DG priority task interval (50 ms). @param ms: (int) interval (in ms) to override with @param reset: (int) 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_dg_ch_id, message_id=MsgIds.MSG_ID_FLOW_DATA_PUBLISH_INTERVAL_OVERRIDE.value, payload=payload) self.logger.debug("Overriding flow sensors 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(mis) self.logger.debug( "Flow sensors data broadcast interval overridden to " + str_res + " ms: " + 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