ohmu/dev/nutikodu-web/app.py

"""Nutikodu andmete veebiliides (FastAPI).

Loeb ClickHouse'i `uus67` skeemist (zigbee andurid) ja serveerib:
  GET /                      -> veebirakendus (static/index.html)
  GET /api/meta              -> seadmed, tz, andmete ulatus
  GET /api/temperatuur       -> aegrida (temp + niiskus), agg=hour|day
  GET /api/ukseandur         -> ukseolekute sündmused (tulevikuks)
  GET /api/vead              -> häirete sündmused (tulevikuks)
  GET /api/latest            -> viimased väärtused + lahtised häired
  GET /health

ClickHouse'i päringud kasutavad `zigbee` kasutajat (SELECT uus67.*) ja
`param_*` parameetreid (SQL-süst välistatud). Tulemused loetakse
`FORMAT JSONEachRow` kujul (üks JSON-objekt rea kohta).
Valikuline kaitse: WEB_ACCESS_TOKEN keskkonnamuutuja (tühi = kaitseta).
"""

from __future__ import annotations

import asyncio
import hmac
import html
import secrets
import json
import os
import re
import threading
import time
from collections import defaultdict
from datetime import datetime, timedelta
from pathlib import Path
from zoneinfo import ZoneInfo

import paho.mqtt.client as mqtt
import requests
from fastapi import Body, Depends, FastAPI, Form, HTTPException, Query, Request
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from starlette.middleware.sessions import SessionMiddleware

ROOT = Path(__file__).parent
STATIC_DIR = ROOT / "static"

TZ = ZoneInfo("Europe/Tallinn")

CH_URL = os.environ.get("CH_URL", "http://clickhouse:8123").rstrip("/")
CH_USER = os.environ.get("CH_USER", "zigbee")
CH_PASS = os.environ.get("CH_PASS", "")
CH_DB = os.environ.get("CH_DB", "uus67")

# Värvisensori juhtimine (MQTT, edge broker). OFF peatab ESP32 publitseerimise
# -> ClickHouse'i ei kirjutata.
VARV_MQTT_HOST = os.environ.get("VARV_MQTT_HOST", "")
VARV_MQTT_PORT = int(os.environ.get("VARV_MQTT_PORT", "8883"))
VARV_MQTT_TLS = os.environ.get("VARV_MQTT_TLS", "true").strip().lower() in ("1", "true", "yes", "on")
VARV_MQTT_USER = os.environ.get("VARV_MQTT_USER", "")
VARV_MQTT_PASS = os.environ.get("VARV_MQTT_PASS", "")
VARV_CMD_TOPIC = os.environ.get("VARV_CMD_TOPIC", "varvus/switch/sensor/command")
VARV_STATE_TOPIC = os.environ.get("VARV_STATE_TOPIC", "varvus/switch/sensor/state")

WEB_ACCESS_TOKEN = os.environ.get("WEB_ACCESS_TOKEN", "")
WEB_PASSWORD = os.environ.get("WEB_PASSWORD", "")
WEB_SECRET_KEY = os.environ.get("WEB_SECRET_KEY", "") or secrets.token_hex(32)
HA_URL = os.environ.get("HA_URL", "http://100.87.1.24:8123").rstrip("/")
HA_TOKEN = os.environ.get("HA_TOKEN", "")
AWAY_ENTITY = os.environ.get("AWAY_ENTITY", "input_boolean.kodust_eemal")
SESSION_MAX_AGE = int(os.environ.get("SESSION_MAX_AGE", "604800"))
LOGIN_MAX_ATTEMPTS = int(os.environ.get("LOGIN_MAX_ATTEMPTS", "5"))
LOGIN_WINDOW_SECONDS = int(os.environ.get("LOGIN_WINDOW_SECONDS", "300"))

DEFAULT_RANGE_DAYS = 7
MAX_RANGE_DAYS = 3660

# --- Transport (bussid + praamid) ---
BUS_GRAPHQL = os.environ.get(
    "BUS_GRAPHQL", "https://api.peatus.ee/routing/v1/routers/estonia/index/graphql"
).rstrip("/")
BUS_STOPS = [s.strip() for s in os.environ.get(
    "BUS_STOPS", "estonia:24703,estonia:27213").split(",") if s.strip()]
PRAAM_API = os.environ.get(
    "PRAAM_API", "https://www.praamid.ee/online/events").rstrip("/")
RII_HOIATUS_URL = os.environ.get(
    "RII_HOIATUS_URL", "https://www.ilmateenistus.ee/ilm/prognoosid/hoiatused/"
)
TRANSPORT_TTL = int(os.environ.get("TRANSPORT_TTL", "60"))
UA = "Mozilla/5.0 (compatible; Nutikodu-web/1.0)"

