#!/usr/bin/env python3 # Copyright (C) 2026 Danilo M. # # 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: / 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 = """ Consumi giornalieri

Consumi giornalieri

{subtitle}

{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='
In attesa del primo giorno completo.
', ) peak = max(r["kwh"] for r in rows) or 1 bars = "".join( '
'.format( h=r["kwh"] / peak * 100, day=r["day"], kwh=r["kwh"], cost=r["cost_allin"] ) for r in rows ) labels = "".join("{}".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='
{}
{}
'.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()