"""Everything that moves, and the one node that puts it on the wire. The device layer used to be Node-RED: twenty-four MQTT topics that only it subscribed to, each stamped with a DMX channel and encoded into one Art-Net universe. Those topics were never a device interface — they were its internal wiring — so they retire with it, and the encoders live here instead. One node owns universe 1. A frame carries all 512 channels, so it has to be one: two senders on one universe overwrite each other every time either of them speaks. """ from __future__ import annotations from typing import Any from .api import Flow from .library import arbiter, motor, stopper # ── the encoders ───────────────────────────────────────────────────────── SWITCHES = '''"""A relay is a channel that is either off or all the way on.""" def process(**fixtures): return {f"dmx_{name}": 255 if value else 0 for name, value in fixtures.items()} ''' LEVELS = '''"""A dimmer channel takes its level raw. The reference hands 0-100 straight to DMX without scaling to 255, so the fixtures on these channels have only ever seen the bottom 39% of their range. Faithful rather than corrected: `scale` is here to fix it deliberately, on a fixture someone has looked at, rather than by surprise on all of them at once. """ def process(scale=None, **fixtures): scale = scale or {} out = {} for name, value in fixtures.items(): level = float(value) * float(scale.get(name, 1.0)) out[f"dmx_{name}"] = max(0, min(255, int(round(level)))) return out ''' RGB = '''"""Hue, saturation and value into three channels of red, green and blue.""" import colorsys def _rgb(value): if not isinstance(value, (list, tuple)) or len(value) < 3: return [0, 0, 0] h, s, v = (float(x) for x in value[:3]) r, g, b = colorsys.hsv_to_rgb((h % 360) / 360.0, s / 100.0, v / 100.0) return [int(round(r * 255)), int(round(g * 255)), int(round(b * 255))] def process(**fixtures): return {f"dmx_{name}": _rgb(value) for name, value in fixtures.items()} ''' RGBW = '''"""The same, for a fixture whose first channel is its own white master. Channel order is A, B, G, R — the reference's, and the A channel takes `v` raw on a 0-100 scale while the colour channels are 0-255. That asymmetry is almost certainly a property of these fixtures rather than a mistake, so it is reproduced deliberately and left adjustable. """ import colorsys def _argb(value, master_raw=True): if not isinstance(value, (list, tuple)) or len(value) < 3: return [0, 0, 0, 0] h, s, v = (float(x) for x in value[:3]) r, g, b = colorsys.hsv_to_rgb((h % 360) / 360.0, s / 100.0, v / 100.0) master = v if master_raw else v * 2.55 return [ max(0, min(255, int(round(master)))), int(round(b * 255)), int(round(g * 255)), int(round(r * 255)), ] def process(master_raw=True, **fixtures): return {f"dmx_{name}": _argb(value, master_raw) for name, value in fixtures.items()} ''' MOTORS = '''"""Two channels per motor: one for each direction, neither for stop. Driving both at once is what a reversal without a stop in between would do, so anything that is not a direction lands as a stop rather than as a guess. `inverted` is the bed shutter, whose two channels are wired the other way round from the door's. That is the hardware, not a preference. """ UP = [255, 0] DOWN = [0, 255] STOP = [0, 0] def process(inverted=(), **fixtures): out = {} for name, value in fixtures.items(): command = str(value).upper() if command in ("UP", "0"): levels = UP elif command in ("DOWN", "100"): levels = DOWN else: levels = STOP if name in inverted and levels is not STOP: levels = DOWN if levels is UP else UP out[f"dmx_{name}"] = levels return out ''' ENCODERS = { "switch": ("switches", "Relays", SWITCHES, "bool", "int"), "level": ("levels", "Dimmers", LEVELS, "float", "int"), "rgb": ("rgb", "Colour fixtures (3 channels)", RGB, "list", "list"), "rgbw": ("rgbw", "Colour fixtures (4 channels)", RGBW, "list", "list"), "motor": ("motors", "Motors", MOTORS, "str", "list"), } #: Which flow provides each fixture's value. The dmx flow reads them all and #: owns none of them: it is a driver, not a decision. SOURCE = { "bath_plug": "plugs", "bed_plug": "plugs", "fridge_pump": "plugs", "fridge_vent": "plugs", "inlet_vent": "plugs", "water_boiler": "boiler", "kitchen_boiler": "boiler", "appliances": "appliances", "floor_heating": "oven", "outdoor_plug": "outdoor", "outdoor_ambient": "lights", "outdoor_direct": "lights", "canopy_light": "lights", "kitchen_direct": "lights", "traverse_spot": "lights", "kitchen_light": "lights", "bed_light": "lights", "living_ambient": "lights", "kitchen_ambient": "lights", "bath_light": "lights", "door_shutter": "shutters", "bed_shutter": "shutters", "window_opener": "window", "canopy": "canopy", } def dmx(h: dict[str, Any]) -> Flow: flow = Flow("dmx", "DMX universe") fixtures = h["dmx"] channels: dict[str, int] = {} artnet_in: list[dict[str, Any]] = [] for kind, (node_id, title, source, in_type, out_type) in ENCODERS.items(): mine = {n: f for n, f in fixtures.items() if f["kind"] == kind} if not mine: continue params: dict[str, Any] = {} if kind == "motor": inverted = [n for n, f in mine.items() if f.get("inverted")] if inverted: params["inverted"] = inverted if kind == "level": scale = { n: f["scale"] for n, f in mine.items() if f.get("scale", 1.0) != 1.0 } if scale: params["scale"] = scale requires = [] provides = [] for name in sorted(mine): spec: dict[str, Any] = { "name": f"{SOURCE[name]}.{name}", "port": name, "dtype": in_type, } if in_type == "list": spec["item"] = "float" requires.append(spec) out: dict[str, Any] = {"name": f"dmx_{name}", "dtype": out_type} if out_type == "list": out["item"] = "int" provides.append(out) channels[f"dmx_{name}"] = mine[name]["ch"] artnet_in.append({**out, "port": f"dmx_{name}"}) flow.add( { "id": node_id, "type": "python", "title": title, "params": params, "requires": requires, "provides": provides, }, source, ) artnet = h["artnet"] flow.add( { "id": "universe", "type": "artnet", "title": f"Art-Net universe {artnet['universe']}", "params": { "host": artnet["host"], "port": artnet["port"], "universe": artnet["universe"], "channels": channels, # Levels the house is already at, so the first frame does not # darken every channel this node is not driving. Fill it in # from what the fixtures are doing before switching over. "baseline": artnet.get("baseline", {}), "poll_interval": artnet.get("refresh_s", 0), # Off until Node-RED's own sender is stopped. "transmit": artnet.get("transmit", False), }, "requires": artnet_in, } ) return flow # ── the motors ─────────────────────────────────────────────────────────── POSITION = '''"""Which way a motor last went, and whether it is still going. The bed shutter being down is how the house knows someone is asleep, and the door shutter moving is why the frost cable stands aside for a minute — both read well outside the flow that moves them. """ import time def process(state): return { "down": state.get("position") == "DOWN", "moving": time.time() < state.get("moving_until", 0.0), } ''' def _motor_chain( flow: Flow, name: str, title: str, spec: dict[str, Any], cmd: str ) -> None: """Command -> run-time -> the one STOP that ends it.""" flow.add( motor( f"{name}_motor", f"{title}: how long to run", cmd=cmd, run=f"{name}_run", run_for=f"{name}_run_for", state=f"{name}_state", up_s=spec["up_s"], down_s=spec["down_s"], ) ) flow.add( stopper( f"{name}_stop", f"{title}: stop when it is there", run=f"{name}_run", run_for=f"{name}_run_for", cmd=name, ) ) flow.input(f"{name}_state", "record", {}) def shutters(h: dict[str, Any]) -> Flow: """The two rollershutters, driven from the screen and nothing else.""" flow = Flow("shutters", "Rollershutters") for name, title in (("door_shutter", "Door"), ("bed_shutter", "Bed")): _motor_chain(flow, name, title, h["dmx"][name], cmd=f"{name}_cmd") flow.input(f"{name}_cmd", "str", "STOP") for name, out in (("bed_shutter", "bed_down"), ("door_shutter", "door_down")): flow.add( { "id": f"{out}_at", "type": "python", "title": f"Is the {name.split('_')[0]} shutter down?", "requires": [ {"name": f"{name}_state", "port": "state", "dtype": "record"} ], "provides": [ {"name": out, "port": "down", "dtype": "bool"}, { "name": f"{name.split('_')[0]}_moving", "port": "moving", "dtype": "bool", }, ], }, POSITION, ) return flow WINDOW = '''"""Whether the storage window should be open. The reference's rules, in the order it applied them. Two of them are about not fighting something else — a stove or a heat pump running with a window open is money out of the window, literally — and the rest is the one genuinely useful idea in the whole installation: open when the air outside is *drier*, not just cooler, because that is what actually takes damp out of a small house. 100 closes and 0 opens, which is the DMX motor's own convention. """ OPEN, CLOSED = "DOWN", "UP" def process( indoor, t_max, top=None, outdoor=None, humidity=None, dewpoint_delta=None, stratified=False, oven_on=False, hvac_on=False, hvac_mode="fan", sleeping=False, high=25.0, lift=0.0, damp=65.0, dry_enough=4.0, ): if oven_on: return {"want": CLOSED, "why": "the stove is lit"} if hvac_on and hvac_mode != "fan": return {"want": CLOSED, "why": "the heat pump is running"} if sleeping: return {"want": CLOSED, "why": "asleep"} t_high = high + lift warm = top if top is not None else indoor if indoor > t_high: if outdoor is not None and outdoor < indoor: return {"want": OPEN, "why": "cooler outside"} return {"want": CLOSED, "why": "no cooler outside"} if ( indoor < t_high - 1 and humidity is not None and humidity > damp and dewpoint_delta is not None and dewpoint_delta > dry_enough ): return {"want": OPEN, "why": "damp inside, drier outside"} return {"want": CLOSED, "why": "nothing to gain"} ''' def window(h: dict[str, Any]) -> Flow: flow = Flow("window", "Window opener") flow.add( { "id": "decide", "type": "python", "title": "Open or closed", "requires": [ {"name": "climate.indoor", "port": "indoor", "dtype": "float"}, { "name": "climate.t_max", "port": "t_max", "dtype": "float", "trigger": False, }, { "name": "climate.top", "port": "top", "dtype": "float", "trigger": False, }, { "name": "climate.lift", "port": "lift", "dtype": "float", "trigger": False, }, { "name": "climate.stratified", "port": "stratified", "dtype": "bool", "trigger": False, }, { "name": "weather.outdoor_temp", "port": "outdoor", "dtype": "float", "trigger": False, }, { "name": "weather.indoor_hum", "port": "humidity", "dtype": "float", "trigger": False, }, { "name": "weather.dewpoint_delta", "port": "dewpoint_delta", "dtype": "float", "trigger": False, }, { "name": "oven.running", "port": "oven_on", "dtype": "bool", "trigger": False, }, { "name": "hvac.running", "port": "hvac_on", "dtype": "bool", "trigger": False, }, { "name": "hvac.reported_mode", "port": "hvac_mode", "dtype": "str", "trigger": False, }, { "name": "presence.sleeping", "port": "sleeping", "dtype": "bool", "trigger": False, }, ], "provides": [ {"name": "want", "dtype": "str"}, {"name": "why", "dtype": "str"}, ], }, WINDOW, ) flow.add( { "id": "settle", "type": "delay", "title": "At most every ten minutes", "params": {"interval": 600.0}, "requires": [{"name": "want", "dtype": "str"}], "provides": [{"name": "settled", "dtype": "str"}], } ) flow.add( arbiter( "arbiter", "Automation or the button", "str", auto="settled", manual="window_manual", command="window_cmd", state="window_arbiter", hold_s=7200.0, ) ) _motor_chain( flow, "window_opener", "Window", h["dmx"]["window_opener"], cmd="window_cmd" ) flow.input("window_manual", "str", "UP") flow.input("window_arbiter", "record", {}) return flow CANOPY = '''"""Whether the awning should be out. An awning is the one thing in the house that the weather can destroy, so this reads as a list of reasons to bring it in and one reason to put it out. Every threshold is the reference's; the wind one especially is not a preference. Trust matters here more than anywhere else: a weather station that has gone quiet is a reason to retract, not a reason to assume it is calm. """ OUT, IN = "DOWN", "UP" def process( home=True, door_down=False, light=None, humidity=None, rainrate=0.0, rain_soon=False, outdoor=10.0, wind=0.0, station_trust=0.0, min_trust=0.3, max_wind=3.0, max_humidity=96.0, min_temp=1.0, sun_temp=26.0, sun_light=80.0, ): if station_trust < min_trust: return {"want": IN, "why": "no recent weather reading"} if not home: return {"want": IN, "why": "nobody in"} if door_down: return {"want": IN, "why": "the house is shut up"} if wind > max_wind: return {"want": IN, "why": "too windy"} if rainrate > 0 or rain_soon: return {"want": IN, "why": "rain"} if humidity is not None and humidity > max_humidity: return {"want": IN, "why": "wet"} if outdoor < min_temp: return {"want": IN, "why": "freezing"} if light is not None and light < 0.1: return {"want": IN, "why": "dark"} if outdoor > sun_temp and light is not None and light > sun_light: return {"want": OUT, "why": "sunny"} return {"want": IN, "why": "nothing to shade"} ''' def canopy(h: dict[str, Any]) -> Flow: flow = Flow("canopy", "Awning") flow.add( { "id": "decide", "type": "python", "title": "Out or in", "requires": [ { "name": "weather.station_trust", "port": "station_trust", "dtype": "float", }, {"name": "presence.home", "port": "home", "dtype": "bool"}, { "name": "shutters.door_down", "port": "door_down", "dtype": "bool", "trigger": False, }, { "name": "weather.light", "port": "light", "dtype": "float", "trigger": False, }, { "name": "weather.outdoor_hum", "port": "humidity", "dtype": "float", "trigger": False, }, { "name": "weather.rainrate", "port": "rainrate", "dtype": "float", "trigger": False, }, { "name": "weather.rain_soon", "port": "rain_soon", "dtype": "bool", "trigger": False, }, { "name": "weather.outdoor_temp", "port": "outdoor", "dtype": "float", "trigger": False, }, { "name": "weather.wind", "port": "wind", "dtype": "float", "trigger": False, }, ], "provides": [ {"name": "want", "dtype": "str"}, {"name": "why", "dtype": "str"}, ], }, CANOPY, ) flow.add( { "id": "settle", "type": "delay", "title": "At most every twenty minutes", "params": {"interval": 1200.0}, "requires": [{"name": "want", "dtype": "str"}], "provides": [{"name": "settled", "dtype": "str"}], } ) flow.add( arbiter( "arbiter", "Automation or the button", "str", auto="settled", manual="canopy_manual", command="canopy_cmd", state="canopy_arbiter", # Shorter than the rest on purpose: an override that outlives the # weather is how an awning ends up in a storm. hold_s=3600.0, ) ) _motor_chain(flow, "canopy", "Awning", h["dmx"]["canopy"], cmd="canopy_cmd") flow.input("canopy_manual", "str", "UP") flow.input("canopy_arbiter", "record", {}) return flow # ── plugs ──────────────────────────────────────────────────────────────── PLUGS = '''"""The four plugs whose rule is a sentence, decided together. They share one node because they share their inputs and each is two lines; five nodes reading the same six messages would be the shape this framework exists to avoid. The fridge pump is the odd one — it is a pulse rather than a state, so what it gets here is permission, and a trigger downstream gives it its thirty seconds. """ def process( home=True, bed_down=False, hour=12, watch=0, oven_on=False, sleeping=False, today_max=15.0, day_from=10, day_to=16, morning_from=6, evening_from=20, vent_warm_c=16.0, ): if watch: # Nothing optional runs while the electricity is in trouble. return {"bath": False, "bed_plug": bed_down, "inlet": 0.0, "fridge": 0.0} if day_from <= hour < day_to: bath = True elif home and (morning_from <= hour < day_from or hour >= evening_from): bath = True else: bath = False # The fans: the reference's table, which is about how much the fridge and # the inlet have to work rather than about comfort. if not home: inlet, fridge = 255.0, 255.0 elif today_max > vent_warm_c: inlet, fridge = 0.0, 150.0 elif oven_on and not sleeping: inlet, fridge = 255.0, 200.0 elif sleeping: inlet, fridge = 0.0, 175.0 else: inlet, fridge = 0.0, 0.0 return { "bath": bath, # The bedroom plug is the bed being down and nothing else. "bed_plug": bool(bed_down), "inlet": inlet, "fridge": fridge, } ''' PUMP_DUE = '''"""The fridge pump wants thirty seconds a day, while nobody is in. Standing water in a pump that never runs is what this is about, so it is once a day rather than on any condition worth reasoning about. """ import time DAY_S = 82800.0 def process(home=True, hour=12, memory=None, after_hour=13, run_s=30.0): memory = dict(memory or {}) now = time.time() last = memory.get("last", 0.0) due = not home and hour >= after_hour and now - last > DAY_S if not due: return None return {"pump": True, "pump_for": run_s, "pump_memory": {"last": now}} ''' def plugs(h: dict[str, Any]) -> Flow: flow = Flow("plugs", "Plugs") flow.add( { "id": "decide", "type": "python", "title": "Bathroom, bedroom and the fans", "requires": [ {"name": "clock.hour", "port": "hour", "dtype": "int"}, {"name": "presence.home", "port": "home", "dtype": "bool"}, {"name": "presence.sleeping", "port": "sleeping", "dtype": "bool"}, {"name": "shutters.bed_down", "port": "bed_down", "dtype": "bool"}, { "name": "power.watch", "port": "watch", "dtype": "int", "trigger": False, }, { "name": "oven.running", "port": "oven_on", "dtype": "bool", "trigger": False, }, { "name": "weather.today_max", "port": "today_max", "dtype": "float", "trigger": False, }, ], "provides": [ {"name": "bath_auto", "port": "bath", "dtype": "bool"}, {"name": "bed_plug", "port": "bed_plug", "dtype": "bool"}, {"name": "inlet_vent", "port": "inlet", "dtype": "float"}, {"name": "fridge_vent", "port": "fridge", "dtype": "float"}, ], }, PLUGS, ) flow.add( arbiter( "bath_arbiter", "Bathroom plug: automation or the switch", "bool", auto="bath_auto", manual="bath_manual", command="bath_plug", state="bath_arbiter", hold_s=3600.0, ) ) flow.add( { "id": "pump_due", "type": "python", "title": "Fridge pump: due?", "requires": [ {"name": "presence.home", "port": "home", "dtype": "bool"}, {"name": "clock.hour", "port": "hour", "dtype": "int"}, { "name": "pump_memory", "port": "memory", "dtype": "record", "trigger": False, }, ], "provides": [ {"name": "pump_start", "port": "pump", "dtype": "bool"}, {"name": "pump_for", "port": "pump_for", "dtype": "float"}, {"name": "pump_memory", "port": "pump_memory", "dtype": "record"}, ], }, PUMP_DUE, ) flow.add( { "id": "pump_run", "type": "trigger", "title": "Fridge pump: thirty seconds", "params": { "first": True, "then": False, "wait": 30.0, "wait_port": "pump_for", "extend": False, }, "requires": [ {"name": "pump_start", "port": "start", "dtype": "bool"}, {"name": "pump_for", "port": "pump_for", "dtype": "float"}, ], "provides": [{"name": "fridge_pump", "dtype": "bool"}], } ) flow.input("bath_manual", "bool", False) flow.input("bath_arbiter", "record", {}) flow.input("pump_memory", "record", {}) # Nothing else provides it, and the plug has to start from somewhere. flow.input("fridge_pump", "bool", False) return flow # ── appliances ─────────────────────────────────────────────────────────── def appliances() -> Flow: """One switch, and a promise that it is off by morning. The reference armed a delay of `24 - hour` hours whenever the plug went on, which is the same intent expressed as arithmetic. A schedule says it better: at two in the morning the plug goes off, and it goes off even if somebody switched it on at midnight — which is exactly the case a hold would otherwise defeat, so this is the one arbiter with a force behind it. """ flow = Flow("appliances", "Appliances") flow.add( { "id": "nightly", "type": "inject", "title": "At two in the morning", "params": {"cron": "0 2 * * *"}, "provides": [{"name": "nightly_off", "dtype": "float"}], } ) flow.add( { "id": "auto", "type": "python", "title": "Nothing to decide", "requires": [{"name": "nightly_off", "port": "tick", "dtype": "float"}], "provides": [{"name": "auto", "dtype": "bool"}], }, '''"""Nobody automates a washing machine; the schedule is the whole rule.""" def process(tick): return {"auto": False} ''', ) flow.add( arbiter( "arbiter", "The switch, until two in the morning", "bool", auto="auto", manual="appliances_manual", command="appliances", state="appliances_arbiter", # Held until the schedule releases it, however long that is. hold_s=0.0, force_at="nightly_off", force_value=False, ) ) flow.input("appliances_manual", "bool", False) flow.input("appliances_arbiter", "record", {}) return flow # ── the outdoor plug ───────────────────────────────────────────────────── OUTDOOR = '''"""One plug, two jobs, and the calendar decides which. In winter it is a frost-protection cable and in summer it is an irrigation valve — the reference relabelled a dashboard button twice a year to say so. The cable is a duty cycle rather than a thermostat: there is no sensor in the pipe, so it runs for part of each hour and for longer the colder it is. That is what the reference expressed as a delay it computed in milliseconds; written against the clock it is one line and it cannot leave the plug on because a message went missing. Watering is the same idea with the season the other way round: five minutes at eight and five at seven, on a day warm enough to be worth it. """ def process( hour=12, minute=0, outdoor=10.0, watering_season=False, frost_season=False, today_max=15.0, out_w=0.0, door_moving=False, memory=None, frost_on_c=0.2, frost_off_c=1.2, duty_min=15.0, duty_max=59.0, duty_from_c=4.0, duty_to_c=-5.0, overload_w=2000.0, water_at=(8, 19), water_min=5.0, water_above_c=17.0, ): memory = dict(memory or {}) if watering_season: wanted = ( today_max > water_above_c and hour in tuple(water_at) and minute < water_min ) return { "want": bool(wanted), "why": "watering" if wanted else "between waterings", "outdoor_memory": memory, } if not frost_season: return {"want": False, "why": "neither season", "outdoor_memory": memory} if out_w > overload_w: return {"want": False, "why": "the inverter is loaded", "outdoor_memory": memory} if door_moving: # They share an inverter phase; the reference kept them apart and the # cheapest way to keep doing that is to wait a minute. return {"want": False, "why": "the door is moving", "outdoor_memory": memory} engaged = bool(memory.get("engaged")) if outdoor < frost_on_c: engaged = True elif outdoor > frost_off_c: engaged = False span = max(0.1, duty_from_c - duty_to_c) duty = duty_min + (duty_max - duty_min) * (duty_from_c - outdoor) / span duty = max(duty_min, min(duty_max, duty)) return { "want": bool(engaged and minute < duty), "why": f"frost cable, {round(duty)} min an hour" if engaged else "not freezing", "outdoor_memory": {"engaged": engaged}, } ''' def outdoor() -> Flow: flow = Flow("outdoor", "Outdoor plug") flow.add( { "id": "decide", "type": "python", "title": "Frost cable or watering", "requires": [ {"name": "clock.minute", "port": "minute", "dtype": "int"}, { "name": "clock.hour", "port": "hour", "dtype": "int", "trigger": False, }, { "name": "weather.outdoor_temp", "port": "outdoor", "dtype": "float", "trigger": False, }, { "name": "weather.watering_season", "port": "watering_season", "dtype": "bool", "trigger": False, }, { "name": "weather.frost_season", "port": "frost_season", "dtype": "bool", "trigger": False, }, { "name": "weather.today_max", "port": "today_max", "dtype": "float", "trigger": False, }, { "name": "power.out_w", "port": "out_w", "dtype": "float", "trigger": False, }, { "name": "shutters.door_moving", "port": "door_moving", "dtype": "bool", "trigger": False, }, { "name": "outdoor_memory", "port": "memory", "dtype": "record", "trigger": False, }, ], "provides": [ {"name": "want", "dtype": "bool"}, {"name": "why", "dtype": "str"}, {"name": "outdoor_memory", "port": "outdoor_memory", "dtype": "record"}, ], }, OUTDOOR, ) flow.add( arbiter( "arbiter", "Automation or the button", "bool", auto="want", manual="outdoor_manual", command="outdoor_plug", state="outdoor_arbiter", # A minute is enough to water something by hand, and an override # that outlasts the frost cable's next hour is a frozen pipe. hold_s=900.0, ) ) flow.input("outdoor_manual", "bool", False) flow.input("outdoor_arbiter", "record", {}) return flow # ── lights ─────────────────────────────────────────────────────────────── AUTO_SCENE = '''"""The scene the house would pick, left to itself. Only on a change of who is in or whether it is dark, because a scene that re-asserts itself every minute is a scene nobody can override by hand — the arbiter downstream would hold, but the house would keep asking, and the log would be unreadable. """ def process(event=None, phase="day", door_down=False): if event == "away": return {"auto": "off"} if event == "asleep": return {"auto": "sleep"} if event == "arrived": return {"auto": "day" if phase == "day" else "outside"} if event in ("home", "working"): return {"auto": "day" if phase == "day" else "night"} return None ''' SCENE = '''"""A scene is five zones and a brightness it starts from. The table is the reference's. What it does not do any more is decide the brightness for good: the level it names is what the house *suggests*, and the slider on the wall overrides it through the same arbiter every other actuator uses — so picking "sleep" dims the house to twenty, and turning it back up afterwards stays turned up. """ ZONES = ("bed", "kitchen", "bath", "workspace", "outside") SCENES = { "off": (0, ()), "sleep": (20, ("bed", "bath")), "day": (70, ("workspace",)), "night": (40, ("kitchen", "workspace", "outside")), "outside": (80, ("workspace", "outside")), "alarm": (100, ZONES), } def process(scene="off"): level, zones = SCENES.get(scene, SCENES["off"]) return {"suggested": float(level), "zones": list(zones)} ''' FIXTURES = '''"""Zones and a brightness into one value for every fixture in the house. Every fixture is named in `DARK` before any zone is applied, which is the whole point of writing it as a table: in the reference a zone that was not selected left its fixtures holding the JavaScript value `undefined`, which went onto the wire as an empty payload and reached the DMX encoder as a NaN. Off is a value here, not an absence. Zones are applied in order and a later one wins — which is how the reference behaved, and what makes `workspace` the zone that decides the living room when both it and `kitchen` are lit. """ DARK = { "living_ambient": [0.0, 0.0, 0.0], "kitchen_ambient": [0.0, 0.0, 0.0], "bath_light": [0.0, 0.0, 0.0], "kitchen_light": [0.0, 0.0, 0.0], "bed_light": [0.0, 0.0, 0.0], "kitchen_direct": 0.0, "traverse_spot": 0.0, "outdoor_ambient": False, "outdoor_direct": False, "canopy_light": False, } def process(zones, level=70.0, door_down=False, spot_above=20.0): out = dict(DARK) level = float(level) half = level / 2.0 for zone in zones: if zone == "bed": out["bed_light"] = [10.0, 100.0, level] out["kitchen_ambient"] = [10.0, 60.0, half] elif zone == "kitchen": out["kitchen_ambient"] = [0.0, 60.0, level] out["living_ambient"] = [10.0, 100.0, half] out["kitchen_direct"] = level elif zone == "bath": out["living_ambient"] = [10.0, 100.0, half] out["bath_light"] = [0.0, 0.0, level] elif zone == "workspace": out["living_ambient"] = [0.0, 20.0, level] out["traverse_spot"] = level if level > spot_above else 0.0 out["kitchen_ambient"] = [0.0, 90.0, half] elif zone == "outside" and not door_down: # Shut up for the night: no point lighting the terrace. out["outdoor_ambient"] = level > 10 out["outdoor_direct"] = level > 70 out["canopy_light"] = level > 30 out["lit"] = ", ".join(zones) if zones else "nothing" return out ''' def lights() -> Flow: flow = Flow("lights", "Lights") flow.add( { "id": "auto", "type": "python", "title": "What the house would pick", "requires": [ {"name": "presence.event", "port": "event", "dtype": "str"}, { "name": "weather.phase", "port": "phase", "dtype": "str", "trigger": False, }, { "name": "shutters.door_down", "port": "door_down", "dtype": "bool", "trigger": False, }, ], "provides": [{"name": "auto", "dtype": "str"}], }, AUTO_SCENE, ) flow.add( arbiter( "scene_arbiter", "The scene: automation or the dial", "str", auto="auto", manual="scene", command="active_scene", state="scene_state", # Long: a scene someone chose should last the evening. hold_s=14400.0, ) ) flow.add( { "id": "scene", "type": "python", "title": "Scene to zones", "requires": [{"name": "active_scene", "port": "scene", "dtype": "str"}], "provides": [ {"name": "suggested", "dtype": "float"}, {"name": "zones", "dtype": "list", "item": "str"}, ], }, SCENE, ) flow.add( arbiter( "brightness_arbiter", "Brightness: the scene or the slider", "float", auto="suggested", manual="brightness", command="level", state="brightness_state", hold_s=14400.0, ) ) flow.add( { "id": "fixtures", "type": "python", "title": "Zones to fixtures", "requires": [ {"name": "zones", "dtype": "list", "item": "str"}, {"name": "level", "dtype": "float"}, { "name": "shutters.door_down", "port": "door_down", "dtype": "bool", "trigger": False, }, ], "provides": [ {"name": "living_ambient", "dtype": "list", "item": "float"}, {"name": "kitchen_ambient", "dtype": "list", "item": "float"}, {"name": "bath_light", "dtype": "list", "item": "float"}, {"name": "kitchen_light", "dtype": "list", "item": "float"}, {"name": "bed_light", "dtype": "list", "item": "float"}, {"name": "kitchen_direct", "dtype": "float"}, {"name": "traverse_spot", "dtype": "float"}, {"name": "outdoor_ambient", "dtype": "bool"}, {"name": "outdoor_direct", "dtype": "bool"}, {"name": "canopy_light", "dtype": "bool"}, {"name": "lit", "dtype": "str"}, ], }, FIXTURES, ) flow.input("scene", "str", "off") flow.input("scene_state", "record", {}) flow.input("brightness", "float", 70.0) flow.input("brightness_state", "record", {}) return flow