esavkin
/
thermostat
forked from M-Labs/thermostat
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Make Ui_MainWindow a superclass of our main window 

Gets rid of the global ui.
This commit is contained in:
atse 2023-07-01 23:46:40 +08:00
parent 299ef7dcc3
commit c6ca2b3490
1 changed files with 101 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

150
pytec/tec_qt.py
View File

@ -1,4 +1,4 @@
from PyQt6 import QtWidgets, uic from PyQt6 import QtWidgets
from PyQt6.QtCore import pyqtSignal, QObject, QSignalBlocker, pyqtSlot from PyQt6.QtCore import pyqtSignal, QObject, QSignalBlocker, pyqtSlot
from pyqtgraph import PlotWidget from pyqtgraph import PlotWidget
from pyqtgraph.parametertree import Parameter, ParameterTree, ParameterItem, registerParameterType from pyqtgraph.parametertree import Parameter, ParameterTree, ParameterItem, registerParameterType
@ -16,11 +16,7 @@ from ui_tec_qt import Ui_MainWindow
tec_client: Client = None tec_client: Client = None
# ui = None
ui: Ui_MainWindow = None
client_watcher = None client_watcher = None
client_watcher_task = None
def get_argparser(): def get_argparser():
@ -62,105 +58,118 @@ class ClientWatcher(QObject):
self.running = False self.running = False
@pyqtSlot() @pyqtSlot()
def set_update_s(self): def set_update_s(self, update_s):
self.update_s = ui.report_refresh_spin.value() self.update_s = update_s
def on_connection_changed(result): class MainWindow(QtWidgets.QMainWindow, Ui_MainWindow):
global client_watcher, client_watcher_task def __init__(self, args):
ui.graph_group.setEnabled(result) super().__init__()
ui.hw_rev_lbl.setEnabled(result)
ui.fan_group.setEnabled(result)
ui.report_group.setEnabled(result)
ui.ip_set_line.setEnabled(not result) self.setupUi(self)
ui.port_set_spin.setEnabled(not result)
ui.status_lbl.setText("Connected" if result else "Disconnected") self.connect_btn.clicked.connect(self.connect)
ui.connect_btn.setText("Disconnect" if result else "Connect") self.fan_power_slider.valueChanged.connect(self.fan_set)
self.fan_auto_box.stateChanged.connect(self.fan_auto_set)
self.client_watcher_task = None
if args.connect:
if args.IP:
self.ip_set_line.setText(args.IP)
if args.PORT:
self.port_set_spin.setValue(int(args.PORT))
self.connect_btn.click()
def _on_connection_changed(self, result):
global client_watcher
self.graph_group.setEnabled(result)
self.hw_rev_lbl.setEnabled(result)
self.fan_group.setEnabled(result)
self.report_group.setEnabled(result)
self.ip_set_line.setEnabled(not result)
self.port_set_spin.setEnabled(not result)
self.status_lbl.setText("Connected" if result else "Disconnected")
self.connect_btn.setText("Disconnect" if result else "Connect")
if not result: if not result:
ui.hw_rev_lbl.setText("Thermostat vX.Y") self.hw_rev_lbl.setText("Thermostat vX.Y")
ui.fan_group.setStyleSheet("") self.fan_group.setStyleSheet("")
if client_watcher: if client_watcher:
client_watcher.stop_watching() client_watcher.stop_watching()
client_watcher = None client_watcher = None
client_watcher_task = None self.client_watcher_task = None
def _hw_rev(self, hw_rev_d: dict):
def hw_rev(hw_rev_d: dict):
logging.debug(hw_rev_d) logging.debug(hw_rev_d)
ui.hw_rev_lbl.setText(f"Thermostat v{hw_rev_d['rev']['major']}.{hw_rev_d['rev']['major']}") self.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"]) self.fan_group.setEnabled(hw_rev_d["settings"]["fan_available"])
if hw_rev_d["settings"]["fan_pwm_recommended"]: if hw_rev_d["settings"]["fan_pwm_recommended"]:
ui.fan_group.setStyleSheet("") self.fan_group.setStyleSheet("")
ui.fan_group.setToolTip("") self.fan_group.setToolTip("")
else: else:
ui.fan_group.setStyleSheet("background-color: yellow") self.fan_group.setStyleSheet("background-color: yellow")
ui.fan_group.setToolTip("Changing the fan settings of not recommended") self.fan_group.setToolTip("Changing the fan settings of not recommended")
def fan_update(self, fan_settings):
def fan_update(fan_settings):
logging.debug(fan_settings) logging.debug(fan_settings)
if fan_settings is None: if fan_settings is None:
return return
with QSignalBlocker(ui.fan_power_slider) as _: with QSignalBlocker(self.fan_power_slider) as _:
ui.fan_power_slider.setValue(fan_settings["fan_pwm"]) self.fan_power_slider.setValue(fan_settings["fan_pwm"])
ui.fan_power_slider.setEnabled(not fan_settings["auto_mode"]) self.fan_power_slider.setEnabled(not fan_settings["auto_mode"])
with QSignalBlocker(ui.fan_auto_box) as _: with QSignalBlocker(self.fan_auto_box) as _:
ui.fan_auto_box.setChecked(fan_settings["auto_mode"]) self.fan_auto_box.setChecked(fan_settings["auto_mode"])
@asyncSlot()
@asyncSlot() async def fan_set(self):
async def fan_set(_):
global tec_client global tec_client
if tec_client is None or ui.fan_auto_box.isChecked(): if tec_client is None or self.fan_auto_box.isChecked():
return return
await tec_client.set_param("fan", ui.fan_power_slider.value()) await tec_client.set_param("fan", self.fan_power_slider.value())
@asyncSlot(int)
@asyncSlot() async def fan_auto_set(self, enabled):
async def fan_auto_set(enabled):
global tec_client global tec_client
if tec_client is None: if tec_client is None:
return return
ui.fan_power_slider.setEnabled(not enabled) self.fan_power_slider.setEnabled(not enabled)
if enabled: if enabled:
await tec_client.set_param("fan", "auto") await tec_client.set_param("fan", "auto")
else: else:
await tec_client.set_param("fan", ui.fan_power_slider.value()) await tec_client.set_param("fan", self.fan_power_slider.value())
@asyncSlot()
async def connect(self):
global tec_client, client_watcher
@asyncSlot() ip, port = self.ip_set_line.text(), self.port_set_spin.value()
async def connect(_):
global tec_client, client_watcher, client_watcher_task
ip, port = ui.ip_set_line.text(), ui.port_set_spin.value()
try: try:
if tec_client: if tec_client:
await tec_client.disconnect() await tec_client.disconnect()
tec_client = None tec_client = None
on_connection_changed(False) self._on_connection_changed(False)
else: else:
ui.status_lbl.setText("Connecting...") self.status_lbl.setText("Connecting...")
tec_client = Client() tec_client = Client()
await tec_client.connect(host=ip, port=port, timeout=30) await tec_client.connect(host=ip, port=port, timeout=30)
on_connection_changed(True) self._on_connection_changed(True)
hw_rev(await tec_client.hw_rev()) self._hw_rev(await tec_client.hw_rev())
# fan_update(await tec_client.fan()) # self.fan_update(await tec_client.fan())
if client_watcher is None: if client_watcher is None:
client_watcher = ClientWatcher(ui.main_widget, ui.report_refresh_spin.value()) client_watcher = ClientWatcher(self.main_widget, self.report_refresh_spin.value())
client_watcher.fan_update.connect(fan_update) client_watcher.fan_update.connect(self.fan_update)
ui.report_apply_btn.clicked.connect( self.report_apply_btn.clicked.connect(
lambda: client_watcher.set_update_s(ui.report_refresh_spin.value()) lambda: client_watcher.set_update_s(self.report_refresh_spin.value())
) )
QtWidgets.QApplication.instance().aboutToQuit.connect(client_watcher.stop_watching) QtWidgets.QApplication.instance().aboutToQuit.connect(client_watcher.stop_watching)
client_watcher_task = asyncio.create_task(client_watcher.run()) self.client_watcher_task = asyncio.create_task(client_watcher.run())
except Exception as e: except Exception as e:
logging.error(f"Failed communicating to the {ip}:{port}: {e}") logging.error(f"Failed communicating to the {ip}:{port}: {e}")
on_connection_changed(False) self._on_connection_changed(False)
async def coro_main(): async def coro_main():
global ui
args = get_argparser().parse_args() args = get_argparser().parse_args()
if args.logLevel: if args.logLevel:
logging.basicConfig(level=getattr(logging, args.logLevel)) logging.basicConfig(level=getattr(logging, args.logLevel))
@ -170,22 +179,7 @@ async def coro_main():
app = QtWidgets.QApplication.instance() app = QtWidgets.QApplication.instance()
app.aboutToQuit.connect(app_quit_event.set) app.aboutToQuit.connect(app_quit_event.set)
main_window = QtWidgets.QMainWindow() main_window = MainWindow(args)
ui = Ui_MainWindow()
ui.setupUi(main_window)
# ui = uic.loadUi('tec_qt.ui', main_window)
ui.connect_btn.clicked.connect(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() main_window.show()
await app_quit_event.wait() await app_quit_event.wait()