stroblmeandClaude Opus 5
8734e51ef1
Sharing after the flows were created left every flow between the owner and the end of the list pointing at a library entry that did not exist yet. A node that fails to load provides nothing, so one missing entry reported as inputs nothing carries, in eighteen places. The library is written *from* a node, so this is one throwaway flow that carries both sources in and is deleted again — the entries outlive it. Order stops mattering, which also covers the case that would have broken the previous fix: two instances of one shared node inside a single flow. Verified against the installation: nineteen flows, zero validation issues. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
496 lines
18 KiB
Python
496 lines
18 KiB
Python
"""Seed this house into a Fluksio installation.
|
|
|
|
make -C app seed-tinyhouse # check everything, then push
|
|
make -C app seed-tinyhouse ARGS=--dry # check everything, push nothing
|
|
|
|
Everything is seeded **stopped** and nothing is armed: the Art-Net node does
|
|
not transmit and the heat pump does not accept commands until `house.json`
|
|
says so. Starting a flow is a decision someone makes while watching, which is
|
|
the whole of the cutover procedure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from typing import Any
|
|
|
|
from . import actuators, control, dashboards, screens, sensing
|
|
from .api import Api, Flow, house, report
|
|
from .library import ARBITER, MOTOR
|
|
|
|
#: Which secrets the flows reference. Nothing runs without them, and nothing
|
|
#: here can invent them.
|
|
SECRETS = {
|
|
"influx_token": "InfluxDB 2.x API token, for the history and the charts",
|
|
"unifi_password": "UniFi controller account (read-only is enough)",
|
|
"owm_appid": "OpenWeatherMap One Call 3.0 key",
|
|
"caldav_password": "Nextcloud app password for the shared calendar",
|
|
}
|
|
|
|
|
|
def build(h: dict[str, Any], commands: bool) -> list[Flow]:
|
|
"""Every flow, in the order it is safe to start them."""
|
|
return [
|
|
sensing.clock(),
|
|
sensing.power(h),
|
|
sensing.weather(h),
|
|
sensing.presence(h),
|
|
sensing.calendar(h),
|
|
screens.history(h),
|
|
control.climate(),
|
|
control.hvac(h, commands),
|
|
control.oven(h),
|
|
control.boiler(h),
|
|
actuators.shutters(h),
|
|
actuators.window(h),
|
|
actuators.canopy(h),
|
|
actuators.lights(),
|
|
actuators.plugs(h),
|
|
actuators.appliances(),
|
|
actuators.outdoor(),
|
|
actuators.dmx(h),
|
|
screens.kiosk(h),
|
|
]
|
|
|
|
|
|
# ── preflight ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _qualify(flow: str, name: str) -> str:
|
|
return name if "." in name else f"{flow}.{name}"
|
|
|
|
|
|
def check(flows: list[Flow]) -> list[str]:
|
|
"""Everything wrong that can be seen without an installation.
|
|
|
|
The API validates too, and better — but it does it one flow at a time as
|
|
they are created, so a cross-flow name that nothing provides is only
|
|
reported once the last flow lands. Reading the whole set first means a
|
|
typo is a message here rather than nineteen flows and a broken house.
|
|
"""
|
|
provided: dict[str, list[str]] = {}
|
|
declared: set[str] = set()
|
|
problems: list[str] = []
|
|
|
|
for flow in flows:
|
|
for spec in flow.inputs:
|
|
declared.add(_qualify(flow.name, spec["spec"]["name"]))
|
|
for node in flow.nodes:
|
|
for spec in node.get("provides", []):
|
|
provided.setdefault(_qualify(flow.name, spec["name"]), []).append(
|
|
f"{flow.name}.{node['id']}"
|
|
)
|
|
|
|
known = set(provided) | declared
|
|
for flow in flows:
|
|
ids = [node["id"] for node in flow.nodes]
|
|
if len(set(ids)) != len(ids):
|
|
problems.append(f"{flow.name}: two nodes share an id")
|
|
for node in flow.nodes:
|
|
for spec in node.get("requires", []):
|
|
name = _qualify(flow.name, spec["name"])
|
|
if name not in known:
|
|
problems.append(
|
|
f"{flow.name}.{node['id']} needs '{name}', "
|
|
"which nothing provides and no flow declares"
|
|
)
|
|
ports = [s.get("port") or s["name"] for s in node.get("requires", [])]
|
|
if len(set(ports)) != len(ports):
|
|
problems.append(f"{flow.name}.{node['id']}: two inputs share a port")
|
|
settings = set(node.get("params", {}))
|
|
clash = settings & set(ports)
|
|
if clash:
|
|
problems.append(
|
|
f"{flow.name}.{node['id']}: setting and port share a name: "
|
|
+ ", ".join(sorted(clash))
|
|
)
|
|
|
|
problems += _check_types(flows)
|
|
problems += _check_cycles(flows, provided)
|
|
problems += _check_widgets(known)
|
|
problems += _check_sources(flows)
|
|
problems += _check_schemas(flows)
|
|
return problems
|
|
|
|
|
|
#: Node types that come from connector packages. `make connectors` installs
|
|
#: them into the image, so a checkout does not have them and their absence
|
|
#: here is not a problem to report.
|
|
CONNECTORS = {"wfrac", "unifi_presence", "ical", "artnet"}
|
|
|
|
#: Ports whose payload has a shape the node reads keys out of. Everything else
|
|
#: is a scalar or an object the node only ever asks politely.
|
|
SHAPES: dict[str, Any] = {
|
|
"chart_request": {"range_s": 3600, "interval_s": 60},
|
|
"rows": {"rows": [], "range_s": 3600, "interval_s": 60},
|
|
"limited": {
|
|
"operation": True,
|
|
"mode": "fan",
|
|
"preset_temp": 22.0,
|
|
"fan_speed": "auto",
|
|
},
|
|
}
|
|
|
|
#: Something of the right shape to hand a port, so a node can be called at all.
|
|
SAMPLE: dict[str, Any] = {
|
|
"float": 1.0,
|
|
"int": 1,
|
|
"str": "x",
|
|
"bool": True,
|
|
"json": {},
|
|
"record": {},
|
|
"list": [],
|
|
"series": {"lines": []},
|
|
}
|
|
|
|
|
|
def _check_sources(flows: list[Flow]) -> list[str]:
|
|
"""Run every Python node once, on values of the shape it declared.
|
|
|
|
Not a test of what they decide — that is what the thresholds are for, and
|
|
they are settings. This catches the class of mistake that is otherwise
|
|
invisible until the house is running: a name that does not exist, an
|
|
argument the function does not take, and above all an output key that is
|
|
not a port, which publishes nothing at all and says nothing about it.
|
|
"""
|
|
problems = []
|
|
for flow in flows:
|
|
for node in flow.nodes:
|
|
if node["type"] != "python":
|
|
continue
|
|
source = flow.sources.get(node["id"])
|
|
if source is None:
|
|
continue
|
|
namespace: dict[str, Any] = {}
|
|
where = f"{flow.name}.{node['id']}"
|
|
try:
|
|
exec(compile(source, where, "exec"), namespace)
|
|
except Exception as exc: # noqa: BLE001 - reported, not raised
|
|
problems.append(f"{where}: will not load: {exc}")
|
|
continue
|
|
process = namespace.get("process")
|
|
if process is None:
|
|
problems.append(f"{where}: has no process()")
|
|
continue
|
|
|
|
ports = {}
|
|
for spec in node.get("requires", []):
|
|
port = spec.get("port") or spec["name"]
|
|
ports[port] = SHAPES.get(port, SAMPLE.get(spec["dtype"]))
|
|
try:
|
|
result = process(**ports, **node.get("params", {}))
|
|
except Exception as exc: # noqa: BLE001 - reported, not raised
|
|
problems.append(f"{where}: {type(exc).__name__} when called: {exc}")
|
|
continue
|
|
|
|
if result is None:
|
|
continue
|
|
if not isinstance(result, dict):
|
|
problems.append(
|
|
f"{where}: returned {type(result).__name__}, not a dict"
|
|
)
|
|
continue
|
|
declared = {
|
|
spec.get("port") or spec["name"] for spec in node.get("provides", [])
|
|
}
|
|
stray = set(result) - declared
|
|
if stray:
|
|
problems.append(
|
|
f"{where}: returns {', '.join(sorted(stray))}, which "
|
|
"no output port carries — it would publish nothing"
|
|
)
|
|
return problems
|
|
|
|
|
|
def _check_cycles(flows: list[Flow], provided: dict[str, list[str]]) -> list[str]:
|
|
"""A waits for B and B waits for A, so neither of them ever runs.
|
|
|
|
The same rule the engine applies: only a *triggering* input creates a
|
|
dependency, and a node reading a message it also writes is carrying state
|
|
rather than waiting for itself. Worth doing here because the engine can
|
|
only see it once every flow exists, and by then the house is half seeded.
|
|
"""
|
|
deps: dict[str, set[str]] = {}
|
|
for flow in flows:
|
|
for node in flow.nodes:
|
|
me = f"{flow.name}.{node['id']}"
|
|
mine = {_qualify(flow.name, s["name"]) for s in node.get("provides", [])}
|
|
deps[me] = {
|
|
producer
|
|
for spec in node.get("requires", [])
|
|
if spec.get("trigger", True)
|
|
for producer in provided.get(_qualify(flow.name, spec["name"]), [])
|
|
if producer != me and _qualify(flow.name, spec["name"]) not in mine
|
|
}
|
|
|
|
# Kahn's algorithm; whatever is left over is in a loop.
|
|
pending = {node: set(edges) for node, edges in deps.items()}
|
|
while True:
|
|
ready = [node for node, edges in pending.items() if not edges]
|
|
if not ready:
|
|
break
|
|
for node in ready:
|
|
pending.pop(node)
|
|
for edges in pending.values():
|
|
edges.difference_update(ready)
|
|
if pending:
|
|
return [
|
|
"these depend on each other in a loop, so none of them can run: "
|
|
+ ", ".join(sorted(pending))
|
|
]
|
|
return []
|
|
|
|
|
|
def _check_types(flows: list[Flow]) -> list[str]:
|
|
"""One message, one type — whoever writes it and whoever reads it."""
|
|
types: dict[str, tuple[str, str]] = {}
|
|
problems = []
|
|
for flow in flows:
|
|
for spec in flow.inputs:
|
|
types.setdefault(
|
|
_qualify(flow.name, spec["spec"]["name"]),
|
|
(spec["spec"]["dtype"], f"{flow.name} inputs"),
|
|
)
|
|
for node in flow.nodes:
|
|
for spec in node.get("provides", []) + node.get("requires", []):
|
|
name = _qualify(flow.name, spec["name"])
|
|
where = f"{flow.name}.{node['id']}"
|
|
seen = types.setdefault(name, (spec["dtype"], where))
|
|
if seen[0] != spec["dtype"]:
|
|
problems.append(
|
|
f"'{name}' is {seen[0]} in {seen[1]} but "
|
|
f"{spec['dtype']} in {where}"
|
|
)
|
|
return problems
|
|
|
|
|
|
def _check_widgets(known: set[str]) -> list[str]:
|
|
"""A tile bound to a message nothing carries draws nothing, quietly."""
|
|
problems = []
|
|
for _, title, _, widgets in dashboards.SCREENS:
|
|
for widget in widgets:
|
|
config = widget["config"]
|
|
for key in ("message", "target", "request"):
|
|
name = config.get(key)
|
|
if name and name not in known:
|
|
problems.append(
|
|
f"{title}: '{widget['id']}' binds {key} to '{name}', "
|
|
"which no flow carries"
|
|
)
|
|
return problems
|
|
|
|
|
|
# ── pushing it ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def _check_schemas(flows: list[Flow]) -> list[str]:
|
|
"""Hand every node to the engine's own models before the API sees them.
|
|
|
|
The API would say the same thing, one flow at a time, after creating the
|
|
ones before it. Importing the models is free and answers it here — a
|
|
misspelled setting on a connector is otherwise found halfway through a
|
|
house.
|
|
"""
|
|
try:
|
|
from fluksio.flow.controller import NODE_TYPES
|
|
from fluksio.flow.schemas import FlowDef
|
|
except ImportError:
|
|
return []
|
|
|
|
problems = []
|
|
missing: set[str] = set()
|
|
for flow in flows:
|
|
try:
|
|
FlowDef.model_validate(
|
|
{
|
|
"name": flow.name,
|
|
"title": flow.title,
|
|
"nodes": flow.nodes,
|
|
"inputs": flow.inputs,
|
|
}
|
|
)
|
|
except Exception as exc: # noqa: BLE001 - reported, not raised
|
|
problems.append(f"{flow.name}: {exc}")
|
|
continue
|
|
|
|
for node in flow.nodes:
|
|
spec = NODE_TYPES.get(node["type"])
|
|
if spec is None:
|
|
# Connectors are installed into the image, not into the venv
|
|
# this runs in, so their absence here says nothing. A type that
|
|
# is neither built in nor a connector this house uses is a
|
|
# typo, and that is what is worth saying.
|
|
if node["type"] not in CONNECTORS:
|
|
problems.append(
|
|
f"{flow.name}.{node['id']}: no node type '{node['type']}'"
|
|
)
|
|
else:
|
|
missing.add(node["type"])
|
|
continue
|
|
if spec.free_params:
|
|
continue
|
|
params = getattr(spec.cls, "Params", None)
|
|
if params is None:
|
|
continue
|
|
try:
|
|
params.model_validate(_without_secrets(node.get("params", {})))
|
|
except Exception as exc: # noqa: BLE001 - reported, not raised
|
|
problems.append(f"{flow.name}.{node['id']} settings: {exc}")
|
|
|
|
if missing:
|
|
print(f" (from the image, not checked here: {', '.join(sorted(missing))})")
|
|
return problems
|
|
|
|
|
|
def _without_secrets(params: Any) -> Any:
|
|
"""A secret reference is an ordinary string by the time a node is built."""
|
|
if isinstance(params, dict):
|
|
if set(params) == {"$secret"}:
|
|
return "from-the-store"
|
|
return {k: _without_secrets(v) for k, v in params.items()}
|
|
if isinstance(params, list):
|
|
return [_without_secrets(v) for v in params]
|
|
return params
|
|
|
|
|
|
#: Settings these flows depend on that a node type gained for this port. A node
|
|
#: type accepts an unknown setting and ignores it — which is right for a node
|
|
#: someone is editing and wrong here, where it would mean a rollershutter
|
|
#: running for the default minute instead of the twenty-six seconds it takes.
|
|
NEEDED = {
|
|
"trigger": {
|
|
"wait_port": "a shutter would run for the default wait, not its own",
|
|
"passthrough": "a motor would be commanded with 'true' rather than a direction",
|
|
},
|
|
"mqtt": {
|
|
"json_key": "every Victron reading would arrive as an object, not a number",
|
|
},
|
|
"http": {
|
|
"query": "the weather key would have to sit in the flow file in clear",
|
|
},
|
|
"artnet": {
|
|
"baseline": "the first frame would darken what it is not driving",
|
|
},
|
|
}
|
|
|
|
|
|
def _engine_too_old(api: Api) -> list[tuple[str, str]]:
|
|
"""Settings this installation's node types do not know about yet."""
|
|
types = {t["type"]: t for t in api("GET", "/flows/node-types")}
|
|
stale = []
|
|
for type_name, settings in NEEDED.items():
|
|
schema = types.get(type_name, {}).get("params_schema") or {}
|
|
known = set(schema.get("properties") or {})
|
|
if not known:
|
|
continue
|
|
for setting, why in settings.items():
|
|
if setting not in known:
|
|
stale.append((f"{type_name}.{setting}", why))
|
|
return stale
|
|
|
|
|
|
LIBRARY = {"arbiter": ARBITER, "motor": MOTOR}
|
|
|
|
|
|
def seed_library(api: Api) -> None:
|
|
"""Write the shared sources before anything references them.
|
|
|
|
A node pointing at a library entry that does not exist yet fails to load,
|
|
and a node that failed to load provides nothing — so one missing entry
|
|
reports as inputs nothing carries, in every flow downstream of it. The
|
|
library is written *from* a node, though, so this is one throwaway flow
|
|
that carries the sources in and is deleted again; the entries outlive it.
|
|
"""
|
|
existing = {node["name"] for node in api("GET", "/flows/library")}
|
|
wanted = {name: code for name, code in LIBRARY.items() if name not in existing}
|
|
if not wanted:
|
|
return
|
|
|
|
bootstrap = Flow("librarybootstrap", "Library bootstrap")
|
|
for name in wanted:
|
|
bootstrap.add(
|
|
{
|
|
"id": name,
|
|
"type": "python",
|
|
"title": name,
|
|
"requires": [{"name": "unused", "dtype": "json"}],
|
|
"provides": [{"name": f"{name}_unused", "dtype": "json"}],
|
|
},
|
|
wanted[name],
|
|
)
|
|
bootstrap.input("unused", "json", {})
|
|
|
|
api.put_flow(bootstrap)
|
|
for name in wanted:
|
|
api.share(bootstrap.name, name, name)
|
|
print(f" library: '{name}' written")
|
|
api.drop(f"/flows/{bootstrap.name}")
|
|
|
|
|
|
def push(api: Api, flows: list[Flow]) -> None:
|
|
"""The library first, then every flow, then the screens."""
|
|
seed_library(api)
|
|
for flow in flows:
|
|
report(flow.name, len(flow.nodes), api.put_flow(flow))
|
|
for name, title, glyph, widgets in dashboards.SCREENS:
|
|
api.put_dashboard(name, title, glyph, widgets)
|
|
print(f" dashboard '{name}': {len(widgets)} widgets, published")
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
dry = "--dry" in argv
|
|
h = house()
|
|
flows = build(h, commands=bool(h["aircon"].get("commands", False)))
|
|
|
|
problems = check(flows)
|
|
nodes = sum(len(f.nodes) for f in flows)
|
|
print(f"{len(flows)} flows, {nodes} nodes, {len(dashboards.SCREENS)} dashboards")
|
|
for problem in problems:
|
|
print(f" ! {problem}", file=sys.stderr)
|
|
if problems:
|
|
print(f"\n{len(problems)} problems; nothing was pushed.", file=sys.stderr)
|
|
return 1
|
|
if dry:
|
|
print("\nEverything checks out. Nothing was pushed (--dry).")
|
|
return 0
|
|
|
|
api = Api()
|
|
stale = _engine_too_old(api)
|
|
if stale:
|
|
print(
|
|
"\nThis installation's engine does not have what these flows need:",
|
|
file=sys.stderr,
|
|
)
|
|
for setting, why in stale:
|
|
print(f" {setting} — {why}", file=sys.stderr)
|
|
print(
|
|
"\nAn unknown setting is accepted and ignored rather than refused, so\n"
|
|
"seeding now would look like it worked and run every shutter for a\n"
|
|
"minute. Rebuild the image first:\n"
|
|
' docker compose -p fluksio-app --env-file "$PWD/.env" \\\n'
|
|
" -f docker/compose.yml up --build -d backend",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
have = set(api("GET", "/secrets/")["data"])
|
|
missing = [name for name in SECRETS if name not in have]
|
|
if missing:
|
|
print("\nAdd these under Secrets first:", file=sys.stderr)
|
|
for name in missing:
|
|
print(f" {name} — {SECRETS[name]}", file=sys.stderr)
|
|
return 1
|
|
|
|
push(api, flows)
|
|
print(
|
|
"\nEvery flow is published and stopped. Start the reading ones first:\n"
|
|
" clock, power, weather, presence, calendar, history\n"
|
|
"Nothing moves until 'dmx' is started, and it only transmits once\n"
|
|
"house.json says so and Node-RED's own Art-Net sender is stopped."
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|