PRAAM_DIRECTIONS = {"HR": "Heltermaa → Rohuküla", "RH": "Rohuküla → Heltermaa"}
SHIP_NAMES = {"TI": "Tiiu", "LE": "Leiger", "RE": "Regu"}
PRAAM_STATUS = {"CONFIRMED": "plaanipärane", "CANCELLED": "tühistatud", "SUSPENDED": "seisatud"}
SEA_AREAS = ["Väinameri", "Hiiu maakond", "Lääne maakond"]
WIND_TYPES = ["tugev tuul", "väga tugev tuul", "torm", "merehoiatus", "tugev tuul ja lained"]

_cache: dict[str, tuple[float, object]] = {}
_cache_lock = threading.Lock()


def cached(key: str, ttl: int, producer):
    """Lihtne mälusisene TTL-vahemälu väliste API-de jaoks."""
    now = time.monotonic()
    with _cache_lock:
        hit = _cache.get(key)
        if hit and now - hit[0] < ttl:
            return hit[1]
    value = producer()
    with _cache_lock:
        _cache[key] = (time.monotonic(), value)
    return value


app = FastAPI(title="Nutikodu", version="0.2.0")
app.add_middleware(
    SessionMiddleware,
    secret_key=WEB_SECRET_KEY,
    https_only=False,
    same_site="lax",
    max_age=SESSION_MAX_AGE,
)


@app.middleware("http")
async def add_no_cache(request: Request, call_next):
    """Staatika ja API ilma vahemäluta — muidu jääb brauserisse vana JS/HTML."""
    response = await call_next(request)
    path = request.url.path
    if (
        path == "/"
        or path.startswith("/api/")
        or path.endswith((".js", ".css", ".html", ".webmanifest", ".svg"))
    ):
        response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
        response.headers["Pragma"] = "no-cache"
    return response


def ch_query(query: str, params: dict[str, str] | None = None) -> list[dict]:
    """Käivita ClickHouse'i päring ja tagasta read JSON-objektidena."""
    q = {"database": CH_DB, "query": f"{query} FORMAT JSONEachRow"}
    for key, value in (params or {}).items():
        q[f"param_{key}"] = value
    try:
        resp = requests.post(
            f"{CH_URL}/",
            params=q,
            auth=(CH_USER, CH_PASS),
            timeout=20,
        )
    except requests.RequestException as exc:
        raise HTTPException(status_code=502, detail=f"ClickHouse'i ühendus: {exc}") from exc
    if resp.status_code != 200:
        raise HTTPException(status_code=502, detail=f"ClickHouse: {resp.text.strip()[:400]}")
    text = resp.text.strip()
    if not text:
        return []
    try:
        return [json.loads(line) for line in text.splitlines() if line.strip()]
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=502, detail=f"ClickHouse'i vastus: {text[:200]}") from exc


_login_failures: dict[str, list[float]] = defaultdict(list)
_login_lock = threading.Lock()


def client_ip(request: Request) -> str:
    forwarded = request.headers.get("x-forwarded-for")
    if forwarded:
        return forwarded.split(",")[0].strip()
    return request.client.host if request.client else "?"


def retry_after_seconds(ip: str) -> int:
    now = time.monotonic()
    with _login_lock:
        history = [t for t in _login_failures.get(ip, []) if now - t < LOGIN_WINDOW_SECONDS]
        _login_failures[ip] = history
        if len(history) >= LOGIN_MAX_ATTEMPTS:
            return int(LOGIN_WINDOW_SECONDS - (now - history[0])) + 1
    return 0


def register_login_failure(ip: str) -> None:
    with _login_lock:
        _login_failures[ip].append(time.monotonic())


def reset_login_failures(ip: str) -> None:
    with _login_lock:
        _login_failures.pop(ip, None)


def is_authed(request: Request) -> bool:
    """Kas päring on autenditud: Bearer token VÕI sessiooniküpsis."""
    auth = request.headers.get("authorization", "")
    if auth.lower().startswith("bearer "):
        token = auth[7:].strip()
        if WEB_ACCESS_TOKEN and secrets.compare_digest(token, WEB_ACCESS_TOKEN):
            return True
    try:
        if request.session.get("auth") is True:
            return True
    except Exception:
        pass
    return False


def require_auth(request: Request) -> None:
    if not is_authed(request):
        raise HTTPException(
            status_code=401,
            detail="Autentimine nõutud (Bearer token või sisselogimine)",
            headers={"WWW-Authenticate": "Bearer"},
        )


def parse_dt(value: str | None, default: datetime) -> str:
    """Võta vastu 'YYYY-MM-DD[ HH:MM[:SS]]' või ISO 'T'-eraldajaga."""
    if not value:
        return default.strftime("%Y-%m-%d %H:%M:%S")
    cleaned = value.strip().replace("T", " ")
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            return datetime.strptime(cleaned, fmt).strftime("%Y-%m-%d %H:%M:%S")
        except ValueError:
            continue
    raise HTTPException(status_code=422, detail=f"Vigane kuupäev: {value}")


