#!/usr/bin/env python3 """Dashboard read-only do boiler Tuya com histórico SQLite e consumo estimado.""" from __future__ import annotations import asyncio import csv import hashlib import hmac import io import json import os import sqlite3 import time import urllib.parse import urllib.request from datetime import date, datetime, time as dt_time, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo from aiohttp import web DEVICE_ID = os.environ["TUYA_DEVICE_ID"] ACCESS_ID = os.environ["TUYA_ACCESS_ID"] ACCESS_KEY = os.environ["TUYA_ACCESS_KEY"] ENDPOINT = os.getenv("TUYA_API_ENDPOINT", "https://openapi.tuyaeu.com") DB_PATH = Path(os.getenv("BOILER_DB_PATH", "/home/admin/boiler-energy/data/boiler.db")) PORT = int(os.getenv("BOILER_PORT", "8788")) POWER_KW = float(os.getenv("BOILER_POWER_KW", "2.3")) UNIT_RATE = float(os.getenv("BOILER_RATE_EUR_KWH", "0.3652")) POLL_SECONDS = int(os.getenv("BOILER_POLL_SECONDS", "300")) TZ = ZoneInfo(os.getenv("BOILER_TIMEZONE", "Europe/Dublin")) class TuyaOpenAPI: """Minimal Tuya REST client; avoids the unused Pulsar/Crypto dependency.""" def __init__(self, endpoint: str, access_id: str, access_key: str) -> None: self.endpoint = endpoint.rstrip("/") self.access_id = access_id self.access_key = access_key self.token = "" self.token_expires_ms = 0 def _request(self, method: str, path: str, params: dict | None = None) -> dict: params = params or {} canonical_query = "&".join(f"{key}={params[key]}" for key in sorted(params)) canonical_path = path + (f"?{canonical_query}" if canonical_query else "") body_hash = hashlib.sha256(b"").hexdigest() string_to_sign = f"{method}\n{body_hash}\n\n{canonical_path}" now_ms = int(time.time() * 1000) message = f"{self.access_id}{self.token}{now_ms}{string_to_sign}" signature = hmac.new( self.access_key.encode(), message.encode(), hashlib.sha256 ).hexdigest().upper() query = urllib.parse.urlencode(sorted(params.items())) url = self.endpoint + path + (f"?{query}" if query else "") request = urllib.request.Request( url, method=method, headers={ "client_id": self.access_id, "sign": signature, "sign_method": "HMAC-SHA256", "access_token": self.token, "t": str(now_ms), "lang": "en", }, ) with urllib.request.urlopen(request, timeout=20) as response: return json.loads(response.read()) def connect(self) -> dict: self.token = "" result = self._request("GET", "/v1.0/token", {"grant_type": 1}) if result.get("success"): payload = result.get("result", {}) self.token = payload.get("access_token", "") self.token_expires_ms = int(time.time() * 1000) + int( payload.get("expire_time", 7200) ) * 1000 return result def get(self, path: str, params: dict | None = None) -> dict: if not self.token or self.token_expires_ms - 60000 < int(time.time() * 1000): connected = self.connect() if not connected.get("success"): return connected return self._request("GET", path, params) def iso_local(ms: int) -> str: return datetime.fromtimestamp(ms / 1000, timezone.utc).astimezone(TZ).isoformat() class Store: def __init__(self) -> None: DB_PATH.parent.mkdir(parents=True, exist_ok=True) self.db = sqlite3.connect(DB_PATH) self.db.row_factory = sqlite3.Row self.db.execute("PRAGMA journal_mode=WAL") self.db.executescript( """ CREATE TABLE IF NOT EXISTS switch_events ( event_time_ms INTEGER NOT NULL, value INTEGER NOT NULL CHECK(value IN (0,1)), event_from TEXT NOT NULL DEFAULT '', received_at TEXT NOT NULL, PRIMARY KEY(event_time_ms, value, event_from) ); CREATE INDEX IF NOT EXISTS idx_switch_events_time ON switch_events(event_time_ms); CREATE TABLE IF NOT EXISTS device_events ( event_time_ms INTEGER NOT NULL, event_id INTEGER NOT NULL, event_from TEXT NOT NULL DEFAULT '', code TEXT NOT NULL DEFAULT '', value TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT '', received_at TEXT NOT NULL, PRIMARY KEY(event_time_ms, event_id, event_from, code, value) ); CREATE INDEX IF NOT EXISTS idx_device_events_time ON device_events(event_time_ms); CREATE INDEX IF NOT EXISTS idx_device_events_type ON device_events(event_id, event_time_ms); CREATE TABLE IF NOT EXISTS sync_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, started_at TEXT NOT NULL, finished_at TEXT, ok INTEGER NOT NULL DEFAULT 0, fetched INTEGER NOT NULL DEFAULT 0, inserted INTEGER NOT NULL DEFAULT 0, error TEXT ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); """ ) self.db.commit() def add_events(self, logs: list[dict]) -> int: before = self.db.total_changes now = datetime.now(timezone.utc).isoformat() for item in logs: self.db.execute( "INSERT OR IGNORE INTO device_events VALUES (?,?,?,?,?,?,?)", ( int(item["event_time"]), int(item.get("event_id", 0)), str(item.get("event_from", "")), str(item.get("code", "")), str(item.get("value", item.get("event_value", ""))), str(item.get("status", "")), now, ), ) if item.get("code") != "switch_1": continue raw = str(item.get("value", "")).lower() if raw not in {"true", "false", "1", "0", "on", "off"}: continue value = 1 if raw in {"true", "1", "on"} else 0 self.db.execute( "INSERT OR IGNORE INTO switch_events VALUES (?,?,?,?)", (int(item["event_time"]), value, str(item.get("event_from", "")), now), ) self.db.commit() return self.db.total_changes - before def normalized_events(self, start_ms: int | None = None, end_ms: int | None = None): # Include the last event before the range to establish initial state. rows: list[sqlite3.Row] = [] if start_ms is not None: previous = self.db.execute( "SELECT * FROM switch_events WHERE event_time_ms < ? " "ORDER BY event_time_ms DESC LIMIT 1", (start_ms,), ).fetchone() if previous: rows.append(previous) query = "SELECT * FROM switch_events WHERE 1=1" args: list[int] = [] if start_ms is not None: query += " AND event_time_ms >= ?" args.append(start_ms) if end_ms is not None: query += " AND event_time_ms < ?" args.append(end_ms) query += " ORDER BY event_time_ms, event_from" rows.extend(self.db.execute(query, args).fetchall()) result: list[dict] = [] for row in rows: event = { "ms": row["event_time_ms"], "on": bool(row["value"]), "source_code": str(row["event_from"]), } # Tuya often emits the same transition twice from cloud/device. if result and event["on"] == result[-1]["on"]: continue result.append(event) return result @staticmethod def source_label(code: str) -> str: return { "1": "dispositivo", "2": "app/cliente", "3": "plataforma/automação", "4": "nuvem", "-1": "desconhecido", }.get(str(code), "desconhecido") @staticmethod def weekday_name(value: date) -> str: return [ "segunda-feira", "terça-feira", "quarta-feira", "quinta-feira", "sexta-feira", "sábado", "domingo", ][value.weekday()] def connectivity(self, start_ms: int, end_ms: int) -> dict: previous = self.db.execute( "SELECT * FROM device_events WHERE event_id IN (1,2) AND event_time_ms < ? " "ORDER BY event_time_ms DESC LIMIT 1", (start_ms,), ).fetchone() rows = self.db.execute( "SELECT * FROM device_events WHERE event_id IN (1,2) " "AND event_time_ms >= ? AND event_time_ms < ? ORDER BY event_time_ms", (start_ms, end_ms), ).fetchall() offline_at = start_ms if previous and previous["event_id"] == 2 else None incidents: list[dict] = [] disconnects = 0 for row in rows: event_ms = int(row["event_time_ms"]) if row["event_id"] == 2: disconnects += 1 if offline_at is None: offline_at = event_ms elif row["event_id"] == 1 and offline_at is not None: begin = datetime.fromtimestamp( offline_at / 1000, timezone.utc ).astimezone(TZ) finish = datetime.fromtimestamp( event_ms / 1000, timezone.utc ).astimezone(TZ) incidents.append( { "offline_at": begin.isoformat(timespec="seconds"), "online_at": finish.isoformat(timespec="seconds"), "weekday": self.weekday_name(begin.date()), "duration_min": round((event_ms - offline_at) / 60000, 1), "ongoing": False, } ) offline_at = None if offline_at is not None: stop_ms = min(int(time.time() * 1000), end_ms) begin = datetime.fromtimestamp(offline_at / 1000, timezone.utc).astimezone(TZ) incidents.append( { "offline_at": begin.isoformat(timespec="seconds"), "online_at": None, "weekday": self.weekday_name(begin.date()), "duration_min": round(max(0, stop_ms - offline_at) / 60000, 1), "ongoing": True, } ) return { "disconnects": disconnects, "offline_min": round(sum(x["duration_min"] for x in incidents), 1), "incidents": incidents, } def report(self, start_day: date, end_day: date) -> dict: start = datetime.combine(start_day, dt_time.min, TZ) end = datetime.combine(end_day + timedelta(days=1), dt_time.min, TZ) start_ms = int(start.timestamp() * 1000) end_ms = int(end.timestamp() * 1000) events = self.normalized_events(start_ms, end_ms) sessions: list[dict] = [] on_at: int | None = None on_source = "" for event in events: if event["on"] and on_at is None: on_at = max(event["ms"], start_ms) on_source = event.get("source_code", "") elif not event["on"] and on_at is not None: stop = min(event["ms"], end_ms) if stop > on_at: begin = datetime.fromtimestamp( on_at / 1000, timezone.utc ).astimezone(TZ) finish = datetime.fromtimestamp( stop / 1000, timezone.utc ).astimezone(TZ) duration_min = (stop - on_at) / 60000 kwh = duration_min / 60 * POWER_KW sessions.append( { "on_at": begin.isoformat(timespec="seconds"), "off_at": finish.isoformat(timespec="seconds"), "calculation_end_at": finish.isoformat(timespec="seconds"), "date": begin.date().isoformat(), "weekday": self.weekday_name(begin.date()), "duration_min": round(duration_min, 1), "kwh": round(kwh, 3), "cost_eur": round(kwh * UNIT_RATE, 2), "source": self.source_label(on_source), "source_code": on_source, "ongoing": False, } ) on_at = None if on_at is not None: stop_ms = min(int(time.time() * 1000), end_ms) if stop_ms > on_at: begin = datetime.fromtimestamp( on_at / 1000, timezone.utc ).astimezone(TZ) calculation_end = datetime.fromtimestamp( stop_ms / 1000, timezone.utc ).astimezone(TZ) duration_min = (stop_ms - on_at) / 60000 kwh = duration_min / 60 * POWER_KW sessions.append( { "on_at": begin.isoformat(timespec="seconds"), "off_at": None, "calculation_end_at": calculation_end.isoformat(timespec="seconds"), "date": begin.date().isoformat(), "weekday": self.weekday_name(begin.date()), "duration_min": round(duration_min, 1), "kwh": round(kwh, 3), "cost_eur": round(kwh * UNIT_RATE, 2), "source": self.source_label(on_source), "source_code": on_source, "ongoing": True, } ) daily: dict[date, dict] = {} cursor = start_day while cursor <= end_day: daily[cursor] = { "date": cursor.isoformat(), "weekday": self.weekday_name(cursor), "seconds": 0.0, "starts": 0, "disconnects": 0, "offline_min": 0.0, } cursor += timedelta(days=1) for session in sessions: begin = datetime.fromisoformat(session["on_at"]) finish = datetime.fromisoformat(session["calculation_end_at"]) daily[begin.date()]["starts"] += 1 point = begin while point < finish: midnight = datetime.combine(point.date() + timedelta(days=1), dt_time.min, TZ) segment_end = min(finish, midnight) if point.date() in daily: daily[point.date()]["seconds"] += (segment_end - point).total_seconds() point = segment_end connection = self.connectivity(start_ms, end_ms) for incident in connection["incidents"]: incident_day = datetime.fromisoformat(incident["offline_at"]).date() if incident_day in daily: daily[incident_day]["disconnects"] += 1 daily[incident_day]["offline_min"] += incident["duration_min"] rows = [] for item in daily.values(): hours = item.pop("seconds") / 3600 kwh = hours * POWER_KW rows.append( { **item, "offline_min": round(item["offline_min"], 1), "runtime_min": round(hours * 60, 1), "kwh": round(kwh, 3), "cost_eur": round(kwh * UNIT_RATE, 2), } ) total_min = sum(row["runtime_min"] for row in rows) total_kwh = sum(row["kwh"] for row in rows) weekday_summary: list[dict] = [] for weekday in [ "segunda-feira", "terça-feira", "quarta-feira", "quinta-feira", "sexta-feira", "sábado", "domingo", ]: matching = [row for row in rows if row["weekday"] == weekday] weekday_summary.append( { "weekday": weekday, "starts": sum(row["starts"] for row in matching), "runtime_min": round(sum(row["runtime_min"] for row in matching), 1), "disconnects": sum(row["disconnects"] for row in matching), } ) return { "ok": True, "from": start_day.isoformat(), "to": end_day.isoformat(), "power_kw": POWER_KW, "rate_eur_kwh": UNIT_RATE, "measurement": "estimated_from_on_time", "days": rows, "sessions": sessions, "connectivity": connection, "weekdays": weekday_summary, "user_identity_available": False, "user_identity_note": ( "A API informa apenas a origem (dispositivo, app/cliente, " "plataforma/automação ou nuvem), não o nome/email do utilizador." ), "totals": { "starts": sum(row["starts"] for row in rows), "runtime_min": round(total_min, 1), "kwh": round(total_kwh, 3), "cost_eur": round(total_kwh * UNIT_RATE, 2), "disconnects": connection["disconnects"], "offline_min": connection["offline_min"], }, } class TuyaMonitor: def __init__(self, store: Store) -> None: self.store = store self.api = TuyaOpenAPI(ENDPOINT, ACCESS_ID, ACCESS_KEY) self.connected = False self.latest = {"ok": False, "error": "a iniciar"} self.lock = asyncio.Lock() def _connect(self) -> None: result = self.api.connect() if not result.get("success"): raise RuntimeError(result.get("msg", "falha de autenticação Tuya")) self.connected = True def _fetch_status(self) -> dict: if not self.connected: self._connect() response = self.api.get(f"/v1.0/iot-03/devices/{DEVICE_ID}/status") if not response.get("success"): self.connected = False raise RuntimeError(response.get("msg", "falha ao consultar status")) values = {x["code"]: x["value"] for x in response.get("result", [])} return values def _fetch_logs(self, days: int) -> list[dict]: if not self.connected: self._connect() now_ms = int(time.time() * 1000) logs: list[dict] = [] row_key = "" seen_keys: set[str] = set() for _page in range(20): params = { "type": "1,2,5,7,9,10", "start_time": now_ms - days * 86400 * 1000, "end_time": now_ms, "size": 100, } if row_key: params["start_row_key"] = row_key response = self.api.get(f"/v1.0/devices/{DEVICE_ID}/logs", params) if not response.get("success"): self.connected = False raise RuntimeError(response.get("msg", "falha ao consultar logs")) result = response.get("result", {}) logs.extend(result.get("logs", [])) if not result.get("has_next"): break next_key = result.get("next_row_key", "") if not next_key or next_key in seen_keys: break seen_keys.add(next_key) row_key = next_key return logs async def sync(self, days: int = 14) -> dict: async with self.lock: try: # tuya-connector keeps the access token on the client; serialize its calls. status = await asyncio.to_thread(self._fetch_status) logs = await asyncio.to_thread(self._fetch_logs, days) inserted = self.store.add_events(logs) now = datetime.now(TZ) today = self.store.report(now.date(), now.date()) self.latest = { "ok": True, "ts": now.isoformat(timespec="seconds"), "device_id": DEVICE_ID, "name": "Boiler Shower", "is_on": bool(status.get("switch_1")), "online": True, "power_kw_assumed": POWER_KW, "watts_estimated": POWER_KW * 1000 if status.get("switch_1") else 0, "today": today["totals"], "events_fetched": len(logs), "events_inserted": inserted, "measurement": "ESTIMATIVA por tempo ON; este Tuya não mede watts/kWh", } return self.latest except Exception as exc: self.latest = { "ok": False, "online": False, "error": f"{type(exc).__name__}: {exc}", } return self.latest async def loop(self) -> None: await self.sync(days=90) while True: await asyncio.sleep(POLL_SECONDS) await self.sync(days=7) INDEX_HTML = r""" Boiler · Tuya
Boiler Shower · Tuya
Interruptor do boiler
a iniciar…
—
Potência configurada: 2,3 kW · apenas leitura
Hoje
—
kWh estimados
Custo hoje
—
36,52 c/kWh
Tempo hoje
—
relé ON
Ligações hoje
—
ciclos detectados
Desconexões
—
Wi-Fi/nuvem hoje
Tempo offline
—
detectado hoje
Relatório por período
CSV
—
Resumo diário
Data/diaONminkWh€quedas
Cada ligação
DiaLigouDesligouminOrigem
Desconexões Wi-Fi/nuvem
DiaOfflineOnlinemin
Eventos por dia da semana
Estimativa: este modelo Tuya não possui medidor elétrico. kWh = 2,3 kW × tempo em que o relé ficou ON. Um termóstato interno pode reduzir o consumo real.
Quem ligou: a API Tuya mostra apenas a origem (app/cliente, automação, dispositivo ou nuvem); não fornece nome/email do utilizador.
""" store = Store() monitor = TuyaMonitor(store) def parse_range(request: web.Request) -> tuple[date, date]: today = datetime.now(TZ).date() start = date.fromisoformat(request.query.get("from", (today - timedelta(days=29)).isoformat())) end = date.fromisoformat(request.query.get("to", today.isoformat())) if end < start: raise ValueError("data final anterior à inicial") if (end - start).days > 730: raise ValueError("intervalo máximo: 730 dias") return start, end async def index(_request: web.Request) -> web.Response: return web.Response(text=INDEX_HTML, content_type="text/html") async def boiler_redirect(_request: web.Request) -> web.Response: raise web.HTTPPermanentRedirect("/boiler/") async def api_status(_request: web.Request) -> web.Response: if not monitor.latest.get("ok"): await monitor.sync(days=14) return web.json_response(monitor.latest) async def api_sync(_request: web.Request) -> web.Response: return web.json_response(await monitor.sync(days=90)) async def api_report(request: web.Request) -> web.Response: try: start, end = parse_range(request) return web.json_response(store.report(start, end)) except Exception as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) async def api_report_csv(request: web.Request) -> web.Response: try: start, end = parse_range(request) report = store.report(start, end) output = io.StringIO() writer = csv.writer(output) writer.writerow( [ "tipo", "data", "dia_semana", "ligou_offline", "desligou_online", "origem", "ligacoes", "minutos", "kwh_estimados", "custo_eur", "desconexoes", ] ) for row in report["days"]: writer.writerow( [ "dia", row["date"], row["weekday"], "", "", "", row["starts"], row["runtime_min"], row["kwh"], row["cost_eur"], row["disconnects"], ] ) for session in report["sessions"]: writer.writerow( [ "ligacao", session["date"], session["weekday"], session["on_at"], session["off_at"] or "", session["source"], 1, session["duration_min"], session["kwh"], session["cost_eur"], "", ] ) for incident in report["connectivity"]["incidents"]: writer.writerow( [ "desconexao", incident["offline_at"][:10], incident["weekday"], incident["offline_at"], incident["online_at"] or "", "", "", incident["duration_min"], "", "", 1, ] ) return web.Response( text=output.getvalue(), content_type="text/csv", headers={ "Content-Disposition": f'attachment; filename="boiler_{start}_{end}.csv"' }, ) except Exception as exc: return web.json_response({"ok": False, "error": str(exc)}, status=400) async def on_start(app: web.Application) -> None: app["collector"] = asyncio.create_task(monitor.loop()) async def on_stop(app: web.Application) -> None: app["collector"].cancel() def main() -> None: app = web.Application() app.router.add_get("/", index) app.router.add_get("/api/status", api_status) app.router.add_post("/api/sync", api_sync) app.router.add_get("/api/report", api_report) app.router.add_get("/api/report.csv", api_report_csv) app.router.add_get("/boiler", boiler_redirect) app.router.add_get("/boiler/", index) app.router.add_get("/boiler/api/status", api_status) app.router.add_get("/boiler/api/report", api_report) app.router.add_get("/boiler/api/report.csv", api_report_csv) app.on_startup.append(on_start) app.on_cleanup.append(on_stop) web.run_app(app, host="0.0.0.0", port=PORT, print=None) if __name__ == "__main__": main()