########################################################################### # # Copyright (c) 2019-2021 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 blood_leak.py # # @author (last) Quang Nguyen # @date (last) 11-Aug-2021 # @author (original) Peman Montazemi # @date (original) 15-Apr-2021 # ############################################################################ import struct 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.conversions import integer_to_bytearray, float_to_bytearray class HDBloodLeak(AbstractSubSystem): """ HDBloodLeak Hemodialysis Delivery (HD) Dialin API sub-class for blood leak related commands. """ # Blood leak detector status BLOOD_LEAK_DETECTED = 0 # Blood detected NO_BLOOD_LEAK_DETECTED = 1 # No blood detected # Blood leak detector state machine states BLOOD_LEAK_INIT_STATE = 0 # Initial state BLOOD_LEAK_ZERO_STATE = 1 # Zero state BLOOD_LEAK_SELF_TEST_STATE = 2 # Self-test state BLOOD_LEAK_NORMAL_STATE = 3 # Normal state def __init__(self, can_interface, 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: channel_id = DenaliChannels.hd_sync_broadcast_ch_id msg_id = MsgIds.MSG_ID_HD_BLOOD_LEAK_DATA.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_blood_leak_sync) self.blood_leak_status = self.NO_BLOOD_LEAK_DETECTED self.blood_leak_state = self.BLOOD_LEAK_INIT_STATE self.blood_leak_zeroed_status_counter = 0 self.blood_leak_counter = 0 self.blood_leak_zeroed_status = 0 self.blood_leak_detect_set_point = 0 self.blood_leak_detect_level = 0 self.blood_leak_st_count = 0 self.blood_leak_led_intensity = 0 self.blood_leak_register_counter = 0 def get_blood_leak_status(self): """ Gets the current blood leak status @return: List containing blood leak status: [detected, undetected] """ return self.blood_leak_status def get_blood_leak_state(self): """ Gets the current blood leak state @return: integer - blood leak state (0: init, 1: zeroing, 2: self-test, 3: normal) """ return self.blood_leak_state def get_blood_leak_zero_status_counter(self): """ Gets the current blood leak zero status counter @return: integer - blood leak zero status counter """ return self.blood_leak_zeroed_status_counter def get_blood_leak_counter(self): """ Gets the current blood leak counter @return: integer - blood leak counter """ return self.blood_leak_counter def get_blood_leak_zeroed_status(self): """ Gets the current blood leak zeroed status @return: integer - blood leak zeroed status """ return self.blood_leak_zeroed_status def get_blood_leak_detect_set_point(self): """ Gets the current blood leak detect set point @return: integer - blood leak detect set point """ return self.blood_leak_detect_set_point def get_blood_leak_detect_level(self): """ Gets the current blood leak detect level @return: integer - blood leak detect level """ return self.blood_leak_detect_level def get_blood_leak_st_count(self): """ Gets the current blood leak st count @return: integer - blood leak st count """ return self.blood_leak_st_count def get_blood_leak_led_intensity(self): """ Gets the current blood leak LED intensity @return: integer - blood leak LED intensity """ return self.blood_leak_led_intensity def get_blood_leak_register_counter(self): """ Gets the current blood leak register counter @return: integer - blood leak register counter """ return self.blood_leak_register_counter @publish("blood_leak_status") def _handler_blood_leak_sync(self, message): """ Handles published blood leak status messages. Blood leak status is captured for reference. @param message: published blood leak status message @return: None """ self.blood_leak_status = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1]))[0] self.blood_leak_state = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2]))[0] self.blood_leak_zeroed_status_counter = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_3:MsgFieldPositions.END_POS_FIELD_3]))[0] self.blood_leak_counter = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_4:MsgFieldPositions.END_POS_FIELD_4]))[0] self.blood_leak_zeroed_status = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_5:MsgFieldPositions.END_POS_FIELD_5]))[0] self.blood_leak_detect_set_point = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_6:MsgFieldPositions.END_POS_FIELD_6]))[0] self.blood_leak_detect_level = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_7:MsgFieldPositions.END_POS_FIELD_7]))[0] self.blood_leak_st_count = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_8:MsgFieldPositions.END_POS_FIELD_8]))[0] self.blood_leak_led_intensity = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_9:MsgFieldPositions.END_POS_FIELD_9]))[0] self.blood_leak_register_counter = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_10:MsgFieldPositions.END_POS_FIELD_10]))[0] def cmd_blood_leak_detector_override(self, detected, reset=NO_RESET): """ Constructs and sends the blood leak detector state override command Constraints: Must be logged into HD. @param detected: unsigned int - detected (0=detected, 1=undetected) to override detector with @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ rst = integer_to_bytearray(reset) det = float_to_bytearray(detected) payload = rst + det message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_HD_BLOOD_LEAK_STATUS_OVERRIDE.value, payload=payload) self.logger.debug("Override blood leak detector state value") # 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("Timeout!!!!") return False def cmd_blood_leak_zero_request(self): """ Request blood leak zeroing Constraints: Must be logged into HD. @return: 1 if successful, zero otherwise """ message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_HD_BLOOD_LEAK_ZERO_REQUEST.value) self.logger.debug("Request blood leak zeroing") # 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("Timeout!!!!") return False def cmd_blood_leak_data_broadcast_interval_override(self, ms, reset=NO_RESET): """ Constructs and sends the blood leak data broadcast interval override command Constraints: Must be logged into HD. Given interval must be non-zero and a multiple of the HD 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 """ rst = integer_to_bytearray(reset) mis = integer_to_bytearray(ms) payload = rst + mis message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_HD_BLOOD_LEAK_DATA_SEND_INTERVAL_OVERRIDE.value, payload=payload) self.logger.debug("Override HD blood leak data broadcast interval") # Send message received_message = self.can_interface.send(message) # If there is content... if received_message is not None: if reset == RESET: str_res = "reset back to normal: " else: str_res = str(ms) + " ms: " self.logger.debug("Blood leak data broadcast interval overridden to " + str_res + 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