def resolve_range(from_: str | None, to: str | None) -> tuple[str, str]:
    now = datetime.now(TZ)
    to_s = parse_dt(to, now)
    to_dt = datetime.strptime(to_s, "%Y-%m-%d %H:%M:%S")
    from_s = parse_dt(from_, to_dt - timedelta(days=DEFAULT_RANGE_DAYS))
    from_dt = datetime.strptime(from_s, "%Y-%m-%d %H:%M:%S")
    if from_dt > to_dt:
        raise HTTPException(status_code=422, detail="'from' peab olema enne 'to'")
    if (to_dt - from_dt).days > MAX_RANGE_DAYS:
        raise HTTPException(status_code=422, detail="Vahemik on liiga pikk")
    return from_s, to_s


@app.get("/api/meta")
async def meta() -> dict:
    devices = [row["device"] for row in ch_query(
        f"SELECT DISTINCT device FROM {CH_DB}.temperatuur ORDER BY device"
    )]
    bounds = ch_query(
        f"SELECT toString(min(hour)) AS min_hour, toString(max(hour)) AS max_hour, count() AS rows "
        f"FROM {CH_DB}.temperatuur"
    )
    first = bounds[0] if bounds else {"min_hour": None, "max_hour": None, "rows": 0}
    return {
        "tz": "Europe/Tallinn",
        "db": CH_DB,
        "devices": devices,
        "min": first["min_hour"],
        "max": first["max_hour"],
        "rows": first["rows"],
        "default_range_days": DEFAULT_RANGE_DAYS,
    }


@app.get("/api/temperatuur")
async def temperatuur(
    from_: str | None = Query(None, alias="from"),
    to: str | None = Query(None),
    device: str | None = Query(None),
    agg: str = Query("hour", pattern="^(hour|day)$"),
) -> dict:
    from_s, to_s = resolve_range(from_, to)
    params = {"from": from_s, "to": to_s}
    where = (
        f"hour BETWEEN toDateTime({{from:String}}, 'Europe/Tallinn') "
        f"AND toDateTime({{to:String}}, 'Europe/Tallinn')"
    )
    if device:
        where += " AND device = {device:String}"
        params["device"] = device
    if agg == "day":
        select = (
            "toUInt64(toUnixTimestamp(toStartOfDay(hour))) * 1000 AS t, device, "
            "round(avg(temperature), 2) AS temperature, round(avg(humidity), 2) AS humidity, "
            "max(taidetud) AS taidetud"
        )
        group = "GROUP BY t, device"
    else:
        select = (
            "toUInt64(toUnixTimestamp(hour)) * 1000 AS t, device, "
            "temperature, humidity, taidetud"
        )
        group = ""
    rows = ch_query(
        f"SELECT {select} FROM {CH_DB}.temperatuur WHERE {where} {group} ORDER BY t",
        params,
    )

    series: dict[str, dict] = {}
    for row in rows:
        s = series.setdefault(
            row["device"], {"t": [], "temperature": [], "humidity": [], "taidetud": []}
        )
        s["t"].append(int(row["t"]))
        s["temperature"].append(row["temperature"])
        s["humidity"].append(row["humidity"])
        s["taidetud"].append(bool(row.get("taidetud")))
    return {"from": from_s, "to": to_s, "agg": agg, "series": series}


@app.get("/api/valisilm")
async def valisilm(
    from_: str | None = Query(None, alias="from"),
    to: str | None = Query(None),
    agg: str = Query("hour", pattern="^(hour|day)$"),
) -> dict:
    from_s, to_s = resolve_range(from_, to)
    params = {"from": from_s, "to": to_s}
    where = (
        f"hour BETWEEN toDateTime({{from:String}}, 'Europe/Tallinn') "
        f"AND toDateTime({{to:String}}, 'Europe/Tallinn')"
    )
    if agg == "day":
        select = (
            "toUInt64(toUnixTimestamp(toStartOfDay(hour))) * 1000 AS t, "
            "round(avg(temperature), 2) AS temperature, round(avg(humidity), 2) AS humidity, "
            "round(avg(wind), 2) AS wind, round(sum(rain), 2) AS rain, max(taidetud) AS taidetud"
        )
        group = "GROUP BY t"
    else:
        select = (
            "toUInt64(toUnixTimestamp(hour)) * 1000 AS t, "
            "temperature, humidity, wind, rain, taidetud"
        )
        group = ""
    rows = ch_query(
        f"SELECT {select} FROM {CH_DB}.valisilm FINAL WHERE {where} {group} ORDER BY t",
        params,
    )
    series: dict[str, list] = {
        "t": [],
        "temperature": [],
        "humidity": [],
        "wind": [],
        "rain": [],
        "taidetud": [],
    }
    for row in rows:
        series["t"].append(int(row["t"]))
        series["temperature"].append(row["temperature"])
        series["humidity"].append(row["humidity"])
        series["wind"].append(row["wind"])
        series["rain"].append(row["rain"])
        series["taidetud"].append(bool(row.get("taidetud")))
    return {"from": from_s, "to": to_s, "agg": agg, "series": series}


