HVAC Runtime-vs-Temperature Characterization (WS2) Implementation Plan¶

# tests/analysis/test_hvac_correlation.py
import math
import importlib.util
from pathlib import Path

_spec = importlib.util.spec_from_file_location(
    "hvac_corr",
    Path(__file__).resolve().parents[2] / "scripts" / "analysis" / "hvac_runtime_correlation.py",
)
hvac = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(hvac)


def test_pair_by_day_inner_joins_and_drops_unmatched():
    runtime = {"2026-07-01": 8.0, "2026-07-02": 10.0, "2026-07-03": 6.0}
    temps = {"2026-07-01": 96.0, "2026-07-02": 100.0, "2026-07-09": 88.0}
    pairs = hvac.pair_by_day(runtime, temps)
    # only the two overlapping days survive; ordered by day
    assert pairs == [(96.0, 8.0), (100.0, 10.0)]


def test_analyze_strong_positive_correlation_is_ok():
    # runtime rises perfectly with temperature
    pairs = [(80.0, 4.0), (85.0, 6.0), (90.0, 8.0), (95.0, 10.0), (100.0, 12.0)]
    out = hvac.analyze(pairs)
    assert out["n"] == 5
    assert out["r"] is not None and out["r"] > 0.99
    assert out["slope"] is not None and out["slope"] > 0
    assert out["verdict"] == "OK"


def test_analyze_no_coupling_is_flagged():
    # runtime flat/uncorrelated with temperature -> weak r
    pairs = [(80.0, 7.0), (85.0, 7.1), (90.0, 6.9), (95.0, 7.0), (100.0, 7.05)]
    out = hvac.analyze(pairs)
    assert out["verdict"] == "FLAG"
    assert out["r"] is None or abs(out["r"]) < hvac.FLAG_THRESHOLD


def test_analyze_insufficient_samples():
    out = hvac.analyze([(90.0, 8.0), (95.0, 10.0)])
    assert out["verdict"] == "INSUFFICIENT"
    assert out["n"] == 2

# scripts/analysis/hvac_runtime_correlation.py
"""WS2 HVAC diagnostic: correlate daily AC condenser runtime with outdoor daily high.

Pure analysis functions live here and are unit-tested. The InfluxDB fetch + CLI
(added in Task 2) shells out to `ssh haos` so no DB credentials touch this machine.
"""
from __future__ import annotations

import statistics

# r below this => runtime does not track outdoor temperature (possible stuck relay,
# thermostat fault, or refrigerant problem). Tuned once against real data.
FLAG_THRESHOLD: float = 0.4

# (display name, InfluxDB entity_id tag for *_condenser_runtime_today — no 'sensor.' prefix)
CONDENSERS: list[tuple[str, str]] = [
    ("Floor 1", "first_floor_condenser_runtime_today"),
    ("Floor 2", "second_floor_condenser_runtime_today"),
    ("Floor 3", "third_floor_condenser_runtime_today"),
]


def pair_by_day(runtime: dict[str, float], temps: dict[str, float]) -> list[tuple[float, float]]:
    """Inner-join {day: value} maps into [(outdoor_high, runtime_hours), ...], sorted by day."""
    days = sorted(set(runtime) & set(temps))
    return [(temps[d], runtime[d]) for d in days]


def analyze(pairs: list[tuple[float, float]]) -> dict:
    """Pearson r + OLS slope of runtime (y) vs outdoor high (x), with a verdict."""
    n = len(pairs)
    result = {
        "n": n, "r": None, "slope": None, "intercept": None,
        "temp_min": None, "temp_max": None, "verdict": "INSUFFICIENT",
    }
    if n < 3:
        return result
    xs = [x for x, _ in pairs]
    ys = [y for _, y in pairs]
    result["temp_min"], result["temp_max"] = min(xs), max(xs)
    # zero variance on either axis => correlation undefined
    if statistics.pstdev(xs) == 0 or statistics.pstdev(ys) == 0:
        result["verdict"] = "INSUFFICIENT"
        return result
    r = statistics.correlation(xs, ys)
    slope, intercept = statistics.linear_regression(xs, ys)
    result["r"] = r
    result["slope"] = slope
    result["intercept"] = intercept
    result["verdict"] = "OK" if r >= FLAG_THRESHOLD else "FLAG"
    return result

