Claude Code transcript - page 1/2

[project]
name = "heatwires"
version = "0.1.0"
description = "Automated weather-based control of roof heat-cable switches (TP-Link Kasa KP400)"
readme = "README.md"
requires-python = ">=3.10"
dependencies = [
    "python-kasa>=0.7,<0.12",
    "requests>=2.28",
    "PyYAML>=6.0",
]

[project.scripts]
heatwires = "heatwires.cli:main"

[build-system]
requires = ["setuptools>=64"]
build-backend = "setuptools.build_meta"

[tool.setuptools]
packages = ["heatwires"]

# Local config and runtime state (never commit these)
config.yaml
state.json

# Python
__pycache__/
*.py[cod]
*.egg-info/
.eggs/
build/
dist/
.venv/
venv/

# Copy this to config.yaml and edit for your site:  cp config.example.yaml config.yaml

# --- Location -----------------------------------------------------------------
# Decimal degrees for your roof. (Default below is Bozeman, MT.)
location:
  latitude: 45.6770
  longitude: -111.0429

# --- Weather (US National Weather Service, api.weather.gov) --------------------
weather:
  # NWS requires a User-Agent identifying you. An email is ideal.
  user_agent_contact: "you@example.com"
  # How long to cache the NWS gridpoint lookup (it never really changes).
  points_cache_seconds: 86400

# --- Switches -----------------------------------------------------------------
# Your Kasa KP400 switch(es). Give each a reserved/static IP in your router so
# the address never changes. Run `heatwires -c config.yaml discover` to find them.
switches:
  - host: "192.168.1.50"
    outlets: all          # "all", or a list of outlet indexes, e.g. [0] or [0, 1]
  # - host: "192.168.1.51"
  #   outlets: [0]

# --- Icing logic --------------------------------------------------------------
# Ice dams form in a melt/refreeze window near freezing WITH snow present.
icing:
  temp_on_low_f: 15.0      # below this: too cold to melt/refreeze — cables OFF
  temp_on_high_f: 38.0     # above this: snow sheds on its own — cables OFF
  hysteresis_f: 3.0        # once ON, widen the band by this much to avoid flapping
  require_snow: true       # only run when frozen precip is present/forecast
  min_switch_minutes: 30   # minimum dwell time between state changes

# --- Fail-safe ----------------------------------------------------------------
# If the weather can't be fetched, bias toward safety during winter months.
fail_safe:
  enabled: true
  action: "on"                     # "on" (safe) or "off"
  months: [11, 12, 1, 2, 3, 4]     # only fail-safe during these months

# --- Misc ---------------------------------------------------------------------
# Relative paths are resolved next to this config file.
state_file: "state.json"
logging:
  level: "INFO"

"""Automated weather-based control of roof heat-cable switches."""

__version__ = "0.1.0"

"""Load and validate the YAML configuration."""

from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Union

import yaml


@dataclass
class Location:
    latitude: float
    longitude: float


@dataclass
class WeatherConfig:
    user_agent_contact: str
    points_cache_seconds: int = 86400


@dataclass
class SwitchConfig:
    host: str
    outlets: Union[str, List[int]] = "all"  # "all" or a list of outlet indexes


@dataclass
class IcingConfig:
    temp_on_low_f: float = 15.0
    temp_on_high_f: float = 38.0
    hysteresis_f: float = 3.0
    require_snow: bool = True
    min_switch_minutes: int = 30


@dataclass
class FailSafeConfig:
    enabled: bool = True
    action: str = "on"  # "on" or "off"
    months: List[int] = field(default_factory=lambda: [11, 12, 1, 2, 3, 4])


@dataclass
class Config:
    location: Location
    weather: WeatherConfig
    switches: List[SwitchConfig]
    icing: IcingConfig
    fail_safe: FailSafeConfig
    state_file: Path
    log_level: str
    path: Path


class ConfigError(Exception):
    """Raised when the configuration is missing or invalid."""


def _require(mapping: dict, key: str, where: str):
    if key not in mapping:
        raise ConfigError(f"missing required '{key}' under {where}")
    return mapping[key]


def _parse_outlets(value) -> Union[str, List[int]]:
    if value in (None, "all"):
        return "all"
    if isinstance(value, int):
        return [value]
    if isinstance(value, list) and all(isinstance(i, int) for i in value):
        return list(value)
    raise ConfigError(f"'outlets' must be 'all' or a list of integers, got {value!r}")