@app.get("/api/ukseandur", dependencies=[Depends(require_auth)])
async def ukseandur(
    from_: str | None = Query(None, alias="from"),
    to: str | None = Query(None),
    device: str | None = Query(None),
) -> dict:
    from_s, to_s = resolve_range(from_, to)
    params = {"from": from_s, "to": to_s}
    where = (
        f"ts BETWEEN toDateTime({{from:String}}, 'Europe/Tallinn') "
        f"AND toDateTime({{to:String}}, 'Europe/Tallinn')"
    )
    if device:
        where += " AND device = {device:String}"
        params["device"] = device
    rows = ch_query(
        f"SELECT toUInt64(toUnixTimestamp(ts)) * 1000 AS t, device, contact, olek "
        f"FROM {CH_DB}.ukseandur WHERE {where} ORDER BY ts",
        params,
    )
    return {
        "from": from_s,
        "to": to_s,
        "events": [
            {"t": int(r["t"]), "device": r["device"], "contact": bool(r["contact"]), "olek": r["olek"]}
            for r in rows
        ],
    }


@app.get("/api/vead", dependencies=[Depends(require_auth)])
async def vead(
    from_: str | None = Query(None, alias="from"),
    to: str | None = Query(None),
    device: str | None = Query(None),
) -> dict:
    from_s, to_s = resolve_range(from_, to)
    params = {"from": from_s, "to": to_s}
    where = (
        f"ts BETWEEN toDateTime({{from:String}}, 'Europe/Tallinn') "
        f"AND toDateTime({{to:String}}, 'Europe/Tallinn')"
    )
    if device:
        where += " AND device = {device:String}"
        params["device"] = device
    rows = ch_query(
        f"SELECT toUInt64(toUnixTimestamp(ts)) * 1000 AS t, device, tyyp, olek, detail, vaartus "
        f"FROM {CH_DB}.vead WHERE {where} ORDER BY ts",
        params,
    )
    return {
        "from": from_s,
        "to": to_s,
        "events": [
            {
                "t": int(r["t"]),
                "device": r["device"],
                "tyyp": r["tyyp"],
                "olek": r["olek"],
                "detail": r["detail"],
                "vaartus": r["vaartus"],
            }
            for r in rows
        ],
    }




@app.get("/api/liikumine", dependencies=[Depends(require_auth)])
async def liikumine(
    from_: str | None = Query(None, alias="from"),
    to: str | None = Query(None),
    device: str | None = Query(None),
) -> dict:
    from_s, to_s = resolve_range(from_, to)
    params = {"from": from_s, "to": to_s}
    where = (
        f"ts BETWEEN toDateTime({{from:String}}, 'Europe/Tallinn') "
        f"AND toDateTime({{to:String}}, 'Europe/Tallinn')"
    )
    if device:
        where += " AND device = {device:String}"
        params["device"] = device
    rows = ch_query(
        f"SELECT toUInt64(toUnixTimestamp(ts)) * 1000 AS t, device, occupancy, illumination, olek "
        f"FROM {CH_DB}.liikumine WHERE {where} ORDER BY ts",
        params,
    )
    return {
        "from": from_s,
        "to": to_s,
        "events": [
            {
                "t": int(r["t"]),
                "device": r["device"],
                "occupancy": bool(r["occupancy"]),
                "illumination": r["illumination"],
                "olek": r["olek"],
            }
            for r in rows
        ],
    }


@app.get("/api/energia", dependencies=[Depends(require_auth)])
async def energia(
    from_: str | None = Query(None, alias="from"),
    to: str | None = Query(None),
    device: str | None = Query(None),
    agg: str = Query("hour", pattern="^(hour|day)$"),
) -> dict:
    from_s, to_s = resolve_range(from_, to)
    params = {"from": from_s, "to": to_s}
    where = (
        f"ts BETWEEN toDateTime({{from:String}}, 'Europe/Tallinn') "
        f"AND toDateTime({{to:String}}, 'Europe/Tallinn')"
    )
    if device:
        where += " AND device = {device:String}"
        params["device"] = device
    if agg == "day":
        select = (
            "toUInt64(toUnixTimestamp(toStartOfDay(ts))) * 1000 AS t, device, "
            "round(avg(power), 2) AS power, round(avg(voltage), 1) AS voltage, "
            "round(avg(current), 3) AS current, max(energy) AS energy"
        )
        group = "GROUP BY t, device"
    else:
        select = (
            "toUInt64(toUnixTimestamp(ts)) * 1000 AS t, device, "
            "power, voltage, current, energy"
        )
        group = ""
    rows = ch_query(
        f"SELECT {select} FROM {CH_DB}.energia WHERE {where} {group} ORDER BY t",
        params,
    )
    series: dict[str, dict] = {}
    for row in rows:
        s2 = series.setdefault(
            row["device"], {"t": [], "power": [], "voltage": [], "current": [], "energy": []}
        )
        s2["t"].append(int(row["t"]))
        s2["power"].append(row["power"])
        s2["voltage"].append(row["voltage"])
        s2["current"].append(row["current"])
        s2["energy"].append(row["energy"])
    return {"from": from_s, "to": to_s, "agg": agg, "series": series}




