diff --git a/Makefile b/Makefile index 30ab93f..3b4b937 100644 --- a/Makefile +++ b/Makefile @@ -3,7 +3,7 @@ # The workspace root delegates to these (see ../Makefile). .PHONY: dev-utils dev dev-local up down update install dev-backend dev-frontend \ - generate-client seed-example seed-demo test test-backend test-frontend soak bench-startup lint lint-backend \ + generate-client seed-example seed-demo seed-hosted-demo test test-backend test-frontend soak bench-startup lint lint-backend \ lint-frontend umami clean help COMPOSE_ROOT := $(CURDIR) @@ -97,6 +97,20 @@ seed-example: ## Seed the querying-chart example (needs a running stack + Influ seed-demo: ## Seed the training-run example: a batch flow and its dashboard cd backend && uv run python ../scripts/seed_demo_training.py +# Operators of the hosted demo only — NOT part of any deployment, and nothing a +# self-hosted instance needs. It wipes and recreates its three flows and its +# dashboard, so re-running it is how the public demo is reset. +# `uv run` does not read .env, so the credentials are lifted out of it here; +# anything already exported wins, which is how a remote instance is targeted: +# API_URL=https://api.example.com make seed-hosted-demo +seed-hosted-demo: ## Seed the hosted demo panel (operators only; API_URL selects the instance) + @user=$$(grep -E '^FIRST_SUPERUSER=' $(COMPOSE_ROOT)/.env 2>/dev/null | head -1 | cut -d= -f2-); \ + pass=$$(grep -E '^FIRST_SUPERUSER_PASSWORD=' $(COMPOSE_ROOT)/.env 2>/dev/null | head -1 | cut -d= -f2-); \ + cd backend && \ + FIRST_SUPERUSER="$${FIRST_SUPERUSER:-$$user}" \ + FIRST_SUPERUSER_PASSWORD="$${FIRST_SUPERUSER_PASSWORD:-$$pass}" \ + uv run python ../scripts/seed_demo.py + # ── Testing ─────────────────────────────────────────────────────── test: test-backend test-frontend ## Run all tests (backend + frontend) diff --git a/scripts/seed_demo.py b/scripts/seed_demo.py new file mode 100644 index 0000000..e12cc5e --- /dev/null +++ b/scripts/seed_demo.py @@ -0,0 +1,1297 @@ +#!/usr/bin/env python +"""Seed the hosted demo: a house panel, the flows behind it, and a batch run. + +This is an **operational script for the hosted instance**, not part of any +deployment. Nothing installs it, nothing calls it on first run, and a +self-hosted fluksio never sees it — it exists so whoever operates the public +demo can point it at that instance and get the same panel back every time. + +What it builds:: + + home the house: readings, the controls, the week ahead + home_history answers the temperature chart's questions + pv_model a batch flow that fits the array's yield curve + demo one dashboard, three sections, every widget type + +The panel is a small-house energy and climate display — the thing a person +would actually hang in a hallway — and it is also the widget gallery: all +fifteen types are on it, including the variants no editor field can write +(a switch drawn as a button, a segmented dropdown, a bar with a nested +reading, an icon ladder with labels, fixed chart axes, decimal places). + +It **wipes and recreates**. The three flows and the dashboard are deleted +first, which also drops the values and history they left behind, so running it +twice gives the same panel — and a scheduled reset is just another run. + +Nothing here needs a database, a broker or a connector package. The querying +chart is answered by a plain Python node, which is the whole point of the +exchange: the widget asks for a window and draws what comes back, and what +serves it is the flow's business. + +Run it against a stack that is already up:: + + make -C app seed-hosted-demo # local + API_URL=https://api.example.com make -C app seed-hosted-demo + +Environment: + API_URL default http://api.localhost; https works + FIRST_SUPERUSER, FIRST_SUPERUSER_PASSWORD +""" + +from __future__ import annotations + +import os +import sys +import time +from typing import Any + +import httpx + +API = os.environ.get("API_URL", "http://api.localhost") +EMAIL = os.environ.get("FIRST_SUPERUSER", "") +PASSWORD = os.environ.get("FIRST_SUPERUSER_PASSWORD", "") + +HOUSE = "home" +HISTORY = "home_history" +MODEL = "pv_model" +PANEL = "demo" + +#: The alert channel this seed owns. Anything else configured is left alone. +CHANNEL = "demo_panel" + +#: What the notification widget says when nothing has gone wrong. +ALL_QUIET = { + "title": "Engine healthy", + "body": "Nothing has failed since this panel was seeded.", + "severity": "info", +} + + +def msg(flow: str, name: str) -> str: + """A message name as the engine qualifies it.""" + return f"{flow}.{name}" + + +# ----------------------------------------------------------------------------- +# home — the house +# ----------------------------------------------------------------------------- + +HOUSE_SOURCE = '''"""The house, read every five seconds. + +Nothing here talks to a device: the readings are a plausible curve over the +clock, so the panel has something to show wherever it is stood up. In a real +installation this node is an MQTT or Modbus one and everything downstream of it +is unchanged — which is what it means for the message to be the wiring. + +The three control inputs do not wake this node (``trigger: false``); they are +read on the next tick, which is what a setpoint is for. +""" + +import math +import time + +#: What the array does at noon in June, in watts. +PEAK_PV = 6400.0 +#: What the house draws with nothing switched on. +STANDBY = 320.0 +#: Where each mode puts the room, relative to the slider. +MODES = {"eco": -1.5, "comfort": 0.0, "boost": 1.5} + + +def _sun(hour): + """How much of the array's peak the sun is worth at this hour.""" + if hour < 6.5 or hour > 20.5: + return 0.0 + return math.sin((hour - 6.5) / 14.0 * math.pi) ** 1.4 + + +def _peak(hour, at, width): + """A meal, or anything else that happens around one time of day.""" + return math.exp(-((hour - at) ** 2) / (2 * width * width)) + + +def process(tick, mode="comfort", away=False, setpoint=21.0, params=None): + clock = time.localtime() + hour = clock.tm_hour + clock.tm_min / 60.0 + clock.tm_sec / 3600.0 + seconds = time.time() + + # A fast wobble and a slow drift, so a window of any width has some shape. + wobble = math.sin(seconds / 210.0 * 2 * math.pi) + drift = math.sin(seconds / 5400.0 * 2 * math.pi) + + pv = PEAK_PV * _sun(hour) * (0.84 + 0.16 * wobble) + + # What the controls are worth. Away drops the room and stops the + # appliances; the mode moves the target the slider set. + target = float(setpoint) + MODES.get(str(mode), 0.0) - (3.0 if away else 0.0) + + outdoor = 11.5 + 7.5 * math.sin((hour - 9.5) / 24.0 * 2 * math.pi) + 0.6 * drift + indoor = target - 0.3 + 0.5 * wobble + humidity = 55.0 + 5.0 * drift - 0.9 * (indoor - 21.0) + + kitchen = 0.0 + if not away: + kitchen = 1400.0 * _peak(hour, 7.5, 0.8) + 1900.0 * _peak(hour, 18.5, 1.1) + heating = max(0.0, target - outdoor) * 95.0 * (0.55 if away else 1.0) + load = STANDBY + kitchen + heating + + # The battery fills through the afternoon and is drawn down overnight. + battery = 58.0 - 34.0 * math.cos((hour - 5.0) / 24.0 * 2 * math.pi) + + return { + "pv_watts": round(pv, 1), + "load_watts": round(load, 1), + "grid_watts": round(load - pv, 1), + "battery": round(battery, 1), + "indoor": round(indoor, 2), + "outdoor": round(outdoor, 2), + "humidity": round(humidity, 1), + } +''' + +METER_SOURCE = '''"""What the meter says: the cost, and how much of it the roof paid. + +The tariff is a message like any other, so the field on the dashboard changes +this the moment it is committed. Exporting costs nothing, which is why only the +import is charged for. + +``self_use_kw`` is the part of the house's draw the roof covered, which is a +share of it by construction — that is what lets the panel draw one inside the +other and have the picture mean something. +""" + + +def process(pv_kw, load_kw, grid_kw, tariff=32.0, params=None): + return { + "cost_now": round(max(0.0, grid_kw) * float(tariff) / 100.0, 2), + "self_use_kw": round(min(pv_kw, load_kw), 2), + } +''' + +PLAN_SOURCE = '''"""The week ahead: the sky, and what is written in the calendar. + +Both are lists of records, which is the shape the forecast and agenda widgets +read — every column is ``{label, icon, value}``, every entry ``{title, ts}``. +The flow decides what a day is called; the widget only draws it. A real +installation puts an HTTP node where these constants are and changes nothing +else. + +Run once when the flow starts and again at six every morning, from a cron on +the inject beside it. The current climate is read but does not wake it: a +forecast is a daily thing. +""" + +import time + +DAY_S = 86400 + +#: A week of sky, and how far each day sits from what it is doing now. +WEATHER = [ + ("sun", "primary", 2.0), + ("sun", "primary", 1.0), + ("cloudy", "default", -1.0), + ("cloud-rain", "primary", -3.0), + ("wind", "muted", -2.0), + ("cloud-drizzle", "default", -1.5), + ("sun", "primary", 0.5), +] + +#: The household calendar: days from today, hour, minute, what it is, all-day. +DIARY = [ + (0, 18, 30, "Dishwasher on the night tariff", False), + (1, 8, 0, "Chimney sweep", False), + (2, 0, 0, "Bin day — paper", True), + (4, 19, 0, "Read the meter", False), + (6, 0, 0, "Service the heat pump", True), +] + + +def process(day_start, climate=None, params=None): + now = time.time() + # Anchored on what it is doing outside, once the flow has got that far; the + # first plan after a cold start has nothing to anchor to yet. + outside = float((climate or {}).get("outdoor") or 12.0) + midnight = time.mktime(time.localtime(now)[:3] + (0, 0, 0, 0, 0, -1)) + + forecast = [] + for index, (icon, colour, delta) in enumerate(WEATHER): + when = time.localtime(midnight + index * DAY_S) + forecast.append( + { + "label": "Today" if index == 0 else time.strftime("%a", when), + "icon": icon, + "color": colour, + "value": "{:.0f}°".format(outside + delta), + } + ) + + agenda = [ + { + "title": title, + "ts": midnight + days * DAY_S + hour * 3600 + minute * 60, + "all_day": all_day, + } + for days, hour, minute, title, all_day in DIARY + ] + return {"forecast": forecast, "agenda": agenda} +''' + +HOUSE_INPUTS = [ + # What the controls on the panel write to. The initial values are what the + # dashboard reads back before anyone touches anything, which is why the + # segmented picker opens on "Comfort" rather than on nothing. + {"spec": {"name": "mode", "dtype": "str"}, "initial": "comfort"}, + {"spec": {"name": "away", "dtype": "bool"}, "initial": False}, + {"spec": {"name": "setpoint", "dtype": "float"}, "initial": 21.0}, + {"spec": {"name": "tariff", "dtype": "float"}, "initial": 32.0}, + # False rather than nothing: a declared message with no starting value is + # a node that can never run, and the engine says so on the canvas. + {"spec": {"name": "boost", "dtype": "bool"}, "initial": False}, + # Where the engine's own failures land. Declared here rather than appearing + # from nowhere: a dashboard channel writes to a message a flow owns. + {"spec": {"name": "engine_alert", "dtype": "record"}, "initial": ALL_QUIET}, +] + +HOUSE_NODES = [ + { + "id": "every_5s", + "type": "inject", + "title": "Every five seconds", + "params": {"interval": 5, "at_start": True, "payload": 1}, + "provides": [{"name": "tick", "dtype": "float"}], + }, + { + "id": "at_six", + "type": "inject", + "title": "Six every morning", + # A schedule rather than an interval: the plan below is a daily thing. + "params": {"cron": "0 6 * * *", "at_start": True, "start_delay": 2.0}, + "provides": [{"name": "day_start", "dtype": "float"}], + }, + { + "id": "house", + "type": "python", + "title": "Read the house", + "requires": [ + {"name": "tick", "dtype": "float"}, + # Read on the next tick, never a reason to run: a setpoint is a + # standing instruction, not an event. + {"name": "mode", "dtype": "str", "trigger": False}, + {"name": "away", "dtype": "bool", "trigger": False}, + {"name": "setpoint", "dtype": "float", "trigger": False}, + ], + "provides": [ + {"name": "pv_watts", "dtype": "float"}, + {"name": "load_watts", "dtype": "float"}, + {"name": "grid_watts", "dtype": "float"}, + {"name": "battery", "dtype": "float"}, + {"name": "indoor", "dtype": "float"}, + # The probe outside is worth reading once a minute; everything + # downstream of it is paced by this rather than by the tick. + {"name": "outdoor", "dtype": "float", "interval": 60}, + {"name": "humidity", "dtype": "float"}, + ], + }, + { + "id": "to_kw", + "type": "change", + "title": "Watts to kilowatts", + "params": {"scale": 0.001, "round_to": 2}, + "requires": [ + {"name": "pv_watts", "dtype": "float"}, + {"name": "load_watts", "dtype": "float"}, + {"name": "grid_watts", "dtype": "float"}, + ], + "provides": [ + {"name": "pv_kw", "dtype": "float"}, + {"name": "load_kw", "dtype": "float"}, + {"name": "grid_kw", "dtype": "float"}, + ], + }, + { + "id": "meter", + "type": "python", + "title": "Read the meter", + "requires": [ + {"name": "pv_kw", "dtype": "float"}, + {"name": "load_kw", "dtype": "float"}, + {"name": "grid_kw", "dtype": "float"}, + {"name": "tariff", "dtype": "float", "trigger": False}, + ], + "provides": [ + {"name": "cost_now", "dtype": "float"}, + {"name": "self_use_kw", "dtype": "float"}, + ], + }, + { + "id": "settle", + "type": "rbe", + "title": "Only when it moves", + "params": {"deadband": 0.1}, + "requires": [{"name": "indoor", "dtype": "float"}], + "provides": [{"name": "indoor_stable", "dtype": "float"}], + }, + { + "id": "climate", + "type": "join", + "title": "One climate reading", + # Synchronous by nature: every input has to be fresh, so this runs at + # the pace of the slowest of them — the minute-limited probe outside. + "params": {"mode": "object"}, + "requires": [ + {"name": "indoor_stable", "dtype": "float"}, + {"name": "outdoor", "dtype": "float"}, + {"name": "humidity", "dtype": "float"}, + ], + "provides": [{"name": "climate", "dtype": "record"}], + }, + { + "id": "plan", + "type": "python", + "title": "The week ahead", + "requires": [ + {"name": "day_start", "dtype": "float"}, + {"name": "climate", "dtype": "record", "trigger": False}, + ], + "provides": [ + {"name": "forecast", "dtype": "list", "item": "record"}, + {"name": "agenda", "dtype": "list", "item": "record"}, + ], + }, + { + "id": "water", + "type": "trigger", + "title": "Boost, then let it go", + # The whole shape of a trigger: on now, off again once things have been + # quiet for a quarter of an hour. A second press starts the wait over. + "params": {"first": True, "then": False, "wait": 900, "extend": True}, + "requires": [{"name": "boost", "dtype": "bool"}], + "provides": [{"name": "hot_water", "dtype": "bool"}], + }, + { + "id": "water_label", + "type": "change", + "title": "Say it in words", + "params": {"mapping": {"True": "Heating water", "False": "Idle"}}, + "requires": [{"name": "hot_water", "dtype": "bool"}], + "provides": [{"name": "water_state", "dtype": "str"}], + }, +] + +# ----------------------------------------------------------------------------- +# home_history — the querying chart's other half +# ----------------------------------------------------------------------------- + +HISTORY_SOURCE = '''"""Answer a chart's question about the last however-long. + +There is no database in this demo, so the window is computed the same way the +house's readings are — from the clock. The part worth looking at is the shape of +the exchange rather than where the numbers came from: the chart publishes the +window and the resolution it wants, this answers with a series, and the answer +carries the window it was computed for. A widget throws away an answer to a +different question, which is what makes the range picker trustworthy. + +The target line comes from the *other* flow — ``home.setpoint``, the message the +slider on the panel publishes — so moving the slider redraws this chart too. +""" + +import math +import time + + +def _outdoor(when): + """The same curve the house node reads, evaluated anywhere in the past.""" + clock = time.localtime(when) + hour = clock.tm_hour + clock.tm_min / 60.0 + clock.tm_sec / 3600.0 + return ( + 11.5 + + 7.5 * math.sin((hour - 9.5) / 24.0 * 2 * math.pi) + + 0.6 * math.sin(when / 5400.0 * 2 * math.pi) + ) + + +def process(chart_request, setpoint=21.0, params=None): + span = int(chart_request["range_s"]) + every = max(1, int(chart_request["interval_s"])) + now = int(time.time()) + stamps = list(range(now - span, now + 1, every)) + target = float(setpoint) + + def line(label, at): + return {"label": label, "points": [[ts, round(at(ts), 2)] for ts in stamps]} + + return { + "climate_series": { + # The echo the widget matches against its own request. Without it + # every answer looks like an answer to whatever was last asked. + "range_s": chart_request["range_s"], + "interval_s": chart_request["interval_s"], + "lines": [ + line( + "Inside", + lambda ts: target - 0.3 + 0.4 * math.sin(ts / 5400.0 * 2 * math.pi), + ), + line("Outside", _outdoor), + line("Target", lambda ts: target), + ], + } + } +''' + +HISTORY_INPUTS = [ + # The request arrives from the panel, not from a node upstream. Saying so + # is what stops the canvas reporting this node as waiting on something + # nothing provides. The initial is the widget's own default window, so the + # flow has an answer ready before anyone opens the dashboard. + { + "spec": {"name": "chart_request", "dtype": "record"}, + "initial": {"range_s": 3600, "interval_s": 60}, + } +] + +HISTORY_NODES = [ + { + "id": "answer", + "type": "python", + "title": "Draw the window", + "requires": [ + {"name": "chart_request", "dtype": "record"}, + # Another flow's message, named in full. This is the only thing + # that makes the canvas draw the two flows as related. + {"name": msg(HOUSE, "setpoint"), "dtype": "float"}, + ], + "provides": [{"name": "climate_series", "dtype": "series"}], + } +] + +# ----------------------------------------------------------------------------- +# pv_model — the batch half +# +# The same flow `scripts/seed_demo_training.py` builds, dressed for this panel: +# a supervised fit of the array's yield against normalised irradiance. Nothing +# structural is different, because the structure is the demonstration — a batch +# flow, a streaming output, artifacts between the stages, and a node bound to a +# GPU worker that falls back to the engine when there is none. +# ----------------------------------------------------------------------------- + +PREPARE_SOURCE = '''"""Make the training set out of the logged readings, as an artifact. + +Runs on the engine: no device, so it goes to the local worker pool. Only the +standard library, because the engine's venv is its own and this node has no +business asking it for anything. + +The samples are a fortnight of the array's own history — irradiance as a +deviation from the mean, and what the inverter made of it. +""" + +import json +import random + +import fluksio + +#: What the array is actually worth, and what the fit has to recover. +KW_PER_UNIT = 3.0 +KW_AT_MEAN_LIGHT = 2.0 + + +def process(seed, noise, samples, params): + rng = random.Random(int(seed)) + rows = [ + [x, KW_PER_UNIT * x + KW_AT_MEAN_LIGHT + rng.gauss(0.0, float(noise))] + for x in (rng.uniform(-1.0, 1.0) for _ in range(int(samples))) + ] + payload = json.dumps( + { + "rows": rows, + "truth": {"slope": KW_PER_UNIT, "intercept": KW_AT_MEAN_LIGHT}, + } + ).encode() + # Data between stages is an artifact, not a message: the reference is what + # travels, and it stays valid on whichever machine opens it. + return { + "dataset": fluksio.save_artifact(payload, "readings.json", "application/json"), + "sample_count": len(rows), + } +''' + +TRAIN_SOURCE = '''"""Fit the yield curve, publishing the loss as it goes. + +A generator, so this node produces values over time: every ``yield`` is a dict +keyed by output port and is published the moment it happens, and what the +function returns at the end is the node's result. Nothing here logs anything — +the loss is an output of the graph, which is why a chart can bind to it. + +Bound to ``device: gpu``. With a worker carrying that label attached it runs +there; without one it falls back to the engine, and ``trained_on`` says which +happened. numpy is used when the machine it landed on has it, which is how the +two environments tell themselves apart. +""" + +import json +import platform +import time + +import fluksio + +try: + import numpy as np +except ImportError: # The engine's own venv has no numpy; a GPU box will. + np = None + + +def process(dataset, learning_rate, epochs, pace, params): + with open(fluksio.load_artifact(dataset)) as handle: + data = json.load(handle) + + xs = [row[0] for row in data["rows"]] + ys = [row[1] for row in data["rows"]] + count = len(xs) + total = int(epochs) + rate = float(learning_rate) + slope, intercept = 0.0, 0.0 + + if np is not None: + axis_x, axis_y = np.array(xs), np.array(ys) + + def report(epoch): + # Published from inside a helper, where a yield cannot reach — the + # shape a training framework's callback has. Same port, same checking. + fluksio.emit(progress=round(100.0 * (epoch + 1) / total, 1)) + + loss = 0.0 + for epoch in range(total): + if np is not None: + error = (slope * axis_x + intercept) - axis_y + loss = float((error**2).mean()) + slope -= rate * float((2 * error * axis_x).mean()) + intercept -= rate * float((2 * error).mean()) + else: + error = [slope * x + intercept - y for x, y in zip(xs, ys)] + loss = sum(e * e for e in error) / count + slope -= rate * sum(2 * e * x for e, x in zip(error, xs)) / count + intercept -= rate * sum(2 * e for e in error) / count + + yield {"loss": round(loss, 6)} + report(epoch) + # Only so the curve is watchable; a real epoch takes as long as it takes. + if float(pace) > 0: + time.sleep(float(pace)) + + weights = json.dumps({"slope": slope, "intercept": intercept}).encode() + return { + "weights": fluksio.save_artifact(weights, "weights.json", "application/json"), + "final_loss": round(loss, 6), + "trained_on": "{} ({})".format( + platform.node(), + "numpy " + np.__version__ if np is not None else "pure python", + ), + } +''' + +EVALUATE_SOURCE = '''"""Score the fit, back on the engine. + +Opens two artifacts: the readings this run prepared and the weights the +training produced — which may have been written on another machine entirely. A +reference names content, so where it came from does not matter. +""" + +import json + +import fluksio + + +def process(weights, dataset, params): + with open(fluksio.load_artifact(weights)) as handle: + fit = json.load(handle) + with open(fluksio.load_artifact(dataset)) as handle: + data = json.load(handle) + + rows = data["rows"] + truth = data["truth"] + mean = sum(y for _x, y in rows) / len(rows) + total = sum((y - mean) ** 2 for _x, y in rows) + residual = sum((y - (fit["slope"] * x + fit["intercept"])) ** 2 for x, y in rows) + r2 = 1.0 - residual / total if total else 0.0 + + good = r2 > 0.9 + return { + "accuracy": round(100.0 * max(0.0, r2), 2), + "report": { + "title": "{:.2f} kW per unit of light".format(fit["slope"]), + "body": ( + "The array is worth {} kW per unit and {} kW at mean light. " + "R² = {:.4f} over {} readings.".format( + truth["slope"], truth["intercept"], r2, len(rows) + ) + ), + "severity": "info" if good else "warning", + }, + } +''' + +MODEL_NODES = [ + { + "id": "prepare", + "type": "python", + "title": "Take the readings", + "requires": [ + {"name": "seed", "dtype": "int"}, + {"name": "noise", "dtype": "float"}, + {"name": "samples", "dtype": "int"}, + ], + "provides": [ + {"name": "dataset", "dtype": "artifact"}, + {"name": "sample_count", "dtype": "int"}, + ], + }, + { + "id": "train", + "type": "python", + "title": "Fit the curve", + # Runs on a worker carrying this label; falls back here without one, + # which is what makes the demo work before a GPU box is attached. + "device": "gpu", + "device_policy": "prefer", + # Generous, and an *idle* timeout: a node that keeps publishing keeps + # its deadline reset, so this is how long it may go quiet. + "timeout": 300, + "requires": [ + {"name": "dataset", "dtype": "artifact"}, + {"name": "learning_rate", "dtype": "float"}, + {"name": "epochs", "dtype": "int"}, + {"name": "pace", "dtype": "float"}, + ], + "provides": [ + # The curve. `stream` says this port publishes repeatedly while the + # node runs, and is what makes a run keep every value it takes. + {"name": "loss", "dtype": "float", "stream": True}, + {"name": "progress", "dtype": "float", "stream": True}, + {"name": "weights", "dtype": "artifact"}, + {"name": "final_loss", "dtype": "float"}, + {"name": "trained_on", "dtype": "str"}, + ], + }, + { + "id": "evaluate", + "type": "python", + "title": "Score it", + "requires": [ + {"name": "weights", "dtype": "artifact"}, + {"name": "dataset", "dtype": "artifact"}, + ], + "provides": [ + {"name": "report", "dtype": "record"}, + {"name": "accuracy", "dtype": "float"}, + ], + }, +] + +#: The run's parameters, with the values a run gets when it names none. +MODEL_INPUTS = [ + {"spec": {"name": "seed", "dtype": "int"}, "initial": 1}, + {"spec": {"name": "noise", "dtype": "float"}, "initial": 0.25}, + {"spec": {"name": "samples", "dtype": "int"}, "initial": 200}, + {"spec": {"name": "learning_rate", "dtype": "float"}, "initial": 0.2}, + {"spec": {"name": "epochs", "dtype": "int"}, "initial": 40}, + # Only so a human can watch the curve arrive; set it to 0 for a sweep. + {"spec": {"name": "pace", "dtype": "float"}, "initial": 0.15}, +] + +#: What a run reports as its result. Everything else the flow computed stays in +#: the run's own state and is dropped with it. +MODEL_OUTPUTS = ["report", "accuracy", "final_loss", "trained_on", "sample_count"] + + +# ----------------------------------------------------------------------------- +# demo — the panel +# ----------------------------------------------------------------------------- + +#: Sixteen columns on a 27" landscape screen, said out loud rather than left to +#: the defaults: an arrangement belongs to the panel it will hang on. +COLUMNS = 16 +CANVAS = (2560, 1600) + + +def at(x: int, y: int, w: int, h: int) -> dict[str, Any]: + """Where a widget sits. Only ``lg`` is read; the client derives the rest. + + Every widget says where it goes: the grid ignores placement entirely when + nothing has been placed, and lays the page out itself instead. + """ + return {"lg": {"x": x, "y": y, "w": w, "h": h}} + + +NOW_WIDGETS = [ + { + "id": "wall", + "type": "clock", + "title": "", + "layout": at(0, 0, 3, 2), + # Binds nothing at all: it reads the wall, like a clock. + "config": {}, + }, + { + "id": "solar", + "type": "icon", + "title": "Solar", + "layout": at(3, 0, 2, 2), + "config": { + "message": msg(HOUSE, "pv_kw"), + "dtype": "float", + # A ladder, read top down: the first row whose threshold the + # reading has passed wins. Every row names itself in words as well, + # because a colour is not a label. + "rules": [ + {"at": 4.0, "icon": "sun", "color": "primary", "label": "Strong"}, + {"at": 1.5, "icon": "cloudy", "color": "default", "label": "Good"}, + {"at": 0.2, "icon": "cloud-fog", "color": "muted", "label": "Weak"}, + {"at": 0.0, "icon": "moon", "color": "muted", "label": "Dark"}, + ], + # Drawn when nothing has arrived yet, or nothing matched. + "icon": "house", + }, + }, + { + "id": "outside", + "type": "stat", + "title": "Outside", + "layout": at(5, 0, 3, 2), + "config": { + "message": msg(HOUSE, "outdoor"), + "dtype": "float", + "precision": 1, + "unit": "°C", + }, + }, + { + "id": "inside", + "type": "stat", + "title": "Inside", + "layout": at(8, 0, 3, 2), + "config": { + "message": msg(HOUSE, "indoor"), + "dtype": "float", + "precision": 1, + "unit": "°C", + }, + }, + { + "id": "damp", + "type": "stat", + "title": "Humidity", + "layout": at(11, 0, 2, 2), + "config": { + "message": msg(HOUSE, "humidity"), + "dtype": "float", + "precision": 0, + "unit": "%", + }, + }, + { + "id": "water", + "type": "stat", + "title": "Hot water", + "layout": at(13, 0, 3, 2), + "config": {"message": msg(HOUSE, "water_state"), "dtype": "str"}, + }, + { + "id": "week", + "type": "forecast", + "title": "The week", + "layout": at(0, 2, 6, 2), + "config": {"message": msg(HOUSE, "forecast"), "dtype": "list", "count": 5}, + }, + { + "id": "diary", + "type": "agenda", + "title": "Coming up", + "layout": at(6, 2, 5, 2), + "config": {"message": msg(HOUSE, "agenda"), "dtype": "list", "count": 4}, + }, + { + "id": "draw", + "type": "bar", + "title": "House draw, and the share the roof covered", + "layout": at(11, 2, 5, 2), + "config": { + "message": msg(HOUSE, "load_kw"), + "dtype": "float", + # A second reading nested inside the first, on the same scale, so + # containment is what the picture shows rather than something to + # work out. It only means anything because the flow computed a + # share: the roof's whole output is not part of the house's draw. + "inner": msg(HOUSE, "self_use_kw"), + "inner_dtype": "float", + "min": 0, + "max": 9, + "precision": 2, + "unit": " kW", + }, + }, +] + +ENERGY_WIDGETS = [ + { + "id": "power", + "type": "chart", + "title": "Power", + "layout": at(0, 0, 7, 4), + "config": { + "series": [ + {"message": msg(HOUSE, "pv_kw"), "dtype": "float", "label": "Roof"}, + {"message": msg(HOUSE, "load_kw"), "dtype": "float", "label": "House"}, + {"message": msg(HOUSE, "grid_kw"), "dtype": "float", "label": "Grid"}, + ], + "history": {"points": 720}, + "unit": " kW", + "y_label": "kW", + # Fixed, so the picture does not rescale under you every time the + # sun goes behind something. Negative is exporting. + "y_min": -7, + "y_max": 10, + }, + }, + { + "id": "climate", + "type": "chart", + "title": "Temperature", + "layout": at(7, 0, 6, 4), + "config": { + # The other kind of chart: it asks, rather than reading the ring + # the engine keeps. The range picker at the top of the tile is what + # publishes the request. + "source": "query", + "request": msg(HISTORY, "chart_request"), + "request_dtype": "record", + "message": msg(HISTORY, "climate_series"), + "dtype": "series", + "range_s": 3600, + "unit": "°C", + "y_label": "°C", + "y_min": 0, + "y_max": 32, + }, + }, + { + "id": "battery", + "type": "gauge", + "title": "Battery", + "layout": at(13, 0, 3, 4), + "config": { + "message": msg(HOUSE, "battery"), + "dtype": "float", + "min": 0, + "max": 100, + "precision": 0, + "unit": "%", + }, + }, + { + "id": "away", + "type": "switch", + "title": "Away", + "layout": at(0, 4, 2, 2), + # Drawn as a control that stays in rather than a track — the shape that + # reads across a room. + "config": {"target": msg(HOUSE, "away"), "dtype": "bool", "style": "button"}, + }, + { + "id": "mode", + "type": "dropdown", + "title": "Mode", + "layout": at(2, 4, 4, 2), + "config": { + "target": msg(HOUSE, "mode"), + "dtype": "str", + # Every choice on show at once, for the same reason. + "style": "segmented", + "options": [ + {"label": "Eco", "value": "eco"}, + {"label": "Comfort", "value": "comfort"}, + {"label": "Boost", "value": "boost"}, + ], + }, + }, + { + "id": "target", + "type": "slider", + "title": "Target", + "layout": at(6, 4, 4, 2), + "config": { + "target": msg(HOUSE, "setpoint"), + "dtype": "float", + "min": 16, + "max": 25, + "step": 0.5, + "unit": "°C", + }, + }, + { + "id": "tariff", + "type": "input", + "title": "Tariff, ct/kWh", + "layout": at(10, 4, 2, 2), + "config": {"target": msg(HOUSE, "tariff"), "dtype": "float"}, + }, + { + "id": "spend", + "type": "stat", + # "Buying" rather than "Costing": while the roof covers the whole draw + # this reads zero, and zero is the answer rather than a broken tile. + "title": "Buying", + "layout": at(12, 4, 2, 2), + "config": { + "message": msg(HOUSE, "cost_now"), + "dtype": "float", + "precision": 2, + "unit": "€/h", + }, + }, + { + "id": "boost", + "type": "button", + "title": "Hot water", + "layout": at(14, 4, 2, 2), + "config": { + "target": msg(HOUSE, "boost"), + "dtype": "bool", + "value": True, + # What the button says, rather than the tile's own title. + "label": "Boost 15 min", + }, + }, +] + +MODEL_WIDGETS = [ + { + "id": "loss", + "type": "chart", + "title": "Training loss", + "layout": at(0, 0, 6, 5), + "config": { + # Bound to the port the node yields on — the same binding a + # temperature uses. Nothing here knows what a metric is. + "series": [ + {"message": msg(MODEL, "loss"), "dtype": "float", "label": "MSE"} + ], + "history": {"points": 600}, + "y_label": "MSE", + }, + }, + { + "id": "fit", + "type": "gauge", + "title": "Fit quality (R²)", + "layout": at(6, 0, 3, 3), + "config": { + "message": msg(MODEL, "accuracy"), + "dtype": "float", + "min": 0, + "max": 100, + "precision": 1, + "unit": "%", + }, + }, + { + "id": "final", + "type": "stat", + "title": "Final loss", + "layout": at(6, 3, 3, 2), + "config": { + "message": msg(MODEL, "final_loss"), + "dtype": "float", + "precision": 4, + }, + }, + { + "id": "engine", + "type": "notification", + "title": "Engine", + "layout": at(9, 0, 4, 3), + # Where the alerting configuration below delivers. Nothing on the panel + # polls for this: a failure is pushed here when it happens. + "config": {"message": msg(HOUSE, "engine_alert"), "dtype": "record"}, + }, + { + "id": "where", + "type": "stat", + "title": "Trained on", + "layout": at(9, 3, 4, 2), + "config": {"message": msg(MODEL, "trained_on"), "dtype": "str"}, + }, + { + "id": "samples", + "type": "stat", + "title": "Readings used", + "layout": at(13, 0, 3, 2), + "config": {"message": msg(MODEL, "sample_count"), "dtype": "int"}, + }, + { + "id": "how", + "type": "markdown", + "title": "", + "layout": at(13, 2, 3, 3), + "config": { + # Headings and bullets; the widget renders nothing inline. + "content": ( + "## This panel\n" + "- Everything on it comes from three flows you can open" + " and edit.\n" + "- The controls publish into the graph; the readings answer.\n" + "- The temperature chart asks a flow for its window — no" + " database involved.\n" + "- Press Run on pv_model and watch the curve fill in." + ) + }, + }, +] + +PAGES = [ + { + "id": "main", + "title": "House", + # Several sections on one page rather than several pages: a wall panel + # opens one URL and shows what is on it, and a second page would need + # someone standing there to click it. + "sections": [ + {"id": "now", "title": "Right now", "widgets": NOW_WIDGETS}, + {"id": "energy", "title": "Energy and comfort", "widgets": ENERGY_WIDGETS}, + {"id": "model", "title": "Yield model", "widgets": MODEL_WIDGETS}, + ], + } +] + + +# ----------------------------------------------------------------------------- +# The API +# ----------------------------------------------------------------------------- + + +class Api: + def __init__(self) -> None: + self.http = httpx.Client(base_url=f"{API}/api/v1", timeout=60) + token = self.http.post( + "/login/access-token", + data={"username": EMAIL, "password": PASSWORD}, + ).json()["access_token"] + self.http.headers["Authorization"] = f"Bearer {token}" + + def __call__(self, method: str, path: str, body: Any = None) -> Any: + response = self.http.request(method, path, json=body) + response.raise_for_status() + return response.json() if response.content else None + + def drop(self, path: str) -> None: + """Delete something that may not be there.""" + try: + self("DELETE", path) + except httpx.HTTPStatusError: + pass + + +def seed_flow( + api: Api, + name: str, + title: str, + nodes: list, + sources: dict, + inputs: list | None = None, + mode: str = "live", + outputs: list | None = None, +) -> None: + """Write a flow and publish it, replacing whatever was there before. + + Deleting first is what makes this a reset rather than a merge: it takes the + flow's values and history with it, so the second run of this script leaves + exactly what the first one did. + """ + api.drop(f"/flows/{name}") + api( + "PUT", + f"/flows/{name}", + { + "name": name, + "title": title, + "mode": mode, + "outputs": outputs or [], + "nodes": nodes, + "inputs": inputs or [], + }, + ) + for node_id, code in sources.items(): + api("PUT", f"/flows/{name}/nodes/{node_id}/source", {"code": code}) + version = api("GET", f"/flows/{name}?draft=true")["definition"]["version"] + published = api("POST", f"/flows/{name}/publish", {"version": version}) + print(f" flow '{name}': {len(nodes)} nodes, published") + for issue in published.get("issues") or []: + print(f" ! {issue['message']}") + + +def seed_dashboard(api: Api) -> None: + """Write the panel and publish it — a wall reads only the published one.""" + api.drop(f"/dashboards/{PANEL}") + api("POST", f"/dashboards/{PANEL}") + current = api("GET", f"/dashboards/{PANEL}") + api( + "PUT", + f"/dashboards/{PANEL}", + { + **current, + "title": "Home", + "columns": COLUMNS, + "canvas_width": CANVAS[0], + "canvas_height": CANVAS[1], + "pages": PAGES, + }, + ) + version = api("GET", f"/dashboards/{PANEL}?draft=true")["version"] + api("POST", f"/dashboards/{PANEL}/publish", {"version": version}) + + count = sum(len(section["widgets"]) for section in PAGES[0]["sections"]) + print( + f" dashboard '{PANEL}': {count} widgets in " + f"{len(PAGES[0]['sections'])} sections, published" + ) + + +def wire_alerts(api: Api) -> None: + """Point the engine's own failures at the notification tile. + + Merged into whatever is already configured rather than replacing it: this + script owns one channel and one rule, and an operator's ntfy or mail + channel is none of its business. + """ + config = api("GET", "/alerts/config") or {} + channels = [c for c in config.get("channels") or [] if c["name"] != CHANNEL] + rules = [ + r for r in config.get("rules") or [] if CHANNEL not in (r.get("channels") or []) + ] + channels.append( + { + "name": CHANNEL, + "kind": "dashboard", + "enabled": True, + "config": {"message": msg(HOUSE, "engine_alert")}, + } + ) + # No event list: everything worth waking someone for goes to the panel. + rules.append({"events": [], "channels": [CHANNEL], "cooldown_s": 900}) + api( + "PUT", + "/alerts/config", + {"enabled": True, "channels": channels, "rules": rules}, + ) + + # The only way to prove the channel works is to use it — and the only way + # to leave the panel reading sensibly afterwards is to say so again. + api("POST", f"/alerts/test/{CHANNEL}") + api("POST", f"/messages/{msg(HOUSE, 'engine_alert')}", {"value": ALL_QUIET}) + print(f" alerts: '{CHANNEL}' delivers to {msg(HOUSE, 'engine_alert')}, tested") + + +def run_the_model(api: Api) -> None: + """Submit one run, so the panel has a curve and a score to show. + + A batch flow computes nothing until something asks it to, and waiting for + it here is the difference between a training section and three empty tiles. + + A run's messages live in a state of its own — that isolation is what lets + two runs of one flow go in parallel — and it expires with the run. The + panel reads the live namespace, so the finished run's result is put there + afterwards: the same numbers, where a wall can see them. Press Run again + and the tiles follow that one as it happens. + """ + run = api("POST", f"/runs/flows/{MODEL}", {"params": {}}) + deadline = time.time() + 180 + status = run["status"] + while status in ("queued", "running") and time.time() < deadline: + time.sleep(2) + status = api("GET", f"/runs/{run['id']}")["status"] + if status != "ok": + print(f" run {run['id']}: {status} — the model section will be empty") + return + + result = api("GET", f"/runs/{run['id']}").get("result") or {} + for name in ("sample_count", "accuracy", "final_loss", "trained_on"): + if name in result: + api("POST", f"/messages/{msg(MODEL, name)}", {"value": result[name]}) + + # The curve is a streaming output, so the run kept every value it took. + curve = api("GET", f"/runs/{run['id']}/metrics?name={msg(MODEL, 'loss')}") + for point in curve: + api("POST", f"/messages/{msg(MODEL, 'loss')}", {"value": point["value"]}) + # Spaced out, because the chart's axis is time: forty values inside one + # second draw the right curve under labels nobody can read. + time.sleep(0.2) + + print(f" run {run['id']}: finished, {len(curve)} epochs on the panel") + + +def main() -> int: + if not EMAIL or not PASSWORD: + print( + "FIRST_SUPERUSER / FIRST_SUPERUSER_PASSWORD are unset. Either run\n" + " make -C app seed-hosted-demo\n" + "which reads them from app/.env, or export them yourself:\n" + ' export FIRST_SUPERUSER="admin@example.com"\n' + ' export FIRST_SUPERUSER_PASSWORD="..."\n' + ' export API_URL="https://api.example.com" # optional', + file=sys.stderr, + ) + return 1 + + print(f"Seeding the demo at {API}\n") + api = Api() + + seed_flow( + api, + HOUSE, + "House", + HOUSE_NODES, + {"house": HOUSE_SOURCE, "meter": METER_SOURCE, "plan": PLAN_SOURCE}, + inputs=HOUSE_INPUTS, + ) + seed_flow( + api, + HISTORY, + "House history", + HISTORY_NODES, + {"answer": HISTORY_SOURCE}, + inputs=HISTORY_INPUTS, + ) + seed_flow( + api, + MODEL, + "PV yield model", + MODEL_NODES, + { + "prepare": PREPARE_SOURCE, + "train": TRAIN_SOURCE, + "evaluate": EVALUATE_SOURCE, + }, + inputs=MODEL_INPUTS, + mode="batch", + outputs=MODEL_OUTPUTS, + ) + + seed_dashboard(api) + + # The hot water is off until someone presses the button. Saying so is the + # difference between a tile reading "Idle" and one reading a dash. + api("POST", f"/messages/{msg(HOUSE, 'water_state')}", {"value": "Idle"}) + + wire_alerts(api) + run_the_model(api) + + print( + f"\nOpen /view/{PANEL}.\n" + "The live chart fills in over the first few minutes; the forecast and\n" + "the agenda land within a couple of seconds. Every control on it\n" + "publishes for real — the tariff field changes what the meter costs,\n" + "the slider moves the target line on the temperature chart." + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())