Claude Code transcript - page 1/2

"""Unicast-sweep a /24 for Kasa devices (WSL2 NAT blocks broadcast discovery)."""
import asyncio
import sys

from kasa import Discover
from kasa.exceptions import KasaException, TimeoutError as KasaTimeoutError

SUBNET = sys.argv[1] if len(sys.argv) > 1 else "192.168.1"
SEM = asyncio.Semaphore(64)


async def probe(ip: str):
    async with SEM:
        try:
            dev = await Discover.discover_single(ip, timeout=2, discovery_timeout=2)
            if dev is None:
                return None
            await dev.update()
            info = (ip, dev.alias, dev.model, getattr(dev, "children", []))
            await dev.disconnect()
            return info
        except (KasaException, KasaTimeoutError, asyncio.TimeoutError, OSError):
            return None


async def main():
    tasks = [probe(f"{SUBNET}.{i}") for i in range(1, 255)]
    results = [r for r in await asyncio.gather(*tasks) if r]
    for ip, alias, model, children in results:
        print(f"{ip}  model={model}  alias={alias!r}")
        for child in children:
            print(f"    child: {child.alias!r}  on={child.is_on}")
    if not results:
        print("No Kasa devices found on subnet.")


asyncio.run(main())

# heatwires configuration

[location]
# NOTE: defaulted to Bozeman, MT city center — adjust to the actual building
# location (elevation matters for temperature accuracy).
latitude = 45.68
longitude = -111.04

[thresholds]
# Turn wires ON when temp is inside [on_temp_min_f, on_temp_max_f] AND there
# is a snow signal. The melt-refreeze danger zone.
on_temp_max_f = 35.0
on_temp_min_f = 8.0
# Turn OFF when temp rises past off_temp_high_f (meltwater drains freely)
# or drops below off_temp_low_f (nothing is melting; wires are wasted power).
# Gaps between on/off thresholds provide temperature hysteresis.
off_temp_high_f = 38.0
off_temp_low_f = 5.0

# Snow signal: snowfall observed in the last N hours, or forecast in the next
# M hours, exceeding the trigger amount — or standing snow depth.
snow_lookback_hours = 48
snow_forecast_hours = 24
snow_trigger_inches = 0.5
snow_depth_trigger_inches = 1.0

# Minimum time wires stay ON before the program will turn them OFF.
# Turning ON is never delayed.
min_on_minutes = 60

[failsafe]
# If weather data has been unavailable for this many consecutive runs,
# assume the worst during winter months and turn the wires ON.
max_weather_failures = 3
winter_months = [10, 11, 12, 1, 2, 3, 4]

[state]
# Where run-to-run state (hysteresis timers, failure counts) is kept.
file = "state.json"

# Devices to control. `alias` is verified against the live device before any
# actuation — if an IP has moved to a different device, it is skipped with an
# error rather than toggled. For KP400 (multi-outlet) devices, `child_alias`
# selects the outlet; omit it for single-relay devices like the HS200.
# Give these static DHCP reservations in your router.

[[devices]]
host = "192.168.1.64"
alias = "Switch #5"
child_alias = "Shop gutter heat wire"

[[devices]]
host = "192.168.1.71"
alias = "Switch #6"
child_alias = "Shop roof extra wires"

[[devices]]
host = "192.168.1.155"
alias = "House Roof Heat Wires"

"""Automated control of roof heat wires based on weather conditions."""

__version__ = "0.1.0"

"""Configuration loading."""

from __future__ import annotations

import tomllib
from dataclasses import dataclass, field
from pathlib import Path


@dataclass
class Device:
    host: str
    alias: str
    child_alias: str | None = None

    @property
    def label(self) -> str:
        return self.child_alias or self.alias


@dataclass
class Thresholds:
    on_temp_max_f: float = 35.0
    on_temp_min_f: float = 8.0
    off_temp_high_f: float = 38.0
    off_temp_low_f: float = 5.0
    snow_lookback_hours: int = 48
    snow_forecast_hours: int = 24
    snow_trigger_inches: float = 0.5
    snow_depth_trigger_inches: float = 1.0
    min_on_minutes: int = 60


