Let a connector write, and publish strings bare

Two things stopped the engine commanding this house. ConnectorNode hardwired
its node function to a no-op, so an input message reaching a connector was
discarded and Art-Net's packet builder was unreachable; write() now carries
the input ports, which is additive so the contract version holds. And the MQTT
publisher JSON-encoded every payload, so "ON" went on the wire quoted and the
devices on a shared broker, which speak bare values, ignored it.

seed_house_control.py is the rig: a flow that drives the washing machine plug,
a dimmer and a colour fixture over MQTT, carries the same two as DMX on an
Art-Net node with transmit still off, and a dashboard to drive it by hand.
This commit is contained in:
2026-08-20 21:57:27 +02:00
parent 40e9db2d56
commit 906203da1e
7 changed files with 116 additions and 12 deletions
+4 -1
View File
@@ -3,7 +3,7 @@
# The workspace root delegates to these (see ../Makefile). # The workspace root delegates to these (see ../Makefile).
.PHONY: dev-utils dev dev-local up down update install dev-backend dev-frontend \ .PHONY: dev-utils dev dev-local up down update install dev-backend dev-frontend \
generate-client seed-example seed-demo seed-hosted-demo test test-backend test-frontend soak bench-startup lint lint-backend \ generate-client seed-example seed-demo seed-house seed-hosted-demo test test-backend test-frontend soak bench-startup lint lint-backend \
lint-frontend umami clean help lint-frontend umami clean help
COMPOSE_ROOT := $(CURDIR) COMPOSE_ROOT := $(CURDIR)
@@ -97,6 +97,9 @@ 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 seed-demo: ## Seed the training-run example: a batch flow and its dashboard
cd backend && uv run python ../scripts/seed_demo_training.py cd backend && uv run python ../scripts/seed_demo_training.py
seed-house: ## Seed the house write-path rig (needs the real broker reachable)
cd backend && uv run python ../scripts/seed_house_control.py
# Operators of the hosted demo only — NOT part of any deployment, and nothing a # 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 # 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. # dashboard, so re-running it is how the public demo is reset.
+25 -1
View File
@@ -60,9 +60,33 @@ Postgres + Redis + git-files split stays; the actionable part is durability.
Needs someone watching the real hardware, so it is not a background task. This Needs someone watching the real hardware, so it is not a background task. This
is what M4 still waits on, together with porting the flows. is what M4 still waits on, together with porting the flows.
- FEAT/NODE: the connectors only read. Enable the write paths with someone watching: WF-RAC `setAirconStat` (needs an operatorId registered with the unit first, which is itself a write) and Art-Net `transmit`. Art-Net can write now: `ConnectorNode.write` carries a node's input ports, a
per-port `channels` map places each on its own DMX channel, and `transmit`
still gates the socket. Verified on the wire against a listener (channel 33 =
255, channel 31 = 60, nothing else set) and the MQTT half was driven end to
end against the house broker. The rig is the `house_control` flow and its
dashboard, seeded by `make -C app seed-house`.
- FEAT/NODE: Art-Net against the real fixtures is still untried. The house's own dmxnet sender re-emits universe 1 every 1000 ms, so fluksio and Node-RED overwrite each other; the test needs Node-RED's Art-Net sender stopped, and while it is stopped every channel fluksio does not set is dark.
- FEAT/NODE: the WF-RAC write path stays gated. `setAirconStat` needs an operatorId registered with the unit first, which is itself a write, and the aircon is not on the safe-to-control list.
- FEAT/NODE: the second WF-RAC unit (the one Node-RED addresses with operatorId "0") closes the connection on an anonymous read. It likely wants an account registered; the first unit answers without one. - FEAT/NODE: the second WF-RAC unit (the one Node-RED addresses with operatorId "0") closes the connection on an anonymous read. It likely wants an account registered; the first unit answers without one.
- CHORE/NODE: `wfrac` reports `mode` as "unknown" while the unit is off, because the mode bits hold a value outside the known set. Faithful to the reference decoder, but "off" would read better. - CHORE/NODE: `wfrac` reports `mode` as "unknown" while the unit is off, because the mode bits hold a value outside the known set. Faithful to the reference decoder, but "off" would read better.
- PERF/NODE: `ArtNetOut.write` sends one frame per input port, so a node with two ports emits two frames per run. The last one carries both channels, so the end state is right; folding them into one send would halve the traffic.
- CHORE/NODE: the Art-Net node starts from an all-zero universe and has no way to learn what the fixtures are currently at — Art-Net has no read-back. Taking over a universe therefore blanks everything the flow does not drive. A baseline setting, or driving every channel, is what a real cutover needs.
### Porting the Node-RED flows
What the reference actually does, extracted while building the write-path rig.
One `Art-Net Out` node in 865 covers every physical device: `mqtt in <topic>`
-> `change` (msg.topic = DMX channel) -> an `nCH` encoder function -> Art-Net,
to the Art-Net node's address universe 1. Payloads are bare: `ON`/`OFF`, a number, `[h,s,v]`,
`UP`/`DOWN`. Nothing is retained, so state lives only in Node-RED globals and
is lost on its restart.
- CHORE/FLOW: three DMX channel collisions in the reference — ch 9 (`light/bathRoomLight` vs `light/bathRoomSinkLight`), ch 28 (`actor/windowOpenerStorage` 28-29 vs `light/traverseAmbientLight` 28-30), ch 129 (`actor/canopy` 129-130 vs an orphaned 1CH mapping). Decide these deliberately rather than porting them.
- CHORE/FLOW: 1CH values are not scaled. `light/traverseSpotLight` and `light/kitchenDirectLight` receive 0-100 and that number lands on DMX as-is, so those fixtures never go above 100/255. The 4CH "A" channel gets `v` raw for the same reason. Faithful is ugly; deliberate is better.
- BUG/FLOW: the reference's `function 9`/`function 10` publish the string `"undefined"` for unselected zones (`var a, b, c = [0,0,0]` only initialises `c`), which reaches the 3CH/4CH encoders and produces `NaN` DMX values. A port should emit an explicit `OFF`.
- CHORE/FLOW: dead in the reference and not worth porting — the `AmbientModeToLight` chain, the four Dashboard toggle chains, `light/generalLight` (written, no consumer), `light/outdoorPavillonLight` (no wiring), `Color Adapt`.
### Bugs found while building the screens ### Bugs found while building the screens
+9 -3
View File
@@ -73,9 +73,15 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend M
`connector-skeleton/`. The registry follows later `connector-skeleton/`. The registry follows later
- [x] First real connectors written against that contract from outside the engine: - [x] First real connectors written against that contract from outside the engine:
WF-RAC aircon, calendar, UniFi presence and Art-Net, in `connectors/`. WF-RAC aircon, calendar, UniFi presence and Art-Net, in `connectors/`.
Built by `make connectors` and installed into the image. Reading only for Built by `make connectors` and installed into the image. Three of the four
now — the aircon package cannot produce a command and Art-Net keeps its read only; the aircon package still cannot produce a command
packets off the wire until `transmit` is switched on - [x] The other direction of the contract: `ConnectorNode.write` receives the
node's input ports, so a connector can command something rather than only
read it. Additive, so `CONTRACT_VERSION` stays at 1 — before this the base
class discarded every message reaching a connector, which made `artnet`'s
packet builder unreachable. Art-Net now sends: a per-port `channels` map
puts each input on its own DMX channel, `transmit` still gates the socket,
and one node owns one universe because a frame carries all 512 levels
- [x] Node lifecycle as a protocol (`start`/`stop`/`report_health` on `Node`), - [x] Node lifecycle as a protocol (`start`/`stop`/`report_health` on `Node`),
replacing the controller's per-type isinstance chains — the same hooks a replacing the controller's per-type isinstance chains — the same hooks a
connector implements, validated on the built-in nodes first connector implements, validated on the built-in nodes first
+20 -6
View File
@@ -14,7 +14,10 @@ What a connector gets from the base class:
* :meth:`Node.report_health`, so a connection problem shows on the node rather * :meth:`Node.report_health`, so a connection problem shows on the node rather
than only in the log; than only in the log;
* the lifecycle hooks the controller drives, so nothing device-specific has to * the lifecycle hooks the controller drives, so nothing device-specific has to
be known by the engine. be known by the engine;
* :meth:`ConnectorNode.write`, the other direction — values arriving on the
node's input ports, for a connector that commands something rather than only
reading it.
The message schemas and the parameter model are the rest of the contract, and The message schemas and the parameter model are the rest of the contract, and
they are the same ones the built-in nodes use. See ``docs/connectors/`` for the they are the same ones the built-in nodes use. See ``docs/connectors/`` for the
@@ -81,16 +84,15 @@ class ConnectorNode(Node):
__slots__ = ("config", "_poll_task", "_stop_event", "_last_published") __slots__ = ("config", "_poll_task", "_stop_event", "_last_published")
def __init__(self, **kwargs: Any) -> None: def __init__(self, **kwargs: Any) -> None:
super().__init__(f=self._unused, **kwargs) super().__init__(f=self._dispatch, **kwargs)
self.config = type(self).Params(**self.params) self.config = type(self).Params(**self.params)
self._poll_task: asyncio.Task[None] | None = None self._poll_task: asyncio.Task[None] | None = None
self._stop_event: asyncio.Event | None = None self._stop_event: asyncio.Event | None = None
self._last_published: dict[str, Any] = {} self._last_published: dict[str, Any] = {}
@staticmethod def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None:
def _unused(**_: Any) -> None: """The scheduler's entry point. Settings are already on ``self.config``."""
"""A connector publishes from its own loop, not from the scheduler.""" return self.write(**ports)
return None
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# What a connector implements # What a connector implements
@@ -104,6 +106,18 @@ class ConnectorNode(Node):
""" """
return None return None
def write(self, **ports: Any) -> dict[str, Any] | None:
"""Send incoming values to the device. Values arrive keyed by input port.
A connector that only reads leaves this alone — the default discards
whatever reaches it, which is what a node with no inputs gets anyway.
Return ``None`` unless the device answers something worth publishing,
in which case return it keyed by output port like :meth:`poll` does.
This runs on the scheduler's thread, so it must not block for long.
"""
return None
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
# What the engine drives # What the engine drives
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
+5 -1
View File
@@ -379,7 +379,11 @@ class MqttNode(Node):
) )
continue continue
payload = json.dumps(value) # A string goes on the wire as it stands. Devices on a shared
# broker expect bare values, and the subscriber below already
# falls back to the raw text when it is not JSON, so a
# fluksio-to-fluksio round trip is unaffected.
payload = value if isinstance(value, str) else json.dumps(value)
await client.publish( await client.publish(
topic, topic,
payload=payload, payload=payload,
+28
View File
@@ -100,6 +100,34 @@ def test_a_failing_poll_reports_down_and_keeps_going():
assert health[-1][0] == "ok" assert health[-1][0] == "ok"
class Actuator(ConnectorNode):
"""A connector that commands something instead of reading it."""
contract = CONTRACT_VERSION
title = "Test actuator"
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.commands: list[dict[str, Any]] = []
def write(self, **ports: Any) -> None:
self.commands.append(ports)
return None
def test_an_incoming_message_reaches_a_connector_that_writes():
node = Actuator(requires=[MessageSpec(name="level", dtype=DType.INT)])
node.assign_flow("demo", "actuator")
assert node.execute({"demo.level": 255}) is None
assert node.commands == [{"level": 255}]
def test_a_read_only_connector_ignores_what_reaches_it():
node = a_sensor([])
assert node.execute({}) is None
def test_a_credential_param_is_marked_for_the_editor(): def test_a_credential_param_is_marked_for_the_editor():
schema = Sensor.Params.model_json_schema() schema = Sensor.Params.model_json_schema()
assert schema["properties"]["poll_interval"]["default"] == 0 assert schema["properties"]["poll_interval"]["default"] == 0
+25
View File
@@ -64,3 +64,28 @@ def test_a_full_publish_queue_drops_the_oldest():
assert health == [("degraded", "publish queue full")] assert health == [("degraded", "publish queue full")]
asyncio.run(scenario()) asyncio.run(scenario())
def test_a_string_goes_on_the_wire_bare():
"""Devices on a shared broker expect `ON`, not `"ON"`."""
class Recorder:
def __init__(self) -> None:
self.published: list[tuple[str, str]] = []
async def publish(self, topic, payload, **_):
self.published.append((topic, payload))
node = MqttNode(
requires=[
MessageSpec(name="plug", port="plug", dtype=DType.STR),
MessageSpec(name="level", port="level", dtype=DType.INT),
],
params={"topic": {"plug": "actor/plug", "level": "light/level"}},
)
node.assign_flow("house", "out")
client = Recorder()
asyncio.run(node._publish_with(client, {"plug": "ON", "level": 60}))
assert client.published == [("actor/plug", "ON"), ("light/level", "60")]