Dateien nach "app" hochladen

app/main.py

@@ -0,0 +1,498 @@
+"""
+openHAB AI Bridge (simple) – text dialog + session context + Edge TTS.
+
+Flow:
+  openHAB text → POST /api/v1/dialog → SQLite session
+  → MQTT notify → openHAB Cloud push → iOS Shortcut
+  → Ask Claude → POST /api/v1/sessions/{id}/answer
+  → POST /api/v1/audio/speak → Pi speaker (Edge TTS)
+"""
+
+import os
+import json
+import uuid
+import logging
+import sqlite3
+import subprocess
+import time
+from pathlib import Path
+from contextlib import contextmanager
+from datetime import datetime, timezone
+
+from fastapi import FastAPI, HTTPException, Header, Depends
+from pydantic import BaseModel, Field
+import paho.mqtt.client as mqtt
+
+logging.basicConfig(
+    level=os.getenv("LOG_LEVEL", "INFO").upper(),
+    format="%(asctime)s %(levelname)s %(name)s %(message)s",
+)
+log = logging.getLogger("openhab-ai-bridge")
+
+HOST = os.getenv("HOST", "0.0.0.0")
+PORT = int(os.getenv("PORT", "8090"))
+API_TOKEN = os.environ["API_TOKEN"]
+DB_PATH = os.getenv("DB_PATH", "/data/bridge.db")
+
+MAX_HISTORY = int(os.getenv("MAX_HISTORY_MESSAGES", "16"))
+SESSION_TTL = int(os.getenv("SESSION_TTL_SECONDS", "7200"))
+MAX_SESSIONS = int(os.getenv("MAX_SESSIONS", "20"))
+
+MQTT_HOST = os.getenv("MQTT_HOST", "192.168.178.60")
+MQTT_PORT = int(os.getenv("MQTT_PORT", "1884"))
+MQTT_PREFIX = os.getenv("MQTT_PREFIX", "openhab/ai_bridge").strip("/")
+MQTT_USER = os.getenv("MQTT_USERNAME", "")
+MQTT_PASS = os.getenv("MQTT_PASSWORD", "")
+MQTT_TLS = os.getenv("MQTT_TLS", "false").lower() == "true"
+
+ALSA_PLAYBACK = os.getenv("ALSA_PLAYBACK_DEVICE", "default")
+TTS_VOICE = os.getenv("TTS_VOICE", "de-DE-ConradNeural")
+TTS_RATE = os.getenv("TTS_RATE", "+0%")
+TTS_VOLUME = os.getenv("TTS_VOLUME", "+0%")
+
+Path(DB_PATH).parent.mkdir(parents=True, exist_ok=True)
+app = FastAPI(title="openHAB AI Bridge", version="3.0.0-simple")
+
+
+def now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+@contextmanager
+def db():
+    c = sqlite3.connect(DB_PATH, timeout=10)
+    c.row_factory = sqlite3.Row
+    try:
+        c.execute("PRAGMA journal_mode=WAL")
+        c.execute("PRAGMA busy_timeout=5000")
+        yield c
+        c.commit()
+    finally:
+        c.close()
+
+
+def init_db():
+    with db() as c:
+        c.execute(
+            """CREATE TABLE IF NOT EXISTS sessions(
+            id TEXT PRIMARY KEY,
+            title TEXT NOT NULL,
+            prompt TEXT NOT NULL,
+            status TEXT NOT NULL,
+            created_at TEXT NOT NULL,
+            updated_at TEXT NOT NULL,
+            last_response TEXT NOT NULL DEFAULT '')"""
+        )
+        c.execute(
+            """CREATE TABLE IF NOT EXISTS messages(
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            session_id TEXT NOT NULL,
+            role TEXT NOT NULL,
+            content TEXT NOT NULL,
+            created_at TEXT NOT NULL)"""
+        )
+        c.execute(
+            "CREATE INDEX IF NOT EXISTS idx_msg ON messages(session_id, id)"
+        )
+
+
+def mqtt_pub(topic: str, payload: str, retain: bool = False):
+    cl = mqtt.Client(
+        mqtt.CallbackAPIVersion.VERSION2,
+        client_id=f"openhab-ai-bridge-{uuid.uuid4().hex[:8]}",
+    )
+    try:
+        if MQTT_USER:
+            cl.username_pw_set(MQTT_USER, MQTT_PASS)
+        if MQTT_TLS:
+            cl.tls_set()
+        cl.connect(MQTT_HOST, MQTT_PORT, 10)
+        info = cl.publish(topic, payload, qos=1, retain=retain)
+        info.wait_for_publish(5)
+        cl.disconnect()
+    except Exception as e:
+        log.error("MQTT publish failed topic=%s error=%s", topic, e)
+
+
+def auth(authorization: str | None = Header(default=None)):
+    if authorization != f"Bearer {API_TOKEN}":
+        raise HTTPException(401, "Unauthorized")
+
+
+def cleanup():
+    cutoff = time.time() - SESSION_TTL
+    with db() as c:
+        rows = c.execute(
+            "SELECT id, updated_at FROM sessions WHERE status NOT IN ('ended','expired')"
+        ).fetchall()
+        for r in rows:
+            try:
+                if datetime.fromisoformat(r["updated_at"]).timestamp() < cutoff:
+                    c.execute(
+                        "UPDATE sessions SET status='expired', updated_at=? WHERE id=?",
+                        (now(), r["id"]),
+                    )
+            except Exception:
+                pass
+        c.execute(
+            """DELETE FROM messages WHERE session_id IN
+            (SELECT id FROM sessions WHERE status IN ('ended','expired')
+             AND updated_at < datetime('now','-7 days'))"""
+        )
+        c.execute(
+            """DELETE FROM sessions WHERE status IN ('ended','expired')
+            AND updated_at < datetime('now','-7 days')"""
+        )
+
+
+def session_dict(r) -> dict:
+    return {
+        "session_id": r["id"],
+        "title": r["title"],
+        "prompt": r["prompt"],
+        "status": r["status"],
+        "created_at": r["created_at"],
+        "updated_at": r["updated_at"],
+        "last_response": r["last_response"],
+    }
+
+
+def publish_active_session(sid: str | None):
+    mqtt_pub(f"{MQTT_PREFIX}/active_session", sid or "", True)
+
+
+class SessionCreate(BaseModel):
+    title: str = Field(default="KI-Anfrage", max_length=200)
+    prompt: str = Field(min_length=1, max_length=8000)
+    notify_text: str | None = Field(default=None, max_length=1000)
+
+
+class Answer(BaseModel):
+    content: str = Field(min_length=1, max_length=12000)
+
+
+class ContinueReq(BaseModel):
+    content: str = Field(min_length=1, max_length=8000)
+
+
+class DialogReq(BaseModel):
+    text: str = Field(min_length=1, max_length=8000)
+    title: str = Field(default="openHAB Dialog", max_length=200)
+    notify_text: str | None = Field(default="KI_DIALOG", max_length=1000)
+
+
+class SpeakReq(BaseModel):
+    text: str = Field(min_length=1, max_length=12000)
+    voice: str | None = Field(default=None, max_length=100)
+
+
+@app.on_event("startup")
+def startup():
+    init_db()
+    mqtt_pub(f"{MQTT_PREFIX}/availability", "ONLINE", True)
+    mqtt_pub(f"{MQTT_PREFIX}/status", "IDLE", True)
+    log.info(
+        "started port=%s mqtt=%s:%s prefix=%s",
+        PORT,
+        MQTT_HOST,
+        MQTT_PORT,
+        MQTT_PREFIX,
+    )
+
+
+@app.on_event("shutdown")
+def shutdown():
+    mqtt_pub(f"{MQTT_PREFIX}/availability", "OFFLINE", True)
+
+
+@app.get("/health")
+def health():
+    return {"ok": True, "service": "openhab-ai-bridge", "version": "3.0.0-simple"}
+
+
+@app.get("/api/v1/pending")
+def pending(_: None = Depends(auth)):
+    cleanup()
+    with db() as c:
+        r = c.execute(
+            "SELECT * FROM sessions WHERE status='pending' ORDER BY created_at LIMIT 1"
+        ).fetchone()
+    return {"pending": bool(r), **(session_dict(r) if r else {})}
+
+
+@app.get("/api/v1/active")
+def active(_: None = Depends(auth)):
+    cleanup()
+    with db() as c:
+        r = c.execute(
+            """SELECT * FROM sessions
+               WHERE status IN ('pending','processing','active')
+               ORDER BY updated_at DESC LIMIT 1"""
+        ).fetchone()
+    return {"active": bool(r), **(session_dict(r) if r else {})}
+
+
+def _create_session(title: str, prompt: str, notify_text: str | None, source: str) -> str:
+    cleanup()
+    with db() as c:
+        n = c.execute(
+            "SELECT COUNT(*) FROM sessions WHERE status IN ('pending','processing','active')"
+        ).fetchone()[0]
+        if n >= MAX_SESSIONS:
+            raise HTTPException(429, "Too many active sessions")
+        sid = str(uuid.uuid4())
+        ts = now()
+        c.execute(
+            "INSERT INTO sessions VALUES(?,?,?,?,?,?,?)",
+            (sid, title, prompt, "pending", ts, ts, ""),
+        )
+        c.execute(
+            "INSERT INTO messages(session_id, role, content, created_at) VALUES(?,?,?,?)",
+            (sid, "user", prompt, ts),
+        )
+    mqtt_pub(f"{MQTT_PREFIX}/status", "PENDING", True)
+    publish_active_session(sid)
+    mqtt_pub(
+        f"{MQTT_PREFIX}/session/new",
+        json.dumps(
+            {
+                "session_id": sid,
+                "title": title,
+                "prompt": prompt,
+                "notify_text": notify_text,
+                "source": source,
+            },
+            ensure_ascii=False,
+        ),
+    )
+    # Notify topic for openHAB → Cloud Push
+    mqtt_pub(
+        f"{MQTT_PREFIX}/notify",
+        json.dumps(
+            {
+                "session_id": sid,
+                "kind": "dialog",
+                "text": prompt[:200],
+                "notify_text": notify_text or "KI_DIALOG",
+            },
+            ensure_ascii=False,
+        ),
+    )
+    log.info("session created id=%s source=%s title=%r", sid, source, title)
+    return sid
+
+
+@app.post("/api/v1/sessions")
+def create(req: SessionCreate, _: None = Depends(auth)):
+    sid = _create_session(req.title, req.prompt, req.notify_text, "api")
+    return {
+        "session_id": sid,
+        "status": "pending",
+        "title": req.title,
+        "prompt": req.prompt,
+    }
+
+
+@app.get("/api/v1/sessions/{sid}")
+def get_session(sid: str, _: None = Depends(auth)):
+    with db() as c:
+        r = c.execute("SELECT * FROM sessions WHERE id=?", (sid,)).fetchone()
+    if not r:
+        raise HTTPException(404, "Session not found")
+    return session_dict(r)
+
+
+@app.get("/api/v1/sessions/{sid}/context")
+def context(sid: str, _: None = Depends(auth)):
+    with db() as c:
+        s = c.execute("SELECT * FROM sessions WHERE id=?", (sid,)).fetchone()
+        if not s:
+            raise HTTPException(404, "Session not found")
+        m = c.execute(
+            "SELECT role, content, created_at FROM messages "
+            "WHERE session_id=? ORDER BY id DESC LIMIT ?",
+            (sid, MAX_HISTORY),
+        ).fetchall()
+    return {
+        "session_id": sid,
+        "title": s["title"],
+        "status": s["status"],
+        "messages": [dict(x) for x in reversed(m)],
+    }
+
+
+@app.post("/api/v1/sessions/{sid}/answer")
+def answer(sid: str, req: Answer, _: None = Depends(auth)):
+    with db() as c:
+        s = c.execute("SELECT id FROM sessions WHERE id=?", (sid,)).fetchone()
+        if not s:
+            raise HTTPException(404, "Session not found")
+        ts = now()
+        c.execute(
+            "UPDATE sessions SET status='active', updated_at=?, last_response=? WHERE id=?",
+            (ts, req.content, sid),
+        )
+        c.execute(
+            "INSERT INTO messages(session_id, role, content, created_at) VALUES(?,?,?,?)",
+            (sid, "assistant", req.content, ts),
+        )
+    mqtt_pub(f"{MQTT_PREFIX}/last_response", req.content, True)
+    mqtt_pub(f"{MQTT_PREFIX}/status", "ACTIVE", True)
+    publish_active_session(sid)
+    mqtt_pub(f"{MQTT_PREFIX}/session/{sid}/answer", req.content)
+    log.info("answer id=%s chars=%d", sid, len(req.content))
+    return {"ok": True, "session_id": sid, "status": "active"}
+
+
+@app.post("/api/v1/sessions/{sid}/continue")
+def cont(sid: str, req: ContinueReq, _: None = Depends(auth)):
+    with db() as c:
+        s = c.execute("SELECT status FROM sessions WHERE id=?", (sid,)).fetchone()
+        if not s:
+            raise HTTPException(404, "Session not found")
+        if s["status"] in ("ended", "expired"):
+            raise HTTPException(409, "Session ended")
+        ts = now()
+        c.execute(
+            "UPDATE sessions SET status='processing', updated_at=? WHERE id=?",
+            (ts, sid),
+        )
+        c.execute(
+            "INSERT INTO messages(session_id, role, content, created_at) VALUES(?,?,?,?)",
+            (sid, "user", req.content, ts),
+        )
+    mqtt_pub(f"{MQTT_PREFIX}/status", "PROCESSING", True)
+    publish_active_session(sid)
+    mqtt_pub(f"{MQTT_PREFIX}/session/{sid}/user", req.content)
+    mqtt_pub(
+        f"{MQTT_PREFIX}/notify",
+        json.dumps(
+            {
+                "session_id": sid,
+                "kind": "dialog",
+                "text": req.content[:200],
+                "notify_text": "KI_DIALOG",
+            },
+            ensure_ascii=False,
+        ),
+    )
+    return {"ok": True, "session_id": sid, "status": "processing"}
+
+
+@app.post("/api/v1/dialog")
+def dialog(req: DialogReq, _: None = Depends(auth)):
+    """
+    openHAB entry point.
+    Reuses active session if present, otherwise creates a new one.
+    Publishes MQTT notify so openHAB can trigger the Cloud push.
+    """
+    cleanup()
+    with db() as c:
+        s = c.execute(
+            """SELECT * FROM sessions
+               WHERE status IN ('active','processing','pending')
+               ORDER BY updated_at DESC LIMIT 1"""
+        ).fetchone()
+
+    if s:
+        sid = s["id"]
+        cont(sid, ContinueReq(content=req.text))
+        mqtt_pub(
+            f"{MQTT_PREFIX}/dialog/input",
+            json.dumps({"session_id": sid, "text": req.text}, ensure_ascii=False),
+        )
+        return {"session_id": sid, "status": "processing", "reused": True}
+
+    sid = _create_session(req.title, req.text, req.notify_text, "openhab")
+    mqtt_pub(
+        f"{MQTT_PREFIX}/dialog/input",
+        json.dumps({"session_id": sid, "text": req.text}, ensure_ascii=False),
+    )
+    return {"session_id": sid, "status": "pending", "reused": False}
+
+
+@app.post("/api/v1/sessions/{sid}/end")
+def end(sid: str, _: None = Depends(auth)):
+    with db() as c:
+        if not c.execute("SELECT id FROM sessions WHERE id=?", (sid,)).fetchone():
+            raise HTTPException(404, "Session not found")
+        c.execute(
+            "UPDATE sessions SET status='ended', updated_at=? WHERE id=?",
+            (now(), sid),
+        )
+    mqtt_pub(f"{MQTT_PREFIX}/status", "IDLE", True)
+    publish_active_session("")
+    mqtt_pub(f"{MQTT_PREFIX}/session/{sid}/ended", "ENDED")
+    return {"ok": True, "session_id": sid, "status": "ended"}
+
+
+@app.post("/api/v1/audio/speak")
+async def speak(req: SpeakReq, _: None = Depends(auth)):
+    """Edge TTS (free) → WAV → physical Pi speaker via ALSA."""
+    outdir = Path("/data/audio")
+    outdir.mkdir(parents=True, exist_ok=True)
+    ident = uuid.uuid4().hex
+    mp3 = outdir / f"{ident}.mp3"
+    wav = outdir / f"{ident}.wav"
+    voice = req.voice or TTS_VOICE
+    try:
+        import edge_tts
+
+        communicate = edge_tts.Communicate(
+            req.text, voice, rate=TTS_RATE, volume=TTS_VOLUME
+        )
+        await communicate.save(str(mp3))
+        p = subprocess.run(
+            [
+                "ffmpeg",
+                "-y",
+                "-loglevel",
+                "error",
+                "-i",
+                str(mp3),
+                "-ar",
+                "48000",
+                "-ac",
+                "2",
+                str(wav),
+            ],
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        if p.returncode:
+            raise HTTPException(500, p.stderr[-1000:] or "TTS conversion failed")
+        p = subprocess.run(
+            ["aplay", "-D", ALSA_PLAYBACK, str(wav)],
+            capture_output=True,
+            text=True,
+            timeout=180,
+        )
+        if p.returncode:
+            raise HTTPException(500, p.stderr[-1000:] or "ALSA playback failed")
+        return {"ok": True, "voice": voice, "speaker": "pi-alsa"}
+    finally:
+        for f in (mp3, wav):
+            try:
+                f.unlink()
+            except FileNotFoundError:
+                pass
+
+
+@app.get("/api/v1/audio/devices")
+def audio_devices(_: None = Depends(auth)):
+    out = {}
+    for k, cmd in {
+        "playback": ["aplay", "-l"],
+        "capture": ["arecord", "-l"],
+    }.items():
+        try:
+            p = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
+            out[k] = {
+                "returncode": p.returncode,
+                "stdout": p.stdout,
+                "stderr": p.stderr,
+            }
+        except Exception as e:
+            out[k] = {"error": str(e)}
+    return out