mkdir -p tests/analysis
: > tests/analysis/__init__.py

git add scripts/analysis/hvac_runtime_correlation.py tests/analysis/__init__.py tests/analysis/test_hvac_correlation.py
git commit -m "test(energy): HVAC runtime-vs-temp analysis core + tests (#29)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"

import json
import subprocess
import sys

# Daily-reduced InfluxQL. $timeFilter is replaced by the shell wrapper with an explicit
# range (the script runs curl directly, not through Grafana, so no macro is available).
_TEMP_Q = (
    "SELECT max(\"temperature\") FROM \"state\" "
    "WHERE \"entity_id\"='forecast_home' AND time > now()-{days}d "
    "GROUP BY time(1d) fill(none)"
)
_RUNTIME_Q = (
    "SELECT max(\"value\") FROM \"h\" "
    "WHERE \"entity_id\"='{eid}' AND time > now()-{days}d "
    "GROUP BY time(1d) fill(none)"
)


def _influx_query(q: str) -> dict:
    """Run one InfluxQL query on the HAOS box via SSH; creds read from .storage there.

    Returns the parsed InfluxDB HTTP JSON. Raises on SSH/curl failure.
    """
    remote = (
        'PW=$(jq -r \'.data.entries[] | select(.domain=="influxdb") | .data.password\' '
        '/config/.storage/core.config_entries); '
        'curl -s -m 15 -G "http://a0d7b954-influxdb:8086/query" '
        '-u "homeassistant:$PW" --data-urlencode "db=homeassistant" '
        '--data-urlencode "q=' + q.replace('"', '\\"') + '"'
    )
    proc = subprocess.run(
        ["ssh", "-o", "ConnectTimeout=20", "haos", remote],
        capture_output=True, text=True, timeout=90,
    )
    if proc.returncode != 0:
        raise RuntimeError(f"ssh/curl failed: {proc.stderr.strip()}")
    return json.loads(proc.stdout)


def _series_to_daymap(payload: dict) -> dict[str, float]:
    """Turn an InfluxDB single-series result into {YYYY-MM-DD: value}, skipping nulls."""
    out: dict[str, float] = {}
    results = payload.get("results", [{}])
    series = (results[0] or {}).get("series")
    if not series:
        return out
    for ts, val in series[0].get("values", []):
        if val is None:
            continue
        out[ts[:10]] = float(val)
    return out


def fetch(days: int = 30) -> dict[str, dict[str, float]]:
    """Fetch {'temps': daymap, '<eid>': daymap, ...} for the last `days` days."""
    data = {"temps": _series_to_daymap(_influx_query(_TEMP_Q.format(days=days)))}
    for _name, eid in CONDENSERS:
        data[eid] = _series_to_daymap(_influx_query(_RUNTIME_Q.format(eid=eid, days=days)))
    return data


def main(argv: list[str] | None = None) -> int:
    argv = sys.argv[1:] if argv is None else argv
    days = int(argv[0]) if argv else 30
    data = fetch(days=days)
    temps = data["temps"]
    print(f"HVAC runtime vs outdoor-high correlation — last {days} days "
          f"({len(temps)} days of outdoor data)\n")
    print(f"{'Condenser':<10} {'n':>3} {'r':>7} {'slope(h/°F)':>12} "
          f"{'temp range':>12}  verdict")
    print("-" * 62)
    for name, eid in CONDENSERS:
        pairs = pair_by_day(data[eid], temps)
        a = analyze(pairs)
        r = "  n/a " if a["r"] is None else f"{a['r']:+.3f}"
        slope = "     n/a" if a["slope"] is None else f"{a['slope']:+.4f}"
        rng = ("  n/a" if a["temp_min"] is None
               else f"{a['temp_min']:.0f}-{a['temp_max']:.0f}°F")
        print(f"{name:<10} {a['n']:>3} {r:>7} {slope:>12} {rng:>12}  {a['verdict']}")
    return 0


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

