########################################################################### # # Copyright (c) 2019-2020 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 hemodialysis_device.py # # @author (last) Peter Lucia # @date (last) 10-Nov-2020 # @author (original) Peter Lucia # @date (original) 02-Apr-2020 # ############################################################################ import struct from .accelerometer import HDAccelerometer from .air_bubbles import HDAirBubbles from .air_trap import HDAirTrap from .alarms import HDAlarms from .blood_leak import HDBloodLeak from .buttons import HDButtons from .fluid_leak import HDFluidLeak from .ui_proxy import HDUIProxy from .watchdog import HDWatchdog from .rtc import HDRTC from .blood_flow import HDBloodFlow from .dialysate_inlet_flow import HDDialysateInletFlow from .dialysate_outlet_flow import HDDialysateOutletFlow from .treatment import HDTreatment from .valves import HDValves from .voltages import HDVoltages from .calibration_record import HDCalibrationNVRecord from .system_record import HDSystemNVRecords from .service_record import HDServiceNVRecords from .pressure_occlusion import HDPressureOcclusion from .pretreatment import HDPreTreatment from .syringe_pump import HDSyringePump from ..protocols.CAN import (DenaliMessage, DenaliCanMessenger, DenaliChannels) from ..utils.conversions import integer_to_bytearray from ..utils.base import _AbstractSubSystem, _publish, _LogManager from .constants import NO_RESET from ..common.msg_defs import MsgIds, MsgFieldPositions from ..common.hd_defs import HDOpModes class HD(_AbstractSubSystem): """ Hemodialysis Device (HD) Dialin object API. It provides the basic interface to communicate with the HD firmware. """ # HD login password HD_LOGIN_PASSWORD = '123' def __init__(self, can_interface="can0", log_level=None): """ HD object provides test/service commands for the HD sub-system. >> hd_object = HD('can0') >> hd_object = HD(can_interface='can0', log_level="DEBUG") Possible log levels: ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "CAN_ONLY", "PRINT_ONLY"] @param can_interface: (str) CANBus interface name, e.g. "can0" @param log_level: (str) Logging level, defaults to None """ super().__init__() self._log_manager = _LogManager(log_level=log_level, log_filepath=self.__class__.__name__+".log") self.logger = self._log_manager.logger # Create listener self.can_interface = DenaliCanMessenger(can_interface=can_interface, logger=self.logger, log_can=self._log_manager.log_level == "CAN_ONLY") self.can_interface.start() # register handler for HD operation mode broadcast messages if self.can_interface is not None: channel_id = DenaliChannels.hd_sync_broadcast_ch_id msg_id = MsgIds.MSG_ID_HD_OP_MODE.value self.can_interface.register_receiving_publication_function(channel_id, msg_id, self._handler_hd_op_mode_sync) # create properties self.hd_operation_mode = HDOpModes.MODE_INIT.value self.hd_operation_sub_mode = 0 self.hd_logged_in = False self.hd_set_logged_in_status(False) # Create command groups self.accel = HDAccelerometer(self.can_interface, self.logger) self.air_trap = HDAirTrap(self.can_interface, self.logger) self.air_bubbles = HDAirBubbles(self.can_interface, self.logger) self.alarms = HDAlarms(self.can_interface, self.logger) self.buttons = HDButtons(self.can_interface, self.logger) self.blood_leak = HDBloodLeak(self.can_interface, self.logger) self.fluid_leak = HDFluidLeak(self.can_interface, self.logger) self.ui = HDUIProxy(self.can_interface, self.logger) self.rtc = HDRTC(self.can_interface, self.logger) self.watchdog = HDWatchdog(self.can_interface, self.logger) self.bloodflow = HDBloodFlow(self.can_interface, self.logger) self.dialysate_inlet_flow = HDDialysateInletFlow(self.can_interface, self.logger) self.dialysate_outlet_flow = HDDialysateOutletFlow(self.can_interface, self.logger) self.treatment = HDTreatment(self.can_interface, self.logger) self.pretreatment = HDPreTreatment(self.can_interface, self.logger) self.pressure_occlusion = HDPressureOcclusion(self.can_interface, self.logger) self.syringe_pump = HDSyringePump(self.can_interface, self.logger) self.valves = HDValves(self.can_interface, self.logger) self.voltages = HDVoltages(self.can_interface, self.logger) self.calibration_record = HDCalibrationNVRecord(self.can_interface, self.logger) self.system_record = HDSystemNVRecords(self.can_interface, self.logger) self.service_record = HDServiceNVRecords(self.can_interface, self.logger) def get_operation_mode(self): """ Gets the HD operation mode @return: The hd operation mode """ return self.hd_operation_mode def get_hd_logged_in(self): """ Gets the logged in status of the HD @return: True if HD is logged in, False if not """ return self.hd_logged_in @_publish(["hd_logged_in"]) def hd_set_logged_in_status(self, logged_in:bool=False): """ Callback for HD logged in status change. @param logged_in boolean logged in status for HD @return: none """ self.hd_logged_in = logged_in @_publish(["hd_operation_mode"]) def _handler_hd_op_mode_sync(self, message): """ Handles published HD operation mode messages. Current HD operation mode is captured for reference. @param message: published HD operation mode broadcast message @return: None """ mode = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_1:MsgFieldPositions.END_POS_FIELD_1])) smode = struct.unpack('i', bytearray( message['message'][MsgFieldPositions.START_POS_FIELD_2:MsgFieldPositions.END_POS_FIELD_2])) self.hd_operation_mode = mode[0] self.hd_operation_sub_mode = smode[0] def cmd_log_in_to_hd(self, resend: bool = False) -> int: """ Constructs and sends a login command via CAN bus. Login required before \n other commands can be sent to the HD. @param resend: (bool) if False (default), try to login once. Otherwise, tries to login indefinitely @return: 1 if logged in, 0 if log in failed """ message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_TESTER_LOGIN_REQUEST.value, payload=list(map(int, map(ord, self.HD_LOGIN_PASSWORD)))) self.logger.debug("Logging in...") # Send message received_message = self.can_interface.send(message, resend=resend) if received_message is not None: if received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] == 1: self.logger.debug("Success: Logged In") self.hd_set_logged_in_status(True) else: self.logger.debug("Failure: Log In Failed.") return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] else: self.logger.debug("Login Timeout!!!!") return False def cmd_hd_request_calibration_data(self) -> int: """ Constructs and sends an HD calibration data request command via CAN bus. Constraints: Must be logged into HD. \returns response message if received, False if no response received @return: 1 if successful, zero otherwise """ message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_HD_GET_CALIBRATION_RECORD.value) self.logger.debug("requesting HD calibration data.") # Send message received_message = self.can_interface.send(message) if received_message is not None: if received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] == 1: self.logger.debug("HD calibration data request accepted.") else: self.logger.debug("HD calibration data request failed.") return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] else: self.logger.debug("Timeout!!!!") return False def cmd_hd_set_operation_mode(self, newMode: int = 0) -> int: """ Constructs and sends a set operation mode request command via CAN bus. Constraints: Must be logged into HD. Transition from current to requested op mode must be legal. @param newMode: ID of operation mode to transition to HD_OP_MODE_FAULT = 0 HD_OP_MODE_SERVICE = 1 HD_OP_MODE_INIT_POST = 2 HD_OP_MODE_STANDBY = 3 HD_OP_MODE_TREATMENT_PARAMS = 4 HD_OP_MODE_PRE_TREATMENT = 6 HD_OP_MODE_TREATMENT = 7 HD_OP_MODE_POST_TREATMENT = 8 @return: 1 if successful, zero otherwise """ payload = integer_to_bytearray(newMode) message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_HD_SET_OP_MODE_REQUEST.value, payload=payload) self.logger.debug("Requesting HD mode change to " + str(newMode)) # Send message received_message = self.can_interface.send(message) if received_message is not None: if received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] == 1: self.logger.debug("Success: Mode change accepted") else: self.logger.debug("Failure: Mode change rejected.") return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] else: self.logger.debug("HD mode change request Timeout!!!!") return False def cmd_hd_safety_shutdown_override(self, active: bool = True, reset: int = NO_RESET) -> int: """ Constructs and sends an HD safety shutdown override command via CAN bus. Constraints: Must be logged into HD. \returns response message if received, False if no response received @param active: boolean - True to activate safety shutdown, False to deactivate @param reset: integer - 1 to reset a previous override, 0 to override @return: 1 if successful, zero otherwise """ if active: sft=1 else: sft=0 rst = integer_to_bytearray(reset) saf = integer_to_bytearray(sft) payload = rst + saf message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_HD_SAFETY_SHUTDOWN_OVERRIDE.value, payload=payload) self.logger.debug("overriding HD safety shutdown") # Send message received_message = self.can_interface.send(message) if received_message is not None: if received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] == 1: self.logger.debug("Safety shutdown signal overridden") else: self.logger.debug("Safety shutdown signal override failed.") return received_message['message'][DenaliMessage.PAYLOAD_START_INDEX] else: self.logger.debug("Timeout!!!!") return False def cmd_hd_software_reset_request(self) -> None: """ Constructs and sends an HD software reset request via CAN bus. Constraints: Must be logged into HD. @return: None """ message = DenaliMessage.build_message(channel_id=DenaliChannels.dialin_to_hd_ch_id, message_id=MsgIds.MSG_ID_HD_SOFTWARE_RESET_REQUEST.value) self.logger.debug("requesting HD software reset") # Send message self.can_interface.send(message, 0) self.logger.debug("Sent request to HD to reset...") self.hd_set_logged_in_status(False)