From 7b11b9641d232f62789fe85972ac3dfb5fcabb4f Mon Sep 17 00:00:00 2001 From: Sujith Quintelier Date: Tue, 3 Nov 2026 16:31:14 +0100 Subject: [PATCH] Add Solar Buddy app Home Assistant solar companion with demo and live modes. Live mode works on the badge (socket client) and in the Badge Simulator canvas (browser fetch client with a browser-clock fallback when CORS hides the Date header). Credentials stay in a git-ignored private file copied to /state. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- badge/apps/solar_buddy/.gitignore | 2 + badge/apps/solar_buddy/README.md | 118 ++++++++ badge/apps/solar_buddy/__init__.py | 68 +++++ badge/apps/solar_buddy/advisor.py | 45 +++ badge/apps/solar_buddy/browser_http.py | 140 ++++++++++ badge/apps/solar_buddy/character.py | 62 ++++ badge/apps/solar_buddy/clock.py | 15 + badge/apps/solar_buddy/config.py | 191 +++++++++++++ badge/apps/solar_buddy/controller.py | 155 ++++++++++ badge/apps/solar_buddy/demo.py | 46 +++ badge/apps/solar_buddy/energy.py | 159 +++++++++++ badge/apps/solar_buddy/ha_client.py | 122 ++++++++ badge/apps/solar_buddy/hardware.py | 73 +++++ badge/apps/solar_buddy/http_client.py | 264 ++++++++++++++++++ badge/apps/solar_buddy/icon.png | Bin 0 -> 181 bytes badge/apps/solar_buddy/settings.example.json | 20 ++ .../solar_buddy_secrets.example.json | 11 + badge/apps/solar_buddy/ui.py | 228 +++++++++++++++ 18 files changed, 1719 insertions(+) create mode 100644 badge/apps/solar_buddy/.gitignore create mode 100644 badge/apps/solar_buddy/README.md create mode 100644 badge/apps/solar_buddy/__init__.py create mode 100644 badge/apps/solar_buddy/advisor.py create mode 100644 badge/apps/solar_buddy/browser_http.py create mode 100644 badge/apps/solar_buddy/character.py create mode 100644 badge/apps/solar_buddy/clock.py create mode 100644 badge/apps/solar_buddy/config.py create mode 100644 badge/apps/solar_buddy/controller.py create mode 100644 badge/apps/solar_buddy/demo.py create mode 100644 badge/apps/solar_buddy/energy.py create mode 100644 badge/apps/solar_buddy/ha_client.py create mode 100644 badge/apps/solar_buddy/hardware.py create mode 100644 badge/apps/solar_buddy/http_client.py create mode 100644 badge/apps/solar_buddy/icon.png create mode 100644 badge/apps/solar_buddy/settings.example.json create mode 100644 badge/apps/solar_buddy/solar_buddy_secrets.example.json create mode 100644 badge/apps/solar_buddy/ui.py diff --git a/badge/apps/solar_buddy/.gitignore b/badge/apps/solar_buddy/.gitignore new file mode 100644 index 00000000..0b9a483c --- /dev/null +++ b/badge/apps/solar_buddy/.gitignore @@ -0,0 +1,2 @@ +solar_buddy_secrets.json +settings.json \ No newline at end of file diff --git a/badge/apps/solar_buddy/README.md b/badge/apps/solar_buddy/README.md new file mode 100644 index 00000000..d466c77c --- /dev/null +++ b/badge/apps/solar_buddy/README.md @@ -0,0 +1,118 @@ +# Solar Buddy + +An animated solar companion that reads Home Assistant power sensors and recommends appliances after stable surplus. + +This app targets the GitHub Universe 2026 Badgeware runtime at 320 x 240. Copy the +folder into `/system/apps/solar_buddy` through USB mass-storage mode, leaving out +`solar_buddy_secrets.json` and `__pycache__` (see below). + +It starts with simulated data and requires no network in demo mode. Left/right +switch screens, up/down select a scenario or appliance, Select opens appliance +details or toggles demo/live in Settings, Back returns to Buddy, Menu opens Settings, +and Home returns to the launcher. Shake briefly wakes the face. + +Copy `settings.example.json` to `settings.json` to customize thresholds, estimated +appliance loads, polling, freshness, and the 30-second stability period. The default +safety reserve is 100 W. All recommendations are read-only and apply to one load +at a time; they do not guarantee a solar-only appliance cycle. + +An explicit offline/stale state suppresses advice when data is missing or invalid. +The app assumes there is no home battery. Only Solar Superstar is implemented; +daily statistics achievements need additional sensors. + +## Home Assistant credentials + +Live mode reads a private JSON file from `/state/solar_buddy_secrets.json` on the +badge or in the canvas. It never reads credentials from the app folder, and Wi-Fi +uses the badge's existing Wi-Fi settings (`/system/secrets.py` or the Settings app). + +**On your computer**, keep the private copy at +`badge\apps\solar_buddy\solar_buddy_secrets.json`. The app's `.gitignore` excludes +it from Git; check with `git check-ignore -v badge/apps/solar_buddy/solar_buddy_secrets.json`. +Start from the committed, credential-free template and fill in your Home Assistant +URL, IPv4 address, long-lived access token, and three instantaneous power sensor +IDs. Replace the template's example sensor IDs if they do not match your setup; +never put a real token in the `.example.json` file. + +```powershell +Copy-Item badge\apps\solar_buddy\solar_buddy_secrets.example.json ` + badge\apps\solar_buddy\solar_buddy_secrets.json +``` + +Create the token under your Home Assistant profile, **Security**, **Long-lived +access tokens**. Use three different instantaneous power sensors reported in W, kW, +or MW (find their IDs in **Developer tools**, **States**). `connect_ip` is required +unless `base_url` already uses an IPv4 address, so the badge avoids blocking DNS. +`allow_insecure_http` must be `true` for `http://` because the token travels +unencrypted; use it only on a trusted LAN. For `https://` on the badge, copy the +CA certificate to the badge (for example `/state/solar_buddy_ca.pem`), set +`"ca_file"` to that path, and make sure the badge clock is correct. + +**On the badge**, copy the file over USB serial (`/state` is writable; `/system` +is not), then reset. Run one `mpremote` command at a time: + +```powershell +mpremote devs +mpremote connect COM7 fs cp badge\apps\solar_buddy\solar_buddy_secrets.json :/state/solar_buddy_secrets.json +mpremote connect COM7 reset +``` + +Replace `COM7` with the port from `mpremote devs`. If `/state` is missing, run +`mpremote connect COM7 fs mkdir :/state` first. Do not copy +`solar_buddy_secrets.json` into `apps/solar_buddy` on the BADGER volume: the +Badge Simulator **Deploy** button and `deploy_app.py` copy the whole folder, so +delete `apps/solar_buddy/solar_buddy_secrets.json` from BADGER if it was copied. +Open Solar Buddy, press Menu, then Select to switch to live mode. + +**In the Badge Simulator canvas**, the private file in the app folder is staged +(read-only) at `/system/apps/solar_buddy/solar_buddy_secrets.json`. Ask Copilot +to copy it to `/state`, or run this with the canvas `exec` action. It validates +the private copy made from the example above and does not print its contents: + +```python +import json, os, config +with open("/system/apps/solar_buddy/solar_buddy_secrets.json") as f: + data = f.read() +config.validate_secrets(json.loads(data)) +try: + os.mkdir("/state") +except OSError: + pass +with open(config.DEVICE_SECRETS, "w") as f: + f.write(data) +print("Copied to", config.DEVICE_SECRETS) +``` + +Then reload the app and switch to live mode in Settings. Repeat the copy after a +full canvas runtime restart (for example after files are added to, removed from, +or renamed in the app folder) or after editing the private file. A missing copy +shows `Live setup: No /state/solar_buddy_secrets.json` and Offline. See the next +section for canvas networking requirements. + +## Badge Simulator canvas + +Live mode automatically uses asynchronous browser HTTP on the canvas's +WebAssembly runtime. Physical badges and the desktop preview retain the +socket client. Requests remain read-only, time out, reject redirects, limit +responses to 16 KiB, and can be cancelled without blocking badge controls. + +`/state/solar_buddy_secrets.json` lives in the canvas's **virtual** filesystem, +not on Windows; copy it there as described above. `settings.json` contains only +app settings, not connection details. + +Browser networking uses `base_url` and browser DNS, not `connect_ip` (which is +still validated). `localhost` +means the computer hosting the canvas. Allow the canvas's exact origin (including +its port, shown in the canvas URL) in Home Assistant's `http.cors_allowed_origins`. +Browsers hide the HTTP `Date` header unless the server sends +`Access-Control-Expose-Headers: Date`, which Home Assistant's CORS settings +cannot configure. When `Date` is hidden, the canvas checks sensor freshness +against the computer's UTC clock instead and prints a one-time console notice. +Keep the computer and Home Assistant clocks synchronised (for example with NTP): +a skewed clock makes readings look stale, or rejects them as being ahead of the +clock. The physical badge always uses Home Assistant's `Date` header. +CORS, unreachable servers, authentication failures, and stale readings +are reported as errors; they never fall back to fabricated live data. +HTTPS uses browser certificate trust; `ca_file` is not supported in the canvas. +See [Home Assistant HTTP configuration](https://www.home-assistant.io/integrations/http/) +for CORS configuration. Do not disable browser security or TLS verification. diff --git a/badge/apps/solar_buddy/__init__.py b/badge/apps/solar_buddy/__init__.py new file mode 100644 index 00000000..ccdfd48d --- /dev/null +++ b/badge/apps/solar_buddy/__init__.py @@ -0,0 +1,68 @@ +"""Universe 2026 Badgeware entry point. Desktop users run preview.py.""" + +import os +import sys + +APP_DIRS = ("/remote/apps/solar_buddy", "/system/apps/solar_buddy", + "/apps/solar_buddy", "/solar_buddy") +APP_DIR = next(path for path in APP_DIRS if is_dir(path)) +os.chdir(APP_DIR) +sys.path.insert(0, APP_DIR) + +from badgeware import State +from clock import ticks_ms +from config import DEVICE_SECRETS, load_secrets, load_settings +from controller import App +from hardware import Feedback +from ui import BadgePainter, UI + +badge.mode(HIRES | VSYNC) +screen.font = font.sins +settings = load_settings("settings.json") +feedback = Feedback(badge) +saved = {"mode": settings["mode"], "scenario": 0, "superstar": False} +State.load("solar_buddy", saved) + + +def live_source(): + credentials = load_secrets( + DEVICE_SECRETS, "No " + DEVICE_SECRETS + + ". See README: Home Assistant credentials.") + from ha_client import BadgeWifi, HomeAssistant + return HomeAssistant(settings, credentials, wifi=BadgeWifi()) + + +def save_state(value): + try: + State.save("solar_buddy", value) + except OSError as exc: + app.hardware_note = "Cannot save preferences: " + str(exc) + print("Solar Buddy: " + app.hardware_note) + + +app = App(settings, live_source, saved, save_state) +ui = UI(BadgePainter(screen, shape, color)) +buttons = ((BUTTON_LEFT, "left"), (BUTTON_RIGHT, "right"), (BUTTON_UP, "up"), + (BUTTON_DOWN, "down"), (BUTTON_SELECT, "select"), (BUTTON_BACK, "back"), + (BUTTON_MENU, "menu")) + + +def update(): + now = ticks_ms() + for button, action in buttons: + if badge.pressed(button): + app.input(action, now) + app.tick(now) + feedback.update(app, now) + ui.draw(app, now, badge.ticks_delta) + + +def on_exit(): + app.close() + feedback.close() + + +try: + run(update) +finally: + on_exit() diff --git a/badge/apps/solar_buddy/advisor.py b/badge/apps/solar_buddy/advisor.py new file mode 100644 index 00000000..4262f377 --- /dev/null +++ b/badge/apps/solar_buddy/advisor.py @@ -0,0 +1,45 @@ +"""Advice is based on a continuous series of fresh samples, never a UI timer.""" + +from clock import elapsed + + +class Advisor: + def __init__(self, settings): + self.settings = settings + self.items = settings["appliances"] + self.max_gap_ms = settings["poll_interval_ms"] + 3 * settings["request_timeout_ms"] + 1000 + self.invalidate() + + def invalidate(self): + self.since = [None] * len(self.items) + self.statuses = ["UNKNOWN"] * len(self.items) + self.last_sample = None + + def update(self, sample, now): + if sample is None: + self.invalidate() + return + if self.last_sample is not None: + gap = elapsed(now, self.last_sample) + if gap <= 0: + return + if gap > self.max_gap_ms: + self.invalidate() + self.last_sample = now + usable = sample["surplus_w"] - self.settings["reserve_w"] + importing = sample["import_w"] > self.settings["import_threshold_w"] + for index, item in enumerate(self.items): + if importing or usable < item["watts"]: + self.since[index] = None + self.statuses[index] = "WAIT" + elif self.since[index] is None: + self.since[index] = now + self.statuses[index] = "WAIT" + elif elapsed(now, self.since[index]) >= self.settings["stability_ms"]: + self.statuses[index] = "READY" + + def remaining_s(self, index, now): + if self.since[index] is None: + return None + left = self.settings["stability_ms"] - elapsed(now, self.since[index]) + return max(0, (left + 999) // 1000) diff --git a/badge/apps/solar_buddy/browser_http.py b/badge/apps/solar_buddy/browser_http.py new file mode 100644 index 00000000..929b699d --- /dev/null +++ b/badge/apps/solar_buddy/browser_http.py @@ -0,0 +1,140 @@ +"""Incremental Home Assistant GETs through the canvas worker's browser fetch.""" + +import json + +from clock import elapsed +from config import parse_url, validate_secrets + + +# Keep network work in the worker's event loop so badge frames never wait on I/O. +FETCH = r""" +(function (url, headers, timeoutMs) { + const controller = new AbortController(); + const job = { + state: "pending", status: 0, headers: "{}", body: "", error: "", + clock: "server", + cancel() { + if (this.state === "pending") this.state = "cancelled"; + clearTimeout(timer); + controller.abort(); + } + }; + function fail(message) { + if (job.state !== "pending") return; + job.error = message; + job.state = "error"; + clearTimeout(timer); + controller.abort(); + } + const timer = setTimeout( + () => fail("Home Assistant request timed out"), timeoutMs); + (async () => { + try { + const response = await fetch(url, { + method: "GET", headers: JSON.parse(headers), + signal: controller.signal, redirect: "error", + credentials: "omit", cache: "no-store" + }); + if (job.state !== "pending") return; + job.status = response.status; + if (response.status !== 200) { + job.state = "done"; + controller.abort(); + return; + } + const exposedHeaders = {}; + let headerSize = 0; + response.headers.forEach((value, key) => { + exposedHeaders[key.toLowerCase()] = value; + headerSize += key.length + value.length + 4; + }); + if (headerSize > 4096) return fail("HTTP headers exceed the memory limit"); + if (!exposedHeaders.date) { + // Home Assistant's CORS setup cannot expose Date to browsers. + exposedHeaders.date = new Date().toUTCString(); + job.clock = "browser"; + } + job.headers = JSON.stringify(exposedHeaders); + const decoder = new TextDecoder("utf-8", {fatal: true}); + const reader = response.body.getReader(); + let size = 0; + while (job.state === "pending") { + const chunk = await reader.read(); + if (job.state !== "pending") return; + if (chunk.done) { + job.body += decoder.decode(); + job.state = "done"; + return; + } + size += chunk.value.byteLength; + if (size > 16384) return fail("HTTP body exceeds the memory limit"); + job.body += decoder.decode(chunk.value, {stream: true}); + } + } catch (_error) { + fail("Canvas HTTP request failed; check the URL, CORS and browser certificate trust"); + } finally { + clearTimeout(timer); + } + })(); + return job; +}) +""" + + +class HttpClient: + def __init__(self, credentials, timeout_ms): + import js + + validate_secrets(credentials) + if credentials.get("ca_file"): + raise ValueError( + "The canvas uses browser certificate trust, not ca_file") + scheme, _host, _port, prefix, authority = parse_url(credentials["base_url"]) + self.base_url = scheme + "://" + authority + prefix + self.headers = json.dumps({ + "Authorization": "Bearer " + credentials["token"], + "Accept": "application/json", + }) + self.timeout_ms = timeout_ms + self.request = js.eval(FETCH) + self.job = None + self.browser_clock = False + + def start(self, path, now): + if self.job is not None: + raise RuntimeError("An HTTP request is already in progress") + self.started = now + self.job = self.request(self.base_url + path, self.headers, self.timeout_ms) + + def step(self, now): + if self.job is None: + return None + if elapsed(now, self.started) >= self.timeout_ms: + self.close() + raise OSError("Home Assistant request timed out") + state = str(self.job.state) + if state == "pending": + return None + try: + if state == "error": + raise OSError(str(self.job.error)) + if state != "done": + raise OSError("Home Assistant request was cancelled") + code = int(self.job.status) + if code != 200: + hint = {401: "check token", 403: "sensor access denied", + 404: "check entity ID", 429: "rate limited", + 503: "Home Assistant unavailable"}.get(code, "request failed") + raise ValueError("HTTP %d: %s" % (code, hint)) + if str(self.job.clock) == "browser" and not self.browser_clock: + self.browser_clock = True + print("Solar Buddy: CORS hides HTTP Date; " + "checking freshness with the browser clock") + return json.loads(str(self.job.body)), json.loads(str(self.job.headers)) + finally: + self.close() + + def close(self): + job, self.job = self.job, None + if job is not None: + job.cancel() diff --git a/badge/apps/solar_buddy/character.py b/badge/apps/solar_buddy/character.py new file mode 100644 index 00000000..3e7d3177 --- /dev/null +++ b/badge/apps/solar_buddy/character.py @@ -0,0 +1,62 @@ +"""Small vector/pixel character: no sprite downloads or large assets.""" + +import math + + +class Buddy: + def __init__(self): + self.phase = 0.0 + self.blink_ms = 900.0 + + def advance(self, delta_ms): + self.phase = (self.phase + max(0, delta_ms) / 1000.0) % (2 * math.pi * 100) + self.blink_ms = (self.blink_ms + max(0, delta_ms)) % 4200 + + def draw(self, painter, x, y, mood, awake=False): + p = painter + accent = {"excited": "#f2cc60", "happy": "#e3b341", "neutral": "#c9a64d", + "sleepy": "#7891b0", "offline": "#8b949e"}[mood] + speed = 5 if mood == "excited" else 2 + bob = math.sin(self.phase * speed) * (4 if mood == "excited" else 2) + y += bob + p.rect(x - 24, y + 34 - bob, 48, 3, "#21262d") + rays = ((-34, -22, 7, 7), (-5, -38, 10, 9), (27, -22, 7, 7), + (-42, -4, 9, 8), (33, -4, 9, 8), + (-34, 20, 7, 7), (-5, 29, 10, 7), (27, 20, 7, 7)) + for rx, ry, w, h in rays: + p.rect(x + rx, y + ry, w, h, accent) + p.rect(x - 22, y - 28, 44, 56, accent) + p.rect(x - 28, y - 22, 56, 44, accent) + p.rect(x - 20, y - 24, 40, 4, "#ffe8a3" if mood != "offline" else "#b1bac4") + p.rect(x - 27, y + 12, 4, 10, "#26303c") + p.rect(x + 23, y + 12, 4, 10, "#26303c") + ink = "#0d1117" + closed = self.blink_ms < 130 or (mood == "sleepy" and not awake) + for eye_x in (x - 11, x + 11): + if mood == "offline": + p.line(eye_x - 3, y - 7, eye_x + 3, y - 1, ink) + p.line(eye_x - 3, y - 1, eye_x + 3, y - 7, ink) + elif closed: + p.rect(eye_x - 4, y - 3, 8, 2, ink) + else: + p.rect(eye_x - 3, y - 8, 6, 9, ink) + p.rect(eye_x - 2, y - 7, 2, 3, "#ffffff") + if mood in ("excited", "happy") or awake: + p.rect(x - 9, y + 8, 3, 4, ink) + p.rect(x + 6, y + 8, 3, 4, ink) + p.rect(x - 6, y + 12, 12, 3, ink) + if mood == "excited": + p.rect(x - 5, y + 9, 10, 3, ink) + elif mood == "sleepy": + p.rect(x - 3, y + 9, 6, 5, ink) + if not awake: + p.text("z Z", x + 45, y - 25, "#8b949e") + elif mood == "offline": + p.rect(x - 7, y + 12, 14, 2, ink) + p.text("?", x + 44, y - 28, "#f85149") + else: + p.rect(x - 7, y + 10, 14, 2, ink) + if mood == "excited": + for dx, dy in ((-59, -17), (56, 16)): + p.line(x + dx - 3, y + dy, x + dx + 3, y + dy, "#f2cc60") + p.line(x + dx, y + dy - 3, x + dx, y + dy + 3, "#f2cc60") diff --git a/badge/apps/solar_buddy/clock.py b/badge/apps/solar_buddy/clock.py new file mode 100644 index 00000000..7a8f9103 --- /dev/null +++ b/badge/apps/solar_buddy/clock.py @@ -0,0 +1,15 @@ +"""Wrap-safe badge timing, with a CPython implementation for the preview.""" + +import time + + +def ticks_ms(): + if hasattr(time, "ticks_ms"): + return time.ticks_ms() + return int(time.monotonic() * 1000) + + +def elapsed(now, then): + if hasattr(time, "ticks_diff"): + return time.ticks_diff(now, then) + return now - then diff --git a/badge/apps/solar_buddy/config.py b/badge/apps/solar_buddy/config.py new file mode 100644 index 00000000..43778977 --- /dev/null +++ b/badge/apps/solar_buddy/config.py @@ -0,0 +1,191 @@ +"""Non-secret settings and validation shared by the badge and host tools.""" + +import json +import math + +DEVICE_SECRETS = "/state/solar_buddy_secrets.json" +# ENOENT is 2 on the badge and CPython; the canvas's WebAssembly runtime reports 44. +MISSING_FILE_ERRNOS = (2, 44) + + +def defaults(): + return { + "mode": "demo", + "poll_interval_ms": 10000, + "request_timeout_ms": 4000, + "stale_after_ms": 45000, + "sensor_max_age_s": 180, + "stability_ms": 30000, + "reserve_w": 100, + "happy_w": 800, + "excited_w": 2000, + "mood_hysteresis_w": 100, + "import_threshold_w": 25, + "milestone_w": 4000, + "appliances": [ + {"name": "Dishwasher", "watts": 1500}, + {"name": "Washing machine", "watts": 2000}, + {"name": "EV charging", "watts": 2300}, + {"name": "Small appliance", "watts": 500}, + ], + } + + +def number(value, label, minimum=0, maximum=1000000): + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(label + " must be a number") + if not math.isfinite(value) or not minimum <= value <= maximum: + raise ValueError(label + " is outside its allowed range") + return value + + +def read_json(path, optional=False): + try: + with open(path, "r") as handle: + result = json.load(handle) + except OSError as exc: + if optional and exc.args and exc.args[0] in MISSING_FILE_ERRNOS: + return None + raise + if not isinstance(result, dict): + raise ValueError("Configuration must be a JSON object") + return result + + +def load_settings(path=None): + settings = defaults() + if path: + supplied = read_json(path, optional=True) + if supplied is not None: + for key, value in supplied.items(): + if key not in settings: + raise ValueError("Unknown setting: " + key) + settings[key] = value + return validate_settings(settings) + + +def validate_settings(settings): + if settings["mode"] not in ("demo", "live"): + raise ValueError("mode must be demo or live") + for key in ("poll_interval_ms", "request_timeout_ms", "stale_after_ms", + "sensor_max_age_s", "stability_ms"): + number(settings[key], key, 1, 3600000) + if int(settings[key]) != settings[key]: + raise ValueError(key + " must be a whole number") + if settings["poll_interval_ms"] < 1000: + raise ValueError("poll_interval_ms must be at least 1000") + if settings["request_timeout_ms"] < 100: + raise ValueError("request_timeout_ms must be at least 100") + cycle_budget = settings["poll_interval_ms"] + 3 * settings["request_timeout_ms"] + if settings["stale_after_ms"] <= cycle_budget: + raise ValueError("stale_after_ms must exceed poll interval + 3 timeouts") + for key in ("reserve_w", "happy_w", "excited_w", "mood_hysteresis_w", + "import_threshold_w", "milestone_w"): + number(settings[key], key) + if not 0 < settings["happy_w"] < settings["excited_w"]: + raise ValueError("Mood thresholds must satisfy 0 < happy_w < excited_w") + if settings["mood_hysteresis_w"] >= min( + settings["happy_w"], settings["excited_w"] - settings["happy_w"]): + raise ValueError("mood_hysteresis_w is too large for the thresholds") + appliances = settings["appliances"] + if not isinstance(appliances, list) or not 1 <= len(appliances) <= 12: + raise ValueError("Configure between 1 and 12 appliances") + names = [] + for appliance in appliances: + if not isinstance(appliance, dict) or set(appliance) != {"name", "watts"}: + raise ValueError("Each appliance needs only name and watts") + name = appliance["name"] + if (not isinstance(name, str) or not 1 <= len(name) <= 24 + or not all(32 <= ord(ch) < 127 for ch in name) or not name.strip()): + raise ValueError("Appliance names must be 1-24 printable ASCII characters") + if name in names: + raise ValueError("Appliance names must be unique") + names.append(name) + number(appliance["watts"], "Appliance watts", 1) + return settings + + +def is_ipv4(value): + if not isinstance(value, str): + return False + pieces = value.split(".") + return (len(pieces) == 4 and all( + p and all("0" <= ch <= "9" for ch in p) and len(p) <= 3 + and 0 <= int(p) <= 255 and (p == "0" or not p.startswith("0")) + for p in pieces)) + + +def parse_url(value): + if not isinstance(value, str) or not value: + raise ValueError("Set the Home Assistant base_url") + if any(ord(ch) <= 32 or ord(ch) >= 127 for ch in value): + raise ValueError("base_url must contain only printable ASCII without spaces") + if any(ch in value for ch in "@?#\\[]"): + raise ValueError("base_url cannot contain credentials, query, fragment, or IPv6") + parts = value.rstrip("/").split("://") + if len(parts) != 2 or parts[0] not in ("http", "https"): + raise ValueError("base_url must begin with http:// or https://") + scheme, rest = parts + authority, separator, path = rest.partition("/") + pieces = authority.split(":") + if len(pieces) > 2: + raise ValueError("Only IPv4 addresses and DNS hostnames are supported") + host = pieces[0] + allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-." + if not host or any(ch not in allowed for ch in host): + raise ValueError("Invalid Home Assistant hostname") + port = 443 if scheme == "https" else 80 + if len(pieces) == 2: + if not pieces[1] or not all("0" <= ch <= "9" for ch in pieces[1]): + raise ValueError("Invalid Home Assistant port") + port = int(pieces[1]) + number(port, "Home Assistant port", 1, 65535) + return scheme, host, port, "/" + path if separator else "", authority + + +def validate_entity(value): + allowed = "abcdefghijklmnopqrstuvwxyz0123456789_" + if (not isinstance(value, str) or not value.startswith("sensor.") + or not value[7:] or any(ch not in allowed for ch in value[7:])): + raise ValueError("Select an actual sensor entity ID from Home Assistant") + return value + + +def validate_token(token): + if (not isinstance(token, str) or not 1 <= len(token) <= 4096 + or any(not 32 < ord(ch) < 127 for ch in token)): + raise ValueError("Set a nonempty Home Assistant token without whitespace") + return token + + +def validate_secrets(secrets): + if not isinstance(secrets, dict): + raise ValueError("Run configure_home_assistant.py first") + allowed = {"base_url", "token", "connect_ip", "entities", "allow_insecure_http", + "ca_file"} + if set(secrets) - allowed: + raise ValueError("Unknown field in Home Assistant configuration") + scheme, host, _port, _prefix, _authority = parse_url(secrets.get("base_url")) + validate_token(secrets.get("token")) + ip = secrets.get("connect_ip", host) + if not is_ipv4(ip): + raise ValueError("Set connect_ip to a numeric IPv4 address to avoid blocking DNS") + if scheme == "http" and secrets.get("allow_insecure_http") is not True: + raise ValueError("HTTP exposes the token; explicitly allow it on a trusted LAN") + entities = secrets.get("entities") + if not isinstance(entities, dict) or set(entities) != {"solar", "import", "export"}: + raise ValueError("Configure solar, import, and export entity IDs") + values = [validate_entity(entities[key]) for key in ("solar", "import", "export")] + if len(set(values)) != 3: + raise ValueError("Solar, import, and export must be three different power sensors") + ca_file = secrets.get("ca_file") + if ca_file is not None and (not isinstance(ca_file, str) or not ca_file): + raise ValueError("ca_file must be a certificate file path") + return secrets + + +def load_secrets(path, missing="Run configure_home_assistant.py first"): + secrets = read_json(path, optional=True) + if secrets is None: + raise ValueError(missing) + return validate_secrets(secrets) diff --git a/badge/apps/solar_buddy/controller.py b/badge/apps/solar_buddy/controller.py new file mode 100644 index 00000000..c38f61cf --- /dev/null +++ b/badge/apps/solar_buddy/controller.py @@ -0,0 +1,155 @@ +"""Application state, navigation, freshness, and mode separation.""" + +from advisor import Advisor +from clock import elapsed +from demo import Demo, SCENARIOS +from energy import Mood + +PAGES = ("Buddy", "Energy", "Appliances", "Settings") + + +class App: + def __init__(self, settings, live_factory=None, saved=None, save=None): + self.settings = settings + self.live_factory = live_factory + self.save_callback = save + saved = saved or {} + self.page = 0 + self.selected = 0 + self.detail = False + self.demo_index = saved.get("scenario", 0) + if not isinstance(self.demo_index, int) or not 0 <= self.demo_index < len(SCENARIOS): + self.demo_index = 0 + self.superstar = saved.get("superstar") is True + self.demo_superstar = False + self.mood = Mood(settings) + self.advisor = Advisor(settings) + self.sample = None + self.last_good = None + self.valid = False + self.error = "" + self.status = "OFFLINE" + self.source = None + self.mode = "demo" + self.celebration = None + self.awake_until_start = None + self.hardware_note = "" + self._logged_error = None + initial_mode = saved.get("mode", settings["mode"]) + if initial_mode not in ("demo", "live"): + initial_mode = settings["mode"] + self.set_mode(initial_mode, persist=False) + + def persist(self): + if self.save_callback: + self.save_callback({"mode": self.mode, "scenario": self.demo_index, + "superstar": self.superstar}) + + def invalidate(self, message, status="OFFLINE"): + self.valid = False + self.error = message + self.status = status + self.advisor.invalidate() + self.mood.update(None) + if message != self._logged_error: + print("Solar Buddy: " + message) + self._logged_error = message + + def set_mode(self, mode, persist=True): + if self.source is not None: + self.source.close() + self.source = None + self.mode = mode + self.sample = None + self.last_good = None + self.celebration = None + self.valid = False + self.error = "" + self.advisor.invalidate() + self.mood.update(None) + self.status = "DEMO" if mode == "demo" else "OFFLINE" + if mode == "demo": + self.source = Demo(self.demo_index) + else: + try: + if self.live_factory is None: + raise ValueError("Run the Home Assistant setup script first") + self.source = self.live_factory() + self.error = "Connecting to Home Assistant" + except (OSError, ValueError, ImportError, AttributeError) as exc: + self.invalidate("Live setup: " + str(exc)) + if persist: + self.persist() + + def tick(self, now): + if (self.valid and self.last_good is not None + and elapsed(now, self.last_good) >= self.settings["stale_after_ms"]): + self.invalidate("No fresh reading; advice paused", "STALE") + event = self.source.tick(now) if self.source is not None else None + if event is None: + return + kind, payload = event + if kind != "reading": + self.invalidate(payload, "STALE" if kind == "stale" or "stale" in payload.lower() + else "OFFLINE") + return + self.sample = payload + self.last_good = now + self.valid = True + self.error = "" + self._logged_error = None + self.status = "DEMO" if self.mode == "demo" else "LIVE" + self.mood.update(payload) + self.advisor.update(payload, now) + if payload["surplus_w"] > self.settings["milestone_w"]: + earned = self.demo_superstar if self.mode == "demo" else self.superstar + if not earned: + self.celebration = now + if self.mode == "demo": + self.demo_superstar = True + else: + self.superstar = True + self.persist() + + def input(self, action, now): + if action in ("left", "right"): + self.page = (self.page + (1 if action == "right" else -1)) % len(PAGES) + self.detail = False + elif action == "menu": + self.page = 3 + self.detail = False + elif action == "back": + if self.detail: + self.detail = False + else: + self.page = 0 + elif action in ("up", "down"): + delta = -1 if action == "up" else 1 + if self.page == 2: + self.selected = (self.selected + delta) % len(self.advisor.items) + self.detail = False + elif self.mode == "demo" and self.page in (0, 1): + self.demo_index = (self.demo_index + delta) % len(SCENARIOS) + self.set_mode("demo") + elif action == "select": + if self.page == 2: + self.detail = not self.detail + elif self.page == 3: + self.set_mode("live" if self.mode == "demo" else "demo") + elif self.mode == "live": + if self.source is None: + self.set_mode("live", persist=False) + else: + self.source.refresh() + else: + self.wake(now) + + def wake(self, now): + self.awake_until_start = now + + def is_awake(self, now): + return self.awake_until_start is not None and elapsed(now, self.awake_until_start) < 2500 + + def close(self): + if self.source is not None: + self.source.close() diff --git a/badge/apps/solar_buddy/demo.py b/badge/apps/solar_buddy/demo.py new file mode 100644 index 00000000..658b20ba --- /dev/null +++ b/badge/apps/solar_buddy/demo.py @@ -0,0 +1,46 @@ +"""Deterministic, explicitly simulated scenarios that never use the network.""" + +import math + +from clock import elapsed +from energy import reading + +SCENARIOS = ("Automatic", "Sunny", "Solar peak", "Cloudy", "Busy home", + "Night", "No network", "Missing sensor", "Stale sensor") + + +class Demo: + def __init__(self, scenario=0): + self.scenario = scenario % len(SCENARIOS) + self.last = None + self.started = None + self.active_name = SCENARIOS[self.scenario] + + def tick(self, now): + if self.last is not None and elapsed(now, self.last) < 1000: + return None + if self.started is None: + self.started = now + self.last = now + age = elapsed(now, self.started) + index = self.scenario + if index == 0: + sequence = (1, 2, 3, 4, 5, 6, 1) + index = sequence[(age // 45000) % len(sequence)] + self.active_name = SCENARIOS[index] + if index == 6: + return "error", "Demo: Home Assistant connection lost" + if index == 7: + return "error", "Demo: a configured sensor is unavailable" + if index == 8: + return "stale", "Demo: sensor report is stale" + solar, home = {1: (3400, 2200), 2: (6500, 1500), 3: (1100, 700), + 4: (1200, 4000), 5: (0, 420)}[index] + if solar: + solar += math.sin(age / 6000) * 70 + home += math.sin(age / 8500) * 25 + difference = solar - home + return "reading", reading(solar, max(0, -difference), max(0, difference)) + + def close(self): + return diff --git a/badge/apps/solar_buddy/energy.py b/badge/apps/solar_buddy/energy.py new file mode 100644 index 00000000..55b2c65a --- /dev/null +++ b/badge/apps/solar_buddy/energy.py @@ -0,0 +1,159 @@ +"""Instantaneous power, source freshness, and hysteretic moods; no hardware imports.""" + +import math + +UNIT_FACTORS = {"W": 1.0, "kW": 1000.0, "MW": 1000000.0} + + +def to_watts(state, unit): + if unit not in UNIT_FACTORS: + raise ValueError("Expected instantaneous W, kW, or MW; never Wh or kWh") + if isinstance(state, bool) or state is None: + raise ValueError("Power sensor is not numeric") + try: + value = float(state) * UNIT_FACTORS[unit] + except (ValueError, TypeError): + raise ValueError("Power sensor is unknown, unavailable, or non-numeric") + if not math.isfinite(value) or value < 0: + raise ValueError("Power sensor must be finite and nonnegative") + return value + + +def reading(solar_w, import_w, export_w): + values = [to_watts(value, "W") for value in (solar_w, import_w, export_w)] + solar_w, import_w, export_w = values + house_w = solar_w + import_w - export_w + if house_w < 0: + raise ValueError("Inconsistent sensors: export exceeds solar plus import") + return {"solar_w": solar_w, "import_w": import_w, "export_w": export_w, + "house_w": house_w, "surplus_w": export_w} + + +def _leap(year): + return year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) + + +def _unix(year, month, day, hour, minute, second): + if not 1970 <= year <= 2199 or not 1 <= month <= 12: + raise ValueError("Invalid timestamp date") + lengths = (31, 29 if _leap(year) else 28, 31, 30, 31, 30, + 31, 31, 30, 31, 30, 31) + if (not 1 <= day <= lengths[month - 1] or not 0 <= hour <= 23 + or not 0 <= minute <= 59 or not 0 <= second <= 59): + raise ValueError("Invalid timestamp component") + days = sum(366 if _leap(y) else 365 for y in range(1970, year)) + days += sum(lengths[:month - 1]) + day - 1 + return ((days * 24 + hour) * 60 + minute) * 60 + second + + +def iso_timestamp(value): + if not isinstance(value, str) or len(value) < 20: + raise ValueError("Missing or invalid sensor timestamp") + try: + if (value[4] != "-" or value[7] != "-" or value[10] != "T" + or value[13] != ":" or value[16] != ":"): + raise ValueError("Invalid ISO timestamp") + result = _unix(int(value[:4]), int(value[5:7]), int(value[8:10]), + int(value[11:13]), int(value[14:16]), int(value[17:19])) + tail = value[19:] + if tail.startswith("."): + index = 1 + while index < len(tail) and "0" <= tail[index] <= "9": + index += 1 + if index == 1: + raise ValueError("Invalid timestamp fraction") + tail = tail[index:] + if tail == "Z": + return result + if len(tail) != 6 or tail[0] not in "+-" or tail[3] != ":": + raise ValueError("Timestamp must include a timezone") + hours, minutes = int(tail[1:3]), int(tail[4:6]) + if not 0 <= hours <= 23 or not 0 <= minutes <= 59: + raise ValueError("Invalid timestamp timezone") + offset = (hours * 60 + minutes) * 60 + return result - offset if tail[0] == "+" else result + offset + except (ValueError, IndexError): + raise ValueError("Invalid sensor timestamp") + + +def http_timestamp(value): + months = ("Jan", "Feb", "Mar", "Apr", "May", "Jun", + "Jul", "Aug", "Sep", "Oct", "Nov", "Dec") + try: + fields = value.split() + if len(fields) != 6 or fields[-1] != "GMT": + raise ValueError("Invalid Date header") + hour, minute, second = [int(x) for x in fields[4].split(":")] + return _unix(int(fields[3]), months.index(fields[2]) + 1, + int(fields[1]), hour, minute, second) + except (AttributeError, ValueError, IndexError): + raise ValueError("Missing or invalid HTTP Date; cannot verify sensor freshness") + + +def decode_report(payload, entity, headers): + """Return (watts, seconds since the sensor's last report) for a valid power sensor.""" + if not isinstance(payload, dict) or payload.get("entity_id") != entity: + raise ValueError("Home Assistant returned the wrong sensor") + attrs = payload.get("attributes") + if not isinstance(attrs, dict): + raise ValueError(entity + ": missing attributes") + device_class = attrs.get("device_class") + if device_class not in (None, "power"): + raise ValueError(entity + ": device_class is not power") + value = to_watts(payload.get("state"), attrs.get("unit_of_measurement")) + reported = payload.get("last_reported") or payload.get("last_updated") + age = http_timestamp(headers.get("date")) - iso_timestamp(reported) + if age < -30: + raise ValueError(entity + ": timestamp is ahead of the server clock") + return value, age + + +def check_freshness(entities, ages, max_age_s): + """Judge report ages for solar, grid import, grid export; a leading subset is allowed. + + Integrations such as DSMR only write a meter direction when its value changes, so + the direction that is not flowing (usually 0 W) can stay unreported for hours while + the other updates every second. The two grid directions therefore share one + deadline: a recent report from either proves the meter is alive. + """ + if ages[0] > max_age_s: + raise ValueError(entities[0] + ": stale sensor report") + if len(ages) > 2 and min(ages[1], ages[2]) > max_age_s: + raise ValueError(entities[1] + " and " + entities[2] + ": stale sensor reports") + + +class Mood: + def __init__(self, settings): + self.settings = settings + self.state = "offline" + + def update(self, sample): + if sample is None: + self.state = "offline" + return self.state + cfg = self.settings + if sample["import_w"] > cfg["import_threshold_w"]: + self.state = "sleepy" + return self.state + if self.state == "sleepy" and sample["import_w"] > 0: + return self.state + surplus = sample["surplus_w"] + happy, excited, band = cfg["happy_w"], cfg["excited_w"], cfg["mood_hysteresis_w"] + if self.state in ("offline", "sleepy"): + self.state = ("excited" if surplus >= excited else + "happy" if surplus >= happy else "neutral") + elif self.state == "excited": + if surplus < happy - band: + self.state = "neutral" + elif surplus < excited - band: + self.state = "happy" + elif self.state == "happy": + if surplus >= excited + band: + self.state = "excited" + elif surplus < happy - band: + self.state = "neutral" + elif surplus >= excited + band: + self.state = "excited" + elif surplus >= happy + band: + self.state = "happy" + return self.state diff --git a/badge/apps/solar_buddy/ha_client.py b/badge/apps/solar_buddy/ha_client.py new file mode 100644 index 00000000..b9be6504 --- /dev/null +++ b/badge/apps/solar_buddy/ha_client.py @@ -0,0 +1,122 @@ +"""Read-only Home Assistant polling, with independent Wi-Fi recovery.""" + +import sys + +from clock import elapsed +from energy import check_freshness, decode_report, reading + +if sys.platform == "webassembly": + from browser_http import HttpClient +else: + from http_client import HttpClient + + +class BadgeWifi: + def __init__(self): + import network + self.wlan = network.WLAN(network.STA_IF) + self.wlan.active(True) + self.started = None + self.ssid = None + self.password = None + if not self.wlan.isconnected(): + self._credentials() + + def _credentials(self): + import secrets + self.ssid = getattr(secrets, "WIFI_SSID", "") + self.password = getattr(secrets, "WIFI_PASSWORD", "") + if not self.ssid: + raise ValueError("Configure WIFI_SSID in the badge's existing secrets.py") + + def tick(self, now): + if self.wlan.isconnected(): + self.started = None + return None + if self.ssid is None: + self._credentials() + if self.started is None or elapsed(now, self.started) >= 15000: + if self.started is not None: + self.wlan.disconnect() + self.wlan.connect(self.ssid, self.password) + self.started = now + status = self.wlan.status() + if status < 0: + return "Wi-Fi failed (status %d); retrying" % status + return "Connecting to Wi-Fi" + + +class HomeAssistant: + def __init__(self, settings, credentials, wifi=None, http=None): + self.settings = settings + self.credentials = credentials + self.wifi = wifi + self.http = http or HttpClient(credentials, settings["request_timeout_ms"]) + self.entities = [credentials["entities"][key] for key in ("solar", "import", "export")] + self.last_attempt = None + self.delay = 0 + self.failures = 0 + self.reports = [] + self.active = False + self.last_wifi_error = None + + def _fail(self, message): + self.http.close() + self.active = False + self.reports = [] + self.failures = min(self.failures + 1, 5) + self.delay = min(60000, self.settings["poll_interval_ms"] * (2 ** (self.failures - 1))) + return "error", message + + def tick(self, now): + try: + if self.wifi is not None: + problem = self.wifi.tick(now) + if problem: + self.http.close() + self.active = False + self.reports = [] + self.last_attempt = None + if problem != self.last_wifi_error: + self.last_wifi_error = problem + return "error", problem + return None + self.last_wifi_error = None + if not self.active: + if (self.last_attempt is not None + and elapsed(now, self.last_attempt) < self.delay): + return None + self.last_attempt = now + self.reports = [] + self.http.start("/api/states/" + self.entities[0], now) + self.active = True + return None + response = self.http.step(now) + if response is None: + return None + payload, headers = response + entity = self.entities[len(self.reports)] + self.reports.append(decode_report(payload, entity, headers)) + check_freshness(self.entities, [age for _watts, age in self.reports], + self.settings["sensor_max_age_s"]) + if len(self.reports) < 3: + self.http.start("/api/states/" + self.entities[len(self.reports)], now) + return None + sample = reading(*[watts for watts, _age in self.reports]) + self.active = False + self.failures = 0 + self.delay = self.settings["poll_interval_ms"] + self.last_attempt = now + return "reading", sample + except (OSError, ValueError, TypeError, KeyError) as exc: + message = str(exc).replace(self.credentials["token"], "[redacted]") + return self._fail(message) + + def refresh(self): + self.close() + self.last_attempt = None + + def close(self): + self.http.close() + self.active = False + self.reports = [] diff --git a/badge/apps/solar_buddy/hardware.py b/badge/apps/solar_buddy/hardware.py new file mode 100644 index 00000000..9e0f8195 --- /dev/null +++ b/badge/apps/solar_buddy/hardware.py @@ -0,0 +1,73 @@ +"""Optional motion and white case-light feedback using firmware APIs only.""" + +import math + +from clock import elapsed + + +class Feedback: + def __init__(self, badge): + self.badge = badge + self.motion = callable(getattr(badge, "imu", None)) + self.lights = callable(getattr(badge, "caselights", None)) + self.previous = None + self.gravity = None + self.last_motion = None + self.last_wake = None + self.last_light = None + self.saved_lights = None + self.note = "" + if self.lights: + try: + self.saved_lights = badge.caselights() + except (OSError, ValueError) as exc: + self._disable("lights", exc) + if not self.motion: + self.note = "Shake is unavailable on this runtime" + + def _disable(self, feature, exc): + if feature == "lights": + self.lights = False + else: + self.motion = False + self.note = feature + " unavailable: " + str(exc) + print("Solar Buddy: " + self.note) + + def update(self, app, now): + if self.motion and (self.last_motion is None or elapsed(now, self.last_motion) >= 80): + self.last_motion = now + try: + values = self.badge.imu() + if len(values) != 6: + raise ValueError("Expected six IMU values") + current = tuple(float(value) for value in values[:3]) + if not all(math.isfinite(value) for value in current): + raise ValueError("Invalid IMU sample") + magnitude = math.sqrt(sum(value * value for value in current)) + if self.gravity is None and magnitude > 0: + self.gravity = magnitude + if self.previous is not None and self.gravity: + delta = math.sqrt(sum((a - b) ** 2 for a, b in zip(current, self.previous))) + if (delta > self.gravity * 1.2 and + (self.last_wake is None or elapsed(now, self.last_wake) > 2000)): + app.wake(now) + self.last_wake = now + self.gravity = self.gravity * 0.98 + magnitude * 0.02 + self.previous = current + except (OSError, ValueError, TypeError) as exc: + self._disable("motion", exc) + if self.lights: + age = elapsed(now, app.celebration) if app.celebration is not None else -1 + brightness = 0.35 if 0 <= age < 1250 and (age // 250) % 2 == 0 else 0 + if brightness != self.last_light: + try: + self.badge.caselights(brightness) + self.last_light = brightness + except (OSError, ValueError) as exc: + self._disable("lights", exc) + if self.note: + app.hardware_note = self.note + + def close(self): + if self.lights and self.saved_lights is not None: + self.badge.caselights(*self.saved_lights) diff --git a/badge/apps/solar_buddy/http_client.py b/badge/apps/solar_buddy/http_client.py new file mode 100644 index 00000000..92986d4f --- /dev/null +++ b/badge/apps/solar_buddy/http_client.py @@ -0,0 +1,264 @@ +"""Bounded, incremental HTTP/1.1 GETs; no threads, blocking DNS, or redirects.""" + +import errno +import json +import select +import socket + +from clock import elapsed +from config import parse_url, validate_secrets + +MAX_BODY = 16384 +MAX_HEADERS = 4096 +_BUSY = tuple(set(getattr(errno, name, -9999) for name in + ("EAGAIN", "EWOULDBLOCK", "EINPROGRESS", "EALREADY")) + | {10035, 10036, 10037}) + + +class Response: + def __init__(self): + self.headers = None + self.pending = bytearray() + self.body = bytearray() + self.length = None + self.chunked = False + self.chunk_size = None + self.trailers = False + self.done = False + self.received = 0 + + def feed(self, data): + if self.done and data: + raise ValueError("Unexpected data after HTTP response") + self.received += len(data) + if self.received > MAX_BODY * 3 + MAX_HEADERS: + raise ValueError("HTTP response exceeds the memory limit") + self.pending.extend(data) + if self.headers is None: + end = self.pending.find(b"\r\n\r\n") + if end < 0: + if len(self.pending) > MAX_HEADERS: + raise ValueError("HTTP headers exceed the memory limit") + return + if end > MAX_HEADERS: + raise ValueError("HTTP headers exceed the memory limit") + lines = bytes(self.pending[:end]).decode().split("\r\n") + del self.pending[:end + 4] + status = lines[0].split(" ", 2) + if len(status) < 2 or status[0] not in ("HTTP/1.0", "HTTP/1.1"): + raise ValueError("Invalid HTTP status line") + code = int(status[1]) + if code != 200: + hint = {401: "check token", 403: "sensor access denied", + 404: "check entity ID", 429: "rate limited", + 503: "Home Assistant unavailable"}.get(code, "request failed") + raise ValueError("HTTP %d: %s" % (code, hint)) + self.headers = {} + for line in lines[1:]: + name, separator, value = line.partition(":") + if not separator: + raise ValueError("Malformed HTTP header") + name, value = name.strip().lower(), value.strip() + if name in self.headers and name in ("content-length", "transfer-encoding"): + raise ValueError("Duplicate HTTP framing header") + self.headers[name] = value + encoding = self.headers.get("content-encoding", "identity").lower() + if encoding != "identity": + raise ValueError("Compressed HTTP responses are not supported") + transfer = self.headers.get("transfer-encoding", "").lower() + if transfer not in ("", "chunked"): + raise ValueError("Unsupported HTTP transfer encoding") + self.chunked = transfer == "chunked" + if "content-length" in self.headers: + if self.chunked: + raise ValueError("Ambiguous HTTP response framing") + self.length = int(self.headers["content-length"]) + if not 0 <= self.length <= MAX_BODY: + raise ValueError("HTTP body exceeds the memory limit") + self._consume() + + def _consume(self): + if not self.chunked: + self.body.extend(self.pending) + self.pending = bytearray() + if len(self.body) > MAX_BODY: + raise ValueError("HTTP body exceeds the memory limit") + if self.length is not None: + if len(self.body) > self.length: + raise ValueError("HTTP body exceeds Content-Length") + self.done = len(self.body) == self.length + return + while True: + if self.trailers: + if self.pending[:2] == b"\r\n": + self.done = True + elif b"\r\n\r\n" in self.pending: + self.done = True + elif len(self.pending) > MAX_HEADERS: + raise ValueError("HTTP trailers exceed the memory limit") + return + if self.chunk_size is None: + end = self.pending.find(b"\r\n") + if end < 0: + if len(self.pending) > 128: + raise ValueError("Invalid HTTP chunk header") + return + if end > 128: + raise ValueError("Invalid HTTP chunk header") + size = bytes(self.pending[:end]).split(b";", 1)[0] + if not size or any(ch not in b"0123456789abcdefABCDEF" for ch in size): + raise ValueError("Invalid HTTP chunk size") + self.chunk_size = int(size, 16) + del self.pending[:end + 2] + if self.chunk_size + len(self.body) > MAX_BODY: + raise ValueError("HTTP body exceeds the memory limit") + if self.chunk_size == 0: + self.trailers = True + continue + if len(self.pending) < self.chunk_size + 2: + return + if self.pending[self.chunk_size:self.chunk_size + 2] != b"\r\n": + raise ValueError("Invalid HTTP chunk terminator") + self.body.extend(self.pending[:self.chunk_size]) + del self.pending[:self.chunk_size + 2] + self.chunk_size = None + + def eof(self): + if self.headers is None: + raise ValueError("Connection closed before HTTP headers") + if not self.chunked and self.length is None: + self.done = True + if not self.done: + raise ValueError("Truncated HTTP response") + + def result(self): + if not self.done: + raise ValueError("HTTP response is incomplete") + return json.loads(bytes(self.body).decode("utf-8")), self.headers + + +class HttpClient: + def __init__(self, credentials, timeout_ms): + validate_secrets(credentials) + self.scheme, self.host, self.port, self.prefix, self.authority = parse_url( + credentials["base_url"]) + self.ip = credentials.get("connect_ip", self.host) + self.token = credentials["token"] + self.timeout_ms = timeout_ms + self.sock = None + self.phase = None + self.tls = None + self.ssl_module = None + if self.scheme == "https": + import ssl + self.ssl_module = ssl + self.tls = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + self.tls.verify_mode = ssl.CERT_REQUIRED + ca_file = credentials.get("ca_file") + if ca_file: + self.tls.load_verify_locations(cafile=ca_file) + elif hasattr(self.tls, "load_default_certs"): + self.tls.load_default_certs() + else: + raise ValueError("HTTPS on the badge requires ca_file and a correct RTC") + + def start(self, path, now): + if self.sock is not None: + raise RuntimeError("An HTTP request is already in progress") + self.started = now + self.response = Response() + self.sent = 0 + self.request = ( + "GET %s%s HTTP/1.1\r\nHost: %s\r\nAuthorization: Bearer %s\r\n" + "Accept: application/json\r\nAccept-Encoding: identity\r\n" + "Connection: close\r\nUser-Agent: SolarBuddy/1.0\r\n\r\n" + % (self.prefix, path, self.authority, self.token)).encode() + address = socket.getaddrinfo(self.ip, self.port, socket.AF_INET, socket.SOCK_STREAM)[0] + self.sock = socket.socket(address[0], address[1], address[2]) + try: + self.sock.setblocking(False) + try: + self.sock.connect(address[-1]) + except OSError as exc: + if not self._busy(exc): + raise + self.phase = "connect" + except OSError: + self.close() + raise + + def _busy(self, exc): + if self.ssl_module is not None: + for name in ("SSLWantReadError", "SSLWantWriteError"): + cls = getattr(self.ssl_module, name, None) + if cls is not None and isinstance(exc, cls): + return True + return bool(exc.args) and exc.args[0] in _BUSY + + def _writable(self): + if hasattr(select, "poll"): + poller = select.poll() + poller.register(self.sock, select.POLLOUT) + return bool(poller.poll(0)) + _readable, writable, errors = select.select([], [self.sock], [self.sock], 0) + return bool(writable or errors) + + def step(self, now): + if self.sock is None: + return None + if elapsed(now, self.started) >= self.timeout_ms: + self.close() + raise OSError("Home Assistant request timed out") + try: + if self.phase == "connect": + if not self._writable(): + return None + if hasattr(socket, "SO_ERROR") and hasattr(self.sock, "getsockopt"): + error = self.sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR) + if error: + raise OSError(error, "Connection failed") + if self.tls is not None: + self.sock = self.tls.wrap_socket( + self.sock, server_hostname=self.host, do_handshake_on_connect=False) + self.sock.setblocking(False) + self.phase = "send" + return None + if self.phase == "send": + part = memoryview(self.request)[self.sent:self.sent + 1024] + count = self.sock.write(part) if self.tls else self.sock.send(part) + if count is None: + return None + if count == 0: + raise OSError("Connection closed while sending the request") + self.sent += count + if self.sent == len(self.request): + self.phase = "read" + self.request = b"" + return None + data = self.sock.read(1024) if self.tls else self.sock.recv(1024) + if data is None: + return None + if data: + self.response.feed(data) + else: + self.response.eof() + if self.response.done: + result = self.response.result() + self.close() + return result + except OSError as exc: + if self._busy(exc): + return None + self.close() + raise + except (ValueError, TypeError): + self.close() + raise + return None + + def close(self): + sock, self.sock = self.sock, None + self.phase = None + self.request = b"" + if sock is not None: + sock.close() diff --git a/badge/apps/solar_buddy/icon.png b/badge/apps/solar_buddy/icon.png new file mode 100644 index 0000000000000000000000000000000000000000..76a7f3e9cd7a4fbcad72791c13c1bcb11008cb6f GIT binary patch literal 181 zcmeAS@N?(olHy`uVBq!ia0vp^5+KaM1|%Pp+x`GjHJ&bxAr-fhABc+l_xC>Iu;!3| zx=_G}2U{5CGq&g)P<7wJy5z@%4DJiDz1+gNjAr3~>o-d1t38QkOH!I5GM~qen`5cc zA{_;_LmgWg=GuP{n;^x`oXlu=lxbs9ifZtzXa}E%?vq%Tq*-k0&R!*6R-(UdW!yua ezg9EFBp8l*b!`-wJZTot9SokXelF{r5}E)p)ji7q literal 0 HcmV?d00001 diff --git a/badge/apps/solar_buddy/settings.example.json b/badge/apps/solar_buddy/settings.example.json new file mode 100644 index 00000000..d2ff0807 --- /dev/null +++ b/badge/apps/solar_buddy/settings.example.json @@ -0,0 +1,20 @@ +{ + "mode": "demo", + "poll_interval_ms": 10000, + "request_timeout_ms": 4000, + "stale_after_ms": 45000, + "sensor_max_age_s": 180, + "stability_ms": 30000, + "reserve_w": 100, + "happy_w": 800, + "excited_w": 2000, + "mood_hysteresis_w": 100, + "import_threshold_w": 25, + "milestone_w": 4000, + "appliances": [ + {"name": "Dishwasher", "watts": 1500}, + {"name": "Washing machine", "watts": 2000}, + {"name": "EV charging", "watts": 2300}, + {"name": "Small appliance", "watts": 500} + ] +} diff --git a/badge/apps/solar_buddy/solar_buddy_secrets.example.json b/badge/apps/solar_buddy/solar_buddy_secrets.example.json new file mode 100644 index 00000000..03b036dd --- /dev/null +++ b/badge/apps/solar_buddy/solar_buddy_secrets.example.json @@ -0,0 +1,11 @@ +{ + "base_url": "http://localhost:8123", + "token": "", + "connect_ip": "127.0.0.1", + "entities": { + "solar": "sensor.sg5_0rs_local_total_active_power", + "import": "sensor.electricity_meter_power_consumption", + "export": "sensor.electricity_meter_power_production" + }, + "allow_insecure_http": true +} diff --git a/badge/apps/solar_buddy/ui.py b/badge/apps/solar_buddy/ui.py new file mode 100644 index 00000000..7aa010c6 --- /dev/null +++ b/badge/apps/solar_buddy/ui.py @@ -0,0 +1,228 @@ +"""Shared 320x240 layout for Badgeware, Tkinter, and SVG previews.""" + +from character import Buddy +from clock import elapsed +from controller import PAGES +from demo import SCENARIOS + +BG = "#0d1117" +PANEL = "#161b22" +EDGE = "#30363d" +TEXT = "#e6edf3" +MUTED = "#8b949e" +GREEN = "#3fb950" +GOLD = "#e3b341" +BLUE = "#58a6ff" +RED = "#f85149" + + +def watts(value): + return "%.0f W" % value + + +class BadgePainter: + def __init__(self, screen, shape, color): + self.screen, self.shape, self.color = screen, shape, color + self.width, self.height = screen.width, screen.height + self.pens = {} + + def _pen(self, value): + if value not in self.pens: + self.pens[value] = self.color.rgb( + int(value[1:3], 16), int(value[3:5], 16), int(value[5:7], 16)) + self.screen.pen = self.pens[value] + + def clear(self, value): + self._pen(value) + self.screen.clear() + + def rect(self, x, y, width, height, value): + self._pen(value) + self.screen.shape(self.shape.rectangle(x, y, width, height)) + + def line(self, x1, y1, x2, y2, value): + self._pen(value) + self.screen.line(x1, y1, x2, y2) + + def text(self, value, x, y, pen): + self._pen(pen) + self.screen.text(value, x, y) + + def measure(self, value): + return self.screen.measure_text(value)[0] + + +class UI: + def __init__(self, painter): + self.p = painter + self.buddy = Buddy() + + def fit(self, value, width): + if self.p.measure(value) <= width: + return value + while value and self.p.measure(value + "...") > width: + value = value[:-1] + return value + "..." + + def text(self, value, x, y, pen=TEXT, width=None): + width = self.p.width - x - 10 if width is None else width + self.p.text(self.fit(value, width), x, y, pen) + + def right(self, value, y, pen=TEXT): + self.p.text(value, self.p.width - 12 - self.p.measure(value), y, pen) + + def center(self, value, y, pen=TEXT): + value = self.fit(value, self.p.width - 24) + self.p.text(value, (self.p.width - self.p.measure(value)) / 2, y, pen) + + def wrap(self, value, y, lines=3, pen=RED): + wrapped, current = [], "" + for word in value.split(): + candidate = current + (" " if current else "") + word + if self.p.measure(candidate) > self.p.width - 24 and current: + wrapped.append(current) + current = word + else: + current = candidate + if current: + wrapped.append(current) + if len(wrapped) > lines: + wrapped = wrapped[:lines - 1] + [" ".join(wrapped[lines - 1:])] + for index, line in enumerate(wrapped): + self.text(line, 12, y + index * 14, pen) + + def draw(self, app, now, delta_ms): + self.buddy.advance(delta_ms) + p = self.p + p.clear(BG) + p.rect(0, 0, p.width, 29, PANEL) + self.text("SOLAR BUDDY" if app.page == 0 else PAGES[app.page].upper(), 12, 10, + TEXT, p.width - 155) + status = app.status + if app.mode == "demo" and status != "DEMO": + status = "DEMO/" + status + self.right(status, 10, GOLD if app.mode == "demo" else GREEN if app.valid else RED) + if app.page == 0: + self.dashboard(app, now) + elif app.page == 1: + self.energy(app, now) + elif app.page == 2: + self.appliances(app, now) + else: + self.settings(app, now) + p.line(10, p.height - 21, p.width - 10, p.height - 21, EDGE) + hint = "L/R screens" + if app.page == 2: + hint += " | U/D choose | SEL details" + elif app.page == 3: + hint += " | SEL switch demo/live" + elif app.mode == "demo": + hint += " | U/D scenario | SEL wake" + else: + hint += " | SEL retry | MENU setup" + self.text(hint, 10, p.height - 14, MUTED) + + def dashboard(self, app, now): + self.buddy.draw(self.p, self.p.width / 2, 77, app.mood.state, app.is_awake(now)) + messages = {"excited": "PLENTY OF SUNSHINE!", "happy": "SOME SOLAR TO SPARE", + "neutral": "A LITTLE SOLAR AVAILABLE", "sleepy": "USING GRID POWER", + "offline": "WHERE DID MY SUNSHINE GO?"} + message = messages[app.mood.state] + if app.valid and app.sample["surplus_w"] < 100 and app.mood.state == "neutral": + message = "LITTLE OR NO SOLAR TO SPARE" + self.center(message, 122, GOLD if app.valid else MUTED) + self.metric("Solar production", app, "solar_w", 145, GOLD) + self.metric("Home consumption", app, "house_w", 163, TEXT) + key = "import_w" if app.valid and app.sample["import_w"] > 0 else "export_w" + self.metric("Grid import" if key == "import_w" else "Grid export", + app, key, 181, RED if key == "import_w" else GREEN) + self.p.rect(10, 199, self.p.width - 20, 17, PANEL) + if not app.valid: + advice = app.error or "Waiting for a fresh reading" + else: + ready = [item["name"] for index, item in enumerate(app.advisor.items) + if app.advisor.statuses[index] == "READY"] + advice = ready[0] + ": READY" if ready else "Appliances: wait for stable surplus" + self.text(advice, 16, 203, GREEN if "READY" in advice else MUTED, + self.p.width - 32) + + def metric(self, label, app, key, y, pen): + self.text(label, 14, y, MUTED, self.p.width - 115) + self.right(watts(app.sample[key]) if app.valid else "--", y, pen) + + def energy(self, app, now): + for label, key, y, pen in ( + ("Solar production", "solar_w", 44, GOLD), + ("Grid import", "import_w", 66, RED), + ("Grid export / surplus", "export_w", 88, GREEN), + ("House consumption", "house_w", 110, TEXT)): + self.metric(label, app, key, y, pen) + self.text("Home = solar + import - export", 12, 141, BLUE) + if app.last_good is not None: + self.text("Last good sample: %ds ago" % (elapsed(now, app.last_good) // 1000), + 12, 160, MUTED) + if app.mode == "demo": + scenario = app.source.active_name if app.source else SCENARIOS[app.demo_index] + self.text("SIMULATED: " + scenario, 12, 179, GOLD) + else: + self.text("Live REST data | no battery assumed", 12, 179, MUTED) + if not app.valid: + self.text(app.error, 12, 199, RED) + else: + self.text("Read-only advice. No appliance control.", 12, 199, MUTED) + + def appliances(self, app, now): + if app.detail: + item = app.advisor.items[app.selected] + self.text(item["name"], 12, 44, BLUE) + for label, value, y in ( + ("Estimated load", watts(item["watts"]), 71), + ("Available export", watts(app.sample["surplus_w"]) if app.valid else "--", 93), + ("Safety reserve", watts(app.settings["reserve_w"]), 115), + ("Status", app.advisor.statuses[app.selected], 137)): + self.text(label, 12, y, MUTED, self.p.width - 110) + self.right(value, y, GREEN if value == "READY" else TEXT) + remaining = app.advisor.remaining_s(app.selected, now) + note = ("Need more surplus" if remaining is None else + "Stable evidence: %ds remaining" % remaining if remaining else + "Waiting for next sample" if app.advisor.statuses[app.selected] != "READY" + else "Stable surplus observed") + self.text(note if app.valid else "Need fresh sensor readings", 12, 164, GOLD) + self.text("One appliance at a time; loads vary.", 12, 189, MUTED) + self.text("Not a solar-only cycle guarantee.", 12, 204, MUTED) + return + first = (app.selected // 4) * 4 + for row, index in enumerate(range(first, min(first + 4, len(app.advisor.items)))): + item = app.advisor.items[index] + y = 41 + row * 37 + self.p.rect(10, y, self.p.width - 20, 31, "#1c2d3f" if index == app.selected else PANEL) + self.text(item["name"], 17, y + 5, TEXT, self.p.width - 116) + status = app.advisor.statuses[index] + self.right(status, y + 5, GREEN if status == "READY" else GOLD if status == "WAIT" else MUTED) + self.text(watts(item["watts"]) + " estimated", 17, y + 18, MUTED) + self.text("Each READY is an individual recommendation.", 12, 193, MUTED) + self.text("Recheck after starting any appliance.", 12, 207, MUTED) + + def settings(self, app, now): + self.text("Mode: " + ("SIMULATED - no Wi-Fi" if app.mode == "demo" else "LIVE - read only"), + 12, 42, GOLD if app.mode == "demo" else GREEN) + self.text("Polling: %ds | stale after: %ds" % ( + app.settings["poll_interval_ms"] // 1000, app.settings["stale_after_ms"] // 1000), + 12, 62, MUTED) + self.text("READY needs %ds + %.0f W reserve" % ( + app.settings["stability_ms"] // 1000, app.settings["reserve_w"]), 12, 82, MUTED) + self.text("Sensor report limit: %ds" % app.settings["sensor_max_age_s"], 12, 102, MUTED) + earned = app.demo_superstar if app.mode == "demo" else app.superstar + self.text("Solar Superstar: " + ("earned" if earned else "> %.0f W export" % + app.settings["milestone_w"]), 12, 122, + GOLD if earned else MUTED) + if app.mode == "demo": + self.text("Scenario: " + SCENARIOS[app.demo_index], 12, 142, BLUE) + else: + self.text("Credentials kept outside the app folder", 12, 142, BLUE) + note = app.error or app.hardware_note + if note: + self.wrap(note, 165, lines=3, pen=RED if app.error else MUTED) + else: + self.text("SELECT toggles live/demo.", 12, 170, TEXT) + self.text("HOME returns to the badge launcher.", 12, 191, MUTED)