@dataclass
class Failsafe:
    max_weather_failures: int = 3
    winter_months: list[int] = field(
        default_factory=lambda: [10, 11, 12, 1, 2, 3, 4]
    )


@dataclass
class Config:
    latitude: float
    longitude: float
    thresholds: Thresholds
    failsafe: Failsafe
    devices: list[Device]
    state_file: Path


def load(path: Path) -> Config:
    with open(path, "rb") as f:
        raw = tomllib.load(f)

    loc = raw["location"]
    state_file = Path(raw.get("state", {}).get("file", "state.json"))
    if not state_file.is_absolute():
        state_file = path.parent / state_file

    return Config(
        latitude=loc["latitude"],
        longitude=loc["longitude"],
        thresholds=Thresholds(**raw.get("thresholds", {})),
        failsafe=Failsafe(**raw.get("failsafe", {})),
        devices=[Device(**d) for d in raw["devices"]],
        state_file=state_file,
    )

"""Weather retrieval from Open-Meteo (keyless, free for non-commercial use)."""

from __future__ import annotations

import json
import urllib.parse
import urllib.request
from dataclasses import dataclass
from datetime import datetime

from .config import Thresholds

API_URL = "https://api.open-meteo.com/v1/forecast"
USER_AGENT = "heatwires/0.1 (roof heat-wire controller)"


@dataclass
class WeatherSignals:
    """Inputs to the icing decision, derived from hourly weather data."""

    temp_f: float
    snowfall_recent_in: float
    snowfall_forecast_in: float
    snow_depth_in: float
    observed_at: str