def load_config(path: Union[str, Path]) -> Config:
    path = Path(path).expanduser().resolve()
    if not path.exists():
        raise ConfigError(
            f"config file not found: {path}\n"
            "Create one with:  cp config.example.yaml config.yaml"
        )
    raw = yaml.safe_load(path.read_text()) or {}
    base = path.parent

    loc = _require(raw, "location", "top level")
    location = Location(
        latitude=float(_require(loc, "latitude", "location")),
        longitude=float(_require(loc, "longitude", "location")),
    )

    wx = raw.get("weather", {}) or {}
    weather = WeatherConfig(
        user_agent_contact=str(
            _require(wx, "user_agent_contact", "weather")
        ),
        points_cache_seconds=int(wx.get("points_cache_seconds", 86400)),
    )

    raw_switches = _require(raw, "switches", "top level")
    if not isinstance(raw_switches, list) or not raw_switches:
        raise ConfigError("'switches' must be a non-empty list")
    switches = [
        SwitchConfig(
            host=str(_require(s, "host", "switches[]")),
            outlets=_parse_outlets(s.get("outlets", "all")),
        )
        for s in raw_switches
    ]

    ic = raw.get("icing", {}) or {}
    icing = IcingConfig(
        temp_on_low_f=float(ic.get("temp_on_low_f", 15.0)),
        temp_on_high_f=float(ic.get("temp_on_high_f", 38.0)),
        hysteresis_f=float(ic.get("hysteresis_f", 3.0)),
        require_snow=bool(ic.get("require_snow", True)),
        min_switch_minutes=int(ic.get("min_switch_minutes", 30)),
    )
    if icing.temp_on_low_f >= icing.temp_on_high_f:
        raise ConfigError("icing.temp_on_low_f must be below icing.temp_on_high_f")

    fs = raw.get("fail_safe", {}) or {}
    fail_safe = FailSafeConfig(
        enabled=bool(fs.get("enabled", True)),
        action=str(fs.get("action", "on")).lower(),
        months=list(fs.get("months", [11, 12, 1, 2, 3, 4])),
    )
    if fail_safe.action not in ("on", "off"):
        raise ConfigError("fail_safe.action must be 'on' or 'off'")

    state_file = Path(raw.get("state_file", "state.json"))
    if not state_file.is_absolute():
        state_file = (base / state_file).resolve()

    log_level = str((raw.get("logging", {}) or {}).get("level", "INFO")).upper()

    return Config(
        location=location,
        weather=weather,
        switches=switches,
        icing=icing,
        fail_safe=fail_safe,
        state_file=state_file,
        log_level=log_level,
        path=path,
    )

"""Fetch current/near-term conditions from the US National Weather Service.

api.weather.gov is free, keyless, US-only, and requires a descriptive
User-Agent header. Flow:

    /points/{lat},{lon}  ->  forecastHourly + observationStations URLs
    observationStations  ->  nearest station  ->  observations/latest
    forecastHourly       ->  next hour's forecast (fallback + snow signal)
"""

from __future__ import annotations

import logging
import time
from dataclasses import dataclass
from typing import Optional

import requests

log = logging.getLogger("heatwires.weather")

NWS_BASE = "https://api.weather.gov"

# Words in a conditions/forecast string that indicate frozen precipitation.
SNOW_WORDS = (
    "snow",
    "sleet",
    "ice",
    "icy",
    "wintry",
    "flurr",
    "freezing",
    "blizzard",
)


@dataclass
class WeatherSnapshot:
    temp_f: Optional[float]
    snow_signal: bool
    description: str
    ok: bool
    source: str = ""

    def summary(self) -> str:
        if not self.ok:
            return f"UNAVAILABLE ({self.description})"
        t = f"{self.temp_f:.0f}°F" if self.temp_f is not None else "?"
        snow = "snow/ice" if self.snow_signal else "no snow signal"
        return f"{t}, {snow} — {self.description or 'n/a'} [{self.source}]"


def _c_to_f(celsius: float) -> float:
    return celsius * 9.0 / 5.0 + 32.0


