app/scripts/tinyhouse/library.py

"""The two pieces of logic every actuator in the house needs.

Both are shared library nodes rather than copies: one arbiter fix reaches
fifteen actuators, and a rollershutter that learns to stop properly should not
have to learn it four times.
"""

from __future__ import annotations

from typing import Any

ARBITER = '''"""Which value reaches an actuator: what the house decided, or what a person did.

Automation runs on a loop and a person does not. Without something between the
two, a switch someone presses is undone by the next evaluation — so a manual
value wins for a while, and the house takes over again once the hold expires.

The control on the dashboard binds to ``manual`` in both directions: this node
writes back whatever actually reached the actuator, so one tile shows the state
and sets it. ``force_at`` is a pulse that outranks a person — the nightly off,
and later the fire alarm — because a schedule that a forgotten override can
defeat is not a schedule.
"""

import time


def process(auto, manual, state=None, force_at=0.0, hold_s=14400.0, force_value=None):
    state = dict(state or {})
    now = time.time()
    held_until = state.get("override_until", 0.0)

    if not state:
        # First run: adopt what is already there instead of reading the
        # difference between two initial values as somebody pressing something.
        decided = auto
        held_until = 0.0
    elif force_at != state.get("force_at", 0.0):
        decided = auto if force_value is None else force_value
        held_until = 0.0
    elif manual != state.get("manual"):
        # ponytail: a person setting the value it already holds is not seen,
        # so it does not start a hold. Pressing what the screen already shows
        # is not how anyone overrides anything; revisit if a control ever
        # publishes a timestamp of its own.
        decided = manual
        held_until = -1.0 if hold_s <= 0 else now + hold_s
    elif held_until < 0 or now < held_until:
        decided = state.get("manual")
    else:
        decided = auto

    return {
        "command": decided,
        "manual": decided,
        "state": {
            "manual": decided,
            "force_at": force_at,
            "override_until": held_until,
        },
    }
'''

MOTOR = '''"""A rollershutter with no position sensor: run for as long as it takes.

Time is the only feedback there is. The run-time per direction is a setting
because it was measured on this hardware and will drift — a motor that stops
half a turn early is a knob, not a rewrite.

What leaves here is the command and how long to hold it: a trigger node
downstream sends the STOP when the run is over, once, rather than the
reference's forever.
"""

import time

STOP = "STOP"


def process(cmd, state=None, up_s=26.0, down_s=28.0):
    now = time.time()
    state = dict(state or {})
    # Nothing reports the end of a run, so it is over when it has had its time.
    moving = state.get("moving", "") if now < state.get("moving_until", 0.0) else ""
    position = state.get("position", "")

    if cmd == STOP:
        if not moving:
            return None
        run, run_for, target = STOP, 0.0, ""
    elif moving == cmd:
        return None
    elif moving:
        # Reversing mid-travel drives both relays at once on the way past.
        # Stop; the next command moves it.
        run, run_for, target = STOP, 0.0, ""
    elif cmd == position:
        return None
    else:
        run = cmd
        run_for = up_s if cmd == "UP" else down_s
        target = cmd

    return {
        "run": run,
        "run_for": run_for,
        # Written back to the message the control publishes on, so one segmented
        # button both sets the direction and shows which one it is doing — the
        # same trick the arbiter uses. A run that was refused returns nothing at
        # all, so the control keeps what the person put there.
        "cmd": run,
        "state": {
            "moving": "" if run == STOP else run,
            "moving_until": 0.0 if run == STOP else now + run_for,
            # A run stopped part way is at no position anybody knows, so the
            # next command in either direction has to be allowed to move it.
            "position": target,
        },
    }
'''


def arbiter(
    node_id: str,
    title: str,
    dtype: str,
    auto: str,
    manual: str,
    command: str,
    state: str,
    hold_s: float = 14400.0,
    force_at: str = "",
    force_value: Any = None,
    shared: bool = True,
) -> dict[str, Any]:
    """One actuator's arbiter, wired to the messages it decides between."""
    params: dict[str, Any] = {"hold_s": hold_s}
    if force_at:
        params["force_value"] = force_value
    requires = [
        {"name": auto, "port": "auto", "dtype": dtype},
        {"name": manual, "port": "manual", "dtype": dtype},
        {"name": state, "port": "state", "dtype": "record", "trigger": False},
    ]
    if force_at:
        requires.append({"name": force_at, "port": "force_at", "dtype": "float"})
    node = {
        "id": node_id,
        "type": "python",
        "title": title,
        "params": params,
        "requires": requires,
        "provides": [
            {"name": command, "port": "command", "dtype": dtype},
            {"name": manual, "port": "manual", "dtype": dtype},
            {"name": state, "port": "state", "dtype": "record"},
        ],
    }
    if shared:
        node["source_ref"] = "arbiter"
    return node


def motor(
    node_id: str,
    title: str,
    cmd: str,
    run: str,
    run_for: str,
    state: str,
    up_s: float,
    down_s: float,
    shared: bool = True,
) -> dict[str, Any]:
    """One motor's run-time controller."""
    node = {
        "id": node_id,
        "type": "python",
        "title": title,
        "params": {"up_s": up_s, "down_s": down_s},
        "requires": [
            {"name": cmd, "port": "cmd", "dtype": "str"},
            {"name": state, "port": "state", "dtype": "record", "trigger": False},
        ],
        "provides": [
            {"name": run, "port": "run", "dtype": "str"},
            {"name": run_for, "port": "run_for", "dtype": "float"},
            {"name": cmd, "port": "cmd", "dtype": "str"},
            {"name": state, "port": "state", "dtype": "record"},
        ],
    }
    if shared:
        node["source_ref"] = "motor"
    return node


def stopper(
    node_id: str, title: str, run: str, run_for: str, cmd: str
) -> dict[str, Any]:
    """The trigger that sends a motor's STOP when its run is over.

    Passes the command through immediately, then sends STOP once the run-time
    the motor node worked out has elapsed. A run-time of zero — which is what a
    STOP itself carries — sends nothing afterwards, so the relays are released
    once and then left alone.
    """
    return {
        "id": node_id,
        "type": "trigger",
        "title": title,
        "params": {
            "then": "STOP",
            "wait": 60.0,
            "passthrough": True,
            "wait_port": "run_for",
            "extend": True,
        },
        "requires": [
            {"name": run, "port": "run", "dtype": "str"},
            {"name": run_for, "port": "run_for", "dtype": "float"},
        ],
        "provides": [{"name": cmd, "port": "cmd", "dtype": "str"}],
    }