"""What the house wants the temperature to be, and what it runs to get there. Four flows and one idea: `climate` works out the bands, and the three things that can change the temperature — the heat pump, the pellet stove, the water boilers competing for the same electricity — read them rather than each deriving their own. The reference computed a winter factor in four places with four different constants; here the calendar part is the clock's and the rest is one node with the constants as settings. """ from __future__ import annotations from typing import Any from .api import Flow from .library import arbiter from .sensing import _broker # ── climate ────────────────────────────────────────────────────────────── BANDS = '''"""The temperature bands, by who is in and what time of year it is. Every threshold below is the reference's, and every one of them is a setting, because they are opinions about this house rather than physics. The winter factor lifts the whole band as it gets colder outside: at -5 the house is allowed to be warmer before it counts as too warm, which is what stops the heating fighting the weather in January. `indoor` is deliberately not the mean of the two sensors. The bottom sensor is where a person is and the top one is where the heat goes, so the bottom one carries the weight — and when it is silent, which it is while the 433 MHz station is dead, the top one is the whole reading rather than nothing at all. """ def process( top, bottom=None, outdoor=10.0, preset=21.0, sleeping=False, home=True, winter_k=0.6, winter_from=9.5, night_max=23.0, night_min=19.5, away_max=20.0, away_min=18.0, band=1.5, bottom_weight=0.75, ): lift = winter_k * max(0.0, (winter_from - outdoor)) / winter_from indoor = top if bottom is None else (1 - bottom_weight) * top + bottom_weight * bottom if not home: low, high = away_min, away_max elif sleeping: low, high = night_min, night_max else: low, high = preset - band, preset return { "indoor": round(indoor, 2), "top": top, "lift": round(lift, 2), "setpoint": round(preset + lift, 1), "t_min": round(low + lift, 2), "t_max": round(high + lift, 2), "stratified": bottom is not None and (top - bottom) > 6, } ''' def climate() -> Flow: flow = Flow("climate", "Climate") flow.add( { "id": "bands", "type": "python", "title": "Temperature bands", "requires": [ # The aircon reports the top of the room and is polled, so it # is the one input that is always there. {"name": "hvac.indoor_temp", "port": "top", "dtype": "float"}, { "name": "weather.indoor_temp", "port": "bottom", "dtype": "float", "trigger": False, }, { "name": "weather.outdoor_temp", "port": "outdoor", "dtype": "float", "trigger": False, }, {"name": "preset", "dtype": "float", "trigger": False}, {"name": "presence.sleeping", "port": "sleeping", "dtype": "bool"}, {"name": "presence.home", "port": "home", "dtype": "bool"}, ], "provides": [ {"name": "indoor", "dtype": "float"}, {"name": "top", "dtype": "float"}, {"name": "lift", "dtype": "float"}, {"name": "setpoint", "dtype": "float"}, {"name": "t_min", "dtype": "float"}, {"name": "t_max", "dtype": "float"}, {"name": "stratified", "dtype": "bool"}, ], }, BANDS, ) flow.input("preset", "float", 21.0) return flow # ── heat pump ──────────────────────────────────────────────────────────── HVAC = '''"""What to ask of the heat pump, if anything. Ported from the reference's `HVAC Logic`, which is the most conditional thing in that installation and the one worth keeping closest to what it was: every branch is a bill, and the state of charge gates that appear arbitrary are the difference between running the compressor off the sun and running it off the battery at four in the morning. What changed: it decides in one place instead of reading nine globals, the enable switch is honoured rather than being overwritten with true on the way past, and it says why. Modes are the unit's own: 1 cool, 2 heat, 3 fan, 4 dry. """ OFF = {"operation": False, "mode": "fan", "preset_temp": 22.0, "fan_speed": "auto"} def process( indoor, top, t_min, t_max, lift=0.0, humidity=None, stratified=False, soc=100.0, out_w=0.0, in_v=230.0, watch=0, enabled=True, home=True, sleeping=False, oven_on=False, hot=29.0, high=26.0, cool_soc=60.0, cool_soc_away=90.0, fan_soc=45.0, dry_soc=35.0, dry_humidity=69.0, dry_humidity_night=73.0, ): if not enabled or watch: return {"command": OFF, "why": "off: " + ("disabled" if not enabled else "power")} t_hot, t_high = hot + lift, high + lift def ask(mode, temp=None, fan="auto", why=""): return { "command": { "operation": True, "mode": mode, "preset_temp": round(temp if temp is not None else t_max, 1), "fan_speed": fan, }, "why": why, } if not home: # Nobody in: only ever to stop the house cooking, and only on the sun. if top > t_hot and soc > cool_soc_away: return ask("cool", t_hot, why="empty house is too hot") if humidity is not None and humidity > 60 and soc > dry_soc: return ask("dry", t_min + 3, why="empty house is damp") return {"command": OFF, "why": "nobody in"} if stratified: # Warm air at the ceiling and cold feet: move it rather than make more. return ask("fan", fan="high" if oven_on else "low", why="stratified") limit = t_max if not sleeping else t_max + 1.5 if top > limit: if soc > (cool_soc if not sleeping else cool_soc_away): return ask("cool", t_hot, why="too warm") if soc > fan_soc: return ask("fan", why="too warm, saving the battery") return {"command": OFF, "why": "too warm, battery too low to help"} damp = dry_humidity_night if sleeping else dry_humidity if humidity is not None and humidity > damp and in_v > 200: return ask("dry", t_min + 3, fan="low", why="damp") if indoor < t_min and soc > dry_soc: return ask("heat", t_min + 3, why="too cold") return {"command": OFF, "why": "within the band"} ''' HVAC_LIMIT = '''"""Never let the compressor be the thing that trips the inverter.""" def process(command, out_w=0.0, overload_w=2000.0): if command.get("mode") == "cool" and out_w > overload_w: return {"limited": {**command, "mode": "fan"}} return {"limited": command} ''' HVAC_PORTS = '''"""One command record into the four ports the connector writes.""" def process(limited): return { "operation": bool(limited["operation"]), "mode": str(limited["mode"]), "preset_temp": float(limited["preset_temp"]), "fan_speed": str(limited["fan_speed"]), } ''' def hvac(h: dict[str, Any], commands: bool) -> Flow: flow = Flow("hvac", "Heat pump") # Reading and commanding are two nodes on one adapter, not one node doing # both. A node that did both would sit in its own cascade: the temperature # it reports decides the bands, the bands decide the command, and the # command comes back to it — a cycle the canvas would refuse to run. The # connector already keeps two nodes on one unit from talking over each # other, which is what makes the split safe rather than merely legal. flow.add( { "id": "unit", "type": "wfrac", "title": "Aircon (reading)", "params": { "host": h["aircon"]["host"], "poll_interval": 60.0, "commands": False, }, "provides": [ {"name": "indoor_temp", "dtype": "float"}, {"name": "outdoor_temp", "dtype": "float"}, {"name": "reported_temp", "port": "preset_temp", "dtype": "float"}, {"name": "running", "port": "operation", "dtype": "bool"}, {"name": "reported_mode", "port": "mode", "dtype": "str"}, {"name": "electric", "dtype": "float"}, ], } ) flow.add( { "id": "writer", "type": "wfrac", "title": "Aircon (commanding)", "params": { "host": h["aircon"]["host"], # Nothing to poll here; the reader above owns that. "poll_interval": 0.0, # Off until someone is watching: turning a heat pump on has a # bill attached. "commands": commands, }, "requires": [ {"name": "operation", "dtype": "bool"}, {"name": "mode", "dtype": "str"}, {"name": "preset_temp", "dtype": "float"}, {"name": "fan_speed", "dtype": "str"}, ], } ) flow.add( { "id": "second", "type": "wfrac", "title": "Aircon (bedroom, read only)", "params": { "host": h["aircon"]["second_host"], "poll_interval": 300.0, "commands": False, }, "provides": [ {"name": "second_indoor", "port": "indoor_temp", "dtype": "float"}, {"name": "second_running", "port": "operation", "dtype": "bool"}, ], } ) flow.add( { "id": "decide", "type": "python", "title": "What to ask of it", "requires": [ {"name": "climate.indoor", "port": "indoor", "dtype": "float"}, { "name": "climate.top", "port": "top", "dtype": "float", "trigger": False, }, { "name": "climate.t_min", "port": "t_min", "dtype": "float", "trigger": False, }, { "name": "climate.t_max", "port": "t_max", "dtype": "float", "trigger": False, }, { "name": "climate.lift", "port": "lift", "dtype": "float", "trigger": False, }, { "name": "climate.stratified", "port": "stratified", "dtype": "bool", "trigger": False, }, { "name": "weather.indoor_hum", "port": "humidity", "dtype": "float", "trigger": False, }, { "name": "power.soc", "port": "soc", "dtype": "float", "trigger": False, }, { "name": "power.out_w", "port": "out_w", "dtype": "float", "trigger": False, }, { "name": "power.in_v", "port": "in_v", "dtype": "float", "trigger": False, }, {"name": "power.watch", "port": "watch", "dtype": "int"}, { "name": "presence.home", "port": "home", "dtype": "bool", "trigger": False, }, { "name": "presence.sleeping", "port": "sleeping", "dtype": "bool", "trigger": False, }, { "name": "oven.running", "port": "oven_on", "dtype": "bool", "trigger": False, }, {"name": "enabled", "dtype": "bool", "trigger": False}, ], "provides": [ {"name": "wanted", "port": "command", "dtype": "record"}, {"name": "why", "dtype": "str"}, ], }, HVAC, ) flow.add( { "id": "limit", "type": "python", "title": "Not while the inverter is loaded", "requires": [ {"name": "wanted", "port": "command", "dtype": "record"}, { "name": "power.out_w", "port": "out_w", "dtype": "float", "trigger": False, }, ], "provides": [{"name": "limited", "dtype": "record"}], }, HVAC_LIMIT, ) flow.add( { "id": "settle", "type": "delay", "title": "At most every fifteen minutes", # A heat pump that is asked something new every minute never # reaches anything. The reference used the same figure. "params": {"interval": 900.0}, "requires": [{"name": "limited", "dtype": "record"}], "provides": [{"name": "settled", "dtype": "record"}], } ) flow.add( { "id": "changed", "type": "rbe", "title": "Only when it changes", "requires": [{"name": "settled", "port": "settled", "dtype": "record"}], "provides": [{"name": "command", "port": "command", "dtype": "record"}], } ) flow.add( { "id": "ports", "type": "python", "title": "Into the unit's ports", "requires": [{"name": "command", "port": "limited", "dtype": "record"}], "provides": [ {"name": "operation", "dtype": "bool"}, {"name": "mode", "dtype": "str"}, {"name": "preset_temp", "dtype": "float"}, {"name": "fan_speed", "dtype": "str"}, ], }, HVAC_PORTS, ) flow.input("enabled", "bool", True) return flow # ── pellet stove ───────────────────────────────────────────────────────── OVEN = '''"""Whether the pellet stove should be burning. The reference decided this by pushing a vote into a twenty-sample array every three seconds and firing when the mean crossed 0.9 — which is a way of building hysteresis out of a heartbeat. With no heartbeat there is no window to average over, so the hysteresis is written down instead: light it below the bottom of the band, stop it above the top, and do nothing in between. That is what the vote was approximating, and it is legible. None of this ran in the reference at all — the controller had been disconnected from its heartbeat. It is reinstated here because heating the house is what the stove is for. """ def process( indoor, t_min, t_max, running=False, heating_season=True, faulty=False, watch=0, margin=0.0, ): if not heating_season: return {"wanted": False, "why": "not the season"} if faulty: return {"wanted": False, "why": "the stove reports a fault"} if watch >= 4: return {"wanted": False, "why": "mains is down"} if not running and indoor <= t_min - margin: return {"wanted": True, "why": "below the band"} if running and indoor >= t_max + margin: return {"wanted": False, "why": "above the band"} return {"wanted": running, "why": "within the band"} ''' OVEN_STATE = '''"""What the stove says about itself, and whether to believe it. A stove that has claimed to be on for five hours without the room warming up is stuck, and the reference used exactly that to stop the heat pump trying to heat with it. Kept, including the five hours. """ import time def process(onoff=None, power_state=None, memory=None, faulty_after_s=18000.0): state = str(onoff if onoff is not None else power_state or "").upper() if not state: return None running = state in ("ON", "1", "TRUE") memory = dict(memory or {}) now = time.time() since = memory.get("since", now) if running != memory.get("running"): since = now return { "running": running, "faulty": bool(running and now - since > faulty_after_s), "since": since, "oven_memory": {"running": running, "since": since}, } ''' OVEN_CMD = '''"""The stove takes ON to light and 'force' to shut down. Nothing else.""" def process(command): return {"oven_command": "ON" if command else "force"} ''' PUMP = '''"""The floor-heating pump follows the stove, late and then later. The plug labelled microwave in the reference is wired to the pumps that move the water the stove heats. Running them the moment it lights pushes cold water through a cold house, and stopping them the moment it goes out leaves the heat in the stove — so the pump starts a quarter of an hour behind and runs three hours past. Both are settings; they are properties of this plumbing. It reads the clock rather than scheduling anything, which is what makes a stove that lights and goes straight back out harmless: there is no pending message to arrive after the reason for it has gone. """ import time def process(running, since=0.0, tick=0, on_after_s=900.0, off_after_s=10800.0): if not since: return {"floor_heating": False} elapsed = time.time() - since if running: return {"floor_heating": elapsed >= on_after_s} return {"floor_heating": elapsed < off_after_s} ''' def oven(h: dict[str, Any]) -> Flow: flow = Flow("oven", "Pellet stove") topics = h["topics"] flow.add( { "id": "stove_in", "type": "mqtt", "title": "What the stove says", "params": { "topic": { "onoff": topics["oven_state"], "power_state": topics["oven_power_state"], "ambient": topics["oven_ambient"], "fume": topics["oven_fume"], }, **_broker(h, "fluksio-oven"), }, "provides": [ {"name": "onoff", "dtype": "str"}, {"name": "power_state", "dtype": "str"}, {"name": "ambient", "dtype": "float"}, {"name": "fume", "dtype": "float"}, ], } ) flow.add( { "id": "state", "type": "python", "title": "Running, and healthy?", "requires": [ {"name": "onoff", "dtype": "str"}, {"name": "power_state", "dtype": "str", "trigger": False}, { "name": "oven_memory", "port": "memory", "dtype": "record", "trigger": False, }, ], "provides": [ {"name": "running", "dtype": "bool"}, {"name": "faulty", "dtype": "bool"}, {"name": "since", "dtype": "float"}, {"name": "oven_memory", "port": "oven_memory", "dtype": "record"}, ], }, OVEN_STATE, ) flow.add( { "id": "decide", "type": "python", "title": "Should it be burning?", "requires": [ {"name": "climate.indoor", "port": "indoor", "dtype": "float"}, { "name": "climate.t_min", "port": "t_min", "dtype": "float", "trigger": False, }, { "name": "climate.t_max", "port": "t_max", "dtype": "float", "trigger": False, }, {"name": "running", "dtype": "bool", "trigger": False}, { "name": "weather.heating_season", "port": "heating_season", "dtype": "bool", "trigger": False, }, {"name": "faulty", "dtype": "bool", "trigger": False}, { "name": "power.watch", "port": "watch", "dtype": "int", "trigger": False, }, ], "provides": [ {"name": "wanted", "dtype": "bool"}, {"name": "why", "dtype": "str"}, ], }, OVEN, ) flow.add( arbiter( "arbiter", "Automation or the button", "bool", auto="wanted", manual="oven_manual", command="command", state="oven_arbiter", # Lighting a stove by hand is a decision that should stand for the # evening, not be undone by the next reading. hold_s=10800.0, ) ) flow.add( { "id": "settle", "type": "delay", "title": "At most every half hour", "params": {"interval": 1800.0}, "requires": [{"name": "command", "dtype": "bool"}], "provides": [{"name": "settled", "dtype": "bool"}], } ) flow.add( { "id": "changed", "type": "rbe", "title": "Only when it changes", "requires": [{"name": "settled", "port": "settled", "dtype": "bool"}], "provides": [{"name": "to_send", "port": "to_send", "dtype": "bool"}], } ) flow.add( { "id": "as_words", "type": "python", "title": "ON or force", "requires": [{"name": "to_send", "port": "command", "dtype": "bool"}], "provides": [{"name": "oven_command", "dtype": "str"}], }, OVEN_CMD, ) flow.add( { "id": "stove_out", "type": "mqtt", "title": "Tell the stove", "params": { "topic": {"oven_command": topics["oven_command"]}, "qos": 1, **_broker(h, "fluksio-oven-out"), }, "requires": [{"name": "oven_command", "dtype": "str"}], } ) flow.add( { "id": "pump", "type": "python", "title": "Floor heating follows", "requires": [ {"name": "running", "port": "running", "dtype": "bool"}, {"name": "since", "dtype": "float", "trigger": False}, {"name": "clock.minute", "port": "tick", "dtype": "int"}, ], "provides": [{"name": "floor_heating", "dtype": "bool"}], }, PUMP, ) flow.input("oven_manual", "bool", False) flow.input("oven_arbiter", "record", {}) return flow # ── water boilers ──────────────────────────────────────────────────────── HEATED = '''"""Has this boiler had its heating today? A boiler with no thermostat readback: the only evidence it got hot is that it drew power for a while without the inverter being busy. The reference counted ten samples of a three-second heartbeat, which is thirty seconds of drawing below 600 W. Kept as thirty seconds of wall clock, which is the same statement without the heartbeat. Ratcheting on purpose — once heated, heated, until it is reset at five in the morning. A boiler that cools slightly should not send the house looking for sun again at four in the afternoon. """ import time def process(on, out_w=0.0, hour=12, memory=None, quiet_w=600.0, needs_s=30.0, reset_hour=5): memory = dict(memory or {}) now = time.time() if hour == reset_hour and memory.get("day") != reset_hour: memory = {"day": reset_hour} elif hour != reset_hour: memory["day"] = hour heating_since = memory.get("since", 0.0) if on and out_w < quiet_w: heating_since = heating_since or now else: heating_since = 0.0 heated = bool(memory.get("heated")) or bool( heating_since and now - heating_since >= needs_s ) if memory.get("day") == reset_hour: heated = False return { "heated": heated, "heated_memory": {"heated": heated, "since": heating_since, "day": memory.get("day", hour)}, } ''' BOILER = '''"""Whether to put the immersion heater on, and on whose electricity. This is the reference's `Water Boiler Logic`, and the thing worth preserving about it is the shape rather than any single number: by day it heats only on surplus solar, and at night it falls back to the grid — but only if the day did not manage it and tomorrow does not look better than today. Two boilers share the rules and the constants; the kitchen one is gated on the main one being satisfied first, so they never draw together. The battery thresholds are in volts because that is what this bank reports, and the turn-on point moves with the season: 50.4 V in June, 49.8 V in December, because a winter battery that waits for a summer voltage waits all day. """ def process( batt_v, heated=False, out_w=0.0, in_w=0.0, in_v=230.0, watch=0, hour=12, winter=0.0, tomorrow_day=0.0, tomorrow_clouds=100.0, on=False, day_from=9, day_to=18, night_from=1, night_to=4, min_batt_v=48.4, on_batt_v=50.4, winter_batt_drop=0.6, overload_w=2800.0, busy_w=2600.0, low_grid_v=185.0, better_tomorrow_c=25.0, better_tomorrow_clouds=50.0, ): if out_w > overload_w: return {"want": False, "why": "the inverter is overloaded"} if heated: return {"want": False, "why": "already heated today"} day = day_from <= hour < day_to night = night_from <= hour < night_to if not day and not night: return {"want": False, "why": "outside both windows"} if watch: return {"want": False, "why": "power watch"} if out_w > busy_w: return {"want": False, "why": "the house is drawing too much"} if day: if in_w > 0: return {"want": False, "why": "importing rather than exporting"} # Hysteresis: below the floor it stops, above the mark it starts, and # in between it keeps doing whatever it was doing. threshold = on_batt_v - winter_batt_drop * winter if batt_v < min_batt_v: return {"want": False, "why": "battery too low"} if batt_v > threshold: return {"want": True, "why": "surplus solar"} return {"want": on, "why": "holding"} if tomorrow_day > better_tomorrow_c and tomorrow_clouds < better_tomorrow_clouds: return {"want": False, "why": "tomorrow looks better than the grid"} if in_v < 10: threshold = on_batt_v - winter_batt_drop * winter if batt_v > threshold: return {"want": True, "why": "no mains, but the battery has it"} return {"want": False, "why": "no mains and the battery is low"} if in_v < low_grid_v: return {"want": False, "why": "mains too weak"} return {"want": True, "why": "the day did not manage it"} ''' def boiler(h: dict[str, Any]) -> Flow: """Two immersion heaters, one set of rules, one at a time.""" flow = Flow("boiler", "Water boilers") for which, title, day_from, day_to, night_from, night_to, grid_v in ( ("water", "Main boiler", 9, 18, 1, 4, 185.0), ("kitchen", "Kitchen boiler", 9, 16, 2, 4, 180.0), ): state_msg = f"{which}_heated_memory" flow.add( { "id": f"{which}_heated", "type": "python", "title": f"{title}: heated today?", "requires": [ { "name": f"{which}_boiler", "port": "on", "dtype": "bool", "trigger": False, }, {"name": "power.out_w", "port": "out_w", "dtype": "float"}, { "name": "clock.hour", "port": "hour", "dtype": "int", "trigger": False, }, { "name": state_msg, "port": "memory", "dtype": "record", "trigger": False, }, ], "provides": [ {"name": f"{which}_heated", "port": "heated", "dtype": "bool"}, {"name": state_msg, "port": "heated_memory", "dtype": "record"}, ], }, HEATED, ) requires = [ {"name": "power.batt_v", "port": "batt_v", "dtype": "float"}, { "name": f"{which}_heated", "port": "heated", "dtype": "bool", "trigger": False, }, { "name": "power.out_w", "port": "out_w", "dtype": "float", "trigger": False, }, {"name": "power.in_w", "port": "in_w", "dtype": "float", "trigger": False}, {"name": "power.in_v", "port": "in_v", "dtype": "float", "trigger": False}, {"name": "power.watch", "port": "watch", "dtype": "int", "trigger": False}, {"name": "clock.hour", "port": "hour", "dtype": "int"}, { "name": "clock.winter", "port": "winter", "dtype": "float", "trigger": False, }, { "name": "weather.tomorrow_day", "port": "tomorrow_day", "dtype": "float", "trigger": False, }, { "name": "weather.tomorrow_clouds", "port": "tomorrow_clouds", "dtype": "float", "trigger": False, }, { "name": f"{which}_boiler", "port": "on", "dtype": "bool", "trigger": False, }, ] if which == "kitchen": # Only once the main one is satisfied: they share an inverter, and # two immersion heaters is more than it has. requires.append( { "name": "water_heated", "port": "main_heated", "dtype": "bool", "trigger": False, } ) flow.add( { "id": f"{which}_decide", "type": "python", "title": f"{title}: on?", "params": { "day_from": day_from, "day_to": day_to, "night_from": night_from, "night_to": night_to, "low_grid_v": grid_v, }, "requires": requires, "provides": [ {"name": f"{which}_want", "port": "want", "dtype": "bool"}, {"name": f"{which}_why", "port": "why", "dtype": "str"}, ], }, BOILER if which == "water" else KITCHEN_BOILER, ) flow.add( { "id": f"{which}_changed", "type": "rbe", "title": "Only when it changes", "requires": [ {"name": f"{which}_want", "port": "want", "dtype": "bool"} ], "provides": [ {"name": f"{which}_boiler", "port": "on", "dtype": "bool"} ], } ) flow.input(state_msg, "record", {}) flow.input(f"{which}_boiler", "bool", False) return flow KITCHEN_BOILER = BOILER.replace( """def process( batt_v,""", """def process( batt_v, main_heated=False,""", 1, ).replace( """ if out_w > overload_w: return {"want": False, "why": "the inverter is overloaded"}""", """ if out_w > overload_w: return {"want": False, "why": "the inverter is overloaded"} if not main_heated: return {"want": False, "why": "the main boiler comes first"}""", 1, )