Commit c4e45409 authored by Tiago Sarmento's avatar Tiago Sarmento

separated business and display logic

parents 87499648 74ec0065
This diff is collapsed.
......@@ -16,37 +16,46 @@ import sys
from datetime import datetime
from PySide2 import QtCore, QtGui, QtWidgets
from handler_library.handler import Handler
from handler_library.handler import PayloadHandler
import logging
class AlarmHandler(Handler, QtCore.QObject):
class AlarmHandler(PayloadHandler):
UpdateAlarm = QtCore.Signal(dict)
NewAlarm = QtCore.Signal(QtWidgets.QWidget)
RemoveAlarm = QtCore.Signal(QtWidgets.QWidget)
def __init__(self, NativeUI, *args, **kwargs):
super(AlarmHandler, self).__init__(*args, **kwargs)
QtCore.QObject.__init__(self)
super().__init__(['DATA', 'ALARM'],*args, **kwargs)
self.NativeUI = NativeUI
self.alarmDict = {}
self.alarm_list = []
self.oldAlarms = []
def acknowledge_pressed(self):
self.popup.clearAlarms()
self.list.acknowledge_all()
def active_payload(self) -> int:
alarm_data = self.get_db()
outdict = {}
currentAlarms = self.NativeUI.ongoingAlarms # instead of getting database at a particular frequency, this should be triggered when a new alarm arrives
def _set_alarm_list(self, alarm_list:list):
self.alarm_list = alarm_list
def active_payload(self, *args) -> int:
#alarm_data = self.get_db()
#outdict = {}
full_payload = args[0]
#print(full_payload['alarms'])
currentAlarms = full_payload['alarms']#self.NativeUI.ongoingAlarms # instead of getting database at a particular frequency, this should be triggered when a new alarm arrives
self.alarm_list = currentAlarms
#self._set__alarm_list(currentAlarms)
if self.oldAlarms != currentAlarms:
if len(self.oldAlarms) != len(currentAlarms):
self.oldAlarms = currentAlarms
self.UpdateAlarm.emit(currentAlarms)
def handle_newAlarm(self, currentAlarms):
def handle_newAlarm(self, currentAlarms): # if this is combined with active_payload an error arises
for alarm in currentAlarms:
alarmCode = alarm["alarm_code"]
if alarmCode in self.alarmDict:
......@@ -55,42 +64,20 @@ class AlarmHandler(Handler, QtCore.QObject):
else:
newAbstractAlarm = AbstractAlarm(self.NativeUI, alarm)
self.alarmDict[alarmCode] = newAbstractAlarm
self.NewAlarm.emit(newAbstractAlarm)
newAbstractAlarm.alarmExpired.connect(
lambda i=newAbstractAlarm: self.handleAlarmExpiry(i)
)
self.NativeUI.widgets.alarm_popup.addAlarm(newAbstractAlarm)
self.NativeUI.widgets.alarm_list.addAlarm(newAbstractAlarm)
self.NativeUI.widgets.alarm_table.addAlarmRow(newAbstractAlarm)
#
# def update_alarms(self):
# currentAlarms = self.NativeUI.ongoingAlarms # instead of getting database at a particular frequency, this should be triggered when a new alarm arrives
# if self.oldAlarms != currentAlarms:
# if len(self.oldAlarms) != len(currentAlarms):
# self.oldAlarms = currentAlarms
#
# for alarm in currentAlarms:
# alarmCode = alarm["alarm_code"]
# if alarmCode in self.alarmDict:
# self.alarmDict[alarmCode].resetTimer()
# self.alarmDict[alarmCode].calculateDuration()
# else:
# newAbstractAlarm = AbstractAlarm(self.NativeUI, alarm)
# self.alarmDict[alarmCode] = newAbstractAlarm
# newAbstractAlarm.alarmExpired.connect(
# lambda i=newAbstractAlarm: self.handleAlarmExpiry(i)
# )
# self.NativeUI.widgets.alarm_popup.addAlarm(newAbstractAlarm)
# self.NativeUI.widgets.alarm_list.addAlarm(newAbstractAlarm)
# self.NativeUI.widgets.alarm_table.addAlarmRow(newAbstractAlarm)
def handleAlarmExpiry(self, abstractAlarm):
abstractAlarm.freezeTimer()
self.NativeUI.widgets.alarm_popup.removeAlarm(abstractAlarm)
self.NativeUI.widgets.alarm_list.removeAlarm(abstractAlarm)
self.alarmDict.pop(abstractAlarm.alarmPayload["alarm_code"])
abstractAlarm.recordFinishTime()
self.RemoveAlarm.emit(abstractAlarm)
self.alarmDict.pop(abstractAlarm.alarmPayload["alarm_code"])
# abstractAlarm is deleted by itself
class AbstractAlarm(QtWidgets.QWidget):
class AbstractAlarm(QtCore.QObject):
alarmExpired = QtCore.Signal()
......@@ -104,7 +91,7 @@ class AbstractAlarm(QtWidgets.QWidget):
self.finishTime = -1
self.timer = QtCore.QTimer()
self.timer.setInterval(100) # just faster than 60Hz
self.timer.setInterval(2000) # just faster than 60Hz
self.timer.timeout.connect(self.timeoutDelete)
self.timer.start()
......
......@@ -51,8 +51,12 @@ class AlarmWidget(QtWidgets.QWidget):
self.setLayout(self.layout)
if self.alarmPayload["alarm_type"] == "PRIORITY_HIGH":
self.setStyleSheet("background-color:red;")
#self.priority = "PRIORITY_HIGH"
elif self.alarmPayload["alarm_type"] == "PRIORITY_MEDIUM":
self.setStyleSheet("background-color:orange;")
#self.priority = "PRIORITY_LOW"
#self.code =
# self.timer = QtCore.QTimer()
# self.timer.setInterval(500) # just faster than 60Hz
......@@ -66,6 +70,10 @@ class AlarmWidget(QtWidgets.QWidget):
self.NativeUI.alarms_view.alarmButton.click()
return False
def get_priority(self):
return self.alarmPayload["alarm_type"]
# def checkAlarm(self):
# """Check alarm still exists in ongoingAlarms object. If present do nothing, otherwise delete."""
# self.ongoingAlarms = self.NativeUI.ongoingAlarms
......@@ -85,6 +93,7 @@ class AlarmPopup(QtWidgets.QDialog):
self.setParent(NativeUI) # ensures popup closes when main UI does
self.alarmDict = {}
self.NativeUI = NativeUI
self.extraAlarms = AlarmExtrasWidget(NativeUI,self)
self.layout = QtWidgets.QVBoxLayout()
self.layout.setSpacing(0)
......@@ -104,6 +113,7 @@ class AlarmPopup(QtWidgets.QDialog):
self.timer = QtCore.QTimer()
self.timer.setInterval(100) # just faster than 60Hz
self.timer.timeout.connect(self.adjustSize)
#self.timer.timeout.connect(self.refresh_alarm_ordering)
self.timer.start()
self.show()
......@@ -122,17 +132,42 @@ class AlarmPopup(QtWidgets.QDialog):
self.alarmDict[abstractAlarm.alarmPayload["alarm_code"]] = AlarmWidget(
self.NativeUI, abstractAlarm, self
)
self.layout.addWidget(self.alarmDict[abstractAlarm.alarmPayload["alarm_code"]])
self.refresh_alarm_ordering()
#self.layout.addWidget(self.alarmDict[abstractAlarm.alarmPayload["alarm_code"]])
return 0
def removeAlarm(self, abstractAlarm):
"""Creates a new alarmWidget and adds it to the container"""
self.alarmDict[abstractAlarm.alarmPayload["alarm_code"]].setParent(None)
self.alarmDict.pop(abstractAlarm.alarmPayload["alarm_code"])
self.refresh_alarm_ordering()
return 0
def refresh_alarm_ordering(self):
self.layout.removeWidget(self.extraAlarms)
for key in self.alarmDict:
self.layout.removeWidget(self.alarmDict[key])
for key in self.alarmDict:
if self.alarmDict[key].get_priority() == 'PRIORITY_HIGH':
if self.layout.count() == 4:
self.extraAlarms.update_text(1 + len(self.alarmDict) - self.layout.count())
self.layout.addWidget(self.extraAlarms)
break
self.layout.addWidget(self.alarmDict[key])
if self.layout.count() < 5:
for key in self.alarmDict:
if self.layout.count() == 3:
self.extraAlarms.update_text(len(self.alarmDict) - self.layout.count())
self.layout.addWidget(self.extraAlarms)
break
if self.alarmDict[key].get_priority() == 'PRIORITY_LOW':
self.layout.addWidget(self.alarmDict[key])
# def resetTimer(self, alarmPayload):
# self.alarmDict[alarmPayload["alarm_code"]].timer.start()
......@@ -143,3 +178,59 @@ class AlarmPopup(QtWidgets.QDialog):
y = 0 # screen.height() - widget.height()
self.move(x, y)
return 0
class AlarmExtrasWidget(QtWidgets.QWidget):
"""Object containing information particular to one alarm.
Created when alarm received from microcontroller, timeout after alarm signal stops.
Is contained within alarmPopup"""
def __init__(self, NativeUI, alarmCarrier, *args, **kwargs):
super(AlarmExtrasWidget, self).__init__(*args, **kwargs)
self.NativeUI = NativeUI
self.alarmCarrier = alarmCarrier # Needs to refer to its containing object
self.layout = QtWidgets.QHBoxLayout()
self.layout.setSpacing(0)
self.layout.setMargin(0)
#self.alarmPayload = abstractAlarm.alarmPayload
iconLabel = QtWidgets.QLabel()
iconpath_check = os.path.join(
self.NativeUI.iconpath, "exclamation-triangle-solid.png"
)
pixmap = QtGui.QPixmap(iconpath_check).scaledToHeight(40)
iconLabel.setPixmap(pixmap)
self.layout.addWidget(iconLabel)
self.textLabel = QtWidgets.QLabel()
self.textLabel.setText('1 More Alarms')
self.textLabel.setFixedWidth(400)
self.textLabel.setAlignment(QtCore.Qt.AlignCenter)
self.textLabel.setStyleSheet("font-size: " + NativeUI.text_size + ";")
self.layout.addWidget(self.textLabel)
self.setFixedHeight(40)
self.setLayout(self.layout)
self.setStyleSheet("background-color:red;")
#self.priority = "PRIORITY_LOW"
def update_text(self, num):
self.textLabel.setText(str(num)+ ' More Alarms')
#self.code =
# self.timer = QtCore.QTimer()
# self.timer.setInterval(500) # just faster than 60Hz
# self.timer.timeout.connect(self.checkAlarm)
# self.timer.start()
# self.installEventFilter(self)
def eventFilter(self, source, event):
if (event.type() == QtCore.QEvent.MouseButtonPress):
self.NativeUI.leftBar.tab_page_buttons.button_alarms.click()
self.NativeUI.alarms_view.alarmButton.click()
return False
def get_priority(self):
return self.alarmPayload["alarm_type"]
{
"settings":[
[["APNEA", "ms", "APNEA", "SET_THRESHOLD_MIN", "APNEA"],["APNEA", "ms", "APNEA", "SET_THRESHOLD_MAX", "APNEA"]],
[["Check Pressure Patient", "ms", "CHECK_P_PATIENT", "SET_THRESHOLD_MIN", "CHECK_P_PATIENT"],["Check Pressure Patient", "ms", "CHECK_P_PATIENT", "SET_THRESHOLD_MAX", "CHECK_P_PATIENT"]],
[["FIO2", "%", "HIGH_FIO2", "SET_THRESHOLD_MIN", "HIGH_FIO2", -10, 0, 5, 1, 0],["FIO2", "%", "fiO2_percent", "SET_TARGET_", "FIO2_PERCENT", 20, 100, 21, 1, 0],["FIO2", "%", "HIGH_FIO2", "SET_THRESHOLD_MAX", "HIGH_FIO2", 0, 10, 5, 1, 0]],
[["Pressure", " ", "HIGH_PRESSURE", "SET_THRESHOLD_", "HIGH_PRESSURE"],["Pressure", " ", "HIGH_PRESSURE", "SET_THRESHOLD_", "HIGH_PRESSURE"],["Pressure", " ", "HIGH_PRESSURE", "SET_THRESHOLD_", "HIGH_PRESSURE"]],
[["Respiratory Rate", " ", "HIGH_RR", "SET_THRESHOLD_", "HIGH_RR"],["Respiratory Rate", " ", "HIGH_RR", "SET_THRESHOLD_", "HIGH_RR"],["Respiratory Rate", " ", "HIGH_RR", "SET_THRESHOLD_", "HIGH_RR"]],
[["VTE", " ", "HIGH_VTE", "SET_THRESHOLD_", "HIGH_VTE"],["VTE", " ", "HIGH_VTE", "SET_THRESHOLD_", "HIGH_VTE"],["VTE", " ", "HIGH_VTE", "SET_THRESHOLD_", "HIGH_VTE"]],
[["VTI", " ", "HIGH_VTI", "SET_THRESHOLD_", "HIGH_VTI"],["VTI", " ", "HIGH_VTI", "SET_THRESHOLD_", "HIGH_VTI"],["VTI", " ", "HIGH_VTI", "SET_THRESHOLD_", "HIGH_VTI"]],
[["Occlusion", " ", "OCCLUSION","SET_THRESHOLD_", "OCCLUSION"],["Occlusion", " ", "OCCLUSION","SET_THRESHOLD_", "OCCLUSION"],["Occlusion", " ", "OCCLUSION","SET_THRESHOLD_", "OCCLUSION"]],
[["PEEP", " ", "HIGH_PEEP","SET_THRESHOLD_", "HIGH_PEEP"],["PEEP", " ", "HIGH_PEEP","SET_THRESHOLD_", "HIGH_PEEP"],["PEEP", " ", "HIGH_PEEP","SET_THRESHOLD_", "HIGH_PEEP"]]
],
"HighLowLimits": ["High Pressure", "Occlusion"]
}
\ No newline at end of file
{
"settings":[
["APNEA", "ms", "APNEA", "SET_THRESHOLD_MIN", "APNEA"],
["Check Pressure Patient", "ms", "CHECK_P_PATIENT", "SET_THRESHOLD_MIN", "CHECK_P_PATIENT"],
["FIO2", "ms", "HIGH_FIO2", "SET_THRESHOLD_MIN", "HIGH_FIO2"],
["High Pressure", " ", "HIGH_PRESSURE", "SET_THRESHOLD_MIN", "HIGH_PRESSURE"],
["High Respiratory Rate", " ", "HIGH_RR", "SET_THRESHOLD_MIN", "HIGH_RR"],
["High VTE", " ", "HIGH_VTE", "SET_THRESHOLD_MAX", "HIGH_VTE"],
["Low VTE", " ", "LOW_VTE", "SET_THRESHOLD_MIN", "LOW_VTE"],
["High VTI", " ", "HIGH_VTI", "SET_THRESHOLD_MAX", "HIGH_VTI"],
["Low VTI", " ", "LOW_VTI", "SET_THRESHOLD_MIN", "LOW_VTI"],
["Low FIO2", " ", "LOW_FIO2", "SET_THRESHOLD_MIN", "LOW_FIO2"],
["Occlusion", " ", "OCCLUSION","SET_THRESHOLD_MIN", "OCCLUSION"],
["PEEP", " ", "HIGH_PEEP","SET_THRESHOLD_MAX", "HIGH_PEEP"],
["Low PEEP", " ", "LOW_PEEP","SET_THRESHOLD_MIN", "LOW_PEEP"] ],
"HighLowLimits": ["High Pressure", "Occlusion"]
}
\ No newline at end of file
......@@ -8,6 +8,6 @@
["Inhale Volume", "", "volume", "SET_TARGET_", "VOLUME", 200, 800, 400, 20, 0],
["Percentage O2", "", "fiO2_percent", "SET_TARGET_", "FIO2_PERCENT", 20, 100, 21, 1, 0]],
"radioSettings": ["Inhale Time", "IE Ratio"],
"enableDict":{"PC/AC":[1, 0, 1, 1, 0, 1, 0, 1], "PC/AC-PRVC":[1, 1, 0, 1, 0, 1, 1, 1], "PC-PSV":[1, 1, 0, 1, 0, 1, 0, 1], "CPAP":[1, 0, 1, 1, 0, 1, 0, 1]}
"enableDict":{"PC/AC":[1, 0, 1, 1, 0, 1, 0, 1], "PC/AC-PRVC":[1, 1, 0, 1, 0, 1, 1, 1], "PC-PSV":[1, 1, 0, 1, 0, 1, 0, 1], "CPAP":[1, 0, 1, 1, 0, 1, 0, 1]},
"mainPageSettings": ["Inhale Pressure", "Respiratory Rate", "Inhale Time", "IE Ratio", "Percentage O2" ]
}
\ No newline at end of file
......@@ -14,6 +14,11 @@ __status__ = "Prototype"
from PySide2 import QtWidgets, QtGui, QtCore
from widget_library.ok_cancel_buttons_widget import OkButtonWidget, CancelButtonWidget
from widget_library.expert_handler import ExpertHandler
from mode_widgets.personal_handler import PersonalHandler
from mode_widgets.mode_handler import ModeHandler
from mode_widgets.clinical_handler import ClinicalHandler
# from global_widgets.global_ok_cancel_buttons import okButton, cancelButton
import sys
......@@ -24,6 +29,11 @@ class SetConfirmPopup(QtWidgets.QDialog):
"""Popup called when user wants to send new values to microcontroller.
This popup shows changes and asks for confirmation"""
ExpertSend = QtCore.Signal()
ModeSend = QtCore.Signal()
PersonalSend = QtCore.Signal()
ClinicalSend = QtCore.Signal()
def __init__(self, NativeUI, *args, **kwargs):
super().__init__(*args, **kwargs)
# self.setStyleSheet("background-color:rgba(255,0,255,50%);color:rgb(0,255,0)")
......@@ -69,7 +79,10 @@ class SetConfirmPopup(QtWidgets.QDialog):
self.setAttribute(QtCore.Qt.WA_NoSystemBackground, True)
self.setWindowOpacity(0.5)
def populatePopup(self, messageList, commandList ):
def populatePopup(self, handlerWidget, messageList):
print(handlerWidget)
self.handler = handlerWidget
self.clearPopup()
if messageList == []:
messageList = ["no values were set"]
for item in messageList:
......@@ -83,7 +96,7 @@ class SetConfirmPopup(QtWidgets.QDialog):
self.listWidget.update()
self.update()
self.commandList = commandList
self.show()
def clearPopup(self):
self.listWidget.clear()
......@@ -91,10 +104,17 @@ class SetConfirmPopup(QtWidgets.QDialog):
def ok_button_pressed(self):
print('ok button pressed')
"""Send commands when ok button is clicked"""
#self.parentTemplate.liveUpdating = True
for command in self.commandList:
self.NativeUI.q_send_cmd(*command)
if isinstance(self.handler, ExpertHandler):
self.ExpertSend.emit()
elif isinstance(self.handler, ModeHandler):
self.ModeSend.emit()
elif isinstance(self.handler, PersonalHandler):
self.PersonalSend.emit()
elif isinstance(self.handler, ClinicalHandler):
self.ClinicalSend.emit()
self.close()
return 0
......
......@@ -2,13 +2,13 @@
alarm_handler.py
"""
from handler_library.handler import Handler
from handler_library.handler import PayloadHandler
from PySide2.QtCore import QObject
class AlarmHandler(Handler, QObject):
class AlarmHandler(PayloadHandler, QObject):
"""
Subclass of the Handler class (handler.py) to handle alarm data.
Subclass of the PayloadHandler class (handler.py) to handle alarm data.
Inherits from QObject to give us access to pyside2's signal class.
"""
......
import logging
from handler_library.handler import Handler
from handler_library.handler import PayloadHandler
from PySide2.QtCore import Signal, QObject
# TODO: initially we tried to have a check so that the handler only signals the display
# elements when something changeds. Problem: when the UI starts up it takes a few
# seconds for display elements to become active (may be an artefact of X11 forwarding?),
# so for something like the battery which doesn't change fast, the initial signal to the
# display is missed, and no further signal gets sent because nothing is changing.
# Current workaround is to trigger the ative_payload every time get_db completes. Could
# reinstate the check but add an override so every nth payload triggers the signal
# regardless of whether data has changed, or force some update frequency (c.f. plots)?
class BatteryHandler(PayloadHandler):
"""
Subclass of the PayloadHandler class (handler.py) to handle alarm data.
"""
class BatteryHandler(Handler, QObject):
UpdateBatteryDisplay = Signal(dict)
def __init__(self):
super().__init__()
QObject.__init__(
self
) # Give ourselves access to the QObject Signal functionality.
def __init__(self, *args, **kwargs):
super().__init__(["BATTERY"], *args, **kwargs)
def active_payload(self) -> int:
def active_payload(self, *args, **kwargs) -> int:
"""
When battery information is set, interprets it to construct the battery status
and emits the UpdateBatteryDisplay signal containing that status as a dict.
"""
new_status = {}
battery_data = self.get_db()
......@@ -71,5 +67,7 @@ class BatteryHandler(Handler, QObject):
return 85.0
elif battery_data["bat85"] == 0:
return 0.0
return 0.0
else:
raise TypeError(
"Battery Percentage (entry 'bat85' in the battery payload) is not 1 or 2."
)
from handler_library.handler import Handler
from PySide2.QtCore import Signal, QObject, QTimer
from handler_library.handler import PayloadHandler
from PySide2.QtCore import Signal
import numpy as np
from threading import Lock
class DataHandler(Handler, QObject):
class DataHandler(PayloadHandler):
UpdatePlots = Signal(dict)
def __init__(self, plot_history_length=500):
super().__init__()
QObject.__init__(self)
def __init__(self, *args, plot_history_length=500, **kwargs):
super().__init__(["DATA"], *args, **kwargs)
self.__plot_history_length = plot_history_length
self.__plots_database = {
"data": np.zeros((plot_history_length, 4)),
......@@ -23,7 +22,7 @@ class DataHandler(Handler, QObject):
}
self.__plot_lock = Lock()
def active_payload(self):
def active_payload(self, *args, **kwargs):
"""
Take the raw payload information into conveniently plotable forms and store them
in self.__plots_database.
......
......@@ -5,28 +5,29 @@ handler.py
from threading import Lock
from global_widgets.global_spinbox import labelledSpin
from widget_library.ok_cancel_buttons_widget import OkButtonWidget, CancelButtonWidget, OkSendButtonWidget
from PySide2.QtCore import QObject
from PySide2.QtWidgets import QRadioButton
class Handler:
class GenericDataHandler(QObject):
"""
Base class for the data handlers.
Base class for non-payload data handlers.
"""
def __init__(self):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.__database = {}
self.__lock = Lock()
def set_db(self, payload: dict) -> int:
def set_db(self, data: dict) -> int:
"""
Set the content of the __database dictionary.
Copy the contents of 'data' to the internal database.
"""
with self.__lock:
for key in payload:
self.__database[key] = payload[key]
for key in data:
self.__database[key] = data[key]
self.active_payload()
self.on_data_set()
return 0
def get_db(self) -> dict:
......@@ -36,9 +37,53 @@ class Handler:
with self.__lock:
return dict(self.__database)
def active_payload(self):
def on_data_set(self):
"""
Overridable function called after recieving new data.
"""
pass
class PayloadHandler(GenericDataHandler):
"""
Base class for the payload data handlers.
"""
def __init__(self, payload_types: list, *args, **kwargs):
super().__init__(*args, **kwargs)
for key in payload_types:
if not isinstance(key, str):
raise TypeError(
"payload types must be of type 'str', not %s", type(key)
)
self.__database = {}
self.__lock = Lock()
self.__payload_types = payload_types
def set_db(self, payload: dict) -> int:
"""
If the provided database is of the correct type, copy its contents to the database
"""
if payload["type"] not in self.__payload_types:
return 1
with self.__lock:
for key in payload[payload["type"]]:
self.__database[key] = payload[payload["type"]][key]
self.active_payload(payload)
return 0
def get_db(self) -> dict:
"""
Return the content of the __database dictionary.
"""
with self.__lock:
return dict(self.__database)
def active_payload(self, payload: dict):
"""
Overridable function called after recieving new data. Passes in the full payload
so that we have access to the full context of the information.
"""
pass
......@@ -2,25 +2,22 @@
measurement_handler.py
"""
from handler_library.handler import Handler
from PySide2.QtCore import Signal, QObject
from handler_library.handler import PayloadHandler
from PySide2.QtCore import Signal
import logging
class MeasurementHandler(Handler, QObject):
class MeasurementHandler(PayloadHandler):
"""
Subclass of the Handler class (handler.py) to handle cycle and readback data.
Inherits from QObject to give us access to pyside2's signal class.
Subclass of the PayloadHandler class (handler.py) to handle cycle and readback data.
"""
UpdateMeasurements = Signal(dict)
def __init__(self):
super().__init__()
QObject.__init__(self)
def __init__(self, *args, **kwargs):
super().__init__(["CYCLE", "READBACK"], *args, **kwargs)
def active_payload(self) -> int:
def active_payload(self, *args, **kwargs) -> int:
cycle_data = self.get_db()
outdict = {}
......
......@@ -2,15 +2,13 @@
personal_handler.py
"""
from handler_library.handler import Handler
from PySide2.QtCore import Signal, QObject
from handler_library.handler import PayloadHandler
from PySide2.QtCore import Signal
class PersonalHandler(Handler, QObject):
class PersonalHandler(PayloadHandler):
"""
Subclass of the Handler class (handler.py) to handle personal data.
Inherits from QObject to give us access to pyside2's signal class.
Subclass of the PayloadHandler class (handler.py) to handle personal data.
Adds the UpdatePersonalDisplay signal designed to convey information to be displayed
to the display widget.
......@@ -18,11 +16,10 @@ class PersonalHandler(Handler, QObject):
UpdatePersonalDisplay = Signal(dict)
def __init__(self):
super().__init__()
QObject.__init__(self)
def __init__(self, *args, **kwargs):
super().__init__(["PERSONAL"], *args, **kwargs)
def active_payload(self) -> int:
def active_payload(self, *args, **kwargs) -> int:
"""
When the personal data is updated, extract those parameters needed for display
and emit the UpdatePersonalDisplay signal.
......
"""
readback_handler.py
"""
from handler_library.handler import Handler
from PySide2.QtCore import Signal, QObject
class ReadbackHandler(Handler, QObject):
"""
Subclass of the Handler class (handler.py) to handle readback data.
Inherits from QObject to give us access to pyside2's signal class.
"""
ReadbackSignal = Signal(dict)
def __init__(self):
super().__init__()
QObject.__init__(self)
def active_payload(self) -> int:
cycle_data = self.get_db()
outdict = {}
for key in ["peep", "inhale_exhale_ratio"]:
outdict[key] = cycle_data[key]
self.ReadbackSignal.emit(outdict)
return 0
from global_widgets.global_spinbox import labelledSpin
from widget_library.ok_cancel_buttons_widget import OkButtonWidget, CancelButtonWidget, OkSendButtonWidget
#from global_widgets.global_send_popup import SetConfirmPopup
from widget_library.spin_buttons_widget import SpinButton, SpinButtonsWidget
from PySide2 import QtWidgets, QtGui, QtCore
from handler_library.handler import PayloadHandler
import logging
import json
class ClinicalHandler(PayloadHandler):
#modeSwitched = QtCore.Signal(str)
UpdateClinical = QtCore.Signal(dict)
OpenPopup = QtCore.Signal(PayloadHandler, list)
#settingToggle = QtCore.Signal(str)
def __init__(self, NativeUI, *args, **kwargs):
super().__init__(['TARGET'],*args, **kwargs)
#super(TabModes, self).__init__(NativeUI, *args, **kwargs)
self.NativeUI = NativeUI
self.spinDict = {}
self.buttonDict = {}
self.radioDict = {}
self.commandList = []
self.manuallyUpdated = False
with open("NativeUI/configs/clinical_config.json") as json_file:
clinicalDict = json.load(json_file)
#self.relevantKeys = [print(setting) for setting in clinicalDict['settings']]
def add_widget(self, widget, key: str):
if isinstance(widget, labelledSpin):
self.spinDict[key] = widget
if isinstance(widget, OkButtonWidget) or isinstance(widget, CancelButtonWidget) or isinstance(widget, OkSendButtonWidget):
self.buttonDict[key] = widget
if isinstance(widget, QtWidgets.QRadioButton):
self.radioDict[key] = widget
def active_payload(self, *args) -> int:
target_data = self.get_db()
outdict = {}
for key in self.relevantKeys:
try:
outdict[key] = target_data[key]
except KeyError:
logging.debug("Invalid key %s in measurement database", key)
self.UpdateClinical.emit(outdict)
return 0
def handle_okbutton_click(self):
print('ok')
message, command = [], []
for widget in self.spinDict:
if self.spinDict[widget].manuallyUpdated:
print('True')
setVal = self.spinDict[widget].get_value()
message.append("set" + widget + " to " + str(setVal))
command.append(
[
self.spinDict[widget].cmd_type,
self.spinDict[widget].cmd_code,
setVal,
]
)
# create a signal emitting message, command, handler identifier - in nativeui connect to a popup widget
# command sending should occur in handler
self.commandList = command
self.OpenPopup.emit(self,message)
def sendCommands(self):
if self.commandList == []:
a=1
else:
for command in self.commandList:
self.NativeUI.q_send_cmd(*command)
self.commandSent()
return 0
def commandSent(self):
self.commandList = []
for widget in self.spinDict:
self.spinDict[widget].manuallyUpdated = False
self.active_payload()
self.refresh_button_colour()
def handle_manual_change(self, changed_spin_key):
self.active_payload()
self.refresh_button_colour()
def refresh_button_colour(self):
self.manuallyUpdated = False
for spin in dict(self.spinDict):
self.manuallyUpdated = self.manuallyUpdated or self.spinDict[spin].manuallyUpdated
for button in dict(self.buttonDict):
if isinstance(self.buttonDict[button], OkSendButtonWidget):
self.buttonDict[button].setColour(str(int(True)))
else:
self.buttonDict[button].setColour(str(int(self.manuallyUpdated)))
def get_mode(self, key: str):
for mode in self.modeList:
if mode in key:
return mode
from global_widgets.global_spinbox import labelledSpin
from widget_library.ok_cancel_buttons_widget import OkButtonWidget, CancelButtonWidget, OkSendButtonWidget
from global_widgets.global_send_popup import SetConfirmPopup
#from global_widgets.global_send_popup import SetConfirmPopup
from widget_library.spin_buttons_widget import SpinButton, SpinButtonsWidget
from PySide2 import QtWidgets, QtGui, QtCore
from handler_library.handler import Handler
from handler_library.handler import PayloadHandler
import logging
import json
class ModeHandler(Handler, QtCore.QObject): # chose QWidget over QDialog family because easier to modify
class ModeHandler(PayloadHandler):
modeSwitched = QtCore.Signal(str)
UpdateModes = QtCore.Signal(dict)
OpenPopup = QtCore.Signal(PayloadHandler, list)
settingToggle = QtCore.Signal(str)
def __init__(self, NativeUI, confirmPopup, *args, **kwargs):
super().__init__(*args, **kwargs)
QtCore.QObject.__init__(self)
def __init__(self, NativeUI, *args, **kwargs):
super().__init__(['TARGET'],*args, **kwargs)
#super(TabModes, self).__init__(NativeUI, *args, **kwargs)
self.NativeUI = NativeUI
self.spinDict = {}
self.buttonDict = {}
self.radioDict = {}
self.manuallyUpdatedBoolDict = { mode: False for mode in self.NativeUI.modeList }
self.popup = confirmPopup
self.commandList = []
self.mainSpinDict = {}
self.mainButtonDict = {}
self.modeList = self.NativeUI.modeList + ['CURRENT']
self.manuallyUpdatedBoolDict = { mode: False for mode in self.modeList }
self.mainManuallyUpdated = False
self.activeMode = self.modeList[0]
with open("NativeUI/configs/mode_config.json") as json_file:
modeDict = json.load(json_file)
self.relevantKeys = [setting[2] for setting in modeDict['settings']]
def add_widget(self, widget, key: str):
if isinstance(widget, labelledSpin):
self.spinDict[key] = widget
if isinstance(widget, OkButtonWidget) or isinstance(widget, CancelButtonWidget) or isinstance(widget, OkSendButtonWidget):
self.buttonDict[key] = widget
if self.get_mode(key) == 'CURRENT':
self.mainButtonDict[key] = widget
else:
self.buttonDict[key] = widget
if isinstance(widget, QtWidgets.QRadioButton):
self.radioDict[key] = widget
if isinstance(widget, SpinButton):
self.mainSpinDict[key] = widget
def active_payload(self) -> int:
def active_payload(self, *args) -> int:
target_data = self.get_db()
outdict = {}
for key in [
"respiratory_rate",
"fiO2_percent",
"inhale_trigger_threshold",
"exhale_trigger_threshold",
"volume",
"inspiratory_pressure",
"inhale_time",
"ie_ratio",
]:
for key in self.relevantKeys:
try:
outdict[key] = target_data[key]
except KeyError:
......@@ -53,74 +62,91 @@ class ModeHandler(Handler, QtCore.QObject): # chose QWidget over QDialog family
return 0
# def update_values(self):
# db = self.NativeUI.get_db('targets')
# if db == {}:
# return 0 # do nothing
# else:
# self.manuallyUpdatedBoolDict = { mode: False for mode in self.NativeUI.modeList }
# for spin in self.spinDict:
# self.spinDict[spin].update_value(db)
def handle_okbutton_click(self, key):
print('ok')
print(key)
mode = self.get_mode(key)
print(mode)
message, command = [], []
for widget in self.spinDict:
if mode in widget:
if self.spinDict[widget].manuallyUpdated:
setVal = self.spinDict[widget].get_value()
message.append("set" + widget + " to " + str(setVal))
command.append(
[
self.spinDict[widget].cmd_type,
self.spinDict[widget].cmd_code,
setVal,
]
)
self.popup.clearPopup()
self.popup.populatePopup(message, command)
self.popup.okButton.pressed.connect(lambda i=mode: self.commandSent(i))
#self.popup.okButton.pressed.connect(self.modeSwitched.emit())
if 'send' in key:
self.popup.okButton.click()
self.modeSwitched.emit(mode)
if (mode in widget) and self.spinDict[widget].manuallyUpdated:
print('True')
setVal = self.spinDict[widget].get_value()
message.append("set" + widget + " to " + str(setVal))
command.append(
[
self.spinDict[widget].cmd_type,
self.spinDict[widget].cmd_code,
setVal,
]
)
# create a signal emitting message, command, handler identifier - in nativeui connect to a popup widget
# command sending should occur in handler
self.commandList = command
self.OpenPopup.emit(self,message)
def handle_mainokbutton_click(self):
message, command = [], []
for widget in self.mainSpinDict:
if self.mainSpinDict[widget].manuallyUpdated:
setVal = self.mainSpinDict[widget].get_value()
message.append("set" + widget + " to " + str(setVal))
command.append(
[
self.mainSpinDict[widget].cmd_type,
self.mainSpinDict[widget].cmd_code,
setVal,
]
)
# create a signal emitting message, command, handler identifier - in nativeui connect to a popup widget
# command sending should occur in handler
self.commandList = command
self.OpenPopup.emit(message)
def sendCommands(self):
if self.commandList == []:
a=1
else:
self.popup.show()
def commandSent(self, mode):
self.popup.clearPopup()
for command in self.commandList:
self.NativeUI.q_send_cmd(*command)
self.modeSwitched.emit(self.activeMode)
self.commandSent()
return 0
#self.toggleButtons(mode,0)
def commandSent(self):
self.commandList = []
for widget in self.spinDict:
if mode in widget:
if self.activeMode in widget:
self.spinDict[widget].manuallyUpdated = False
for widget in self.mainSpinDict:
self.mainSpinDict[widget].manuallyUpdated = False
self.active_payload()
self.refresh_button_colour()
self.refresh_main_button_colour()
def handle_manual_change(self, changed_spin_key):
self.active_payload()
self.refresh_button_colour()
self.refresh_main_button_colour()
def handle_radio_toggled(self, radioButtonState, radioKey):
"""TODO Docstring"""
mode = self.get_mode(radioKey)
radioButton = self.radioDict[radioKey]
spinKey= radioKey.replace('radio', 'spin')
spinBox = self.spinDict[spinKey]
radioButtonState = radioButton.isChecked()
spinBox.simpleSpin.setEnabled(radioButtonState)
spinBox.simpleSpin.setProperty("bgColour", str(int(radioButtonState)))
spinBox.simpleSpin.setProperty("textColour", str(int(radioButtonState)))
spinBox.simpleSpin.style().unpolish(spinBox.simpleSpin)
spinBox.simpleSpin.style().polish(spinBox.simpleSpin)
spinBox.setEnabled(radioButtonState)
if mode == self.NativeUI.currentMode:
self.NativeUI.widgets.spin_buttons.setStackWidget(spinBox.label)
self.settingToggle.emit(spinBox.label)
def refresh_button_colour(self):
self.manuallyUpdatedBoolDict = { mode: False for mode in self.NativeUI.modeList }
for spin in self.spinDict:
self.manuallyUpdatedBoolDict = { mode: False for mode in self.modeList }
for spin in dict(self.spinDict):
self.manuallyUpdatedBoolDict[self.get_mode(spin)] = self.manuallyUpdatedBoolDict[self.get_mode(spin)] or self.spinDict[spin].manuallyUpdated
for button in self.buttonDict:
for button in dict(self.buttonDict):
mode = str(self.get_mode(button))
if isinstance(self.buttonDict[button], OkSendButtonWidget) and (mode != self.NativeUI.currentMode):
self.buttonDict[button].setColour(str(int(True)))
......@@ -128,7 +154,14 @@ class ModeHandler(Handler, QtCore.QObject): # chose QWidget over QDialog family
self.buttonDict[button].setColour(str(int(self.manuallyUpdatedBoolDict[mode])))
def refresh_main_button_colour(self):
self.manuallyUpdatedBoolDict['CURRENT'] = False
for spin in self.mainSpinDict:
self.manuallyUpdatedBoolDict['CURRENT'] = self.manuallyUpdatedBoolDict['CURRENT'] or self.mainSpinDict[spin].manuallyUpdated
for button in self.mainButtonDict:
self.mainButtonDict[button].setColour(str(int(self.manuallyUpdatedBoolDict['CURRENT'])))
def get_mode(self, key: str):
for mode in self.NativeUI.modeList:
for mode in self.modeList:
if mode in key:
return mode
from global_widgets.global_spinbox import labelledSpin
from widget_library.ok_cancel_buttons_widget import OkButtonWidget, CancelButtonWidget, OkSendButtonWidget
from global_widgets.global_send_popup import SetConfirmPopup
#from global_widgets.global_send_popup import SetConfirmPopup
from widget_library.line_edit_widget import LabelledLineEditWidget
from handler_library.handler import PayloadHandler
from PySide2.QtCore import Signal
from PySide2 import QtWidgets, QtGui, QtCore
class PersonalHandler(QtWidgets.QWidget): # chose QWidget over QDialog family because easier to modify
class PersonalHandler(PayloadHandler): # chose QWidget over QDialog family because easier to modify
UpdatePersonalDisplay = Signal(dict)
def __init__(self, NativeUI, confirmPopup, *args, **kwargs):
super().__init__(*args, **kwargs)
#super(TabModes, self).__init__(NativeUI, *args, **kwargs)
def __init__(self, NativeUI, *args, **kwargs):
super().__init__(['PERSONAL'], *args, **kwargs)
self.NativeUI = NativeUI
self.spinDict = {}
self.buttonDict = {}
self.textDict = {}
self.popup = confirmPopup
def add_widget(self, widget, key: str):
if isinstance(widget, labelledSpin):
......@@ -22,4 +23,16 @@ class PersonalHandler(QtWidgets.QWidget): # chose QWidget over QDialog family b
if isinstance(widget, LabelledLineEditWidget):
self.textDict[key] = widget
if isinstance(widget, OkButtonWidget) or isinstance(widget, CancelButtonWidget) or isinstance(widget, OkSendButtonWidget):
self.buttonDict[key] = widget
\ No newline at end of file
self.buttonDict[key] = widget
def active_payload(self, *args, **kwargs) -> int:
"""
When the personal data is updated, extract those parameters needed for display
and emit the UpdatePersonalDisplay signal.
"""
current_data = self.get_db()
outdict = {}
for key in ["name", "patient_id", "age", "sex", "height", "weight"]:
outdict[key] = current_data[key]
self.UpdatePersonalDisplay.emit(outdict)
return 0
\ No newline at end of file
{
"version": 182,
"timestamp": 0,
"type": "BATTERY",
"bat": 0,
"ok": 1,
"alarm": 0,
"rdy2buf": 0,
"bat85": 0,
"prob_elec": 0,
"dummy": false
}
\ No newline at end of file
{
"on_mains_power": true,
"on_battery_power": false,
"battery_percent": 0.0,
"electrical_problem": null
}
\ No newline at end of file
{
"type": "TEST",
"TEST":{
"version": 182,
"timestamp": 0,
"type": "TEST",
"attribute_1": true,
"attribute_2": 1,
"attribute_3": 0.579
}
}
......@@ -27,121 +27,122 @@ def widget():
# Test default values of databases(no set method involved)
def test_must_return_if_raises_attribute_error_when_false_db_item_is_got_from_get_db(
widget,
):
with pytest.raises(AttributeError):
widget.get_db("__false_db_item")
# Superseeded by handlers.
# def test_must_return_if_raises_attribute_error_when_false_db_item_is_got_from_get_db(
# widget,
# ):
# with pytest.raises(AttributeError):
# widget.get_db("__false_db_item")
def test_must_return_correct_db_item_from_get_db_data(widget):
assert widget.get_db("__data") == {} and widget.get_db("data") == {}
# def test_must_return_correct_db_item_from_get_db_data(widget):
# assert widget.get_db("__data") == {} and widget.get_db("data") == {}
def test_must_return_correct_db_item_from_get_db_readback(widget):
assert widget.get_db("__readback") == {} and widget.get_db("readback") == {}
# def test_must_return_correct_db_item_from_get_db_readback(widget):
# assert widget.get_db("__readback") == {} and widget.get_db("readback") == {}
def test_must_return_correct_db_item_from_get_db_cycle(widget):
assert widget.get_db("__cycle") == {} and widget.get_db("cycle") == {}
# def test_must_return_correct_db_item_from_get_db_cycle(widget):
# assert widget.get_db("__cycle") == {} and widget.get_db("cycle") == {}
def test_must_return_correct_db_item_from_get_db_battery(widget):
assert widget.get_db("__battery") == {} and widget.get_db("battery") == {}
# def test_must_return_correct_db_item_from_get_db_battery(widget):
# assert widget.get_db("__battery") == {} and widget.get_db("battery") == {}
def test_must_return_correct_db_item_from_get_db_plots(widget):
plot_history_length = 1000
plot_dict = {
"data": np.zeros((plot_history_length, 5)),
"timestamp": list(el * (-1) for el in range(plot_history_length))[::-1],
"pressure": list(0 for _ in range(plot_history_length)),
"flow": list(0 for _ in range(plot_history_length)),
"volume": list(0 for _ in range(plot_history_length)),
"pressure_axis_range": [0, 20],
"flow_axis_range": [-40, 80],
"volume_axis_range": [0, 80],
}
assert (
widget.get_db("__plots").keys() == plot_dict.keys()
and widget.get_db("plots").keys() == plot_dict.keys()
)
# def test_must_return_correct_db_item_from_get_db_plots(widget):
# plot_history_length = 1000
# plot_dict = {
# "data": np.zeros((plot_history_length, 5)),
# "timestamp": list(el * (-1) for el in range(plot_history_length))[::-1],
# "pressure": list(0 for _ in range(plot_history_length)),
# "flow": list(0 for _ in range(plot_history_length)),
# "volume": list(0 for _ in range(plot_history_length)),
# "pressure_axis_range": [0, 20],
# "flow_axis_range": [-40, 80],
# "volume_axis_range": [0, 80],
# }
# assert (
# widget.get_db("__plots").keys() == plot_dict.keys()
# and widget.get_db("plots").keys() == plot_dict.keys()
# )
def test_must_return_correct_db_item_from_get_db_alarms(widget):
assert widget.get_db("__alarms") == {} and widget.get_db("alarms") == {}
# def test_must_return_correct_db_item_from_get_db_alarms(widget):
# assert widget.get_db("__alarms") == {} and widget.get_db("alarms") == {}
def test_must_return_correct_db_item_from_get_db_targets(widget):
assert widget.get_db("__targets") == {} and widget.get_db("targets") == {}
# def test_must_return_correct_db_item_from_get_db_targets(widget):
# assert widget.get_db("__targets") == {} and widget.get_db("targets") == {}
def test_must_return_correct_db_item_from_get_db_personal(widget):
assert widget.get_db("__personal") == {} and widget.get_db("personal") == {}
# def test_must_return_correct_db_item_from_get_db_personal(widget):
# assert widget.get_db("__personal") == {} and widget.get_db("personal") == {}
# Test set methods with sample payloads
def test_must_return_0_for_set_data_db(widget):
with open("/home/pi/hev/samples/dataSample.json", "r") as f:
data_payload = json.load(f)
assert widget.__set_db("data", data_payload) == 0
# # Test set methods with sample payloads
# def test_must_return_0_for_set_data_db(widget):
# with open("/home/pi/hev/samples/dataSample.json", "r") as f:
# data_payload = json.load(f)
# assert widget.__set_db("data", data_payload) == 0
def test_must_return_0_for_set_targets_db(widget):
with open("/home/pi/hev/samples/targetSample.json", "r") as g:
target_payload = json.load(g)
assert widget.__set_db("targets", target_payload) == 0
# def test_must_return_0_for_set_targets_db(widget):
# with open("/home/pi/hev/samples/targetSample.json", "r") as g:
# target_payload = json.load(g)
# assert widget.__set_db("targets", target_payload) == 0
def test_must_return_0_for_set_readback_db(widget):
with open("/home/pi/hev/samples/readbackSample.json", "r") as f:
readback_payload = json.load(f)
assert widget.__set_db("readback", readback_payload) == 0
# def test_must_return_0_for_set_readback_db(widget):
# with open("/home/pi/hev/samples/readbackSample.json", "r") as f:
# readback_payload = json.load(f)
# assert widget.__set_db("readback", readback_payload) == 0
def test_must_return_0_for_set_cycle_db(widget):
with open(
"/home/pi/hev/NativeUI/tests/unittests/fixtures/cycleSample.json", "r"
) as f:
cycle_payload = json.load(f)
assert widget.__set_db("cycle", cycle_payload) == 0
# def test_must_return_0_for_set_cycle_db(widget):
# with open(
# "/home/pi/hev/NativeUI/tests/unittests/fixtures/cycleSample.json", "r"
# ) as f:
# cycle_payload = json.load(f)
# assert widget.__set_db("cycle", cycle_payload) == 0
def test_must_return_0_for_set_battery_db(widget):
with open("/home/pi/hev/samples/batterySample.json", "r") as f:
battery_payload = json.load(f)
assert widget.__set_db("battery", battery_payload) == 0
# def test_must_return_0_for_set_battery_db(widget):
# with open("/home/pi/hev/samples/batterySample.json", "r") as f:
# battery_payload = json.load(f)
# assert widget.__set_db("battery", battery_payload) == 0
def test_must_return_0_for_set_plots_db(widget):
with open("/home/pi/hev/samples/dataSample.json", "r") as f:
data_payload = json.load(f)
assert widget.__set_plots_db(data_payload) == 0
# def test_must_return_0_for_set_plots_db(widget):
# with open("/home/pi/hev/samples/dataSample.json", "r") as f:
# data_payload = json.load(f)
# assert widget.__set_plots_db(data_payload) == 0
def test_must_return_error_if_not_data_is_sent_as_payload_for_set_plots_db(widget):
with open("/home/pi/hev/samples/batterySample.json", "r") as f:
battery_payload = json.load(f)
with pytest.raises(KeyError):
widget.__set_plots_db(battery_payload)
# def test_must_return_error_if_not_data_is_sent_as_payload_for_set_plots_db(widget):
# with open("/home/pi/hev/samples/batterySample.json", "r") as f:
# battery_payload = json.load(f)
# with pytest.raises(KeyError):
# widget.__set_plots_db(battery_payload)
def test_must_return_0_when__update_plot_ranges_correctly(widget):
assert widget.__update_plot_ranges() == 0
# def test_must_return_0_when__update_plot_ranges_correctly(widget):
# assert widget.__update_plot_ranges() == 0
def test_must_return_0_for_set_alarms_db(widget):
with open("/home/pi/hev/samples/alarmSample.json", "r") as f:
alarm_payload = json.load(f)
assert widget.__set_db("alarms", alarm_payload) == 0
# def test_must_return_0_for_set_alarms_db(widget):
# with open("/home/pi/hev/samples/alarmSample.json", "r") as f:
# alarm_payload = json.load(f)
# assert widget.__set_db("alarms", alarm_payload) == 0
def test_must_return_0_for_set_personal_db(widget):
with open(
"/home/pi/hev/NativeUI/tests/unittests/fixtures/personalSample.json", "r"
) as f:
personal_payload = json.load(f)
assert widget.__set_db("personal", personal_payload) == 0
# def test_must_return_0_for_set_personal_db(widget):
# with open(
# "/home/pi/hev/NativeUI/tests/unittests/fixtures/personalSample.json", "r"
# ) as f:
# personal_payload = json.load(f)
# assert widget.__set_db("personal", personal_payload) == 0
# Asyncio can handle event loops, but we need to add more interaction i think
......@@ -152,6 +153,9 @@ def test_must_return_0_for_set_personal_db(widget):
def test_get_updates_data_payload(widget):
"""
Currently fails because the dataSample.json is only part of the data payload.
"""
with open("/home/pi/hev/samples/dataSample.json", "r") as f:
data_payload = json.load(f)
widget.get_updates(data_payload)
......
"""
Unit tests for the handler files.
"""
import json
import os
from unittest.mock import MagicMock, patch
from handler_library.handler import Handler
from handler_library.battery_handler import BatteryHandler
def test_handler():
"""
Tests the default handler.
Test for set_db and get_db to set the database from a given payload and compare the
db imported from get_db.
Test for if active_payload gets fired when set_db is called.
"""
# Initalise the handler and import sample test json file
handler = Handler(["TEST"])
test_json_file_path = (
os.environ["PYTHONPATH"].split(os.pathsep)[0]
+ "/tests/unittests/fixtures/testSample.json"
)
test_json = json.load(open(test_json_file_path))
# Set the database for the imported json and get the database imported
set_db_return = handler.set_db(test_json)
db = handler.get_db()
# Check if the input payload and output database are the same
if test_json["TEST"] == db:
payload_database_comparison = True
else:
payload_database_comparison = False
# Mock active_payload to return true if it is called.
handler.active_payload = MagicMock(return_value=True)
# Check whether conditions have been met to pass test
assert set_db_return == 0, "set_db does not return 0 for a valid payload"
assert (
handler.active_payload() is True
), "active_payload was not called when set_db was run."
assert (
payload_database_comparison is True
), "set_db does not set the inputted payload to the database."
def test_battery_handler():
"""
Tests the battery_handler logic by giving active_payload a sample battery payload (NativeUI/tests/unittests/fixtures/batterySample.json) with a known output status (NativeUI/tests/unittests/fixtures/battery_status_output_sample.json).
"""
# Initalise the battery handler and import sample battery json file
batt_handler = BatteryHandler()
batt_json_file_path = (
os.environ["PYTHONPATH"].split(os.pathsep)[0]
+ "/tests/unittests/fixtures/batterySample.json"
)
batt_json = json.load(open(batt_json_file_path))
# Set true/false variables
UpdateBatteryDisplay_activated = False
new_status_correctly_set = False
batt_per_1_correctly_set = False
batt_per_0_correctly_set = False
# Compute battery percent payload information
batt_per_1 = {"bat85": 1}
batt_per_0 = {"bat85": 0}
batt_handler.compute_battery_percent(batt_per_0)
# Mock the get_db function to give the sample input json
batt_handler.get_db = MagicMock(return_value=batt_json)
# Mock function to replace UpdateBatteryDisplay which checks if the function has been called and whether the output status is correct.
def mock_UpdateBatteryDisplay(new_status: dict):
nonlocal UpdateBatteryDisplay_activated
nonlocal new_status_correctly_set
# Set activated variable to true to show that this function was called
UpdateBatteryDisplay_activated = True
# Check whether new_status is the expected output
expected_status_file_path = (
os.environ["PYTHONPATH"].split(os.pathsep)[0]
+ "/tests/unittests/fixtures/battery_status_output_sample.json"
)
expected_status = json.load(open(expected_status_file_path))
if new_status == expected_status:
new_status_correctly_set = True
else:
pass
# Connect to active_payload signal
batt_handler.UpdateBatteryDisplay.connect(mock_UpdateBatteryDisplay)
# Run the battery handler logic
batt_handler.active_payload()
# Run the battery handler compute battery percent logic
if batt_handler.compute_battery_percent(batt_per_1) == 85.0:
batt_per_1_correctly_set = True
if batt_handler.compute_battery_percent(batt_per_0) == 0.0:
batt_per_0_correctly_set = True
# Check whether conditions have been met to pass test
assert (
UpdateBatteryDisplay_activated is True
), "UpdateBatteryDisplay.emit(new_status) is not called."
assert (
new_status_correctly_set is True
), "Output of new_status does not match the expected output."
assert (
batt_per_1_correctly_set is True
), "compute_battery_percent does not return 85% when bat85 is set to 1."
assert (
batt_per_0_correctly_set is True
), "compute_battery_percent does not return 0% when bat85 is set to 0."
......@@ -14,8 +14,8 @@ __maintainer__ = "Benjamin Mummery"
__email__ = "benjamin.mummery@stfc.ac.uk"
__status__ = "Prototype"
from PySide2 import QtWidgets, QtCore
from PySide2.QtGui import QFont
from PySide2 import QtWidgets, QtCore, QtGui
#from PySide2.QtGui import QFont, QSizePolicy
from widget_library.switchable_stack_widget import SwitchableStackWidget
import json
......@@ -70,6 +70,7 @@ class Layout:
self.layout_startup_main(),
self.layout_mode_startup(), # self, settings, mode:str, enableList:list, buttons: bool)
self.layout_mode_personal("startup_", False),
self.layout_startup_confirmation()
]
),
"startup_stack",
......@@ -210,7 +211,7 @@ class Layout:
page_main_bottom_layout = QtWidgets.QHBoxLayout()
center_widgets = [self.widgets.plot_stack]
bottom_widgets = [self.widgets.history_buttons, self.widgets.spin_buttons]
bottom_widgets = [self.widgets.history_buttons, self.layout_main_spin_buttons()]
self.widgets.history_buttons.set_size(None, self.main_page_bottom_bar_height)
self.widgets.history_buttons.setFont(self.NativeUI.text_font)
# TODO spin_buttons sizes
......@@ -239,9 +240,9 @@ class Layout:
[
self.layout_tab_alarm_list(alarm_tab_widgets),
self.layout_tab_alarm_table(alarm_table_tab_widgets),
self.widgets.clinical_tab,
#self.layout_tab_clinical_limits(),
],
["List of Alarms", "Alarm Table", "Clinical Limits"],
["List of Alarms", "Alarm Table"],#, "Clinical Limits"],
)
page_alarms.setFont(self.NativeUI.text_font)
return page_alarms
......@@ -599,3 +600,108 @@ class Layout:
vlayout.addLayout(hButtonLayout)
return expert_tab
def layout_main_spin_buttons(self) -> QtWidgets.QWidget:
hlayout = QtWidgets.QHBoxLayout()
with open("NativeUI/configs/mode_config.json") as json_file:
modeDict = json.load(json_file)
stack = self.NativeUI.widgets.main_mode_stack
for setting in modeDict['settings']:
if setting[0] in modeDict['mainPageSettings']:
attrName = 'CURRENT_' + setting[2]
widg = self.NativeUI.widgets.get_widget(attrName)
if setting[0] in modeDict['radioSettings']:
stack.addWidget(widg)
else:
hlayout.addWidget(widg)
hlayout.addWidget(self.NativeUI.widgets.main_mode_stack)
vbuttonLayout = QtWidgets.QVBoxLayout()
okButton = self.NativeUI.widgets.get_widget('CURRENT_ok_button')
okButton.setSizePolicy(QtGui.QSizePolicy.Expanding,QtGui.QSizePolicy.Fixed)
vbuttonLayout.addWidget(okButton)
cancelButton = self.NativeUI.widgets.get_widget('CURRENT_cancel_button')
cancelButton.setSizePolicy(QtGui.QSizePolicy.Expanding,QtGui.QSizePolicy.Fixed)
vbuttonLayout.addWidget(cancelButton)
hlayout.addLayout(vbuttonLayout)
combined_spin_buttons = QtWidgets.QWidget()
combined_spin_buttons.setLayout(hlayout)
return combined_spin_buttons
def layout_tab_clinical_limits(self):
with open("NativeUI/configs/clinical_config.json") as json_file:
clinicalDict = json.load(json_file)
vlayout = QtWidgets.QVBoxLayout()
for setting in clinicalDict['settings']:
attrName = 'clinical_spin_' + setting[0][2]
hlayout = QtWidgets.QHBoxLayout()
hlayout.addWidget(self.NativeUI.widgets.get_widget(attrName + '_min'))
hlayout.addWidget(self.NativeUI.widgets.get_widget(attrName + '_max'))
vlayout.addLayout(hlayout)
hbuttonlayout = QtWidgets.QHBoxLayout()
hbuttonlayout.addWidget(self.NativeUI.widgets.get_widget('clinical_ok_button'))
hbuttonlayout.addWidget(self.NativeUI.widgets.get_widget('clinical_cancel_button'))
vlayout.addLayout(hbuttonlayout)
# hlayoutMeta.addLayout(vlayout)
# hlayoutMeta.addLayout(vlayout2)
clinical_page = QtWidgets.QWidget()
clinical_page.setLayout(vlayout)
return clinical_page
# def layout_tab_clinical_limits(self):
# with open("NativeUI/configs/clinical_config.json") as json_file:
# clinicalDict = json.load(json_file)
#
#
# grid = QtWidgets.QGridLayout()
# #grid.setHorizontalSpacing(0)
# vlayout = QtWidgets.QVBoxLayout()
# vlayout2 = QtWidgets.QVBoxLayout()
# i = 0
# for setting in clinicalDict['settings']:
# attrName = 'spin_' + setting[2]
# if setting[0] in clinicalDict['HighLowLimits']:
#
# grid.addWidget(self.NativeUI.widgets.get_widget(attrName), int(i / 2), 2 * (i % 2), 1, 1)
# grid.addWidget(self.NativeUI.widgets.get_widget(attrName + '_2'), int(i / 2), 2 * (i % 2) + 1, 1, 1)
# else:
# grid.addWidget(self.NativeUI.widgets.get_widget(attrName), int(i / 2), 2 * (i % 2), 1, 2)
#
# i = i + 1
#
# vlayout.addLayout(grid)
# hbuttonlayout = QtWidgets.QHBoxLayout()
# hbuttonlayout.addWidget(self.NativeUI.widgets.get_widget('clinical_ok_button'))
# hbuttonlayout.addWidget(self.NativeUI.widgets.get_widget('clinical_cancel_button'))
# vlayout.addLayout(hbuttonlayout)
# # hlayoutMeta.addLayout(vlayout)
# # hlayoutMeta.addLayout(vlayout2)
# clinical_page = QtWidgets.QWidget()
# clinical_page.setLayout(vlayout)
# return clinical_page
#
def layout_startup_confirmation(self):
vlayout = QtWidgets.QVBoxLayout()
i = 0
for key, spinBox in self.NativeUI.widgets.get_widget('startup_handler').spinDict.items():
i = i + 1
hlayout = QtWidgets.QHBoxLayout()
nameLabel = QtWidgets.QLabel(key)
valLabel = QtWidgets.QLabel(str(spinBox.get_value()))
hlayout.addWidget(nameLabel)
hlayout.addWidget(valLabel)
vlayout.addLayout(hlayout)
if i == 10:
break
widg = QtWidgets.QWidget()
widg.setLayout(vlayout)
return widg
This diff is collapsed.
from global_widgets.global_spinbox import labelledSpin
from widget_library.ok_cancel_buttons_widget import OkButtonWidget, CancelButtonWidget, OkSendButtonWidget
from global_widgets.global_send_popup import SetConfirmPopup
from PySide2 import QtWidgets, QtGui, QtCore
from handler_library.handler import Handler
from handler_library.handler import PayloadHandler
import logging
import json
class ExpertHandler(Handler, QtCore.QObject): # chose QWidget over QDialog family because easier to modify
class ExpertHandler(PayloadHandler): # chose QWidget over QDialog family because easier to modify
UpdateExpert = QtCore.Signal(dict)
OpenPopup = QtCore.Signal(PayloadHandler,list)
def __init__(self, NativeUI, confirmPopup, *args, **kwargs):
super().__init__(*args, **kwargs)
QtCore.QObject.__init__(self)
def __init__(self, NativeUI, *args, **kwargs):
super().__init__(['READBACK'],*args, **kwargs)
self.NativeUI = NativeUI
self.spinDict = {}
self.buttonDict = {}
self.manuallyUpdated = False
self.popup = confirmPopup
self.popup.okButton.pressed.connect(lambda: self.commandSent()) # not sure why, but lambda is necessary here
self.commandList = []
with open("NativeUI/configs/expert_config.json") as json_file:
controlDict = json.load(json_file)
self.relevantKeys = [list[2] for key in controlDict for list in controlDict[key]]
def add_widget(self, widget, key: str):
if isinstance(widget, labelledSpin):
......@@ -29,29 +31,16 @@ class ExpertHandler(Handler, QtCore.QObject): # chose QWidget over QDialog fami
self.buttonDict[key] = widget
def active_payload(self) -> int:
def active_payload(self, *args) -> int:
readback_data = self.get_db()
outdict = {}
self.UpdateExpert.emit(readback_data)
for key in self.relevantKeys:
try:
outdict[key] = readback_data[key]
except KeyError:
logging.debug("Invalid key %s in measurement database", key)
self.UpdateExpert.emit(outdict)
return 0
# for key in [
# "respiratory_rate",
# "fiO2_percent",
# "inhale_trigger_threshold",
# "exhale_trigger_threshold",
# "volume",
# "inspiratory_pressure",
# "inhale_time",
# "ie_ratio",
# ]:
# try:
# outdict[key] = readback_data[key]
# except KeyError:
# logging.debug("Invalid key %s in measurement database", key)
# self.UpdateExpert.emit(readback_dict)
# return 0
def handle_okbutton_click(self, key):
message, command = [], []
......@@ -66,18 +55,28 @@ class ExpertHandler(Handler, QtCore.QObject): # chose QWidget over QDialog fami
setVal,
]
)
self.popup.clearPopup()
self.popup.populatePopup(message, command)
self.commandList = command
if 'send' in key:
self.popup.okButton.click()
self.sendCommands()
else:
self.popup.show()
self.OpenPopup.emit(self,message)
def sendCommands(self):
if self.commandList == []:
a=1
else:
for command in self.commandList:
self.NativeUI.q_send_cmd(*command)
self.commandSent()
return 0
def commandSent(self):
self.commandList = []
for widget in self.spinDict:
self.spinDict[widget].manuallyUpdated = False
self.refresh_button_colour()
def handle_manual_change(self, changed_spin_key):
self.refresh_button_colour()
......
......@@ -46,7 +46,6 @@ class styledButton(QtWidgets.QPushButton):
# self.setFixedSize(QtCore.QSize(150, 50))
def setColour(self, option):
#print('setting colour again again')
self.setEnabled(bool(float(option)))
self.setProperty("bgColour", str(option))
self.style().polish(self)
......
......@@ -21,7 +21,7 @@ from global_widgets.global_typeval_popup import TypeValuePopup
# from global_widgets.global_ok_cancel_buttons import okButton, cancelButton
from widget_library.ok_cancel_buttons_widget import OkButtonWidget, CancelButtonWidget
from global_widgets.global_spinbox import signallingSpinBox
from global_widgets.global_send_popup import SetConfirmPopup
#from global_widgets.global_send_popup import SetConfirmPopup
class SpinButton(QtWidgets.QFrame):
......@@ -34,7 +34,7 @@ class SpinButton(QtWidgets.QFrame):
# self.setStyleSheet("background-color:blue;")
self.currentVal = 0
self.cmd_type = settings[3]
self.val_code = settings[2]
self.cmd_code = settings[2]
self.NativeUI = NativeUI
self.layout = QtWidgets.QVBoxLayout()
......@@ -107,20 +107,23 @@ class SpinButton(QtWidgets.QFrame):
self.setFixedWidth(300)
#self.setStyleSheet("border:2px solid white; border-radius:4px; padding:0px;")
def update_targets_value(self):
newVal = self.NativeUI.get_db("targets")
if (newVal == {}) or (self.val_code == ""):
def update_value(self, db):
newVal = db
if (newVal == {}) or (self.cmd_code == ""):
a = 1 # do nothing
else:
if not self.manuallyUpdated:
self.simpleSpin.setValue(newVal[self.val_code])
self.simpleSpin.setValue(newVal[self.cmd_code])
self.setTextColour(1)
else:
if int(self.simpleSpin.value()) == int(newVal[self.val_code]):
if int(self.simpleSpin.value()) == int(newVal[self.cmd_code]):
self.manuallyUpdated = False
self.setTextColour(1)
def get_value(self):
return self.simpleSpin.value()
def manualChanged(self):
"""Called when user manually makes a change. Stops value from updating and changes colour"""
self.manuallyUpdated = True
......@@ -185,6 +188,7 @@ class SpinButtonsWidget(QtWidgets.QWidget):
self.spinStack.addWidget(self.spinDict[settings[0]])
else:
self.layout.addWidget(self.spinDict[settings[0]])
self.spinStack.setCurrentIndex(1)
self.layout.addWidget(self.spinStack)
......@@ -198,10 +202,10 @@ class SpinButtonsWidget(QtWidgets.QWidget):
self.setLayout(self.layout)
self.timer = QtCore.QTimer()
self.timer.setInterval(160)
self.timer.timeout.connect(self.update_targets)
self.timer.start()
#self.timer = QtCore.QTimer()
#self.timer.setInterval(160)
#self.timer.timeout.connect(self.update_targets)
#self.timer.start()
def setStackWidget(self, label):
self.spinStack.setCurrentWidget(self.spinDict[label])
......@@ -223,20 +227,6 @@ class SpinButtonsWidget(QtWidgets.QWidget):
self.okButton.setColour(str(int(self.manuallyUpdated)))
self.cancelButton.setColour((str(int(self.manuallyUpdated))))
# targets = self.NativeUI.get_db("targets")
# if targets == {}:
# return
# if targets["mode"] == "CURRENT":
# for spin, code in zip(self.__spins, self.__codes):
# if spin.simpleSpin.value() != float(targets[code]):
# if spin.liveUpdating:
# spin.simpleSpin.setValue(float(targets[code]))
# spin.setTextColour("2")
# else:
# spin.setTextColour("0")
# else:
# spin.setTextColour("2")
def ok_button_pressed(self):
"""Respond to ok button pressed by changing text colour and liveUpdating to True"""
message, command = [], []
......@@ -247,13 +237,13 @@ class SpinButtonsWidget(QtWidgets.QWidget):
command.append(
[
self.spinDict[widget].cmd_type,
self.spinDict[widget].val_code,
self.spinDict[widget].cmd_code,
setVal,
]
)
self.popup = SetConfirmPopup(self, self.NativeUI, message, command)
self.popup.okButton.pressed.connect(self.commandSent)
self.popup.show()
#self.popup = SetConfirmPopup(self, self.NativeUI, message, command)
#self.popup.okButton.pressed.connect(self.commandSent)
#self.popup.show()
return 0
......@@ -271,11 +261,4 @@ class SpinButtonsWidget(QtWidgets.QWidget):
self.spinDict[spin].setTextColour("1")
self.refresh_button_colour()
# targets = self.NativeUI.get_targets_db()
# if targets == {}:
# return
# for spin, label in zip(self.__spins, self.__labels):
# if spin.doubleSpin.value() != float(targets[label]):
# spin.setTextColour("0")
# else:
# spin.setTextColour("2")
......@@ -9,13 +9,18 @@ from PySide2 import QtWidgets, QtGui, QtCore
class StartupHandler(QtWidgets.QWidget): # chose QWidget over QDialog family because easier to modify
def __init__(self, NativeUI, confirmPopup, *args, **kwargs):
UpdateModes = QtCore.Signal(dict)
OpenPopup = QtCore.Signal(list)
settingToggle = QtCore.Signal(str)
def __init__(self, NativeUI, *args, **kwargs):
super().__init__(*args, **kwargs)
self.NativeUI = NativeUI
self.buttonDict = {}
self.spinDict = {}
self.calibDict = {}
self.radioDict = {}
self.popup = confirmPopup
self.modeRadioDict = {}
self.settingsRadioDict = {}
def add_widget(self, widget, key: str):
if isinstance(widget, labelledSpin):
......@@ -26,13 +31,25 @@ class StartupHandler(QtWidgets.QWidget): # chose QWidget over QDialog family be
if isinstance(widget, OkButtonWidget) or isinstance(widget, CancelButtonWidget) or isinstance(widget,OkSendButtonWidget):
self.buttonDict[key] = widget
if isinstance(widget, QRadioButton):
self.radioDict[key] = widget
if widget.text() in self.NativeUI.modeList:
self.modeRadioDict[key] = widget
else:
self.settingsRadioDict[key] = widget
def handle_radiobutton(self, checked, radio):
def handle_mode_radiobutton(self, checked, radio):
if checked:
print(radio.text())
self.NativeUI.currentMode = radio.text()
def handle_settings_radiobutton(self, radioButtonState, radioKey):
"""TODO Docstring"""
mode = self.get_mode(radioKey)
spinKey= radioKey.replace('radio', 'spin')
spinBox = self.spinDict[spinKey]
spinBox.setEnabled(radioButtonState)
if mode == self.NativeUI.currentMode:
self.settingToggle.emit(spinBox.label)
def handle_calibrationPress(self, calibrationWidget):
calibrationWidget.progBar.setValue(100)
calibrationWidget.lineEdit.setText('completed')
......@@ -56,18 +73,13 @@ class StartupHandler(QtWidgets.QWidget): # chose QWidget over QDialog family be
setVal,
]
)
self.popup.clearPopup()
self.popup.populatePopup(message, command)
#self.popup.okButton.pressed.connect(lambda i=mode: self.commandSent(i))
#self.popup.okButton.pressed.connect(self.modeSwitched.emit())
self.popup.okButton.click()
for com in command:
self.NativeUI.q_send_cmd(*com)
self.NativeUI.q_send_cmd(
"SET_MODE", self.NativeUI.currentMode.replace("/", "_").replace("-", "_")
)
def handle_nextbutton(self, stack):
currentIndex = stack.currentIndex()
nextIndex = currentIndex + 1
......@@ -78,18 +90,18 @@ class StartupHandler(QtWidgets.QWidget): # chose QWidget over QDialog family be
else:
self.buttonDict['nextButton'].setColour(1)
self.buttonDict['backButton'].setColour(1)
self.buttonDict['nextButton'].style().polish(self.buttonDict['nextButton'])
def handle_backbutton(self, stack):
currentIndex = stack.currentIndex()
nextIndex = currentIndex - 1
#totalLength = stack.count()
stack.setCurrentIndex(nextIndex)
if nextIndex == 0:
a = 1 # send
self.buttonDict['backButton'].setColour(0)
else:
self.buttonDict['backButton'].setColour(1)
self.buttonDict['nextButton'].setColour(1)
self.buttonDict['backButton'].style().polish(self.buttonDict['backButton'])
\ No newline at end of file
def get_mode(self, key: str):
for mode in self.NativeUI.modeList:
if mode in key:
return mode
......@@ -29,7 +29,8 @@ import logging
import binascii
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s (%(filename)s line %(lineno)d: %(module)s.%(funcName)s)",
)
# VERSIONING
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment