truetype
b078b850f8
Flask app con gauge Chart.js custom, polling Shelly Cloud API v2, storico SQLite, due widget gauge (produzione/consumo) con scala graduata.
152 lines
6.1 KiB
Python
152 lines
6.1 KiB
Python
"""SQLite database manager for ShellyDash."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
|
|
DB_PATH = Path(__file__).parent / "data" / "shellydash.db"
|
|
|
|
|
|
class Database:
|
|
"""Manage readings storage and queries."""
|
|
|
|
def __init__(self, db_path: str | Path = DB_PATH):
|
|
self.db_path = Path(db_path)
|
|
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._init_schema()
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(str(self.db_path))
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
return conn
|
|
|
|
def _init_schema(self):
|
|
with self._connect() as conn:
|
|
conn.executescript("""
|
|
CREATE TABLE IF NOT EXISTS readings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
ts INTEGER NOT NULL, -- unix timestamp
|
|
ch0_act_power REAL,
|
|
ch0_aprt_power REAL,
|
|
ch0_current REAL,
|
|
ch0_voltage REAL,
|
|
ch0_pf REAL,
|
|
ch0_total_act_energy REAL,
|
|
ch0_total_ret_energy REAL,
|
|
ch1_act_power REAL,
|
|
ch1_aprt_power REAL,
|
|
ch1_current REAL,
|
|
ch1_voltage REAL,
|
|
ch1_pf REAL,
|
|
ch1_total_act_energy REAL,
|
|
ch1_total_ret_energy REAL,
|
|
switch_output INTEGER,
|
|
temperature REAL,
|
|
wifi_rssi INTEGER,
|
|
online INTEGER,
|
|
uptime INTEGER
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_readings_ts ON readings(ts);
|
|
""")
|
|
|
|
def insert_reading(self, r: dict) -> int | None:
|
|
ts = int(r.get("unixtime") or time.time())
|
|
with self._connect() as conn:
|
|
cur = conn.execute(
|
|
"""INSERT INTO readings (
|
|
ts, ch0_act_power, ch0_aprt_power, ch0_current, ch0_voltage,
|
|
ch0_pf, ch0_total_act_energy, ch0_total_ret_energy,
|
|
ch1_act_power, ch1_aprt_power, ch1_current, ch1_voltage,
|
|
ch1_pf, ch1_total_act_energy, ch1_total_ret_energy,
|
|
switch_output, temperature, wifi_rssi, online, uptime
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
|
(
|
|
ts,
|
|
r.get("ch0_act_power"),
|
|
r.get("ch0_aprt_power"),
|
|
r.get("ch0_current"),
|
|
r.get("ch0_voltage"),
|
|
r.get("ch0_pf"),
|
|
r.get("ch0_total_act_energy"),
|
|
r.get("ch0_total_ret_energy"),
|
|
r.get("ch1_act_power"),
|
|
r.get("ch1_aprt_power"),
|
|
r.get("ch1_current"),
|
|
r.get("ch1_voltage"),
|
|
r.get("ch1_pf"),
|
|
r.get("ch1_total_act_energy"),
|
|
r.get("ch1_total_ret_energy"),
|
|
1 if r.get("switch_output") else 0,
|
|
r.get("temperature"),
|
|
r.get("wifi_rssi"),
|
|
1 if r.get("online") else 0,
|
|
r.get("uptime"),
|
|
),
|
|
)
|
|
return cur.lastrowid
|
|
|
|
def get_latest(self) -> dict | None:
|
|
"""Return the most recent reading."""
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"SELECT * FROM readings ORDER BY ts DESC LIMIT 1"
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_history(self, hours: int = 24) -> list[dict]:
|
|
"""Return readings for the last N hours, sampled ~ every 60s."""
|
|
since = int(time.time()) - hours * 3600
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""SELECT * FROM readings
|
|
WHERE ts >= ?
|
|
ORDER BY ts ASC""",
|
|
(since,),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def get_energy_today(self) -> dict:
|
|
"""Return total energy produced/consumed today."""
|
|
t = datetime.now(timezone.utc)
|
|
start_of_day = int(t.replace(hour=0, minute=0, second=0, microsecond=0).timestamp())
|
|
with self._connect() as conn:
|
|
row = conn.execute(
|
|
"""SELECT
|
|
MAX(ch0_total_act_energy) - MIN(ch0_total_act_energy) AS ch0_energy,
|
|
MAX(ch0_total_ret_energy) - MIN(ch0_total_ret_energy) AS ch0_ret,
|
|
MAX(ch1_total_act_energy) - MIN(ch1_total_act_energy) AS ch1_energy,
|
|
MAX(ch1_total_ret_energy) - MIN(ch1_total_ret_energy) AS ch1_ret
|
|
FROM readings WHERE ts >= ?""",
|
|
(start_of_day,),
|
|
).fetchone()
|
|
return dict(row) if row else {}
|
|
|
|
def get_monthly_energy(self) -> list[dict]:
|
|
"""Return daily energy sums for the current month."""
|
|
t = datetime.now(timezone.utc)
|
|
start_of_month = int(t.replace(day=1, hour=0, minute=0, second=0, microsecond=0).timestamp())
|
|
with self._connect() as conn:
|
|
rows = conn.execute(
|
|
"""SELECT
|
|
date(ts, 'unixepoch') AS day,
|
|
MAX(ch0_total_act_energy) - MIN(ch0_total_act_energy) AS ch0_prod,
|
|
MAX(ch1_total_act_energy) - MIN(ch1_total_act_energy) AS ch1_cons
|
|
FROM readings
|
|
WHERE ts >= ?
|
|
GROUP BY day
|
|
ORDER BY day ASC""",
|
|
(start_of_month,),
|
|
).fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
def cleanup_old(self, days: int = 365):
|
|
"""Delete readings older than N days."""
|
|
cutoff = int(time.time()) - days * 86400
|
|
with self._connect() as conn:
|
|
conn.execute("DELETE FROM readings WHERE ts < ?", (cutoff,))
|