"""ShellyDash — Flask application with REST API + dashboard.""" from __future__ import annotations import logging import os from pathlib import Path from dotenv import load_dotenv from flask import Flask, jsonify, render_template, request, send_from_directory from database import Database from poller import ShellyPoller from shelly_client import ShellyClient load_dotenv() logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) SERVER = os.getenv("SHELLY_SERVER", "") AUTH_KEY = os.getenv("SHELLY_AUTH_KEY", "") DEVICE_ID = os.getenv("SHELLY_DEVICE_ID", "") FLASK_HOST = os.getenv("FLASK_HOST", "0.0.0.0") FLASK_PORT = int(os.getenv("FLASK_PORT", "5050")) # ── Globals ──────────────────────────────────────────────────────── db = Database() client = ShellyClient(SERVER, AUTH_KEY, DEVICE_ID) poller = ShellyPoller(client, db) app = Flask(__name__, static_folder="static", template_folder="templates") # ── Frontend routes ──────────────────────────────────────────────── @app.route("/") def index(): return render_template("index.html") @app.route("/static/") def static_files(filename): return send_from_directory(Path(app.root_path) / "static", filename) # ── API routes ───────────────────────────────────────────────────── @app.route("/api/status") def api_status(): """Current real-time readings.""" live = poller.latest_parsed if not live: return jsonify({"ok": False, "error": "No data yet"}), 503 ch0_power = live.get("ch0_act_power") or 0 ch1_power = live.get("ch1_act_power") or 0 consumption = abs(ch1_power) if ch1_power < 0 else ch1_power delta = ch0_power - consumption return jsonify({ "ok": True, "ts": live.get("unixtime"), "production_w": round(ch0_power, 1), "consumption_w": round(consumption, 1), "delta_w": round(delta, 1), "voltage": live.get("ch0_voltage"), "ch0_current": live.get("ch0_current"), "ch1_current": live.get("ch1_current"), "ch0_pf": live.get("ch0_pf"), "ch1_pf": live.get("ch1_pf"), "temperature": live.get("temperature"), "wifi_rssi": live.get("wifi_rssi"), "online": live.get("online"), }) @app.route("/api/history") def api_history(): """Historical data. Query param: hours (default 24).""" hours = int(request.args.get("hours", 24)) rows = db.get_history(hours) return jsonify({ "ok": True, "hours": hours, "data": [ { "ts": r["ts"], "production_w": r["ch0_act_power"], "consumption_w": abs(r["ch1_act_power"]) if (r["ch1_act_power"] or 0) < 0 else r["ch1_act_power"], "voltage": r["ch0_voltage"], "ch0_current": r["ch0_current"], "ch1_current": r["ch1_current"], } for r in rows ], }) @app.route("/api/energy/today") def api_energy_today(): """Today's energy totals.""" today = db.get_energy_today() return jsonify({ "ok": True, "production_kwh": round((today.get("ch0_energy") or 0) / 1000, 3), "production_returned_kwh": round((today.get("ch0_ret") or 0) / 1000, 3), "consumption_kwh": round((today.get("ch1_energy") or 0) / 1000, 3), "grid_returned_kwh": round((today.get("ch1_ret") or 0) / 1000, 3), }) @app.route("/api/energy/monthly") def api_energy_monthly(): """Monthly energy breakdown by day.""" days = db.get_monthly_energy() return jsonify({ "ok": True, "data": [ { "day": d["day"], "production_kwh": round((d["ch0_prod"] or 0) / 1000, 2), "consumption_kwh": round((d["ch1_cons"] or 0) / 1000, 2), } for d in days ], }) @app.route("/api/delta") def api_delta(): """Surplus / deficit value for automations.""" live = poller.latest_parsed if not live: return jsonify({"ok": False, "error": "No data"}), 503 prod = live.get("ch0_act_power") or 0 ch1 = live.get("ch1_act_power") or 0 cons = abs(ch1) if ch1 < 0 else ch1 delta = prod - cons return jsonify({ "ok": True, "delta_w": round(delta, 1), "production_w": round(prod, 1), "consumption_w": round(cons, 1), "surplus": delta > 0, }) # ── Main ─────────────────────────────────────────────────────────── if __name__ == "__main__": poller.poll_once() poller.start() app.run(host=FLASK_HOST, port=FLASK_PORT, debug=False)