class WeatherClient:
    def __init__(
        self,
        latitude: float,
        longitude: float,
        user_agent_contact: str,
        points_cache_seconds: int = 86400,
    ):
        self.lat = latitude
        self.lon = longitude
        self.session = requests.Session()
        self.session.headers.update(
            {
                "User-Agent": f"heatwires/0.1 ({user_agent_contact})",
                "Accept": "application/geo+json",
            }
        )
        self._points_cache_seconds = points_cache_seconds
        self._hourly_url: Optional[str] = None
        self._stations_url: Optional[str] = None
        self._points_at = 0.0

    def _get(self, url: str, tries: int = 3) -> dict:
        last_exc: Optional[Exception] = None
        for attempt in range(1, tries + 1):
            try:
                resp = self.session.get(url, timeout=15)
                resp.raise_for_status()
                return resp.json()
            except Exception as exc:  # noqa: BLE001 - retry on any transient failure
                last_exc = exc
                log.debug("GET %s failed (attempt %d/%d): %s", url, attempt, tries, exc)
                if attempt < tries:
                    time.sleep(1.5 * attempt)
        assert last_exc is not None
        raise last_exc

    def _ensure_points(self) -> None:
        fresh = self._hourly_url and (
            time.time() - self._points_at
        ) < self._points_cache_seconds
        if fresh:
            return
        data = self._get(f"{NWS_BASE}/points/{self.lat},{self.lon}")
        props = data["properties"]
        self._hourly_url = props["forecastHourly"]
        self._stations_url = props["observationStations"]
        self._points_at = time.time()

    def _latest_observation(self) -> dict:
        stations = self._get(self._stations_url)
        features = stations.get("features") or []
        if not features:
            raise RuntimeError("no observation stations for this location")
        station_url = features[0]["id"]  # e.g. .../stations/KBZN
        return self._get(f"{station_url}/observations/latest")["properties"]

    def fetch(self) -> WeatherSnapshot:
        """Return a snapshot; never raises — failures come back as ok=False."""
        try:
            self._ensure_points()

            temp_f: Optional[float] = None
            obs_desc = ""
            source = "observation"

            # Current observation is the most authoritative temperature.
            try:
                obs = self._latest_observation()
                celsius = (obs.get("temperature") or {}).get("value")
                if celsius is not None:
                    temp_f = _c_to_f(celsius)
                obs_desc = obs.get("textDescription") or ""
            except Exception as exc:  # noqa: BLE001 - fall back to forecast
                log.debug("observation fetch failed, using forecast: %s", exc)
                source = ""

            # Hourly forecast: fills in temp if observation lacked it, and
            # contributes to the snow/ice signal.
            hourly = self._get(self._hourly_url)
            period = hourly["properties"]["periods"][0]
            if temp_f is None:
                t = period.get("temperature")
                if t is not None:
                    temp_f = (
                        _c_to_f(float(t))
                        if period.get("temperatureUnit") == "C"
                        else float(t)
                    )
                source = "forecast"
            fc_desc = period.get("shortForecast", "") or ""

            blob = f"{obs_desc} {fc_desc}".lower()
            snow = any(word in blob for word in SNOW_WORDS)

            return WeatherSnapshot(
                temp_f=temp_f,
                snow_signal=snow,
                description=(obs_desc or fc_desc).strip(),
                ok=temp_f is not None,
                source=source or "forecast",
            )
        except Exception as exc:  # noqa: BLE001 - report as unavailable
            log.warning("weather fetch failed: %s", exc)
            return WeatherSnapshot(
                temp_f=None,
                snow_signal=False,
                description=f"fetch failed: {exc}",
                ok=False,
            )

"""The icing decision: given weather, should the cables be on?"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

from .config import IcingConfig
from .weather import WeatherSnapshot


@dataclass
class Decision:
    want_on: bool
    reason: str


def decide(
    snapshot: WeatherSnapshot,
    icing: IcingConfig,
    currently_on: bool,
) -> Optional[Decision]:
    """Return the desired state, or None if weather is unavailable.

    Ice dams form in a melt/refreeze window near freezing with snow present,
    so the cables run when the temperature is inside [low, high] AND (if
    required) frozen precipitation is present. Once on, the band is widened by
    ``hysteresis_f`` on each side to avoid rapid cycling around a threshold.
    """
    if snapshot.temp_f is None:
        return None

    margin = icing.hysteresis_f if currently_on else 0.0
    low = icing.temp_on_low_f - margin
    high = icing.temp_on_high_f + margin
    temp = snapshot.temp_f

    in_band = low <= temp <= high
    snow_ok = snapshot.snow_signal or not icing.require_snow
    want_on = in_band and snow_ok

    if want_on:
        snow_note = (
            "with snow/ice signal"
            if snapshot.snow_signal
            else "(snow not required)"
        )
        reason = f"{temp:.0f}°F within [{low:.0f}, {high:.0f}]°F icing band {snow_note}"
    elif not in_band:
        if temp < low:
            reason = f"{temp:.0f}°F below {low:.0f}°F — too cold for melt/refreeze"
        else:
            reason = f"{temp:.0f}°F above {high:.0f}°F — snow sheds naturally"
    else:
        reason = f"{temp:.0f}°F in band but no snow/ice signal"

    return Decision(want_on=want_on, reason=reason)

"""Tiny JSON persistence for run-to-run state (last known switch state, timing)."""

from __future__ import annotations

import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

log = logging.getLogger("heatwires.state")


def load_state(path: Path) -> dict:
    try:
        return json.loads(Path(path).read_text())
    except FileNotFoundError:
        return {}
    except Exception as exc:  # noqa: BLE001 - corrupt state shouldn't crash a run
        log.warning("could not read state %s (%s); starting fresh", path, exc)
        return {}


def save_state(path: Path, state: dict) -> None:
    Path(path).write_text(json.dumps(state, indent=2, sort_keys=True))


def parse_ts(value: Optional[str]) -> Optional[datetime]:
    if not value:
        return None
    try:
        dt = datetime.fromisoformat(value)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except ValueError:
        return None

"""Control TP-Link Kasa switches (e.g. KP400) directly over the LAN.

