"""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 json import os from datetime import datetime, timedelta from pathlib import Path from zoneinfo import ZoneInfo import requests from fastapi import Depends, FastAPI, HTTPException, Query, Request from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles 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") WEB_ACCESS_TOKEN = os.environ.get("WEB_ACCESS_TOKEN", "") DEFAULT_RANGE_DAYS = 7 MAX_RANGE_DAYS = 3660 app = FastAPI(title="Nutikodu", version="0.2.0") @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 def check_token(request: Request) -> None: if not WEB_ACCESS_TOKEN: return token = request.headers.get("x-access-token") or request.query_params.get("token") if token != WEB_ACCESS_TOKEN: raise HTTPException(status_code=403, detail="Vale või puuduv access token") 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", dependencies=[Depends(check_token)]) 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", dependencies=[Depends(check_token)]) 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", dependencies=[Depends(check_token)]) 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(check_token)]) 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(check_token)]) 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/latest", dependencies=[Depends(check_token)]) async def latest() -> 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'" ) 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, } @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)