##### Runtime-vs-outdoor-high correlation — findings (2026-07-07, <N>-day window)

Computed by `scripts/analysis/hvac_runtime_correlation.py` (Pearson r + OLS slope of daily
condenser runtime vs daily outdoor high). Flag threshold r < 0.4.

| Condenser | n | r | slope (h/°F) | temp range | verdict |
|---|---|---|---|---|---|
| Floor 1 | <n> | <r> | <slope> | <range> | <OK/FLAG> |
| Floor 2 | <n> | <r> | <slope> | <range> | <OK/FLAG> |
| Floor 3 | <n> | <r> | <slope> | <range> | <OK/FLAG> |

Interpretation: <one or two sentences — e.g. "all three track outdoor temperature strongly
(r > 0.7), consistent with healthy thermostat-driven cycling; no anomalies." OR call out any
FLAG and the suspected cause.>

git add scripts/analysis/hvac_runtime_correlation.py projects/house_energy_strategy.md
git commit -m "feat(energy): HVAC runtime-vs-temp diagnostic + WS2 findings (#29)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"

def test_grafana_hvac_runtime_panels_present():
    """WS2 (#29): the runtime-vs-temp timeseries + 3 runtime stat cards exist,
    are well-formed, and query the correct InfluxDB series."""
    d = _grafana()
    panels = {p.get("title", ""): p for p in d.get("panels", [])}

    # dual-axis timeseries: 4 targets (3 runtime + 1 outdoor high)
    ts = panels.get("HVAC Runtime vs Outdoor High")
    assert ts is not None, "missing 'HVAC Runtime vs Outdoor High' panel"
    assert ts["type"] == "timeseries"
    queries = [t.get("query", "") for t in ts.get("targets", [])]
    assert len(queries) == 4, f"expected 4 targets, got {len(queries)}"
    assert any("forecast_home" in q for q in queries), "no outdoor-high target"
    assert sum("condenser_runtime_today" in q for q in queries) == 3
    for q in queries:
        assert "$timeFilter" in q, f"target must use $timeFilter: {q}"

    # 3 runtime stat cards
    for floor, eid in (
        ("Floor 1", "first_floor_condenser_runtime_today"),
        ("Floor 2", "second_floor_condenser_runtime_today"),
        ("Floor 3", "third_floor_condenser_runtime_today"),
    ):
        card = panels.get(f"Runtime Today — {floor}")
        assert card is not None, f"missing 'Runtime Today — {floor}' stat card"
        assert card["type"] == "stat"
        q = card["targets"][0].get("query", "")
        assert eid in q and 'FROM "h"' in q

# /tmp/add_hvac_panels.py
import json

PATH = "grafana/dashboard.json"
DS = {"type": "influxdb", "uid": "P14502BC03F6B8823"}
d = json.load(open(PATH))

def runtime_q(eid):
    return (f'SELECT max("value") FROM "h" WHERE "entity_id" = \'{eid}\' '
            f'AND $timeFilter GROUP BY time(1d) fill(none)')

TEMP_Q = ('SELECT max("temperature") FROM "state" WHERE "entity_id" = \'forecast_home\' '
          'AND $timeFilter GROUP BY time(1d) fill(none)')

RUNTIME = [
    ("Floor 1", "first_floor_condenser_runtime_today"),
    ("Floor 2", "second_floor_condenser_runtime_today"),
    ("Floor 3", "third_floor_condenser_runtime_today"),
]

def target(refid, query, alias):
    return {"datasource": DS, "refId": refid, "rawQuery": True, "query": query,
            "resultFormat": "time_series", "alias": alias}

# --- dual-axis timeseries (id 211) ---
ts_targets = [target(chr(65 + i), runtime_q(eid), name)
              for i, (name, eid) in enumerate(RUNTIME)]