Uses python-kasa's discovery handshake so we don't need to hard-code the
device class. The KP400 exposes its two outlets as ``children``; single-outlet
plugs have no children and are controlled as a whole.
"""

from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field
from typing import List, Optional, Sequence

from kasa import Discover

from .config import SwitchConfig

log = logging.getLogger("heatwires.switches")


@dataclass
class OutletState:
    index: int
    alias: str
    is_on: bool


@dataclass
class SwitchState:
    host: str
    ok: bool
    alias: str = ""
    model: str = ""
    error: str = ""
    outlets: List[OutletState] = field(default_factory=list)

    def summary(self) -> str:
        if not self.ok:
            return f"{self.host}: ERROR ({self.error})"
        parts = [
            f"[{o.index}] {o.alias}={'ON' if o.is_on else 'off'}" for o in self.outlets
        ]
        label = self.alias or self.model or "device"
        return f"{self.host} ({label}): " + ", ".join(parts)


def _selected(children: Sequence, outlets) -> List:
    if not children:
        return []
    if outlets == "all":
        return list(children)
    picked = []
    for i in outlets:
        if 0 <= i < len(children):
            picked.append(children[i])
        else:
            log.warning("outlet index %d out of range (device has %d)", i, len(children))
    return picked


async def _read_device(dev) -> SwitchState:
    children = getattr(dev, "children", None) or []
    if children:
        outlets = [
            OutletState(index=i, alias=c.alias or f"outlet{i}", is_on=bool(c.is_on))
            for i, c in enumerate(children)
        ]
    else:
        outlets = [OutletState(index=0, alias=dev.alias or "outlet", is_on=bool(dev.is_on))]
    return SwitchState(
        host=dev.host,
        ok=True,
        alias=dev.alias or "",
        model=getattr(dev, "model", "") or "",
        outlets=outlets,
    )


async def _apply_one(sw: SwitchConfig, turn_on: Optional[bool]) -> SwitchState:
    """Set (or, if turn_on is None, just read) a single switch."""
    try:
        dev = await Discover.discover_single(sw.host)
        await dev.update()
        if turn_on is not None:
            children = getattr(dev, "children", None) or []
            targets = _selected(children, sw.outlets) if children else [dev]
            for target in targets:
                if turn_on:
                    await target.turn_on()
                else:
                    await target.turn_off()
            await dev.update()
        return await _read_device(dev)
    except Exception as exc:  # noqa: BLE001 - surface per-switch, don't abort others
        log.error("switch %s failed: %s", sw.host, exc)
        return SwitchState(host=sw.host, ok=False, error=str(exc))


async def _apply_all(
    switches: Sequence[SwitchConfig], turn_on: Optional[bool]
) -> List[SwitchState]:
    return await asyncio.gather(*(_apply_one(s, turn_on) for s in switches))


def set_switches(
    switches: Sequence[SwitchConfig], turn_on: bool
) -> List[SwitchState]:
    """Turn all configured switches/outlets on or off. Returns resulting state."""
    return asyncio.run(_apply_all(switches, turn_on))


def read_switches(switches: Sequence[SwitchConfig]) -> List[SwitchState]:
    """Read current state without changing anything."""
    return asyncio.run(_apply_all(switches, None))


def discover(timeout: int = 8) -> List[SwitchState]:
    """Discover all Kasa devices on the LAN."""

    async def _run() -> List[SwitchState]:
        found = await Discover.discover(discovery_timeout=timeout)
        states = []
        for dev in found.values():
            try:
                await dev.update()
                states.append(await _read_device(dev))
            except Exception as exc:  # noqa: BLE001
                states.append(SwitchState(host=dev.host, ok=False, error=str(exc)))
        return states

    return asyncio.run(_run())

"""Orchestrate one control cycle: weather -> decision -> switches -> state."""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import List, Optional

from .config import Config
from .logic import decide
from .state import load_state, parse_ts, save_state
from .switches import SwitchState, read_switches, set_switches
from .weather import WeatherClient, WeatherSnapshot

log = logging.getLogger("heatwires.controller")


@dataclass
class CycleResult:
    snapshot: WeatherSnapshot
    want_on: bool
    changed: bool
    reason: str
    applied: bool
    switch_states: List[SwitchState] = field(default_factory=list)


def _failsafe_target(config: Config, now: datetime) -> Optional[bool]:
    """Desired state when weather is unavailable, or None to hold current."""
    fs = config.fail_safe
    if not fs.enabled or now.month not in fs.months:
        return None
    return fs.action == "on"


def run_cycle(config: Config, apply: bool = True) -> CycleResult:
    now = datetime.now(timezone.utc)

    client = WeatherClient(
        latitude=config.location.latitude,
        longitude=config.location.longitude,
        user_agent_contact=config.weather.user_agent_contact,
        points_cache_seconds=config.weather.points_cache_seconds,
    )
    snapshot = client.fetch()

    state = load_state(config.state_file)
    currently_on = state.get("state") == "on"

    decision = decide(snapshot, config.icing, currently_on)
    forced = False

    if decision is None:
        target = _failsafe_target(config, now)
        if target is None:
            want_on = currently_on
            reason = "weather unavailable; holding current state (fail-safe inactive)"
        else:
            want_on = target
            forced = True
            reason = f"weather unavailable; fail-safe -> {'ON' if want_on else 'OFF'}"
    else:
        want_on = decision.want_on
        reason = decision.reason

    # Enforce minimum dwell time between changes (skipped when fail-safe forces).
    changed = want_on != currently_on
    if changed and not forced:
        last_change = parse_ts(state.get("last_change"))
        if last_change is not None:
            elapsed_min = (now - last_change).total_seconds() / 60.0
            if elapsed_min < config.icing.min_switch_minutes:
                want_on = currently_on
                changed = False
                reason += (
                    f" [held {elapsed_min:.0f}m < {config.icing.min_switch_minutes}m "
                    "min dwell]"
                )

    switch_states: List[SwitchState] = []
    if apply:
        switch_states = set_switches(config.switches, want_on)
        unreachable = [s.host for s in switch_states if not s.ok]
        if unreachable:
            log.error("switches unreachable: %s", ", ".join(unreachable))
    else:
        switch_states = read_switches(config.switches)

    new_state = dict(state)
    new_state["state"] = "on" if want_on else "off"
    new_state["last_run"] = now.isoformat()
    new_state["last_reason"] = reason
    new_state["last_temp_f"] = snapshot.temp_f
    if changed or "last_change" not in new_state:
        new_state["last_change"] = now.isoformat()
    if apply:
        save_state(config.state_file, new_state)

    log.info(
        "weather: %s | decision: %s -> %s%s",
        snapshot.summary(),
        "ON" if want_on else "OFF",
        "changed" if changed else "unchanged",
        f" | {reason}",
    )

    return CycleResult(
        snapshot=snapshot,
        want_on=want_on,
        changed=changed,
        reason=reason,
        applied=apply,
        switch_states=switch_states,
    )

"""Command-line entry point.

    heatwires -c config.yaml discover        # find Kasa devices on the LAN
    heatwires -c config.yaml status          # show weather + switches + decision (no change)
    heatwires -c config.yaml run             # one control cycle (for cron)
    heatwires -c config.yaml run --dry-run   # decide + read switches, but don't change them
    heatwires -c config.yaml loop -i 900     # run forever, every 900s (for systemd)