@app.get("/api/uus67_latest", dependencies=[Depends(require_auth)])
async def uus67_latest() -> dict:
    energy = ch_query(
        f"SELECT device, argMax(power, ts) AS power, argMax(voltage, ts) AS voltage, "
        f"argMax(current, ts) AS current, argMax(energy, ts) AS energy, "
        f"toUInt64(toUnixTimestamp(max(ts))) * 1000 AS t "
        f"FROM {CH_DB}.energia GROUP BY device ORDER BY device"
    )
    motion = ch_query(
        f"SELECT device, argMax(occupancy, ts) AS occupancy, "
        f"argMax(illumination, ts) AS illumination, "
        f"toUInt64(toUnixTimestamp(max(ts))) * 1000 AS t "
        f"FROM {CH_DB}.liikumine GROUP BY device ORDER BY device"
    )
    return {
        "energy": [
            {
                "device": r["device"],
                "power": r.get("power"),
                "voltage": r.get("voltage"),
                "current": r.get("current"),
                "energy": r.get("energy"),
                "t": int(r["t"]) if r.get("t") is not None else None,
            }
            for r in energy
        ],
        "motion": [
            {
                "device": r["device"],
                "occupancy": bool(r.get("occupancy")),
                "illumination": r.get("illumination"),
                "t": int(r["t"]) if r.get("t") is not None else None,
            }
            for r in motion
        ],
    }


# --- Värvisensor (uus67.varvus, AS7341) ---

VARV_BANDS = ("f1", "f2", "f3", "f4", "f5", "f6", "f7", "f8")


@app.get("/api/varv/latest")
async def varv_latest(limit: int = Query(60, ge=1, le=300)) -> dict:
    """Värvisensori viimased mõõtmised (avalik)."""
    cols = ", ".join(VARV_BANDS)
    rows = ch_query(
        f"SELECT toString(ts) AS ts, toUInt64(toUnixTimestamp(ts)) * 1000 AS t, "
        f"rgb_hex, r, g, b, round(x_chrom, 4) AS x_chrom, round(y_chrom, 4) AS y_chrom, "
        f"{cols}, clear, nir, gain "
        f"FROM {CH_DB}.varvus ORDER BY ts DESC LIMIT {int(limit)}"
    )
    return {
        "rows": [
            {
                "ts": r["ts"],
                "t": int(r["t"]),
                "rgb_hex": r.get("rgb_hex") or "#000000",
                "r": int(r.get("r") or 0),
                "g": int(r.get("g") or 0),
                "b": int(r.get("b") or 0),
                "x_chrom": r.get("x_chrom"),
                "y_chrom": r.get("y_chrom"),
                "bands": {b: r.get(b) for b in VARV_BANDS},
                "clear": r.get("clear"),
                "nir": r.get("nir"),
                "gain": r.get("gain"),
            }
            for r in rows
        ]
    }


@app.get("/api/varv/history")
async def varv_history(hours: int = Query(24, ge=1, le=168)) -> dict:
    """Värvitelg: tunnipõhine keskmine värv viimaste tundide jooksul (avalik)."""
    rows = ch_query(
        f"SELECT toString(toStartOfHour(ts)) AS bucket, "
        f"toUInt64(toUnixTimestamp(toStartOfHour(ts))) * 1000 AS t, "
        f"toUInt8(round(avg(r))) AS r, toUInt8(round(avg(g))) AS g, toUInt8(round(avg(b))) AS b, "
        f"round(avg(x_chrom), 4) AS x_chrom, round(avg(y_chrom), 4) AS y_chrom, count() AS n "
        f"FROM {CH_DB}.varvus WHERE ts >= now() - INTERVAL {int(hours)} HOUR "
        f"GROUP BY toStartOfHour(ts) ORDER BY t"
    )
    out = []
    for r in rows:
        rv, gv, bv = int(r.get("r") or 0), int(r.get("g") or 0), int(r.get("b") or 0)
        out.append({
            "ts": r["bucket"],
            "t": int(r["t"]),
            "r": rv, "g": gv, "b": bv,
            "rgb_hex": f"#{rv:02X}{gv:02X}{bv:02X}",
            "x_chrom": r.get("x_chrom"),
            "y_chrom": r.get("y_chrom"),
            "n": int(r.get("n") or 0),
        })
    return {"hours": hours, "rows": out}


# --- Värvisensori mõõtmise sisse/välja lüliti (MQTT, ainult sisselogitult) ---


def _varv_mqtt_client() -> mqtt.Client:
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    if VARV_MQTT_TLS:
        client.tls_set()
    if VARV_MQTT_USER:
        client.username_pw_set(VARV_MQTT_USER, VARV_MQTT_PASS)
    return client


def varv_read_state(timeout: float = 4.0) -> str | None:
    """Loe lüliti retained olek edge-brokerist (lühiajaline ühendus)."""
    if not VARV_MQTT_HOST:
        return None
    box: dict[str, str] = {}

    def on_connect(c, userdata, flags, rc, props=None):
        if rc == 0:
            c.subscribe(VARV_STATE_TOPIC)

    def on_message(c, userdata, msg):
        box["state"] = msg.payload.decode().strip()

    client = _varv_mqtt_client()
    client.on_connect = on_connect
    client.on_message = on_message
    try:
        client.connect_async(VARV_MQTT_HOST, VARV_MQTT_PORT, 10)
        client.loop_start()
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline and "state" not in box:
            time.sleep(0.05)
    finally:
        client.loop_stop()
        try:
            client.disconnect()
        except Exception:
            pass
    state = box.get("state")
    return state if state in ("ON", "OFF") else None


