jens
93b5667a64
Embeds a dark-themed matplotlib canvas (3 stacked strip charts: theta ist/soll, heatrate ist/soll, heater power_set/power_eff) in a new Plot tab, sampled once per second from the latest websocket-received values. HeaterTask now also emits a PowerSet message (the pre-discretization target power) alongside the existing Power (effective) message, since the GUI had no way to see the set-point side of figure 1's bottom plot otherwise.
315 lines
11 KiB
Python
Executable File
315 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from main_window import Ui_MainWindow
|
|
from PyQt5 import QtWidgets, QtGui, QtCore
|
|
import json
|
|
import sys
|
|
import time
|
|
import matplotlib
|
|
matplotlib.use("Qt5Agg")
|
|
import matplotlib.style
|
|
matplotlib.style.use("dark_background")
|
|
from matplotlib.backends.backend_qt5agg import FigureCanvasQTAgg
|
|
from matplotlib.figure import Figure
|
|
from ws.client.ws_client import WsClient
|
|
from ws.message import MessageDispatcherSync
|
|
from queue import Queue
|
|
import asyncio
|
|
|
|
|
|
class RealtimePlot(FigureCanvasQTAgg):
|
|
"""Live theta/heatrate/heater-power strip charts, mirroring figure 1 of
|
|
scripts/demos/sud/demo_sud.py but fed from the websocket connection
|
|
instead of an offline simulation run."""
|
|
|
|
def __init__(self):
|
|
figure = Figure()
|
|
FigureCanvasQTAgg.__init__(self, figure)
|
|
|
|
self.ax_temp, self.ax_rate, self.ax_power = figure.subplots(3, 1, sharex=True)
|
|
|
|
self.line_temp_ist, = self.ax_temp.plot([], [], '-c', linewidth=1, label="theta_ist")
|
|
self.line_temp_soll, = self.ax_temp.plot([], [], '-m', linewidth=1, label="theta_soll")
|
|
self.ax_temp.set_ylabel("Temp [°C]")
|
|
self.ax_temp.legend(loc='upper left', fontsize='small')
|
|
self.ax_temp.grid(True)
|
|
|
|
self.line_rate_ist, = self.ax_rate.plot([], [], '-c', linewidth=1, label="heatrate_ist")
|
|
self.line_rate_soll, = self.ax_rate.plot([], [], '-m', linewidth=1, label="heatrate_soll")
|
|
self.ax_rate.set_ylabel("Rate [K/min]")
|
|
self.ax_rate.legend(loc='upper left', fontsize='small')
|
|
self.ax_rate.grid(True)
|
|
|
|
self.line_power_set, = self.ax_power.plot([], [], '-m', linewidth=1, label="power_set")
|
|
self.line_power_eff, = self.ax_power.plot([], [], '-c', linewidth=1, label="power_eff")
|
|
self.ax_power.set_ylabel("Power [W]")
|
|
self.ax_power.set_xlabel("t [s]")
|
|
self.ax_power.legend(loc='upper left', fontsize='small')
|
|
self.ax_power.grid(True)
|
|
|
|
figure.tight_layout()
|
|
self.reset()
|
|
|
|
def reset(self):
|
|
self.t0 = time.monotonic()
|
|
self.t = []
|
|
self.temp_ist = []
|
|
self.temp_soll = []
|
|
self.rate_ist = []
|
|
self.rate_soll = []
|
|
self.power_set = []
|
|
self.power_eff = []
|
|
|
|
def sample(self, temp_ist, temp_soll, rate_ist, rate_soll, power_set, power_eff):
|
|
self.t.append(time.monotonic() - self.t0)
|
|
self.temp_ist.append(temp_ist)
|
|
self.temp_soll.append(temp_soll)
|
|
self.rate_ist.append(rate_ist)
|
|
self.rate_soll.append(rate_soll)
|
|
self.power_set.append(power_set)
|
|
self.power_eff.append(power_eff)
|
|
|
|
self.line_temp_ist.set_data(self.t, self.temp_ist)
|
|
self.line_temp_soll.set_data(self.t, self.temp_soll)
|
|
self.line_rate_ist.set_data(self.t, self.rate_ist)
|
|
self.line_rate_soll.set_data(self.t, self.rate_soll)
|
|
self.line_power_set.set_data(self.t, self.power_set)
|
|
self.line_power_eff.set_data(self.t, self.power_eff)
|
|
|
|
for ax in (self.ax_temp, self.ax_rate, self.ax_power):
|
|
ax.relim()
|
|
ax.autoscale_view()
|
|
|
|
self.draw_idle()
|
|
|
|
|
|
class Window(QtWidgets.QMainWindow, Ui_MainWindow):
|
|
def __init__(self):
|
|
QtWidgets.QMainWindow.__init__(self)
|
|
self.setWindowIcon(QtGui.QIcon('res/beer.png'))
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
self.msg_dispatch = MessageDispatcherSync(auto_subscribe=True)
|
|
self.ws_client = WsClient(listener=self.msg_dispatch, loop=loop)
|
|
self.msg_pot = self.msg_dispatch.msgio_get('Pot')
|
|
self.msg_sensor = self.msg_dispatch.msgio_get('Sensor')
|
|
self.msg_heater = self.msg_dispatch.msgio_get('Heater')
|
|
self.msg_stirrer = self.msg_dispatch.msgio_get('Stirrer')
|
|
self.msg_tempctrl = self.msg_dispatch.msgio_get('TempCtrl')
|
|
self.msg_sud = self.msg_dispatch.msgio_get('Sud')
|
|
|
|
self.msg_pot.set_recv_handler(self.on_pot_changed)
|
|
self.msg_sensor.set_recv_handler(self.on_sensor_changed)
|
|
self.msg_heater.set_recv_handler(self.on_heater_changed)
|
|
self.msg_stirrer.set_recv_handler(self.on_stirrer_changed)
|
|
self.msg_tempctrl.set_recv_handler(self.on_tempctrl_changed)
|
|
self.msg_sud.set_recv_handler(self.on_sud_changed)
|
|
|
|
self.setupUi(self)
|
|
self.actionStart.triggered.connect(self.connect)
|
|
self.actionStop.triggered.connect(self.disconnect)
|
|
self.actionSudDownload.triggered.connect(self.on_action_sud_download)
|
|
self.actionSudUpload.triggered.connect(self.on_action_sud_upload)
|
|
self.Slider_pwr_soll.valueChanged.connect(self.on_slider_pwr_soll_changed)
|
|
self.Slider_temp_soll.valueChanged.connect(self.on_slider_temp_soll_changed)
|
|
self.Slider_speed_soll.valueChanged.connect(self.on_slider_stirrer_speed_soll_changed)
|
|
|
|
self.doubleSpinBox_heatrate_soll.valueChanged.connect(self.on_heatrate_soll_changed)
|
|
|
|
self.checkbox_heater_activate.stateChanged.connect(self.on_checkbox_changed)
|
|
self.checkbox_stirrer_activate.stateChanged.connect(self.on_checkbox_stirrer_activate_changed)
|
|
|
|
self.send_data = Queue()
|
|
self.slider_pwr_initial_update = True
|
|
self.slider_temp_soll_initial_update = True
|
|
self.slider_speed_initial_update = True
|
|
self.checkbox_heater_activate_initial_update = True
|
|
self.checkbox_stirrer_activate_initial_update = True
|
|
self.heatrate_soll_initial_update = True
|
|
self.sud_download_path = None
|
|
|
|
# Latest values sampled by the plot timer - written from the
|
|
# websocket recv thread, read from the GUI thread; plain floats so
|
|
# no locking, consistent with how the rest of the GUI is wired.
|
|
self.plot_temp_ist = 0.0
|
|
self.plot_temp_soll = 0.0
|
|
self.plot_rate_ist = 0.0
|
|
self.plot_rate_soll = 0.0
|
|
self.plot_power_set = 0.0
|
|
self.plot_power_eff = 0.0
|
|
|
|
self.plot = RealtimePlot()
|
|
plot_layout = QtWidgets.QVBoxLayout(self.widget_plot)
|
|
plot_layout.setContentsMargins(0, 0, 0, 0)
|
|
plot_layout.addWidget(self.plot)
|
|
|
|
self.plot_timer = QtCore.QTimer(self)
|
|
self.plot_timer.timeout.connect(self.on_plot_timer)
|
|
self.plot_timer.start(1000)
|
|
|
|
def connect(self):
|
|
self.slider_pwr_initial_update = True
|
|
self.slider_temp_soll_initial_update = True
|
|
self.slider_speed_initial_update = True
|
|
self.checkbox_heater_activate_initial_update = True
|
|
self.checkbox_stirrer_activate_initial_update = True
|
|
self.heatrate_soll_initial_update = True
|
|
self.plot.reset()
|
|
self.ws_client.connect(uri=self.plainTextUri.toPlainText())
|
|
|
|
def on_plot_timer(self):
|
|
self.plot.sample(
|
|
self.plot_temp_ist, self.plot_temp_soll,
|
|
self.plot_rate_ist, self.plot_rate_soll,
|
|
self.plot_power_set, self.plot_power_eff)
|
|
|
|
def disconnect(self):
|
|
self.ws_client.disconnect()
|
|
|
|
def on_slider_temp_soll_changed(self, value):
|
|
print("on_slider_temp_soll_changed {}".format(value))
|
|
self.msg_tempctrl.send({'Soll': {'Temp': value}})
|
|
|
|
def on_heatrate_soll_changed(self, value):
|
|
value = round(value, 1)
|
|
print("on_textbox_heatrate_soll_changed {}".format(value))
|
|
self.msg_tempctrl.send({'Soll': {'Rate': value}})
|
|
|
|
def on_slider_pwr_soll_changed(self, value):
|
|
print("on_slider_pwr_soll_changed {}".format(value))
|
|
self.msg_heater.send({'Power': value})
|
|
|
|
def on_checkbox_changed(self, value):
|
|
print("on_checkbox_changed {}".format(value))
|
|
self.msg_heater.send({'Activate': int(value == 2)})
|
|
|
|
def on_slider_stirrer_speed_soll_changed(self, value):
|
|
print("on_slider_stirrer_speed_soll_changed {}".format(value))
|
|
self.msg_stirrer.send({'Speed': value})
|
|
|
|
def on_checkbox_stirrer_activate_changed(self, value):
|
|
print("on_checkbox_stirrer_activate_changed {}".format(value))
|
|
self.msg_stirrer.send({'Activate': int(value == 2)})
|
|
|
|
def on_action_sud_download(self):
|
|
# Pick the destination up front, on the GUI thread - the actual
|
|
# write happens in on_sud_changed(), which runs off a worker thread
|
|
# and must not touch Qt (e.g. open a file dialog) itself.
|
|
path, _ = QtWidgets.QFileDialog.getSaveFileName(self, "Download Sud", filter="JSON files (*.json)")
|
|
if not path:
|
|
return
|
|
self.sud_download_path = path
|
|
self.msg_sud.send({'Download': True})
|
|
|
|
def on_action_sud_upload(self):
|
|
path, _ = QtWidgets.QFileDialog.getOpenFileName(self, "Upload Sud", filter="JSON files (*.json)")
|
|
if not path:
|
|
return
|
|
with open(path) as f:
|
|
data = json.load(f)
|
|
self.msg_sud.send({'Upload': data})
|
|
|
|
def on_pot_changed(self, msg):
|
|
print("on_pot_changed {}".format(msg))
|
|
for key in msg:
|
|
if "Power" in key:
|
|
self.lcdNumber_power_pot.display(msg['Power'])
|
|
|
|
def on_sensor_changed(self, msg):
|
|
print("on_sensor_changed {}".format(msg))
|
|
|
|
def on_tempctrl_changed(self, msg):
|
|
print("on_tempctrl_changed {}".format(msg))
|
|
for key in msg:
|
|
if 'State' in key:
|
|
self.label_state.setText(msg['State'])
|
|
elif "Soll" in key:
|
|
submsg = msg['Soll']
|
|
if 'Temp' in submsg:
|
|
self.plot_temp_soll = submsg['Temp']
|
|
self.lcdNumber_temp_soll.display(str(submsg['Temp']))
|
|
if self.slider_temp_soll_initial_update:
|
|
self.Slider_temp_soll.setValue(int(round(submsg['Temp'])))
|
|
self.slider_temp_soll_initial_update = False
|
|
if 'Rate' in submsg:
|
|
subsubmsg = submsg['Rate']
|
|
for subkey in subsubmsg:
|
|
if "Current" in subkey:
|
|
self.plot_rate_soll = subsubmsg['Current']
|
|
self.lcdNumber_heatrate_soll.display(str(subsubmsg['Current']))
|
|
if "Set" in subkey:
|
|
if self.heatrate_soll_initial_update:
|
|
self.heatrate_soll_initial_update = False
|
|
self.doubleSpinBox_heatrate_soll.setValue(subsubmsg['Set'])
|
|
if "Ist" in key:
|
|
submsg = msg['Ist']
|
|
if 'Temp' in submsg:
|
|
self.plot_temp_ist = submsg['Temp']
|
|
self.lcdNumber_temp_ist.display(str(submsg['Temp']))
|
|
if 'Rate' in submsg:
|
|
self.plot_rate_ist = submsg['Rate']
|
|
self.lcdNumber_heatrate_ist.display(str(submsg['Rate']))
|
|
|
|
def on_heater_changed(self, msg):
|
|
print("on_heater_changed {}".format(msg))
|
|
for key in msg:
|
|
if "Activate" in key:
|
|
if self.checkbox_heater_activate_initial_update:
|
|
self.checkbox_heater_activate.setCheckState(2 if msg['Activate'] == 1 else 0)
|
|
self.checkbox_heater_activate_initial_update = False
|
|
elif "PowerSet" in key:
|
|
self.plot_power_set = msg['PowerSet']
|
|
elif "Power" in key:
|
|
self.plot_power_eff = msg['Power']
|
|
self.lcdNumber_power_heater.display(msg['Power'])
|
|
if self.slider_pwr_initial_update:
|
|
self.Slider_pwr_soll.setValue(int(round(msg['Power'])))
|
|
self.slider_pwr_initial_update = False
|
|
elif "Capabilities" in key:
|
|
submsg = msg['Capabilities']
|
|
if "Power" in submsg:
|
|
submsg = submsg['Power']
|
|
self.Slider_pwr_soll.setMinimum(int(submsg['Min']))
|
|
self.Slider_pwr_soll.setMaximum(int(submsg['Max']))
|
|
|
|
def on_stirrer_changed(self, msg):
|
|
print("on_stirrer_changed {}".format(msg))
|
|
for key in msg:
|
|
if "Activate" in key:
|
|
if self.checkbox_stirrer_activate_initial_update:
|
|
self.checkbox_stirrer_activate.setCheckState(2 if msg['Activate'] == 1 else 0)
|
|
self.checkbox_stirrer_activate_initial_update = False
|
|
elif "Speed" in key:
|
|
if self.slider_speed_initial_update:
|
|
self.Slider_speed_soll.setValue(int(round(msg['Speed'])))
|
|
self.slider_speed_initial_update = False
|
|
elif "Capabilities" in key:
|
|
submsg = msg['Capabilities']
|
|
if "Power" in submsg:
|
|
submsg = submsg['Power']
|
|
self.Slider_speed_soll.setMinimum(int(submsg['Min']))
|
|
self.Slider_speed_soll.setMaximum(int(submsg['Max']))
|
|
|
|
def on_sud_changed(self, msg):
|
|
print("on_sud_changed {}".format(msg))
|
|
for key in msg:
|
|
if "Json" in key and self.sud_download_path:
|
|
with open(self.sud_download_path, 'w') as f:
|
|
json.dump(msg['Json'], f, indent='\t')
|
|
self.sud_download_path = None
|
|
elif "Name" in key:
|
|
self.statusBar().showMessage("Sud schedule updated: {}".format(msg['Name']), 5000)
|
|
|
|
def closeEvent(self, event):
|
|
print("Exitting gracefully!")
|
|
self.ws_client.disconnect()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app = QtWidgets.QApplication(sys.argv)
|
|
win = Window()
|
|
win.show()
|
|
sys.exit(app.exec_())
|
|
|
|
print ("End of program.")
|