1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
#!/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.
"""Self-check: run `python3 test_proxy.py`. Fakes the plugs, exercises routing and totals."""
import datetime as dt
import json
import os
import tempfile
import urllib.request
os.environ["HISTORY_DB"] = os.path.join(tempfile.mkdtemp(), "test.db")
os.environ["TASMOTA_PLUGS"] = "ac=1.1.1.1,washer=2.2.2.2,dead=3.3.3.3"
os.environ["RATE_MARGINAL"] = "0.1"
os.environ["RATE_ALLIN"] = "0.2"
import tasmota_proxy as p # noqa: E402
SAMPLE = {
"1.1.1.1": {"Power": 1000, "Voltage": 230, "Current": 4.3,
"Today": 2.0, "Yesterday": 1.0, "Total": 10.0},
"2.2.2.2": {"Power": 500, "Voltage": 231, "Current": 2.1,
"Today": 1.0, "Yesterday": 0.5, "Total": 5.0},
}
class FakeResponse:
def __init__(self, host):
if host not in SAMPLE:
raise OSError("plug offline")
self._body = json.dumps({"StatusSNS": {"ENERGY": SAMPLE[host]}})
def read(self):
return self._body.encode()
def __enter__(self):
return self
def __exit__(self, *a):
return False
p.urllib.request.urlopen = lambda url, timeout=None: FakeResponse(
urllib.parse.urlparse(url).hostname
)
one = p.fetch("1.1.1.1")
assert one["power"] == 1000, one
assert one["cost"]["today_allin"] == 0.4, one # 2.0 kWh * 0.2
assert one["cost"]["now_hourly_marginal"] == 0.1, one # 1 kW * 0.1
allp = p.fetch_all()
assert set(allp) == {"ac", "washer", "dead"}, allp
assert "error" in allp["dead"], allp # offline plug isolated
assert "error" not in allp["ac"], allp
t = p.total()
assert t["power"] == 1500, t # dead plug excluded, not zero-filled
assert t["today_kwh"] == 3.0, t
assert t["cost"]["today_allin"] == 0.6, t
assert t["plugs_ok"] == 2 and t["plugs_total"] == 3, t
assert t["offline"] == ["dead"], t
# --- power formatting ---
# 1585 W -> "1.58 kW": banker's rounding on the exact half, not a typo.
for w, want in [(1585, "1.58 kW"), (79.7, "79.7 W"), (0, "0 W"),
(999, "999 W"), (1000, "1.00 kW"), (2340.5, "2.34 kW")]:
assert p.watts(w) == want, (w, p.watts(w), want)
assert one["power_fmt"] == "1.00 kW", one # the 1000 W fake plug
assert p.total()["power_fmt"] == "1.50 kW", p.total()
# --- paired kWh + cost rows ---
assert p.kwh_eur(2.0) == "2 kWh · 0.40 €", p.kwh_eur(2.0) # RATE_ALLIN=0.2 here
assert p.kwh_eur(0) == "0 kWh · 0.00 €", p.kwh_eur(0)
t = p.total()
assert one["today_fmt"] == "2 kWh · 0.40 €", one # per-plug rows
assert one["total_fmt"] == "10 kWh · 2.00 €", one
t = p.total()
assert t["today_fmt"] == "3 kWh · 0.60 €", t # 2.0 + 1.0, dead skipped
assert t["total_fmt"] == "15 kWh · 3.00 €", t
# --- daily history ---
yday = (dt.date.today() - dt.timedelta(days=1)).isoformat()
h = p.history()
assert h == [{"day": yday, "kwh": 1.5, "cost_allin": 0.3}], h # 1.0 + 0.5, dead skipped
p.total() # poll again same day
h = p.history()
assert len(h) == 1 and h[0]["kwh"] == 1.5, h # idempotent, not doubled
# an older day coexists rather than replacing
with p.db() as c:
c.execute("INSERT OR REPLACE INTO daily VALUES (?, ?, ?)",
((dt.date.today() - dt.timedelta(days=3)).isoformat(), "ac", 4.0))
assert [r["kwh"] for r in p.history()] == [4.0, 1.5], p.history() # oldest first
assert p.rolling(7) == 5.5, p.rolling(7)
assert p.rolling(2) == 1.5, p.rolling(2) # window excludes the 3-day-old row
t = p.total()
assert t["last7_kwh"] == 5.5 and t["last7_allin"] == 1.1, t
assert "<div class=bar" in p.graph_html()
assert "5.5 kWh" in p.graph_html()
print("ok")
|