def varv_set_power(on: bool) -> None:
    if not VARV_MQTT_HOST:
        raise HTTPException(status_code=503, detail="Värvisensori MQTT pole seadistatud")
    client = _varv_mqtt_client()
    try:
        client.connect(VARV_MQTT_HOST, VARV_MQTT_PORT, 10)
        client.loop_start()
        info = client.publish(VARV_CMD_TOPIC, "ON" if on else "OFF", qos=1, retain=True)
        info.wait_for_publish(timeout=5)
    finally:
        client.loop_stop()
        try:
            client.disconnect()
        except Exception:
            pass


@app.get("/api/varv/power", dependencies=[Depends(require_auth)])
def varv_power_get() -> dict:
    """Värvisensori mõõtmise olek (ainult sisselogitult)."""
    state = varv_read_state()
    return {
        "available": bool(VARV_MQTT_HOST),
        "state": state,
        "on": None if state is None else (state == "ON"),
    }


@app.post("/api/varv/power", dependencies=[Depends(require_auth)])
async def varv_power_set(payload: dict = Body(...)) -> dict:
    """Lülita värvisensori mõõtmine sisse/välja (ainult sisselogitult).

    OFF peatab ESP32 publitseerimise -> ClickHouse'i ei kirjutata.
    """
    on = bool(payload.get("on"))
    await asyncio.to_thread(varv_set_power, on)
    state = await asyncio.to_thread(varv_read_state, 3.0)
    return {"on": on, "state": state}


@app.get("/api/latest")
async def latest(request: Request) -> dict:
    readings = ch_query(
        f"SELECT device, "
        f"argMax(temperature, hour) AS temperature, "
        f"argMax(humidity, hour) AS humidity, "
        f"toString(max(hour)) AS ts_str, "
        f"toUInt64(toUnixTimestamp(max(hour))) * 1000 AS t "
        f"FROM {CH_DB}.temperatuur GROUP BY device ORDER BY device"
    )
    faults = ch_query(
        f"SELECT device, tyyp, argMax(detail, ts) AS detail_now, "
        f"toString(max(ts)) AS ts_str, argMax(olek, ts) AS olek_now "
        f"FROM {CH_DB}.vead GROUP BY device, tyyp HAVING olek_now = 'aktiveeritud'"
    ) if is_authed(request) else []
    outside_rows = ch_query(
        f"SELECT toString(max(hour)) AS ts_str, "
        f"argMax(temperature, hour) AS temperature, "
        f"argMax(humidity, hour) AS humidity, "
        f"argMax(wind, hour) AS wind, "
        f"argMax(rain, hour) AS rain, "
        f"toUInt64(toUnixTimestamp(max(hour))) * 1000 AS t "
        f"FROM {CH_DB}.valisilm FINAL"
    )
    outside: dict = {
        "temperature": None,
        "humidity": None,
        "wind": None,
        "rain": None,
        "hour": None,
        "t": None,
    }
    if outside_rows:
        r = outside_rows[0]
        outside = {
            "temperature": r.get("temperature"),
            "humidity": r.get("humidity"),
            "wind": r.get("wind"),
            "rain": r.get("rain"),
            "hour": r.get("ts_str"),
            "t": int(r["t"]) if r.get("t") is not None else None,
        }
    return {
        "readings": [
            {
                "device": r["device"],
                "temperature": r["temperature"],
                "humidity": r["humidity"],
                "hour": r["ts_str"],
                "t": int(r["t"]) if r.get("t") is not None else None,
            }
            for r in readings
        ],
        "open_faults": [
            {"device": r["device"], "tyyp": r["tyyp"], "detail": r["detail_now"], "ts": r["ts_str"]}
            for r in faults
        ],
        "outside": outside,
    }



# --- Transport: bussid (peatus.ee / Digitransit GraphQL) ---


