path: root/tasmota_proxy.py

#!/usr/bin/env python3
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

"""Fetch Tasmota energy JSON from several plugs, add cost fields, serve for Homepage customapi.

Paths: /<name> for one plug, /total for the sum, /history for per-day kWh,
/graph for a bar chart page, / for the plug list.

Daily history: Tasmota keeps only Today/Yesterday/Total, so each poll records
yesterday's finished kWh per plug into SQLite, keyed (day, plug) so repeated
polls and restarts overwrite rather than accumulate.
"""

import datetime as dt
import json
import os
import sqlite3
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# name -> host. Override wholesale with TASMOTA_PLUGS="name=host,name=host".
PLUGS = dict(
    p.split("=", 1)
    for p in os.environ.get(
        "TASMOTA_PLUGS",
        "ac=172.16.34.176,washer=172.16.34.127,dishwasher=172.16.34.149,"
        "pc=172.16.34.118,spare=172.16.34.186",
    ).split(",")
)
PORT = int(os.environ.get("PORT", "8099"))
# Rates from invoice: marginal = consumption quota only, all-in = total bill / total kWh
RATE_MARGINAL = float(os.environ.get("RATE_MARGINAL", "0.17598"))
RATE_ALLIN = float(os.environ.get("RATE_ALLIN", "0.2754"))
DB = os.environ.get("HISTORY_DB", "/var/lib/tasmota-proxy/history.db")


def db():
    conn = sqlite3.connect(DB, timeout=5)
    conn.execute(
        "CREATE TABLE IF NOT EXISTS daily ("
        "day TEXT, plug TEXT, kwh REAL, PRIMARY KEY (day, plug))"
    )
    return conn


def record(plugs):
    """Store yesterday's finished kWh per plug. Idempotent: same day+plug overwrites."""
    day = (dt.date.today() - dt.timedelta(days=1)).isoformat()
    rows = [
        (day, name, p["yesterday_kwh"])
        for name, p in plugs.items()
        if "error" not in p
    ]
    if not rows:
        return
    with db() as conn:
        conn.executemany("INSERT OR REPLACE INTO daily VALUES (?, ?, ?)", rows)


def history(days=30):
    """Per-day totals across all plugs, oldest first."""
    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    with db() as conn:
        rows = conn.execute(
            "SELECT day, ROUND(SUM(kwh), 3) FROM daily WHERE day >= ? "
            "GROUP BY day ORDER BY day",
            (since,),
        ).fetchall()
    return [
        {"day": d, "kwh": k, "cost_allin": round(k * RATE_ALLIN, 2)} for d, k in rows
    ]


def rolling(days):
    """kWh summed over the last N recorded days (excludes today, which is unfinished)."""
    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    with db() as conn:
        (kwh,) = conn.execute(
            "SELECT COALESCE(SUM(kwh), 0) FROM daily WHERE day >= ?", (since,)
        ).fetchone()
    return round(kwh, 3)


def watts(power_w):
    """Human-readable power: kW above 1000 W, W below. Preformatted so no locale
    separator can turn 1585 W into an ambiguous "1,585"."""
    if power_w >= 1000:
        return f"{power_w / 1000:.2f} kW"
    return f"{power_w:g} W"


def kwh_eur(kwh):
    """One row's worth of text: "0.87 kWh · 0.24 €". Homepage's customapi has no
    columns, so pairing the two numbers here is what keeps the card two-up."""
    return f"{kwh:g} kWh · {kwh * RATE_ALLIN:.2f} €"


def costs(power_w, today, total):
    return {
        "now_hourly_marginal": round(power_w / 1000 * RATE_MARGINAL, 4),
        "now_hourly_allin": round(power_w / 1000 * RATE_ALLIN, 4),
        "today_marginal": round(today * RATE_MARGINAL, 2),
        "today_allin": round(today * RATE_ALLIN, 2),
        "total_marginal": round(total * RATE_MARGINAL, 2),
        "total_allin": round(total * RATE_ALLIN, 2),
    }


def fetch(host):
    url = f"http://{host}/cm?cmnd=Status%2010"
    with urllib.request.urlopen(url, timeout=5) as r:
        e = json.load(r)["StatusSNS"]["ENERGY"]
    return {
        "power": e["Power"],
        "power_fmt": watts(e["Power"]),
        "voltage": e["Voltage"],
        "current": e["Current"],
        "today_kwh": e["Today"],
        "yesterday_kwh": e["Yesterday"],
        "total_kwh": e["Total"],
        "today_fmt": kwh_eur(e["Today"]),
        "yesterday_fmt": kwh_eur(e["Yesterday"]),
        "total_fmt": kwh_eur(e["Total"]),
        "cost": costs(e["Power"], e["Today"], e["Total"]),
    }


