thermostat/pytec/tec_qt.py

184 lines
5.8 KiB
Python
Raw Normal View History

from PyQt6 import QtWidgets, uic
from PyQt6.QtCore import QThread, QThreadPool, pyqtSignal, QRunnable, QObject, QSignalBlocker, pyqtSlot
from pyqtgraph import PlotWidget
from pyqtgraph.parametertree import Parameter, ParameterTree, ParameterItem, registerParameterType
import pyqtgraph as pg
import sys
import argparse
import logging
from pytec.client import Client
# pyuic6 -x tec_qt.ui -o ui_tec_qt.py
from ui_tec_qt import Ui_MainWindow
tec_client: Client = None
# ui = None
ui: Ui_MainWindow = None
thread_pool = QThreadPool.globalInstance()
connection_watcher = None
def get_argparser():
parser = argparse.ArgumentParser(description="ARTIQ master")
parser.add_argument("--connect", default=None, action="store_true",
help="Automatically connect to the specified Thermostat in IP:port format")
parser.add_argument('IP', metavar="ip", default=None, nargs='?')
parser.add_argument('PORT', metavar="port", default=None, nargs='?')
parser.add_argument("-l", "--log", dest="logLevel", choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
help="Set the logging level")
return parser
class WatchConnectTask(QThread):
connected = pyqtSignal(bool)
hw_rev = pyqtSignal(dict)
connecting = pyqtSignal()
fan_update = pyqtSignal(object)
def __init__(self, ip, port, parent):
self.ip = ip
self.port = port
super().__init__(parent)
def run(self):
global tec_client
try:
if tec_client:
tec_client.disconnect()
tec_client = None
self.connected.emit(False)
else:
self.connecting.emit()
tec_client = Client(host=self.ip, port=self.port, timeout=30)
self.connected.emit(True)
self.hw_rev.emit(tec_client.hw_rev())
self.fan_update.emit(tec_client.fan())
except Exception as e:
logging.error(f"Failed communicating to the {self.ip}:{self.port}: {e}")
self.connected.emit(False)
@pyqtSlot()
def client_disconnected(self):
global tec_client
if tec_client:
tec_client.disconnect()
tec_client = None
self.connected.emit(False)
class ClientTask(QRunnable):
def __init__(self, func, *args, **kwargs):
self.func = func
self.args = args
self.kwargs = kwargs
super().__init__()
def run(self) -> None:
try:
self.func(*self.args, **self.kwargs)
except (TimeoutError, OSError):
logging.warning("Client connection error, disconnecting", exc_info=True)
if connection_watcher:
connection_watcher.client_disconnected()
def connected(result):
ui.graph_group.setEnabled(result)
ui.hw_rev_lbl.setEnabled(result)
ui.fan_group.setEnabled(result)
ui.report_group.setEnabled(result)
ui.ip_set_line.setEnabled(not result)
ui.port_set_spin.setEnabled(not result)
ui.status_lbl.setText("Connected" if result else "Disconnected")
ui.connect_btn.setText("Disconnect" if result else "Connect")
if not result:
ui.hw_rev_lbl.setText("Thermostat vX.Y")
ui.fan_group.setStyleSheet("")
def hw_rev(hw_rev_d: dict):
logging.debug(hw_rev_d)
ui.hw_rev_lbl.setText(f"Thermostat v{hw_rev_d['rev']['major']}.{hw_rev_d['rev']['major']}")
ui.fan_group.setEnabled(hw_rev_d["settings"]["fan_available"])
if hw_rev_d["settings"]["fan_pwm_recommended"]:
ui.fan_group.setStyleSheet("")
ui.fan_group.setToolTip("")
else:
ui.fan_group.setStyleSheet("background-color: yellow")
ui.fan_group.setToolTip("Changing the fan settings of not recommended")
def fan_update(fan_settings):
logging.debug(fan_settings)
if fan_settings is None:
return
with QSignalBlocker(ui.fan_power_slider) as _:
ui.fan_power_slider.setValue(fan_settings["fan_pwm"])
ui.fan_power_slider.setEnabled(not fan_settings["auto_mode"])
with QSignalBlocker(ui.fan_auto_box) as _:
ui.fan_auto_box.setChecked(fan_settings["auto_mode"])
def fan_set():
global tec_client
if tec_client is None or ui.fan_auto_box.isChecked():
return
thread_pool.start(ClientTask(lambda: tec_client.set_param("fan", ui.fan_power_slider.value())))
def fan_auto_set(enabled):
global tec_client
if tec_client is None:
return
ui.fan_power_slider.setEnabled(not enabled)
if enabled:
thread_pool.start(ClientTask(lambda: tec_client.set_param("fan", "auto")))
else:
thread_pool.start(ClientTask(lambda: tec_client.set_param("fan", ui.fan_power_slider.value())))
def connect():
global connection_watcher
connection_watcher = WatchConnectTask(ui.ip_set_line.text(), ui.port_set_spin.value(), ui.main_widget)
connection_watcher.connected.connect(connected)
connection_watcher.connecting.connect(lambda: ui.status_lbl.setText("Connecting..."))
connection_watcher.hw_rev.connect(hw_rev)
connection_watcher.fan_update.connect(fan_update)
connection_watcher.start()
def main():
global ui
args = get_argparser().parse_args()
if args.logLevel:
logging.basicConfig(level=getattr(logging, args.logLevel))
app = QtWidgets.QApplication(sys.argv)
main_window = QtWidgets.QMainWindow()
#ui = Ui_MainWindow()
#ui.setupUi(main_window)
ui = uic.loadUi('tec_qt.ui', main_window)
ui.connect_btn.clicked.connect(lambda _checked: connect())
ui.fan_power_slider.valueChanged.connect(fan_set)
ui.fan_auto_box.stateChanged.connect(fan_auto_set)
if args.connect:
if args.IP:
ui.ip_set_line.setText(args.IP)
if args.PORT:
ui.port_set_spin.setValue(int(args.PORT))
ui.connect_btn.click()
main_window.show()
sys.exit(app.exec())
if __name__ == '__main__':
main()