thermostat/pytec/gui/model/thermostat.py

from pytec.aioclient import Client
from PyQt6.QtCore import pyqtSignal, QObject, pyqtSlot
from qasync import asyncSlot
from gui.model.property import Property, PropertyMeta
import asyncio
import logging


class WrappedClient(QObject, Client):
    connection_error = pyqtSignal()

    async def _read_line(self):
        try:
            return await super()._read_line()
        except (Exception, TimeoutError, asyncio.exceptions.TimeoutError):
            self.connection_error.emit()


class Thermostat(QObject, metaclass=PropertyMeta):
    hw_rev = Property(dict)
    fan = Property(dict)
    thermistor = Property(list)
    pid = Property(list)
    pwm = Property(list)
    postfilter = Property(list)
    interval = Property(list)
    report = Property(list)
    info_box_trigger = pyqtSignal(str, str)

    def __init__(self, parent, client, update_s):
        self._update_s = update_s
        self._client = client
        self._watch_task = None
        self._report_mode_task = None
        self._poll_for_report = True
        super().__init__(parent)

    async def run(self):
        self.task = asyncio.create_task(self.update_params())
        while True:
            if self.task.done():
                if self.task.exception() is not None:
                    try:
                        raise self.task.exception()
                    except (
                        Exception,
                        TimeoutError,
                        asyncio.exceptions.TimeoutError,
                    ):
                        logging.error(
                            "Encountered an error while updating parameter tree.",
                            exc_info=True,
                        )
                _ = self.task.result()
                self.task = asyncio.create_task(self.update_params())
            await asyncio.sleep(self._update_s)

    async def get_hw_rev(self):
        self.hw_rev = await self._client.hw_rev()
        return self.hw_rev

    async def update_params(self):
        self.fan = await self._client.get_fan()
        self.pwm = await self._client.get_pwm()
        if self._poll_for_report:
            self.report = await self._client.report()
            self.interval = [
                self.report[i]["interval"] for i in range(len(self.report))
            ]
        self.pid = await self._client.get_pid()
        self.thermistor = await self._client.get_steinhart_hart()
        self.postfilter = await self._client.get_postfilter()

    def connected(self):
        return self._client.connected

    def connecting(self):
        return self._client.connecting

    def start_watching(self):
        self._watch_task = asyncio.create_task(self.run())

    @asyncSlot()
    async def stop_watching(self):
        if self._watch_task is not None:
            await self.set_report_mode(False)
            self._watch_task.cancel()
            self._watch_task = None
            self.task.cancel()
            self.task = None

    async def set_report_mode(self, enabled: bool):
        self._poll_for_report = not enabled
        if enabled:
            self._report_mode_task = asyncio.create_task(self.report_mode())
        else:
            self._client.stop_report_mode()

    async def report_mode(self):
        async for report in self._client.report_mode():
            self.report_update.emit(report)
            self.interval = [
                self.report[i]["interval"] for i in range(len(self.report))
            ]

    async def disconnect(self):
        await self._client.end_session()

    async def set_ipv4(self, ipv4):
        await self._client.set_param("ipv4", ipv4)

    async def get_ipv4(self):
        return await self._client.ipv4()

    @asyncSlot()
    async def save_cfg(self, ch):
        await self._client.save_config(ch)
        self.info_box_trigger.emit(
            "Config loaded", f"Channel {ch} Config has been loaded from flash."
        )

    @asyncSlot()
    async def load_cfg(self, ch):
        await self._client.load_config(ch)
        self.info_box_trigger.emit(
            "Config loaded", f"Channel {ch} Config has been loaded from flash."
        )

    async def dfu(self):
        await self._client.dfu()

    async def reset(self):
        await self._client.reset()

    @pyqtSlot(float)
    def set_update_s(self, update_s):
        self._update_s = update_s