ts_targets.append(target("D", TEMP_Q, "Outdoor High"))

timeseries = {
    "id": 211,
    "type": "timeseries",
    "title": "HVAC Runtime vs Outdoor High",
    "datasource": DS,
    "gridPos": {"x": 0, "y": 75, "w": 24, "h": 8},
    "targets": ts_targets,
    "fieldConfig": {
        "defaults": {
            "custom": {"drawStyle": "bars", "fillOpacity": 60,
                       "axisPlacement": "left", "lineWidth": 1},
            "unit": "h",
        },
        "overrides": [{
            "matcher": {"id": "byName", "options": "Outdoor High"},
            "properties": [
                {"id": "custom.drawStyle", "value": "line"},
                {"id": "custom.axisPlacement", "value": "right"},
                {"id": "custom.lineWidth", "value": 2},
                {"id": "custom.fillOpacity", "value": 0},
                {"id": "unit", "value": "fahrenheit"},
                {"id": "color", "value": {"mode": "fixed", "fixedColor": "orange"}},
            ],
        }],
    },
    "options": {"legend": {"displayMode": "list", "placement": "bottom"},
                "tooltip": {"mode": "multi"}},
}

# --- 3 runtime stat cards (ids 212-214), 8-wide across the next row ---
def stat_card(pid, floor, eid, x):
    return {
        "id": pid, "type": "stat", "title": f"Runtime Today — {floor}",
        "datasource": DS,
        "gridPos": {"x": x, "y": 83, "w": 8, "h": 4},
        "targets": [target("A", (f'SELECT last("value") FROM "h" '
                                  f'WHERE "entity_id" = \'{eid}\''), floor)],
        "fieldConfig": {"defaults": {"unit": "h", "color": {"mode": "thresholds"},
                                     "thresholds": {"mode": "absolute",
                                                    "steps": [{"color": "green", "value": None}]}}},
        "options": {"graphMode": "area", "colorMode": "none",
                    "reduceOptions": {"calcs": ["lastNotNull"]}, "textMode": "value_and_name"},
    }

cards = [stat_card(212 + i, name, eid, i * 8)
         for i, (name, eid) in enumerate(RUNTIME)]

d["panels"].extend([timeseries, *cards])
json.dump(d, open(PATH, "w"), indent=2)
print("appended", len([timeseries, *cards]), "panels; total now", len(d["panels"]))

git add grafana/dashboard.json tests/test_static.py
git commit -m "feat(grafana): HVAC runtime-vs-temp panel + per-floor runtime stat cards (#29)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"

### HVAC Runtime vs Outdoor High
Dual-axis timeseries. Left axis: daily AC condenser runtime (hours, bars) for all three floors,
`max("value") FROM "h"` per `*_condenser_runtime_today`, `GROUP BY time(1d)`. Right axis: daily
outdoor high (°F, line), `max("temperature") FROM "state"` where `entity_id='forecast_home'`.
Shows at a glance whether runtime tracks temperature; the quantitative correlation (Pearson r) is
produced by `scripts/analysis/hvac_runtime_correlation.py` and recorded in
`projects/house_energy_strategy.md` (WS2).

### Runtime Today — Floor 1 / 2 / 3
Three stat cards, `last("value") FROM "h"` per `*_condenser_runtime_today`, with a daily-history
area sparkline. At-a-glance per-floor cooling runtime for today.

git add docs/grafana.md projects/house_energy_strategy.md
git commit -m "docs(grafana): document HVAC runtime-vs-temp panels; close WS2 dashboard task (#29)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>"

git push
gh run list --workflow=ci.yml --limit 1

gh issue close 29 --comment "WS2 summer slice done: runtime-vs-temp diagnostic (per-condenser Pearson r + slope, findings in house_energy_strategy.md) and Grafana HVAC row panels (runtime-vs-outdoor-high timeseries + 3 runtime stat cards). Winter items (heat-pump crossover, heat/cool conflict, furnace-blower draw) remain out of scope until heating season."