463 lines
21 KiB
Python
463 lines
21 KiB
Python
#!/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 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:
|
||
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"])}
|
||
# Tuya often emits the same transition twice from cloud/device.
|
||
if result and event["on"] == result[-1]["on"]:
|
||
continue
|
||
result.append(event)
|
||
return result
|
||
|
||
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)
|
||
|
||
intervals: list[tuple[datetime, datetime]] = []
|
||
on_at: int | None = None
|
||
for event in events:
|
||
if event["on"] and on_at is None:
|
||
on_at = max(event["ms"], start_ms)
|
||
elif not event["on"] and on_at is not None:
|
||
stop = min(event["ms"], end_ms)
|
||
if stop > on_at:
|
||
intervals.append(
|
||
(
|
||
datetime.fromtimestamp(on_at / 1000, timezone.utc).astimezone(TZ),
|
||
datetime.fromtimestamp(stop / 1000, timezone.utc).astimezone(TZ),
|
||
)
|
||
)
|
||
on_at = None
|
||
if on_at is not None:
|
||
stop_ms = min(int(time.time() * 1000), end_ms)
|
||
if stop_ms > on_at:
|
||
intervals.append(
|
||
(
|
||
datetime.fromtimestamp(on_at / 1000, timezone.utc).astimezone(TZ),
|
||
datetime.fromtimestamp(stop_ms / 1000, timezone.utc).astimezone(TZ),
|
||
)
|
||
)
|
||
|
||
daily: dict[date, dict] = {}
|
||
cursor = start_day
|
||
while cursor <= end_day:
|
||
daily[cursor] = {"date": cursor.isoformat(), "seconds": 0.0, "starts": 0}
|
||
cursor += timedelta(days=1)
|
||
|
||
for begin, finish in intervals:
|
||
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
|
||
|
||
rows = []
|
||
for item in daily.values():
|
||
hours = item.pop("seconds") / 3600
|
||
kwh = hours * POWER_KW
|
||
rows.append(
|
||
{
|
||
**item,
|
||
"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)
|
||
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,
|
||
"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),
|
||
},
|
||
}
|
||
|
||
|
||
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)
|
||
params = {
|
||
"type": "1,2,7",
|
||
"start_time": now_ms - days * 86400 * 1000,
|
||
"end_time": now_ms,
|
||
"size": 100,
|
||
}
|
||
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"))
|
||
return response.get("result", {}).get("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"""<!doctype html>
|
||
<html lang="pt"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
|
||
<title>Boiler · Tuya</title>
|
||
<style>
|
||
:root{--tuya:#ff5a36;--dark:#e94320;--ink:#18242c;--muted:#7b8a94;--page:#eef1f3;--card:#fff}
|
||
*{box-sizing:border-box}body{margin:0;background:#dfe4e8;font-family:Segoe UI,Arial,sans-serif;color:var(--ink);display:flex;justify-content:center;padding:22px 10px}
|
||
.phone{width:420px;max-width:100%;background:var(--page);border-radius:36px;box-shadow:0 20px 55px #24333c44;overflow:hidden;min-height:760px}
|
||
header{padding:22px 20px 12px;text-align:center;font-weight:700}.body{padding:0 16px 28px}
|
||
.hero{background:linear-gradient(150deg,#ff7656,var(--dark));color:white;border-radius:25px;padding:22px;box-shadow:0 12px 28px #e9432050}
|
||
.top{display:flex;justify-content:space-between;align-items:center}.dot{width:12px;height:12px;border-radius:50%;background:#ddd}.dot.on{background:#73f09b;box-shadow:0 0 10px #73f09b}
|
||
.big{font-size:2.25rem;font-weight:750;margin:24px 0 4px}.sub{opacity:.9;font-size:.85rem}
|
||
.grid{display:grid;grid-template-columns:1fr 1fr;gap:10px;margin-top:12px}.card{background:var(--card);border-radius:18px;padding:16px;box-shadow:0 3px 12px #22313a12}
|
||
.label{font-size:.72rem;color:var(--muted);text-transform:uppercase}.value{font-size:1.35rem;font-weight:750;margin-top:5px}
|
||
.report{margin-top:12px}.dates{display:flex;gap:8px;margin:10px 0}.dates input{width:100%;padding:9px;border:1px solid #d9dfe3;border-radius:10px}
|
||
button,a.btn{border:0;border-radius:12px;background:var(--tuya);color:#fff;padding:10px 14px;font-weight:650;text-decoration:none;cursor:pointer}
|
||
table{width:100%;border-collapse:collapse;font-size:.76rem}th,td{padding:8px 4px;border-bottom:1px solid #edf0f2;text-align:right}th:first-child,td:first-child{text-align:left}
|
||
.warning{font-size:.75rem;background:#fff6df;color:#765c14;border-radius:12px;padding:10px;margin-top:12px}.error{color:#b22710}
|
||
</style></head><body><main class="phone"><header>Boiler Shower · Tuya</header><div class="body">
|
||
<section class="hero"><div class="top"><div><b>Interruptor do boiler</b><div class="sub" id="sync">a iniciar…</div></div><span class="dot" id="dot"></span></div>
|
||
<div class="big" id="state">—</div><div class="sub">Potência configurada: 2,3 kW · apenas leitura</div></section>
|
||
<div class="grid"><div class="card"><div class="label">Hoje</div><div class="value" id="todayKwh">—</div><small>kWh estimados</small></div>
|
||
<div class="card"><div class="label">Custo hoje</div><div class="value" id="todayCost">—</div><small>36,52 c/kWh</small></div>
|
||
<div class="card"><div class="label">Tempo hoje</div><div class="value" id="todayTime">—</div><small>relé ON</small></div>
|
||
<div class="card"><div class="label">Ligações hoje</div><div class="value" id="todayStarts">—</div><small>ciclos detectados</small></div></div>
|
||
<section class="card report"><b>Relatório por período</b><div class="dates"><input type="date" id="from"><input type="date" id="to"></div>
|
||
<button id="run">Gerar</button> <a class="btn" id="csv">CSV</a>
|
||
<div class="value" id="total" style="margin-top:14px">—</div><div class="sub" id="summary"></div>
|
||
<div style="overflow:auto;max-height:300px"><table><thead><tr><th>Data</th><th>ON</th><th>min</th><th>kWh</th><th>€</th></tr></thead><tbody id="rows"></tbody></table></div></section>
|
||
<div class="warning"><b>Estimativa:</b> 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.</div>
|
||
</div></main><script>
|
||
const $=id=>document.getElementById(id), eur=n=>new Intl.NumberFormat('pt-PT',{style:'currency',currency:'EUR'}).format(n);
|
||
const today=new Date(), ago=new Date(Date.now()-29*864e5); $('to').value=today.toISOString().slice(0,10); $('from').value=ago.toISOString().slice(0,10);
|
||
async function status(){try{const s=await (await fetch('api/status')).json(); if(!s.ok)throw Error(s.error); $('state').textContent=s.is_on?'LIGADO':'DESLIGADO'; $('dot').className='dot '+(s.is_on?'on':''); $('sync').textContent='Sincronizado '+new Date(s.ts).toLocaleString('pt-PT'); $('todayKwh').textContent=s.today.kwh.toFixed(3); $('todayCost').textContent=eur(s.today.cost_eur); $('todayTime').textContent=s.today.runtime_min.toFixed(0)+' min'; $('todayStarts').textContent=s.today.starts}catch(e){$('sync').innerHTML='<span class=error>'+e.message+'</span>'}}
|
||
async function report(){const f=$('from').value,t=$('to').value,r=await (await fetch(`api/report?from=${f}&to=${t}`)).json(); if(!r.ok){$('total').textContent=r.error;return}$('total').textContent=`${r.totals.kwh.toFixed(3)} kWh · ${eur(r.totals.cost_eur)}`;$('summary').textContent=`${r.totals.runtime_min.toFixed(0)} min ligados · ${r.totals.starts} ligações`;$('rows').innerHTML=r.days.map(d=>`<tr><td>${d.date}</td><td>${d.starts}</td><td>${d.runtime_min.toFixed(0)}</td><td>${d.kwh.toFixed(3)}</td><td>${eur(d.cost_eur)}</td></tr>`).join('');$('csv').href=`api/report.csv?from=${f}&to=${t}`}
|
||
$('run').onclick=report; status();report();setInterval(status,60000);
|
||
</script></body></html>"""
|
||
|
||
|
||
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(["data", "ligacoes", "minutos_on", "kwh_estimados", "custo_eur"])
|
||
for row in report["days"]:
|
||
writer.writerow(
|
||
[row["date"], row["starts"], row["runtime_min"], row["kwh"], row["cost_eur"]]
|
||
)
|
||
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()
|