def fetch(latitude: float, longitude: float, th: Thresholds,
          timeout: float = 30.0) -> WeatherSignals:
    """Fetch weather and reduce it to decision signals. Raises on any failure."""
    params = urllib.parse.urlencode({
        "latitude": latitude,
        "longitude": longitude,
        "current": "temperature_2m",
        "hourly": "temperature_2m,snowfall,snow_depth",
        "past_days": max(1, -(-th.snow_lookback_hours // 24)),
        "forecast_days": max(1, -(-th.snow_forecast_hours // 24)),
        "temperature_unit": "fahrenheit",
        "precipitation_unit": "inch",
        "timezone": "UTC",
    })
    req = urllib.request.Request(
        f"{API_URL}?{params}", headers={"User-Agent": USER_AGENT}
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        data = json.load(resp)

    now = datetime.fromisoformat(data["current"]["time"])
    times = [datetime.fromisoformat(t) for t in data["hourly"]["time"]]
    snowfall = data["hourly"]["snowfall"]
    snow_depth = data["hourly"]["snow_depth"]  # feet, per hourly_units

    lookback = th.snow_lookback_hours
    forecast = th.snow_forecast_hours
    recent = sum(
        s or 0.0
        for t, s in zip(times, snowfall)
        if 0 <= (now - t).total_seconds() / 3600 <= lookback
    )
    ahead = sum(
        s or 0.0
        for t, s in zip(times, snowfall)
        if 0 < (t - now).total_seconds() / 3600 <= forecast
    )
    # Snow depth: latest non-null value at or before now, feet -> inches.
    depth_ft = 0.0
    for t, d in zip(times, snow_depth):
        if t <= now and d is not None:
            depth_ft = d

    return WeatherSignals(
        temp_f=data["current"]["temperature_2m"],
        snowfall_recent_in=recent,
        snowfall_forecast_in=ahead,
        snow_depth_in=depth_ft * 12.0,
        observed_at=data["current"]["time"],
    )

"""Icing decision logic. Pure functions — no I/O — so it is unit-testable.

Ice dams form from melt-refreeze cycles: snow on the roof, heat loss melting
the underside, meltwater refreezing at the cold eaves. The wires should run
when there is snow AND the temperature is in the melt-refreeze band — not
merely when it is cold.
"""

from __future__ import annotations

from dataclasses import dataclass

from .config import Failsafe, Thresholds
from .weather import WeatherSignals


@dataclass
class Decision:
    wires_on: bool
    reason: str


def snow_signal(w: WeatherSignals, th: Thresholds) -> bool:
    return (
        w.snowfall_recent_in >= th.snow_trigger_inches
        or w.snowfall_forecast_in >= th.snow_trigger_inches
        or w.snow_depth_in >= th.snow_depth_trigger_inches
    )


def decide(
    w: WeatherSignals,
    th: Thresholds,
    currently_on: bool,
    minutes_since_on: float | None,
) -> Decision:
    """Decide desired wire state.

    minutes_since_on: how long ago the wires were last turned on by this
    program (None if unknown/never). Used for the minimum-on hold, which only
    delays turning OFF — turning ON is never delayed.
    """
    snow = snow_signal(w, th)

    if snow and th.on_temp_min_f <= w.temp_f <= th.on_temp_max_f:
        return Decision(True, (
            f"melt-refreeze conditions: {w.temp_f:.0f}°F with snow "
            f"(recent {w.snowfall_recent_in:.1f}\", "
            f"forecast {w.snowfall_forecast_in:.1f}\", "
            f"depth {w.snow_depth_in:.1f}\")"
        ))

    if currently_on:
        held = (
            minutes_since_on is not None
            and minutes_since_on < th.min_on_minutes
        )
        if held and w.temp_f < th.off_temp_high_f:
            return Decision(True, (
                f"holding on: only {minutes_since_on:.0f} of "
                f"{th.min_on_minutes} minimum minutes elapsed"
            ))

    if not snow:
        return Decision(False, "no snow signal")
    if w.temp_f > th.on_temp_max_f:
        if w.temp_f < th.off_temp_high_f and currently_on:
            return Decision(True, (
                f"hysteresis: {w.temp_f:.0f}°F is above on-threshold "
                f"{th.on_temp_max_f:.0f}°F but below off-threshold "
                f"{th.off_temp_high_f:.0f}°F"
            ))
        return Decision(False, f"warm enough to drain ({w.temp_f:.0f}°F)")
    if w.temp_f < th.on_temp_min_f:
        if w.temp_f > th.off_temp_low_f and currently_on:
            return Decision(True, (
                f"hysteresis: {w.temp_f:.0f}°F is below on-threshold "
                f"{th.on_temp_min_f:.0f}°F but above off-threshold "
                f"{th.off_temp_low_f:.0f}°F"
            ))
        return Decision(False, f"too cold for melt ({w.temp_f:.0f}°F)")

    return Decision(False, "no icing conditions")


def decide_failsafe(
    month: int, consecutive_failures: int, fs: Failsafe, currently_on: bool
) -> Decision:
    """Weather data unavailable. Bias toward ON in winter: a false 'on' costs
    electricity, a false 'off' can cost the roof."""
    if consecutive_failures >= fs.max_weather_failures:
        if month in fs.winter_months:
            return Decision(True, (
                f"FAILSAFE: weather unavailable {consecutive_failures} runs "
                f"in a row during winter — wires on as a precaution"
            ))
        return Decision(False, (
            f"weather unavailable {consecutive_failures} runs, but not "
            f"winter — wires off"
        ))
    return Decision(currently_on, (
        f"weather fetch failed ({consecutive_failures} of "
        f"{fs.max_weather_failures} tolerated) — keeping last state"
    ))

"""Kasa switch actuation over the local network."""

from __future__ import annotations

import logging
from dataclasses import dataclass

from kasa import Device as KasaDevice
from kasa import Discover

from .config import Device

log = logging.getLogger("heatwires")


@dataclass
class DeviceState:
    device: Device
    is_on: bool | None  # None = unreachable or alias mismatch
    error: str | None = None


def _resolve_target(dev: KasaDevice, cfg: Device):
    """Return the switchable module (device or child outlet), verifying
    aliases so a moved IP address never toggles the wrong equipment."""
    if dev.alias != cfg.alias:
        raise ValueError(
            f"alias mismatch at {cfg.host}: expected {cfg.alias!r}, "
            f"found {dev.alias!r} — refusing to touch it"
        )
    if cfg.child_alias is None:
        return dev
    for child in dev.children:
        if child.alias == cfg.child_alias:
            return child
    raise ValueError(
        f"no child outlet named {cfg.child_alias!r} on {cfg.host} "
        f"(has: {[c.alias for c in dev.children]})"
    )


async def read_state(cfg: Device, timeout: int = 10) -> DeviceState:
    try:
        dev = await Discover.discover_single(
            cfg.host, timeout=timeout, discovery_timeout=timeout
        )
        try:
            await dev.update()
            target = _resolve_target(dev, cfg)
            return DeviceState(cfg, target.is_on)
        finally:
            await dev.disconnect()
    except Exception as exc:  # noqa: BLE001 - report, don't crash the run
        return DeviceState(cfg, None, error=str(exc))


async def set_state(cfg: Device, on: bool, timeout: int = 10) -> DeviceState:
    """Set one device/outlet, returning its confirmed state afterwards."""
    try:
        dev = await Discover.discover_single(
            cfg.host, timeout=timeout, discovery_timeout=timeout
        )
        try:
            await dev.update()
            target = _resolve_target(dev, cfg)
            if target.is_on != on:
                if on:
                    await target.turn_on()
                else:
                    await target.turn_off()
                await dev.update()
                target = _resolve_target(dev, cfg)
                if target.is_on != on:
                    raise RuntimeError(
                        f"commanded {'on' if on else 'off'} but device "
                        f"still reports {'on' if target.is_on else 'off'}"
                    )
                log.info("%s: turned %s", cfg.label, "ON" if on else "OFF")
            else:
                log.debug("%s: already %s", cfg.label, "ON" if on else "OFF")
            return DeviceState(cfg, target.is_on)
        finally:
            await dev.disconnect()
    except Exception as exc:  # noqa: BLE001
        log.error("%s: FAILED to set %s: %s", cfg.label,
                  "ON" if on else "OFF", exc)
        return DeviceState(cfg, None, error=str(exc))

"""Run-to-run state persisted to a JSON file (hysteresis timers, failure
counts, manual override)."""

from __future__ import annotations

import json
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path


@dataclass
class State:
    wires_on: bool = False          # last state this program commanded
    last_on_at: str | None = None   # ISO time wires were last turned on
    consecutive_failures: int = 0   # weather fetches failed in a row
    override: str | None = None     # "on" | "off" | None
    override_until: str | None = None  # ISO expiry for the override
    last_run_at: str | None = None
    last_reason: str | None = None

    def minutes_since_on(self, now: datetime) -> float | None:
        if self.last_on_at is None:
            return None
        return (now - datetime.fromisoformat(self.last_on_at)).total_seconds() / 60

    def active_override(self, now: datetime) -> str | None:
        if self.override is None:
            return None
        if self.override_until is not None:
            if now >= datetime.fromisoformat(self.override_until):
                return None
        return self.override


def load(path: Path) -> State:
    try:
        return State(**json.loads(path.read_text()))
    except (FileNotFoundError, json.JSONDecodeError, TypeError):
        return State()


def save(path: Path, state: State) -> None:
    tmp = path.with_suffix(".tmp")
    tmp.write_text(json.dumps(asdict(state), indent=2))
    tmp.replace(path)


def utcnow() -> datetime:
    return datetime.now(timezone.utc)

"""CLI entry point.

    python -m heatwires run [--dry-run]     evaluate weather and set switches
    python -m heatwires status              show weather, decision, and switch states
    python -m heatwires override on|off [--hours N]
    python -m heatwires override clear
"""

from __future__ import annotations

import argparse
import asyncio
import logging
import sys
from datetime import timedelta
from pathlib import Path

from . import config as config_mod
from . import decision as decision_mod
from . import state as state_mod
from . import weather as weather_mod
from .control import read_state, set_state

log = logging.getLogger("heatwires")


def setup_logging(log_file: Path | None) -> None:
    fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    stream = logging.StreamHandler()
    stream.setFormatter(fmt)
    root.addHandler(stream)
    if log_file is not None:
        fh = logging.FileHandler(log_file)
        fh.setFormatter(fmt)
        root.addHandler(fh)


def get_decision(cfg, st, now):
    """Fetch weather and decide, handling overrides and fetch failures.
    Returns (decision, weather_or_none)."""
    override = st.active_override(now)
    if override is not None:
        until = st.override_until or "cleared"
        return decision_mod.Decision(
            override == "on", f"manual override {override!r} until {until}"
        ), None

    try:
        w = weather_mod.fetch(cfg.latitude, cfg.longitude, cfg.thresholds)
    except Exception as exc:  # noqa: BLE001 - fail-safe path handles it
        st.consecutive_failures += 1
        log.error("weather fetch failed (%d in a row): %s",
                  st.consecutive_failures, exc)
        return decision_mod.decide_failsafe(
            now.month, st.consecutive_failures, cfg.failsafe, st.wires_on
        ), None

    st.consecutive_failures = 0
    log.info(
        "weather: %.0f°F, snow recent %.1f\" / forecast %.1f\" / depth %.1f\"",
        w.temp_f, w.snowfall_recent_in, w.snowfall_forecast_in, w.snow_depth_in,
    )
    return decision_mod.decide(
        w, cfg.thresholds, st.wires_on, st.minutes_since_on(now)
    ), w


async def cmd_run(cfg, st, dry_run: bool) -> int:
    now = state_mod.utcnow()
    dec, _ = get_decision(cfg, st, now)
    log.info("decision: wires %s — %s", "ON" if dec.wires_on else "OFF",
             dec.reason)

    errors = 0
    if dry_run:
        log.info("dry run: not touching switches")
    else:
        results = await asyncio.gather(
            *(set_state(d, dec.wires_on) for d in cfg.devices)
        )
        errors = sum(1 for r in results if r.error is not None)
        if dec.wires_on and not st.wires_on:
            st.last_on_at = now.isoformat()
        st.wires_on = dec.wires_on
        st.last_run_at = now.isoformat()
        st.last_reason = dec.reason
        state_mod.save(cfg.state_file, st)

    if errors:
        log.error("%d of %d devices failed — check them in the Kasa app",
                  errors, len(cfg.devices))
    return 1 if errors else 0


async def cmd_status(cfg, st) -> int:
    now = state_mod.utcnow()
    dec, w = get_decision(cfg, st, now)
    if w is not None:
        print(f"weather:  {w.temp_f:.0f}°F, snow recent "
              f"{w.snowfall_recent_in:.1f}\" / forecast "
              f"{w.snowfall_forecast_in:.1f}\" / depth {w.snow_depth_in:.1f}\"")
    print(f"decision: wires {'ON' if dec.wires_on else 'OFF'} — {dec.reason}")
    print(f"last run: {st.last_run_at or 'never'}"
          + (f" ({st.last_reason})" if st.last_reason else ""))
    results = await asyncio.gather(*(read_state(d) for d in cfg.devices))
    for r in results:
        actual = ("ERROR: " + r.error if r.error
                  else "ON" if r.is_on else "OFF")
        print(f"  {r.device.label:<28} {actual}")
    return 1 if any(r.error for r in results) else 0


def cmd_override(cfg, st, mode: str, hours: float | None) -> int:
    now = state_mod.utcnow()
    if mode == "clear":
        st.override = None
        st.override_until = None
        print("override cleared; automatic control resumes on next run")
    else:
        st.override = mode
        st.override_until = (
            (now + timedelta(hours=hours)).isoformat() if hours else None
        )
        until = st.override_until or "manually cleared"
        print(f"override: wires {mode.upper()} until {until}")
        print("(takes effect on the next scheduled run, or run "
              "`python -m heatwires run` now)")
    state_mod.save(cfg.state_file, st)
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(prog="heatwires", description=__doc__)
    parser.add_argument("--config", type=Path,
                        default=Path(__file__).parent.parent / "config.toml")
    parser.add_argument("--log-file", type=Path, default=None)
    sub = parser.add_subparsers(dest="command", required=True)

    p_run = sub.add_parser("run", help="evaluate weather and set switches")
    p_run.add_argument("--dry-run", action="store_true",
                       help="decide and log, but do not touch switches")

    sub.add_parser("status", help="show weather, decision, switch states")

    p_ov = sub.add_parser("override", help="force wires on/off")
    p_ov.add_argument("mode", choices=["on", "off", "clear"])
    p_ov.add_argument("--hours", type=float, default=None,
                      help="auto-expire the override after this many hours")

    args = parser.parse_args()
    setup_logging(args.log_file)

    cfg = config_mod.load(args.config)
    st = state_mod.load(cfg.state_file)

    if args.command == "run":
        return asyncio.run(cmd_run(cfg, st, args.dry_run))
    if args.command == "status":
        return asyncio.run(cmd_status(cfg, st))
    return cmd_override(cfg, st, args.mode, args.hours)


if __name__ == "__main__":
    sys.exit(main())

from heatwires.config import Failsafe, Thresholds
from heatwires.decision import decide, decide_failsafe
from heatwires.weather import WeatherSignals

TH = Thresholds()
FS = Failsafe()


def w(temp, recent=0.0, forecast=0.0, depth=0.0):
    return WeatherSignals(
        temp_f=temp,
        snowfall_recent_in=recent,
        snowfall_forecast_in=forecast,
        snow_depth_in=depth,
        observed_at="2026-01-15T12:00",
    )


def test_on_in_melt_refreeze_band_with_recent_snow():
    assert decide(w(28, recent=2.0), TH, False, None).wires_on


def test_on_with_forecast_snow_only():
    assert decide(w(30, forecast=1.0), TH, False, None).wires_on


def test_on_with_standing_snow_depth():
    assert decide(w(25, depth=4.0), TH, False, None).wires_on


def test_off_when_no_snow_even_if_cold():
    assert not decide(w(20), TH, False, None).wires_on


def test_off_when_warm_despite_snow():
    assert not decide(w(45, recent=3.0), TH, True, 999).wires_on


def test_off_when_extremely_cold():
    assert not decide(w(-10, depth=6.0), TH, True, 999).wires_on


def test_temperature_hysteresis_stays_on_between_thresholds():
    # 36°F: above on-max (35) but below off-high (38)
    d = decide(w(36, depth=4.0), TH, True, 999)
    assert d.wires_on
    # ...but does not turn ON at 36°F from an off state
    assert not decide(w(36, depth=4.0), TH, False, None).wires_on


def test_low_temperature_hysteresis():
    # 6°F: below on-min (8) but above off-low (5) — stays on, won't start
    assert decide(w(6, depth=4.0), TH, True, 999).wires_on
    assert not decide(w(6, depth=4.0), TH, False, None).wires_on


def test_minimum_on_hold_delays_turn_off():
    # Snow stopped mattering, but wires only just came on
    d = decide(w(30), TH, True, 15)
    assert d.wires_on
    assert "holding" in d.reason


def test_minimum_on_hold_expires():
    assert not decide(w(30), TH, True, 90).wires_on


def test_hold_does_not_delay_turn_on():
    assert decide(w(28, recent=2.0), TH, False, None).wires_on


def test_hold_does_not_block_warm_shutoff():
    # Above off_temp_high the hold must not keep wires burning
    assert not decide(w(45), TH, True, 5).wires_on


def test_failsafe_on_in_winter_after_repeated_failures():
    d = decide_failsafe(1, 3, FS, currently_on=False)
    assert d.wires_on
    assert "FAILSAFE" in d.reason


def test_failsafe_off_in_summer():
    assert not decide_failsafe(7, 5, FS, currently_on=True).wires_on


def test_failsafe_keeps_last_state_below_threshold():
    assert decide_failsafe(1, 1, FS, currently_on=True).wires_on
    assert not decide_failsafe(1, 1, FS, currently_on=False).wires_on

[project]
name = "heatwires"
version = "0.1.0"
description = "Automated control of roof heat wires based on weather conditions"
requires-python = ">=3.11"
dependencies = [
    "python-kasa>=0.10",
]

[project.optional-dependencies]
dev = ["pytest"]

[tool.pytest.ini_options]
testpaths = ["tests"]

.venv/
__pycache__/
*.pyc
state.json
*.log

# heatwires

Automated control of roof heat wires based on weather conditions, to prevent
ice dams. Talks directly to TP-Link Kasa switches (KP400, HS200, etc.) over
the local network using [python-kasa](https://github.com/python-kasa/python-kasa)
— no cloud dependency — and gets weather from the free, keyless
[Open-Meteo](https://open-meteo.com) API.

## How it decides

Ice dams form from melt-refreeze cycles, so the wires run when there is snow
AND the temperature is in the melt-refreeze band — not merely when it is cold:

- **ON** when temperature is 8–35°F and there is a snow signal (≥0.5" fallen
  in the last 48h, ≥0.5" forecast in the next 24h, or ≥1" standing snow depth).
- **OFF** when it warms past 38°F (meltwater drains freely), drops below 5°F
  (nothing melts), or the snow signal clears.
- The on/off threshold gaps provide temperature hysteresis, and a 60-minute
  minimum-on hold prevents rapid cycling. Turning ON is never delayed.
- **Fail-safe:** if weather data is unavailable 3 runs in a row during winter
  months, the wires turn ON — a false "on" costs electricity; a false "off"
  can cost the roof.

All thresholds are in `config.toml`.

## Setup

```bash
python3 -m venv .venv
.venv/bin/pip install -e .
```

Edit `config.toml`:

- Set `latitude`/`longitude` to the building's actual location.
- List the switches under `[[devices]]`. The `alias` (and `child_alias` for
  multi-outlet devices like the KP400) is verified against the live device
  before every actuation — if DHCP hands the IP to something else, the run
  refuses to touch it rather than toggling the wrong equipment. **Give the
  switches static DHCP reservations in your router** so this never trips.

## Usage

```bash
.venv/bin/python -m heatwires status           # weather, decision, switch states (read-only)
.venv/bin/python -m heatwires run --dry-run    # decide and log, don't touch switches
.venv/bin/python -m heatwires run              # the real thing — run this on a schedule
.venv/bin/python -m heatwires override on --hours 12   # force on (e.g. ahead of a storm)
.venv/bin/python -m heatwires override off --hours 4   # force off (e.g. roof work)
.venv/bin/python -m heatwires override clear
```

Manual changes made in the Kasa app are overwritten on the next scheduled
run — use `override` instead when you want the program to stand down.

## Scheduling

Run every 15 minutes via cron on any always-on machine on the same LAN as the
switches (Raspberry Pi, NAS, home server):

```cron
*/15 * * * * cd /home/david/projects/play/heatwires && .venv/bin/python -m heatwires --log-file heatwires.log run
```

Note for WSL2: WSL does not reliably run cron while no session is open. Fine
for development, but deploy to an always-on Linux box for the winter. (Local
control requires being on the same LAN; WSL2's NAT blocks broadcast discovery
but unicast connections to known switch IPs work.)

Exit code is non-zero if any switch could not be reached or verified, so you
can wrap the cron line with your favorite alerting if wanted.

## Testing

```bash
.venv/bin/pip install -e ".[dev]"
.venv/bin/pytest
```

The decision logic is pure (no I/O) and covered by unit tests in
`tests/test_decision.py`.

## Firmware caveat

TP-Link has been migrating newer firmware toward an authenticated local
protocol (KLAP) and in some cases disabling local control. These switches
currently speak the legacy local protocol. **Avoid accepting Kasa firmware
updates** once this is working, and re-run `python -m heatwires status` after
any update to confirm the switches still respond.