stroblmeandClaude Opus 5
436ca7e9af
The Node-RED installation this replaces is 865 nodes across three tabs, and roughly a fifth of it is unreachable — the pellet stove's controller, the scene engine and the awning's logic were all disconnected from the heartbeat they ran on. What is here is the intent rather than the wiring: nineteen named flows, 109 nodes, and no heartbeat at all. A sensor value is the event. The device layer moves with it. `actor/*` and `light/*` were never a device interface — Node-RED subscribed to its own topics, stamped a DMX channel on each and encoded one Art-Net universe — so those topics retire with it and the encoders are five nodes in the `dmx` flow. Two shared library nodes carry what every actuator needs. `arbiter` answers the thing this design was missing: a value someone sets on a screen is not undone by the next evaluation. A manual value wins for a hold, the house takes over when it expires, and a schedule can force past both — so "off at two in the morning" still means off. The control binds to the message the arbiter writes back, so one tile shows what reached the fixture and setting it is the override. `motor` is why a stop is now commanded once. A rollershutter has no position sensor, so time is the only feedback: it says how long to run and a trigger sends the single STOP that ends it. The reference sent STOP forever. Everything is seeded stopped, the Art-Net node does not transmit and the heat pump does not accept commands until house.json says so. `--dry` checks the whole set without an installation: names nothing provides, loops, type disagreements, widgets bound to nothing, and every Python node run once on values of the shape it declared — including whether what it returns goes anywhere. That last one has already caught a typo that would have published into silence. house.json holds this installation's addresses, MAC addresses and DMX map and is git-ignored, as the Node-RED inventory is. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
145 lines
5.3 KiB
Python
145 lines
5.3 KiB
Python
"""The two shared nodes, checked the way they will fail.
|
|
|
|
Run it directly — it needs no framework and no installation:
|
|
|
|
python scripts/tinyhouse/test_library.py
|
|
|
|
These two are worth a check because everything else in the house is a table of
|
|
thresholds, while these carry state between runs and decide who wins.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from tinyhouse.library import ARBITER, MOTOR # noqa: E402
|
|
|
|
|
|
def _load(source: str):
|
|
namespace: dict = {}
|
|
exec(compile(source, "<node>", "exec"), namespace)
|
|
return namespace["process"]
|
|
|
|
|
|
arbiter = _load(ARBITER)
|
|
motor = _load(MOTOR)
|
|
|
|
|
|
# ── the arbiter ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_the_house_decides_when_nobody_has_said_otherwise():
|
|
first = arbiter(auto=True, manual=False)
|
|
assert first["command"] is True, "the first run adopts what is there"
|
|
|
|
later = arbiter(auto=False, manual=first["manual"], state=first["state"])
|
|
assert later["command"] is False, "and follows the house afterwards"
|
|
|
|
|
|
def test_a_person_wins_and_keeps_winning_for_the_hold():
|
|
state = arbiter(auto=False, manual=False)["state"]
|
|
|
|
pressed = arbiter(auto=False, manual=True, state=state, hold_s=3600)
|
|
assert pressed["command"] is True
|
|
|
|
# The house asks for off, once a second, for an hour. It does not get it.
|
|
state = pressed["state"]
|
|
for _ in range(5):
|
|
again = arbiter(auto=False, manual=state["manual"], state=state, hold_s=3600)
|
|
assert again["command"] is True, "automation must not undo a person"
|
|
state = again["state"]
|
|
|
|
|
|
def test_the_house_takes_over_once_the_hold_has_run_out():
|
|
state = arbiter(auto=False, manual=False)["state"]
|
|
pressed = arbiter(auto=False, manual=True, state=state, hold_s=3600)
|
|
|
|
expired = dict(pressed["state"], override_until=time.time() - 1)
|
|
after = arbiter(auto=False, manual=expired["manual"], state=expired)
|
|
assert after["command"] is False
|
|
|
|
|
|
def test_a_hold_of_zero_lasts_until_something_forces_it():
|
|
state = arbiter(auto=False, manual=False, hold_s=0)["state"]
|
|
pressed = arbiter(auto=False, manual=True, state=state, hold_s=0)
|
|
assert pressed["state"]["override_until"] == -1.0
|
|
|
|
held = arbiter(
|
|
auto=False, manual=pressed["manual"], state=pressed["state"], hold_s=0
|
|
)
|
|
assert held["command"] is True, "no expiry means no expiry"
|
|
|
|
# Two in the morning arrives as a new timestamp.
|
|
forced = arbiter(
|
|
auto=False,
|
|
manual=held["manual"],
|
|
state=held["state"],
|
|
force_at=1_700_000_000.0,
|
|
force_value=False,
|
|
hold_s=0,
|
|
)
|
|
assert forced["command"] is False, "a schedule outranks a forgotten override"
|
|
assert forced["state"]["override_until"] == 0.0
|
|
|
|
|
|
def test_the_control_shows_what_actually_reached_the_fixture():
|
|
"""One tile reads and writes, so what it publishes is what it displays."""
|
|
state = arbiter(auto=False, manual=False)["state"]
|
|
on = arbiter(auto=True, manual=False, state=state)
|
|
assert on["manual"] == on["command"] is True
|
|
|
|
|
|
# ── the motor ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_a_shutter_runs_for_as_long_as_that_direction_takes():
|
|
down = motor(cmd="DOWN", up_s=26, down_s=28)
|
|
assert (down["run"], down["run_for"]) == ("DOWN", 28)
|
|
assert down["state"]["position"] == "DOWN"
|
|
|
|
settled = dict(down["state"], moving_until=0.0)
|
|
up = motor(cmd="UP", state=settled, up_s=26, down_s=28)
|
|
assert (up["run"], up["run_for"]) == ("UP", 26)
|
|
|
|
|
|
def test_asking_again_for_where_it_already_is_does_nothing():
|
|
down = motor(cmd="DOWN", up_s=26, down_s=28)
|
|
settled = dict(down["state"], moving_until=0.0)
|
|
assert motor(cmd="DOWN", state=settled) is None
|
|
|
|
|
|
def test_a_reversal_mid_travel_stops_rather_than_driving_both_ways():
|
|
down = motor(cmd="DOWN", up_s=26, down_s=28)
|
|
assert down["state"]["moving_until"] > time.time()
|
|
|
|
turn = motor(cmd="UP", state=down["state"], up_s=26, down_s=28)
|
|
assert turn["run"] == "STOP" and turn["run_for"] == 0.0
|
|
assert turn["state"]["moving"] == ""
|
|
|
|
|
|
def test_stop_is_commanded_once_and_then_left_alone():
|
|
"""The reference sent STOP forever. This is the whole reason it does not."""
|
|
down = motor(cmd="DOWN", up_s=26, down_s=28)
|
|
stopped = motor(cmd="STOP", state=down["state"])
|
|
assert (stopped["run"], stopped["run_for"]) == ("STOP", 0.0)
|
|
assert motor(cmd="STOP", state=stopped["state"]) is None, "nothing more to say"
|
|
|
|
|
|
def test_a_run_is_over_when_it_has_had_its_time():
|
|
"""Nothing reports the end of a run, so the clock is the only witness."""
|
|
down = motor(cmd="DOWN", up_s=26, down_s=28)
|
|
expired = dict(down["state"], moving_until=time.time() - 1)
|
|
assert motor(cmd="DOWN", state=expired) is None, "it is already there"
|
|
assert motor(cmd="UP", state=expired)["run"] == "UP", "and free to go back"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
checks = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
|
|
for check in checks:
|
|
check()
|
|
print(f"{len(checks)} checks passed")
|