diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..5c082b7 --- /dev/null +++ b/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