def fetch_bus_departures() -> dict:
    """Tänased/tulevased bussiväljumised Suuremõisa küla peatustest."""
    fields = (
        "name stoptimesWithoutPatterns(numberOfDepartures: 40) { "
        "serviceDay scheduledDeparture realtimeDeparture realtime headsign "
        "trip { route { shortName longName } } }"
    )
    aliases = " ".join(
        f's{i}: stop(id: "{sid}") {{ {fields} }}' for i, sid in enumerate(BUS_STOPS)
    )
    try:
        resp = requests.post(
            BUS_GRAPHQL,
            json={"query": "{ " + aliases + " }"},
            headers={"User-Agent": UA},
            timeout=18,
        )
    except requests.RequestException as exc:
        raise HTTPException(status_code=502, detail=f"Bussi API ühendus: {exc}") from exc
    if resp.status_code != 200:
        raise HTTPException(status_code=502, detail=f"Bussi API: HTTP {resp.status_code}")
    try:
        payload = resp.json()
    except ValueError as exc:
        raise HTTPException(status_code=502, detail="Bussi API: vigane JSON") from exc
    if payload.get("errors"):
        msg = payload["errors"][0].get("message", "viga")
        raise HTTPException(status_code=502, detail=f"Bussi API: {msg}")

    departures = []
    for stop in (payload.get("data") or {}).values():
        if not stop or not stop.get("stoptimesWithoutPatterns"):
            continue
        stop_name = stop.get("name") or "Suuremõisa"
        for st in stop["stoptimesWithoutPatterns"]:
            service_day = st.get("serviceDay")
            if service_day is None:
                continue
            sched_s = st.get("scheduledDeparture") or 0
            real_s = st.get("realtimeDeparture")
            dep_s = real_s if st.get("realtime") and real_s is not None else sched_s
            try:
                dep_dt = datetime.fromtimestamp(int(service_day) + int(dep_s), TZ)
            except (OverflowError, OSError, TypeError, ValueError):
                continue
            route = (st.get("trip") or {}).get("route") or {}
            delay = None
            if st.get("realtime") and real_s is not None:
                delay = round((int(real_s) - int(sched_s)) / 60)
            departures.append({
                "t": int(dep_dt.timestamp() * 1000),
                "time": dep_dt.strftime("%H:%M"),
                "date": dep_dt.strftime("%Y-%m-%d"),
                "stop": stop_name,
                "route": route.get("shortName") or "",
                "headsign": st.get("headsign") or "",
                "realtime": bool(st.get("realtime")),
                "delay_min": delay,
            })
    departures.sort(key=lambda d: d["t"])
    now = datetime.now(TZ)
    return {
        "stops": BUS_STOPS,
        "today": now.strftime("%Y-%m-%d"),
        "departures": departures,
        "updated": now.strftime("%Y-%m-%d %H:%M"),
    }


@app.get("/api/buss")
async def buss(limit: int = Query(12, ge=1, le=40)) -> dict:
    """Järgmised bussiväljumised Suuremõisa külast (avalik)."""
    data = cached("buss", TRANSPORT_TTL, fetch_bus_departures)
    return {**data, "departures": data["departures"][:limit]}


# --- Transport: praamid (praamid.ee) ---