def fetch_all():
    """All plugs in parallel; an offline plug yields {"error": ...} instead of failing the batch."""
    def one(item):
        name, host = item
        try:
            return name, fetch(host)
        except Exception as exc:  # plug offline, bad JSON, timeout
            return name, {"error": str(exc)}

    with ThreadPoolExecutor(max_workers=len(PLUGS)) as pool:
        return dict(pool.map(one, PLUGS.items()))


def total():
    plugs = fetch_all()
    record(plugs)
    ok = {n: p for n, p in plugs.items() if "error" not in p}
    power = sum(p["power"] for p in ok.values())
    today = sum(p["today_kwh"] for p in ok.values())
    tot = sum(p["total_kwh"] for p in ok.values())
    last7, last30 = rolling(7), rolling(30)
    return {
        "today_fmt": kwh_eur(round(today, 3)),
        "yesterday_fmt": kwh_eur(round(sum(p["yesterday_kwh"] for p in ok.values()), 3)),
        "last7_fmt": kwh_eur(last7),
        "last30_fmt": kwh_eur(last30),
        "total_fmt": kwh_eur(round(tot, 3)),
        "power": round(power, 1),
        "power_fmt": watts(power),
        "today_kwh": round(today, 3),
        "yesterday_kwh": round(sum(p["yesterday_kwh"] for p in ok.values()), 3),
        "total_kwh": round(tot, 3),
        "cost": costs(power, today, tot),
        "plugs_ok": len(ok),
        "plugs_total": len(plugs),
        "offline": sorted(set(plugs) - set(ok)),
        "last7_kwh": last7,
        "last30_kwh": last30,
        "last7_allin": round(last7 * RATE_ALLIN, 2),
        "last30_allin": round(last30 * RATE_ALLIN, 2),
    }


GRAPH_PAGE = """<!doctype html><meta charset=utf-8>
<title>Consumi giornalieri</title>
<style>
 body{{background:#1a1c1e;color:#e8eaed;font:14px system-ui,sans-serif;margin:0;padding:24px}}
 h1{{font-size:16px;font-weight:600;margin:0 0 4px}}
 p{{color:#9aa0a6;margin:0 0 24px}}
 .bars{{display:flex;align-items:flex-end;gap:4px;height:260px;
        border-bottom:1px solid #3c4043;padding-bottom:2px}}
 .bar{{flex:1;background:#8ab4f8;border-radius:2px 2px 0 0;min-height:1px}}
 .bar:hover{{background:#aecbfa}}
 .labels{{display:flex;gap:4px;margin-top:6px;color:#9aa0a6;font-size:11px}}
 .labels span{{flex:1;text-align:center;overflow:hidden}}
 .empty{{color:#9aa0a6;padding:40px 0}}
</style>
<h1>Consumi giornalieri</h1>
<p>{subtitle}</p>
{body}
"""


def graph_html():
    rows = history(30)
    if not rows:
        return GRAPH_PAGE.format(
            subtitle="Nessun dato ancora. Una riga per giorno viene registrata "
                     "dal giorno successivo al primo avvio.",
            body='<div class=empty>In attesa del primo giorno completo.</div>',
        )
    peak = max(r["kwh"] for r in rows) or 1
    bars = "".join(
        '<div class=bar style="height:{h:.1f}%" title="{day}: {kwh} kWh - {cost} EUR"></div>'.format(
            h=r["kwh"] / peak * 100, day=r["day"], kwh=r["kwh"], cost=r["cost_allin"]
        )
        for r in rows
    )
    labels = "".join("<span>{}</span>".format(r["day"][8:]) for r in rows)
    tot = round(sum(r["kwh"] for r in rows), 2)
    return GRAPH_PAGE.format(
        subtitle="Ultimi {n} giorni - {tot} kWh - {eur} EUR".format(
            n=len(rows), tot=tot, eur=round(tot * RATE_ALLIN, 2)
        ),
        body='<div class=bars>{}</div><div class=labels>{}</div>'.format(bars, labels),
    )


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        name = self.path.strip("/").split("?")[0]
        try:
            if name == "graph":
                page = graph_html().encode()
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(page)))
                self.end_headers()
                self.wfile.write(page)
                return
            if name == "total":
                body, code = total(), 200
            elif name == "history":
                body, code = {"days": history(30)}, 200
            elif name in PLUGS:
                body, code = fetch(PLUGS[name]), 200
            elif not name:
                body, code = {
                    "plugs": sorted(PLUGS),
                    "paths": ["/total", "/history", "/graph"],
                }, 200
            else:
                body, code = {"error": f"unknown plug {name!r}"}, 404
        except Exception as exc:  # plug offline, bad JSON, timeout
            body, code = {"error": str(exc)}, 502
        body = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *a):
        pass  # ponytail: no access log, journald has the unit status


if __name__ == "__main__":
    ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()