shellydash/app.py

"""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/<path:filename>")
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)