"""

from __future__ import annotations

import argparse
import logging
import sys
import time

from .config import ConfigError, load_config
from .controller import run_cycle
from .logic import decide
from .switches import discover as discover_switches
from .switches import read_switches
from .weather import WeatherClient


def _setup_logging(level: str) -> None:
    logging.basicConfig(
        level=getattr(logging, level, logging.INFO),
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )


def _cmd_discover(args) -> int:
    print(f"Discovering Kasa devices (timeout {args.timeout}s)...")
    devices = discover_switches(timeout=args.timeout)
    if not devices:
        print("No devices found. Ensure this machine is on the same LAN/subnet.")
        return 1
    for dev in devices:
        print("  " + dev.summary())
    return 0


def _cmd_status(args) -> int:
    config = load_config(args.config)
    client = WeatherClient(
        latitude=config.location.latitude,
        longitude=config.location.longitude,
        user_agent_contact=config.weather.user_agent_contact,
        points_cache_seconds=config.weather.points_cache_seconds,
    )
    snapshot = client.fetch()
    print(f"Weather:  {snapshot.summary()}")

    states = read_switches(config.switches)
    print("Switches:")
    for st in states:
        print("  " + st.summary())
    any_on = any(o.is_on for st in states if st.ok for o in st.outlets)

    decision = decide(snapshot, config.icing, currently_on=any_on)
    if decision is None:
        print("Decision: weather unavailable")
    else:
        print(
            f"Decision: cables should be "
            f"{'ON' if decision.want_on else 'OFF'} — {decision.reason}"
        )
    return 0


def _cmd_run(args) -> int:
    config = load_config(args.config)
    result = run_cycle(config, apply=not args.dry_run)
    verb = "would be" if args.dry_run else "set"
    print(
        f"Cables {verb} {'ON' if result.want_on else 'OFF'} "
        f"({'changed' if result.changed else 'unchanged'}) — {result.reason}"
    )
    for st in result.switch_states:
        print("  " + st.summary())
    # Non-zero exit if any switch was unreachable, so cron/systemd can alert.
    return 0 if all(st.ok for st in result.switch_states) else 2


def _cmd_loop(args) -> int:
    config = load_config(args.config)
    log = logging.getLogger("heatwires.loop")
    log.info("starting loop, interval %ds", args.interval)
    while True:
        try:
            run_cycle(config, apply=True)
        except Exception as exc:  # noqa: BLE001 - keep the loop alive
            log.exception("cycle failed: %s", exc)
        try:
            time.sleep(args.interval)
        except KeyboardInterrupt:
            log.info("interrupted, exiting")
            return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="heatwires",
        description="Weather-based control of roof heat-cable switches.",
    )
    parser.add_argument(
        "-c",
        "--config",
        default="config.yaml",
        help="path to config file (default: config.yaml)",
    )
    sub = parser.add_subparsers(dest="command", required=True)

    p_disc = sub.add_parser("discover", help="find Kasa devices on the LAN")
    p_disc.add_argument("-t", "--timeout", type=int, default=8)
    p_disc.set_defaults(func=_cmd_discover)

    p_status = sub.add_parser(
        "status", help="show weather, switch states, and the decision (no change)"
    )
    p_status.set_defaults(func=_cmd_status)

    p_run = sub.add_parser("run", help="run one control cycle (for cron)")
    p_run.add_argument(
        "--dry-run",
        action="store_true",
        help="decide and read switches but do not change them",
    )
    p_run.set_defaults(func=_cmd_run)

    p_loop = sub.add_parser("loop", help="run continuously (for systemd)")
    p_loop.add_argument(
        "-i", "--interval", type=int, default=900, help="seconds between cycles"
    )
    p_loop.set_defaults(func=_cmd_loop)

    return parser


def main(argv=None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        # `discover` doesn't need a config; everything else does.
        if args.command != "discover":
            _setup_logging(load_config(args.config).log_level)
        else:
            _setup_logging("INFO")
        return args.func(args)
    except ConfigError as exc:
        print(f"Config error: {exc}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        return 130


if __name__ == "__main__":
    raise SystemExit(main())

"""Offline tests for the icing decision logic."""

from heatwires.config import IcingConfig
from heatwires.logic import decide
from heatwires.weather import WeatherSnapshot


def snap(temp_f, snow=True):
    return WeatherSnapshot(
        temp_f=temp_f, snow_signal=snow, description="test", ok=temp_f is not None
    )


ICING = IcingConfig(
    temp_on_low_f=15.0,
    temp_on_high_f=38.0,
    hysteresis_f=3.0,
    require_snow=True,
    min_switch_minutes=30,
)


def test_on_inside_band_with_snow():
    d = decide(snap(30, snow=True), ICING, currently_on=False)
    assert d.want_on is True


def test_off_too_warm():
    d = decide(snap(45, snow=True), ICING, currently_on=False)
    assert d.want_on is False
    assert "sheds" in d.reason


def test_off_too_cold():
    d = decide(snap(5, snow=True), ICING, currently_on=False)
    assert d.want_on is False
    assert "too cold" in d.reason


def test_off_in_band_but_no_snow_when_required():
    d = decide(snap(30, snow=False), ICING, currently_on=False)
    assert d.want_on is False
    assert "no snow" in d.reason


def test_on_in_band_no_snow_when_not_required():
    cfg = IcingConfig(**{**ICING.__dict__, "require_snow": False})
    d = decide(snap(30, snow=False), cfg, currently_on=False)
    assert d.want_on is True


def test_hysteresis_keeps_on_just_above_high():
    # 40°F is above 38 but within the widened band (38+3=41) once already on.
    off_state = decide(snap(40, snow=True), ICING, currently_on=False)
    on_state = decide(snap(40, snow=True), ICING, currently_on=True)
    assert off_state.want_on is False
    assert on_state.want_on is True


def test_hysteresis_keeps_on_just_below_low():
    # 13°F is below 15 but within the widened band (15-3=12) once already on.
    assert decide(snap(13, snow=True), ICING, currently_on=False).want_on is False
    assert decide(snap(13, snow=True), ICING, currently_on=True).want_on is True


def test_none_when_weather_unavailable():
    assert decide(snap(None), ICING, currently_on=False) is None

"""Offline tests for NWS response parsing, using a fake HTTP session."""

from heatwires.weather import WeatherClient


class FakeResponse:
    def __init__(self, payload):
        self._payload = payload

    def raise_for_status(self):
        pass

    def json(self):
        return self._payload


class FakeSession:
    """Routes GET calls to canned payloads keyed by URL substring."""

    def __init__(self, routes, fail_on=None):
        self.headers = {}
        self._routes = routes
        self._fail_on = fail_on or ()

    def get(self, url, timeout=None):
        for needle in self._fail_on:
            if needle in url:
                raise RuntimeError(f"boom on {needle}")
        for needle, payload in self._routes.items():
            if needle in url:
                return FakeResponse(payload)
        raise AssertionError(f"unexpected URL: {url}")


POINTS = {
    "properties": {
        "forecastHourly": "https://api.weather.gov/gridpoints/X/1,1/forecast/hourly",
        "observationStations": "https://api.weather.gov/gridpoints/X/1,1/stations",
    }
}
STATIONS = {"features": [{"id": "https://api.weather.gov/stations/KBZN"}]}


def make_client(routes, fail_on=None):
    client = WeatherClient(45.0, -111.0, "test@example.com")
    client.session = FakeSession(routes, fail_on=fail_on)
    return client


def test_uses_observation_temperature_and_detects_snow():
    routes = {
        "/points/": POINTS,
        "/stations": STATIONS,
        "observations/latest": {
            "properties": {
                "temperature": {"value": 0.0},  # 0°C -> 32°F
                "textDescription": "Light Snow",
            }
        },
        "forecast/hourly": {
            "properties": {"periods": [{"temperature": 33, "shortForecast": "Snow"}]}
        },
    }
    snap = make_client(routes).fetch()
    assert snap.ok is True
    assert round(snap.temp_f) == 32
    assert snap.snow_signal is True
    assert snap.source == "observation"


def test_falls_back_to_forecast_when_observation_fails():
    routes = {
        "/points/": POINTS,
        "/stations": STATIONS,
        "forecast/hourly": {
            "properties": {
                "periods": [{"temperature": 28, "shortForecast": "Partly Cloudy"}]
            }
        },
    }
    snap = make_client(routes, fail_on=("observations/latest",)).fetch()
    assert snap.ok is True
    assert round(snap.temp_f) == 28
    assert snap.snow_signal is False
    assert snap.source == "forecast"


def test_reports_unavailable_when_points_fails():
    snap = make_client({}, fail_on=("/points/",)).fetch()
    assert snap.ok is False
    assert snap.temp_f is None


def test_celsius_forecast_is_converted():
    routes = {
        "/points/": POINTS,
        "/stations": STATIONS,
        "forecast/hourly": {
            "properties": {
                "periods": [
                    {"temperature": 0, "temperatureUnit": "C", "shortForecast": "Ice"}
                ]
            }
        },
    }
    snap = make_client(routes, fail_on=("observations/latest",)).fetch()
    assert round(snap.temp_f) == 32
    assert snap.snow_signal is True  # "Ice" matches

# heatwires

Automated, weather-based control of roof **heat cables** (de-icing / ice-dam
prevention) via **TP-Link Kasa KP400** smart outdoor plugs.

It queries the US National Weather Service for your location, decides whether
icing conditions warrant running the cables, and switches your Kasa plugs
**directly over the LAN** (no cloud account, no reverse-engineered web API).

## How it decides

Ice dams form in a **melt/refreeze window near freezing with snow present**:
snow melts, water runs to the cold eaves, and refreezes. So the cables run when:

- the temperature is inside a configurable band (default **15–38 °F**), **and**
- frozen precipitation is present/forecast (can be disabled).

Extras for real-world robustness:

- **Hysteresis** widens the band once the cables are on, to avoid flapping.
- **Minimum dwell time** prevents rapid on/off cycling.
- **Fail-safe**: if weather can't be fetched during winter months, bias to a
  safe state (default: ON).
- **Graceful degradation**: an unreachable switch is logged and reported (and
  makes `run` exit non-zero) without aborting the rest of the cycle.

## Install

Requires Python 3.10+. Run this on an always-on machine **on the same LAN** as
the switches (e.g. a Raspberry Pi).

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

## Configure

```bash
cp config.example.yaml config.yaml
```

Then edit `config.yaml`:

- `location` — your roof's latitude/longitude (decimal degrees).
- `weather.user_agent_contact` — an email; NWS requires an identifying User-Agent.
- `switches` — the IP(s) of your KP400(s) and which `outlets` to control.
- `icing` — the temperature band and behavior tunables.

Give each KP400 a **reserved/static IP** in your router so its address never
changes. Find the switches on your network with:

```bash
heatwires discover
```

`config.yaml` and `state.json` are git-ignored (they're host-specific).

## Use

```bash
heatwires -c config.yaml discover        # list Kasa devices on the LAN
heatwires -c config.yaml status          # show weather + switch state + decision (no change)
heatwires -c config.yaml run             # run one control cycle (turns cables on/off)
heatwires -c config.yaml run --dry-run   # decide + read switches, but don't change them
heatwires -c config.yaml loop -i 900     # run continuously, every 900s
```

## Deploy

**Option A — cron** (runs one cycle every 15 minutes):

```cron
*/15 * * * * /path/to/.venv/bin/heatwires -c /path/to/config.yaml run >> /path/to/heatwires.log 2>&1
```

**Option B — systemd service** (long-running loop, auto-restart):

```ini
# /etc/systemd/system/heatwires.service
[Unit]
Description=Roof heat-cable weather controller
After=network-online.target
Wants=network-online.target

[Service]
ExecStart=/path/to/.venv/bin/heatwires -c /path/to/config.yaml loop -i 900
Restart=always
RestartSec=30
User=pi

[Install]
WantedBy=multi-user.target
```

```bash
sudo systemctl enable --now heatwires
journalctl -u heatwires -f
```

The Kasa app keeps working independently as a manual override at any time.

## Develop

```bash
pip install pytest
pytest -q          # offline tests for the decision logic and NWS parsing
```

## Layout

| File | Purpose |
|---|---|
| `heatwires/weather.py` | NWS client → temperature + snow/ice signal |
| `heatwires/logic.py` | the icing decision (band + snow + hysteresis) |
| `heatwires/switches.py` | Kasa control over the LAN (python-kasa) |
| `heatwires/controller.py` | one cycle: weather → decision → switches → state |
| `heatwires/state.py` | run-to-run state (last state, dwell timing) |
| `heatwires/config.py` | YAML config loading/validation |
| `heatwires/cli.py` | `discover` / `status` / `run` / `loop` commands |

# heatwires

Automated, weather-based control of roof **heat cables** (de-icing / ice-dam
prevention) via **TP-Link Kasa KP400** smart outdoor plugs.

It queries the US National Weather Service for your location, decides whether
icing conditions warrant running the cables, and switches your Kasa plugs
**directly over the LAN** (no cloud account, no reverse-engineered web API).

## How it decides

Ice dams form in a **melt/refreeze window near freezing with snow present**:
snow melts, water runs to the cold eaves, and refreezes. So the cables run when:

- the temperature is inside a configurable band (default **15–38 °F**), **and**
- frozen precipitation is present/forecast (can be disabled).

Extras for real-world robustness:

- **Hysteresis** widens the band once the cables are on, to avoid flapping.
- **Minimum dwell time** prevents rapid on/off cycling.
- **Fail-safe**: if weather can't be fetched during winter months, bias to a
  safe state (default: ON).
- **Graceful degradation**: an unreachable switch is logged and reported (and
  makes `run` exit non-zero) without aborting the rest of the cycle.

## Install

Requires Python 3.10+. Run this on an always-on machine **on the same LAN** as
the switches (e.g. a Raspberry Pi).

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

## Configure

```bash
cp config.example.yaml config.yaml
```

Then edit `config.yaml`:

- `location` — your roof's latitude/longitude (decimal degrees).
- `weather.user_agent_contact` — an email; NWS requires an identifying User-Agent.
- `switches` — the IP(s) of your KP400(s) and which `outlets` to control.
- `icing` — the temperature band and behavior tunables.

Give each KP400 a **reserved/static IP** in your router so its address never
changes. Find the switches on your network with:

```bash
heatwires discover
```

`config.yaml` and `state.json` are git-ignored (they're host-specific).

## Use

```bash
heatwires -c config.yaml discover        # list Kasa devices on the LAN
heatwires -c config.yaml status          # show weather + switch state + decision (no change)
heatwires -c config.yaml run             # run one control cycle (turns cables on/off)
heatwires -c config.yaml run --dry-run   # decide + read switches, but don't change them
heatwires -c config.yaml loop -i 900     # run continuously, every 900s
```

## Deploy

**Option A — cron** (runs one cycle every 15 minutes):

```cron
*/15 * * * * /path/to/.venv/bin/heatwires -c /path/to/config.yaml run >> /path/to/heatwires.log 2>&1
```

**Option B — systemd service** (long-running loop, auto-restart):

```ini
# /etc/systemd/system/heatwires.service
[Unit]
Description=Roof heat-cable weather controller
After=network-online.target
Wants=network-online.target

[Service]
ExecStart=/path/to/.venv/bin/heatwires -c /path/to/config.yaml loop -i 900
Restart=always
RestartSec=30
User=pi

[Install]
WantedBy=multi-user.target
```

```bash
sudo systemctl enable --now heatwires
journalctl -u heatwires -f
```

The Kasa app keeps working independently as a manual override at any time.

## Develop

```bash
pip install pytest
pytest -q          # offline tests for the decision logic and NWS parsing
```

## Layout

| File | Purpose |
|---|---|
| `heatwires/weather.py` | NWS client → temperature + snow/ice signal |
| `heatwires/logic.py` | the icing decision (band + snow + hysteresis) |
| `heatwires/switches.py` | Kasa control over the LAN (python-kasa) |
| `heatwires/controller.py` | one cycle: weather → decision → switches → state |
| `heatwires/state.py` | run-to-run state (last state, dwell timing) |
| `heatwires/config.py` | YAML config loading/validation |
| `heatwires/cli.py` | `discover` / `status` / `run` / `loop` commands |