app/scripts/tinyhouse/test_library.py

"""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.control import BOILER, HVAC  # noqa: E402
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)
hvac = _load(HVAC)
boiler = _load(BOILER)


# ── 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"


# ── the heat pump ────────────────────────────────────────────────────────


def _asked(**kw):
    """What the heat pump would be asked for, in one word."""
    base = dict(indoor=25.7, top=25.7, t_min=19.5, t_max=21.0, soc=88.0, in_v=234.0)
    answer = hvac(**{**base, **kw})
    command = answer["command"]
    return command["mode"] if command["operation"] else "off"


def test_cooling_starts_well_above_the_comfort_band():
    """The band is what heating aims at.

    Cooling to it would run the compressor through every summer afternoon, so
    it has thresholds of its own — which is the distinction the first port of
    this lost, and it wanted to cool the house to 21 in August.
    """
    assert _asked(indoor=25.7, top=25.7) == "off"
    assert _asked(indoor=26.5, top=26.5) == "cool"
    assert _asked(indoor=29.5, top=29.5) == "cool"


def test_at_night_it_starts_sooner_and_only_on_a_full_battery():
    assert _asked(indoor=25.7, top=25.7, sleeping=True) == "fan"
    assert _asked(indoor=25.7, top=25.7, sleeping=True, soc=95) == "cool"


def test_the_heat_pump_does_not_heat_while_the_stove_can():
    """Both running is the expensive one winning an argument with the cheap one."""
    assert _asked(indoor=18.0, top=18.0) == "off"
    assert _asked(indoor=18.0, top=18.0, oven_faulty=True) == "heat"


def test_nothing_optional_runs_on_a_weak_supply():
    assert _asked(indoor=29.5, top=29.5, watch=4) == "off"
    assert _asked(indoor=29.5, top=29.5, enabled=False) == "off"
    assert _asked(indoor=29.5, top=29.5, soc=50) == "fan", "move air, do not make cold"
    assert _asked(indoor=29.5, top=29.5, soc=40) == "off", (
        "battery too low even for that"
    )


# ── the boilers ──────────────────────────────────────────────────────────


def _heats(**kw):
    base = dict(batt_v=50.5, hour=12, in_v=234.0, winter=0.0)
    return boiler(**{**base, **kw})["want"]


def test_by_day_a_boiler_heats_on_surplus_and_not_on_the_grid():
    assert _heats(batt_v=50.5) is True, "battery above the mark"
    assert _heats(batt_v=48.0) is False, "battery below the floor"
    assert _heats(batt_v=50.5, in_w=200.0) is False, "importing, so not surplus"
    assert _heats(batt_v=50.5, heated=True) is False, "already heated today"


def test_the_turn_on_point_drops_as_the_year_does():
    """A winter battery waiting for a summer voltage waits all day."""
    assert _heats(batt_v=50.0, winter=0.0) is False
    assert _heats(batt_v=50.0, winter=1.0) is True


def test_at_night_it_falls_back_to_the_grid_unless_tomorrow_is_better():
    assert _heats(hour=2) is True, "the day did not manage it"
    assert _heats(hour=2, tomorrow_day=27.0, tomorrow_clouds=10.0) is False
    assert _heats(hour=2, in_v=170.0) is False, "mains too weak"
    assert _heats(hour=6) is False, "outside both windows"


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")