|
@@ -0,0 +1,289 @@
|
|
|
|
|
+"""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")
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+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"
|
|
|
|
|
+ )
|
|
|
|
|
+ group = "GROUP BY t, device"
|
|
|
|
|
+ else:
|
|
|
|
|
+ select = "toUInt64(toUnixTimestamp(hour)) * 1000 AS t, device, temperature, humidity"
|
|
|
|
|
+ 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": []})
|
|
|
|
|
+ s["t"].append(int(row["t"]))
|
|
|
|
|
+ s["temperature"].append(row["temperature"])
|
|
|
|
|
+ s["humidity"].append(row["humidity"])
|
|
|
|
|
+ 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'"
|
|
|
|
|
+ )
|
|
|
|
|
+ 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
|
|
|
|
|
+ ],
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+@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)
|