""" DD Calibration UI loader """ # Python import os import can import struct # Qt from PySide2 import QtCore, QtWidgets from PySide2.QtCore import Slot # parent from engine.dynamicloader import DynamicLoader # plugin specific from leahi_dialin.common.msg_defs import MsgIds, MsgFieldPositions from leahi_dialin.protocols import CAN from leahi_dialin.utils import conversions # td simulator from leahi_dialin.ui.dd_messaging import DD_Messaging from leahi_dialin.ui.td_messaging import TD_Messaging class Loader(DynamicLoader): """ DD Calibration UI loader """ def __init__(self): self.dd_interface = DD_Messaging() self.can_interface = self.dd_interface.can_interface self.index = 0 super().__init__(os.path.dirname(__file__)) if self.can_interface is not None: self.can_interface.register_receiving_publication_function( CAN.CanChannels.ui_sync_broadcast_ch_id, MsgIds.MSG_ID_UI_DD_NVM_GET_PRESSURE_SENSOR_CAL_RECORD_REQUEST.value, self.handle_get_calibration_pressures) if self.can_interface is not None: self.can_interface.register_receiving_publication_function( CAN.CanChannels.ui_sync_broadcast_ch_id, MsgIds.MSG_ID_UI_DD_NVM_SET_PRESSURE_SENSOR_CAL_RECORD_REQUEST.value, self.handle_set_calibration_pressures) def _init_loader(self): """ finds and creates widgets :return: none """ self.pressureList = [ "Hydralics Outlet", "Dry Bicarb", "Spent Dialysate", "Fresh Dialysate", "Transmembrane", "D87", "Water Inlet before regulator", "Water Inlet after regulator", "Water Inlet before cond. sensor", "Before RO filter", "After RO filter", ] # Pressures self.tbGetPressureSend = self.find_widget(QtWidgets.QToolButton , 'tbGetPressureSend' ) self.lbGetPressuresReq = self.find_widget(QtWidgets.QLabel , 'lbGetPressuresReq' ) self.cbGetPressureAccepted = self.find_widget(QtWidgets.QCheckBox , 'cbGetPressureAccepted' ) self.sbGetPressureRejectionReason = self.find_widget(QtWidgets.QSpinBox , 'sbGetPressureRejectionReason' ) self.sbGetPressureIndex = self.find_widget(QtWidgets.QSpinBox , 'sbGetPressureIndex' ) self.sbGetPressureGain = self.find_widget(QtWidgets.QDoubleSpinBox , 'sbGetPressureGain' ) self.sbGetPressureOffset = self.find_widget(QtWidgets.QDoubleSpinBox , 'sbGetPressureOffset' ) self.sbGetPressureCalibrationTime = self.find_widget(QtWidgets.QSpinBox , 'sbGetPressureCalibrationTime' ) self.tbSetPressureSend = self.find_widget(QtWidgets.QToolButton , 'tbSetPressureSend' ) self.lbSetPressuresIndex = self.find_widget(QtWidgets.QLabel , 'lbSetPressuresIndex' ) self.lbSetPressureGain = self.find_widget(QtWidgets.QLabel , 'lbSetPressureGain' ) self.lbSetPressureOffset = self.find_widget(QtWidgets.QLabel , 'lbSetPressureOffset' ) self.lbSetPressureCalibrationTime = self.find_widget(QtWidgets.QLabel , 'lbSetPressureCalibrationTime' ) self.cbSetPressureAccepted = self.find_widget(QtWidgets.QCheckBox , 'cbSetPressureAccepted' ) self.sbSetPressureRejectionReason = self.find_widget(QtWidgets.QSpinBox , 'sbSetPressureRejectionReason' ) self.sbSetPressureIndex = self.find_widget(QtWidgets.QSpinBox , 'sbSetPressureIndex' ) self.sbSetPressureGain = self.find_widget(QtWidgets.QSpinBox , 'sbSetPressureGain' ) self.sbSetPressureOffset = self.find_widget(QtWidgets.QSpinBox , 'sbSetPressureOffset' ) self.sbSetPressureCalibrationTime = self.find_widget(QtWidgets.QSpinBox , 'sbSetPressureCalibrationTime' ) def _init_connections(self): """ initializes the widget's connections :return: none """ self.tbGetPressureSend .clicked.connect ( self.do_get_pressures_response ) self.tbSetPressureSend .clicked.connect ( self.do_set_pressures_response ) def getIntValue (self, message): val,self.index = conversions.bytearray_to_integer( message, self.index) return val def getFloatValue (self, message): val,self.index = conversions.bytearray_to_float( message, self.index) return val @Slot() def _init_widgets(self): """ initializes the widget's properties :return: none """ self.init_get_pressures() self.init_set_pressures() def init_get_pressures(self): self.lbGetPressuresReq .setText ("-") self.sbGetPressureRejectionReason .setValue (0) self.sbGetPressureIndex .setValue (0) self.sbGetPressureGain .setValue (0) self.sbGetPressureOffset .setValue (0) self.sbGetPressureCalibrationTime .setValue (0) self.cbGetPressureAccepted .setChecked ( True ) def init_set_pressures(self): self.lbSetPressuresIndex .setText ("-") self.lbSetPressureGain .setText ("-") self.lbSetPressureOffset .setText ("-") self.lbSetPressureCalibrationTime .setText ("-") self.cbSetPressureAccepted .setChecked ( True ) self.sbSetPressureIndex .setValue (0) self.sbSetPressureGain .setValue (0) self.sbSetPressureOffset .setValue (0) self.sbSetPressureCalibrationTime .setValue (0) @Slot() def handle_get_calibration_pressures(self, message, timestamp = 0.0): """ Called when the user sends a rinseback command request to firmware from UI @return: None """ message = message['message'] self.index = MsgFieldPositions.START_POS_FIELD_1 value = self.getIntValue(message) print(value) cmdString = self.pressureList[value] if value >= 0 and value < len(self.pressureList) else "" self.lbGetPressuresReq.setText(f"{cmdString} ({value})") @Slot() def handle_set_calibration_pressures(self, message, timestamp = 0.0): """ Called when the user sends a rinseback command request to firmware from UI @return: None """ message = message['message'] self.index = MsgFieldPositions.START_POS_FIELD_1 value = self.getIntValue(message) gain = f"{self.getFloatValue(message):.1f}" offset = f"{self.getFloatValue(message):.1f}" caltime = str(self.getIntValue(message)) cmdString = self.pressureList[value] if value >= 0 and value < len(self.pressureList) else "" self.lbSetPressuresIndex .setText(f"{cmdString} ({value})") self.lbSetPressureGain .setText(gain) self.lbSetPressureOffset .setText(offset) self.lbSetPressureCalibrationTime .setText(caltime) @Slot() def do_get_pressures_response(self): self.lbGetPressuresReq .setText ("-") self.dd_interface.dd_get_calibration_pressure_sensor_record(self.cbGetPressureAccepted .isChecked() , self.sbGetPressureRejectionReason .value() , self.sbGetPressureIndex .value() , self.sbGetPressureGain .value() , self.sbGetPressureOffset .value() , self.sbGetPressureCalibrationTime .value() ) @Slot() def do_set_pressures_response(self): self.lbSetPressuresIndex .setText ("-") self.lbSetPressureGain .setText ("-") self.lbSetPressureOffset .setText ("-") self.lbSetPressureCalibrationTime .setText ("-") self.dd_interface.dd_set_calibration_pressure_sensor_record(self.cbSetPressureAccepted .isChecked() , self.sbSetPressureRejectionReason .value() , self.sbSetPressureIndex .value() , self.sbSetPressureGain .value() , self.sbSetPressureOffset .value() , self.sbSetPressureCalibrationTime .value() )