def _praam_events(day: str, direction: str) -> list[dict]:
    try:
        resp = requests.get(
            PRAAM_API,
            params={"direction": direction, "departure-date": day, "time-shift": "0"},
            headers={"User-Agent": UA},
            timeout=18,
        )
    except requests.RequestException:
        return []
    if resp.status_code != 200:
        return []
    try:
        items = (resp.json() or {}).get("items") or []
    except ValueError:
        return []
    out = []
    for e in items:
        try:
            dep_dt = datetime.strptime((e.get("dtstart") or "")[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=TZ)
        except ValueError:
            continue
        ship = e.get("ship")
        ship_code = ship.get("code") if isinstance(ship, dict) else ship
        status = e.get("status") or "?"
        out.append({
            "uid": e.get("uid"),
            "direction": direction,
            "direction_name": PRAAM_DIRECTIONS.get(direction, direction),
            "date": dep_dt.strftime("%Y-%m-%d"),
            "time": dep_dt.strftime("%H:%M"),
            "t": int(dep_dt.timestamp() * 1000),
            "ship": SHIP_NAMES.get(ship_code, ship_code or "?"),
            "ship_code": ship_code,
            "status": status,
            "status_et": PRAAM_STATUS.get(status, status),
            "disrupted": status != "CONFIRMED",
        })
    return out


def fetch_weather_warnings() -> list[dict]:
    """RII hoiatusleht -> Väinamere-liiklusega seotud aktiivsed hoiatused (parim pingutus)."""
    try:
        resp = requests.get(RII_HOIATUS_URL, headers={"User-Agent": UA}, timeout=18)
    except requests.RequestException:
        return []
    if resp.status_code != 200:
        return []
    text = html.unescape(resp.text)
    text = re.sub(r"<script.*?</script>", " ", text, flags=re.S)
    text = re.sub(r"<[^>]+>", " ", text)
    text = re.sub(r"\s+", " ", text)
    found, seen = [], set()
    for area in SEA_AREAS:
        i = 0
        while True:
            i = text.find(area, i)
            if i < 0:
                break
            seg = text[max(0, i - 100): i + 700]
            if any(w in seg.lower() for w in WIND_TYPES):
                m = re.search(r"Kehtib[^.]{0,200}", seg)
                valid = m.group(0).strip() if m else ""
                key = (area, valid or seg[:80])
                if key not in seen:
                    seen.add(key)
                    found.append({"area": area, "valid": valid, "text": seg.strip()[:300]})
            i += len(area)
    return found


def fetch_praam(days: int) -> dict:
    now = datetime.now(TZ)
    day_list = [(now + timedelta(days=n)).strftime("%Y-%m-%d") for n in range(days)]
    departures = []
    for day in day_list:
        for direction in ("HR", "RH"):
            departures.extend(_praam_events(day, direction))
    departures.sort(key=lambda d: (d["date"], d["time"]))
    return {
        "date": day_list[0],
        "today": now.strftime("%Y-%m-%d"),
        "departures": departures,
        "warnings": fetch_weather_warnings(),
        "disrupted": sum(1 for d in departures if d["disrupted"]),
        "updated": now.strftime("%Y-%m-%d %H:%M"),
    }


@app.get("/api/praam")
async def praam(days: int = Query(1, ge=1, le=3)) -> dict:
    """Heltermaa–Rohuküla praamiväljumised + ilmahoiatuse kontekst (avalik)."""
    return cached(f"praam:{days}", TRANSPORT_TTL, lambda: fetch_praam(days))


LOGIN_HTML = """<!doctype html>
<html lang="et"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
<title>Nutikodu — login</title>
<style>
  body{font-family:system-ui,sans-serif;background:#0e1116;color:#e6edf3;display:flex;min-height:100vh;align-items:center;justify-content:center;margin:0}
  form{background:#161b22;padding:28px 24px;border-radius:12px;border:1px solid #30363d;min-width:280px}
  h1{font-size:18px;margin:0 0 16px}
  input{width:100%;padding:10px;border-radius:8px;border:1px solid #30363d;background:#0d1117;color:#e6edf3;box-sizing:border-box;font-size:16px}
  button{margin-top:14px;width:100%;padding:10px;border-radius:8px;border:0;background:#1f6feb;color:#fff;font-size:16px;cursor:pointer}
  .err{color:#f85149;font-size:14px;margin-top:10px;min-height:18px}
  a{color:#58a6ff;font-size:13px;text-decoration:none}
</style></head><body>
<form method="post" action="/login">
  <h1>🔒 Nutikodu — sisselogimine</h1>
  <input type="password" name="password" placeholder="Parool" autofocus autocomplete="current-password">
  <button type="submit">Logi sisse</button>
  <div class="err">{{ERR}}</div>
  <div style="margin-top:14px"><a href="/">← Temperatuuri leht (avalik)</a></div>
</form>
</body></html>"""


@app.get("/login")
async def login_page():
    return HTMLResponse(LOGIN_HTML.replace("{{ERR}}", ""))


@app.post("/login")
async def login_submit(request: Request, password: str = Form("")):
    ip = client_ip(request)
    ra = retry_after_seconds(ip)
    if ra:
        return HTMLResponse(
            LOGIN_HTML.replace("{{ERR}}", f"Liiga palju katseid. Proovi {ra} sekundi pärast."),
            status_code=429,
        )
    if WEB_PASSWORD and secrets.compare_digest(password, WEB_PASSWORD):
        reset_login_failures(ip)
        request.session["auth"] = True
        return RedirectResponse("/?auth=1", status_code=303)
    register_login_failure(ip)
    return HTMLResponse(LOGIN_HTML.replace("{{ERR}}", "Vale parool"), status_code=401)


@app.get("/logout")
async def logout(request: Request):
    request.session.clear()
    return RedirectResponse("/", status_code=303)


@app.get("/api/me")
async def me(request: Request) -> dict:
    return {"authed": is_authed(request)}



def _ha_headers() -> dict:
    return {"Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json"}


def ha_get_away() -> bool | None:
    if not HA_TOKEN:
        return None
    try:
        r = requests.get(f"{HA_URL}/api/states/{AWAY_ENTITY}", headers=_ha_headers(), timeout=8)
        if r.status_code == 200:
            return r.json().get("state") == "on"
    except requests.RequestException:
        pass
    return None


def ha_set_away(on: bool) -> None:
    service = "turn_on" if on else "turn_off"
    r = requests.post(
        f"{HA_URL}/api/services/input_boolean/{service}",
        headers=_ha_headers(),
        json={"entity_id": AWAY_ENTITY},
        timeout=8,
    )
    r.raise_for_status()


@app.get("/api/away", dependencies=[Depends(require_auth)])
async def away_state() -> dict:
    """Kohalolu ("kodust eemal") olek."""
    state = ha_get_away()
    return {"away": state, "available": state is not None}


@app.post("/api/away", dependencies=[Depends(require_auth)])
async def away_set(payload: dict = Body(...)) -> dict:
    """Lülita "kodust eemal" sisse/välja (keha {"on": true|false})."""
    on = bool(payload.get("on"))
    try:
        ha_set_away(on)
    except requests.RequestException as exc:
        raise HTTPException(status_code=502, detail=f"Home Assistant viga: {exc}") from exc
    return {"away": on, "available": True}


@app.get("/health")
async def health() -> dict:
    try:
        ch_query("SELECT 1")
        ch_ok = True
    except HTTPException:
        ch_ok = False
    return {"status": "ok", "clickhouse": ch_ok}


@app.get("/")
async def index() -> FileResponse:
    return FileResponse(STATIC_DIR / "index.html", media_type="text/html")


app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("app:app", host="0.0.0.0", port=8095, reload=True)