truetype
b078b850f8
Flask app con gauge Chart.js custom, polling Shelly Cloud API v2, storico SQLite, due widget gauge (produzione/consumo) con scala graduata.
79 lines
2.0 KiB
Python
79 lines
2.0 KiB
Python
"""Background poller that fetches Shelly data periodically."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
from database import Database
|
|
from shelly_client import ShellyClient, ShellyError
|
|
|
|
load_dotenv()
|
|
|
|
logger = logging.getLogger("shellydash.poller")
|
|
|
|
POLL_INTERVAL = int(os.getenv("POLL_INTERVAL", "15"))
|
|
|
|
|
|
class ShellyPoller:
|
|
"""Periodically fetches device status and stores readings."""
|
|
|
|
def __init__(
|
|
self,
|
|
client: ShellyClient,
|
|
db: Database,
|
|
interval: int = POLL_INTERVAL,
|
|
):
|
|
self.client = client
|
|
self.db = db
|
|
self.interval = interval
|
|
self._running = False
|
|
self._thread: threading.Thread | None = None
|
|
self._latest_raw: dict = {}
|
|
self._latest_parsed: dict = {}
|
|
|
|
@property
|
|
def latest_parsed(self) -> dict:
|
|
return self._latest_parsed
|
|
|
|
@property
|
|
def latest_raw(self) -> dict:
|
|
return self._latest_raw
|
|
|
|
def start(self):
|
|
self._running = True
|
|
self._thread = threading.Thread(target=self._loop, daemon=True)
|
|
self._thread.start()
|
|
logger.info("Poller started (interval=%ds)", self.interval)
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
|
|
def poll_once(self) -> bool:
|
|
try:
|
|
status = self.client.get_device_status()
|
|
if not status:
|
|
logger.warning("Empty status from Shelly API")
|
|
return False
|
|
self._latest_raw = status
|
|
parsed = self.client.parse_readings(status)
|
|
self._latest_parsed = parsed
|
|
self.db.insert_reading(parsed)
|
|
return True
|
|
except ShellyError as exc:
|
|
logger.error("Poll error: %s", exc)
|
|
return False
|
|
except Exception as exc:
|
|
logger.exception("Unexpected poll error: %s", exc)
|
|
return False
|
|
|
|
def _loop(self):
|
|
while self._running:
|
|
self.poll_once()
|
|
time.sleep(self.interval)
|