Update app.py

app.py

@@ -1,50 +1,68 @@
-# app.py
-import json, time, os
+import os, json, time, uuid, io
 from pathlib import Path
-from fastapi import FastAPI, Request
-from fastapi.responses import JSONResponse, StreamingResponse
+from typing import Dict, Any, AsyncIterator
+from queue import Queue, Empty
 
-app = FastAPI(title="Brain", version="0.1.0")
+import httpx
+from fastapi import FastAPI, File, UploadFile, Request, Response
+from fastapi.responses import StreamingResponse, JSONResponse
 
-# -----------------------------
-# Writable paths (with fallback)
-# -----------------------------
-def _mk_dir(path: Path) -> Path:
-    try:
-        path.mkdir(parents=True, exist_ok=True)
-        return path
-    except Exception as e:
-        # fallback to /tmp which is always writable on HF Spaces
-        fallback = Path("/tmp/brain_app/files") if "files" in str(path) else Path("/tmp/brain_app")
-        try:
-            fallback.mkdir(parents=True, exist_ok=True)
-        except Exception:
-            pass
-        print(f"[WARN] Could not create {path} ({e}); using {fallback}")
-        return fallback
-
-# Allow overrides via env, else use /tmp
-BASE_DIR = Path(os.getenv("BRAIN_BASE_DIR", "/tmp/brain_app"))
-FILES_DIR = _mk_dir(Path(os.getenv("FILES_DIR", str(BASE_DIR / "files"))))
-LOG_DIR = _mk_dir(Path(os.getenv("LOG_DIR", str(FILES_DIR / "logs"))))
-
-EVENTS_FILE = LOG_DIR / "events.jsonl"
-ERRORS_FILE = LOG_DIR / "errors.jsonl"
-
-# -----------------------------
-# Safe file append
-# -----------------------------
-def _append_jsonl(path: Path, obj: dict):
+# ---------- CONFIG ----------
+BASE_DIR  = Path(os.getenv("BRAIN_BASE_DIR", "/tmp/brain_app"))
+FILES_DIR = BASE_DIR / "files"
+LOGS_DIR  = FILES_DIR / "logs"
+FILES_DIR.mkdir(parents=True, exist_ok=True)
+LOGS_DIR.mkdir(parents=True, exist_ok=True)
+
+EVENTS_FILE = LOGS_DIR / "events.jsonl"
+ERRORS_FILE = LOGS_DIR / "errors.jsonl"
+TELEM_FILE  = LOGS_DIR / "telemetry.jsonl"
+
+ACTUAL_TTS = os.getenv("ACTUAL_TTS", "https://Percy3822-ActualTTS.hf.space")
+ACTUAL_STT = os.getenv("ACTUAL_STT", "https://Percy3822-ActualSTT.hf.space")
+PY_AI      = os.getenv("PY_AI",      "https://Percy3822-Python-ai.hf.space")
+
+# ---------- APP ----------
+app = FastAPI(title="Brain")
+
+# simple fanout queues for SSE
+_log_subs, _err_subs, _telem_subs = set(), set(), set()
+
+def _append(path: Path, obj: Dict[str, Any]):
     try:
-        path.parent.mkdir(parents=True, exist_ok=True)
         with path.open("a", encoding="utf-8") as f:
             f.write(json.dumps(obj, ensure_ascii=False) + "\n")
-    except Exception as e:
-        print(f"[WARN] log write failed: {e}")
+    except Exception:
+        pass
+
+def _broadcast(subs, event):
+    dead = []
+    for q in list(subs):
+        try:
+            q.put_nowait(event)
+        except Exception:
+            dead.append(q)
+    for q in dead:
+        subs.discard(q)
 
-# -----------------------------
-# Endpoints
-# -----------------------------
+def _sse_stream(subs) -> StreamingResponse:
+    def make() -> AsyncIterator[str]:
+        q: Queue = Queue(maxsize=1000)
+        subs.add(q)
+        try:
+            yield f"data: {json.dumps({'ts': time.time(), 'hello': True})}\n\n"
+            while True:
+                try:
+                    item = q.get(timeout=30)
+                except Empty:
+                    yield "data: {}\n\n"  # keepalive
+                    continue
+                yield f"data: {json.dumps(item)}\n\n"
+        finally:
+            subs.discard(q)
+    return StreamingResponse(make(), media_type="text/event-stream")
+
+# ---------- HEALTH ----------
 @app.get("/health")
 def health():
     return {
@@ -53,40 +71,101 @@ def health():
         "time": time.time(),
         "base_dir": str(BASE_DIR),
         "files_dir": str(FILES_DIR),
-        "logs_dir": str(LOG_DIR),
+        "logs_dir": str(LOGS_DIR),
         "events_file": str(EVENTS_FILE),
     }
 
+# ---------- WARMUP ----------
+@app.post("/warmup")
+async def warmup():
+    notes = {}
+    async with httpx.AsyncClient(timeout=20) as s:
+        try:
+            r = await s.get(f"{PY_AI}/health")
+            notes["python_ai"] = "ok" if r.status_code == 200 else f"status {r.status_code}"
+        except Exception:
+            notes["python_ai"] = "down"
+        try:
+            r = await s.get(f"{ACTUAL_TTS}/health")
+            notes["tts"] = "ok" if r.status_code == 200 else f"status {r.status_code}"
+        except Exception:
+            notes["tts"] = "down"
+        try:
+            r = await s.get(f"{ACTUAL_STT}/health")
+            notes["stt"] = "ok" if r.status_code == 200 else f"status {r.status_code}"
+        except Exception:
+            notes["stt"] = "down"
+    return {"ok": True, "notes": notes}
+
+# ---------- GENERIC PROCESS (demo) ----------
 @app.post("/process")
-async def process(req: Request):
-    payload = await req.json()
+async def process(payload: Dict[str, Any], request: Request):
     evt = {"ts": time.time(), "type": "process", "data": payload}
-    _append_jsonl(EVENTS_FILE, evt)
+    _append(EVENTS_FILE, evt)
+    _broadcast(_log_subs, evt)
     return {"ok": True, "received": payload}
 
+# ---------- ERRORS ----------
 @app.post("/log_error")
-async def log_error(req: Request):
-    payload = await req.json()
-    err = {"ts": time.time(), "type": "error", "data": payload}
-    _append_jsonl(ERRORS_FILE, err)
-    _append_jsonl(EVENTS_FILE, {"ts": err["ts"], "type": "error_ref"})
+async def log_error(payload: Dict[str, Any], request: Request):
+    evt = {"ts": time.time(), "type": "error", "data": payload}
+    _append(ERRORS_FILE, evt)
+    _broadcast(_err_subs, evt)
     return {"ok": True}
 
-def _sse_from_file(path: Path):
-    # generator for SSE streaming of an existing file (no tailing daemon)
-    def gen():
-        if not path.exists():
-            yield "data: {}\n\n"
-            return
-        with path.open("r", encoding="utf-8") as f:
-            for line in f:
-                yield f"data: " + line.strip() + "\n\n"
-    return gen
-
 @app.get("/stream/logs")
 def stream_logs():
-    return StreamingResponse(_sse_from_file(EVENTS_FILE)(), media_type="text/event-stream")
+    return _sse_stream(_log_subs)
 
 @app.get("/stream/errors")
 def stream_errors():
-    return StreamingResponse(_sse_from_file(ERRORS_FILE)(), media_type="text/event-stream")
+    return _sse_stream(_err_subs)
+
+# ---------- TELEMETRY ----------
+@app.post("/telemetry")
+async def ingest_telemetry(payload: Dict[str, Any], request: Request):
+    evt = {
+        "ts": time.time(),
+        "type": "telemetry",
+        "client_ip": request.client.host if request and request.client else None,
+        "data": payload,
+    }
+    _append(TELEM_FILE, evt)
+    _broadcast(_telem_subs, evt)
+    return {"ok": True}
+
+@app.get("/stream/telemetry")
+def stream_telemetry():
+    return _sse_stream(_telem_subs)
+
+# ---------- PYTHON AI PROXY ----------
+@app.post("/code_help")
+async def code_help_proxy(payload: Dict[str, Any]):
+    async with httpx.AsyncClient(timeout=60) as s:
+        r = await s.post(f"{PY_AI}/code_help", json=payload)
+        return JSONResponse(status_code=r.status_code, content=r.json())
+
+# ---------- SPEAK (TTS) ----------
+@app.post("/speak")
+async def speak(payload: Dict[str, Any]):
+    # payload: { text, voice? }
+    async with httpx.AsyncClient(timeout=120) as s:
+        r = await s.post(f"{ACTUAL_TTS}/speak", json=payload)
+        data = r.json()
+        # Always return fully-qualified URL
+        if "audio_url" in data:
+            url = data["audio_url"]
+            if url.startswith("/"):
+                data["audio_url"] = f"{ACTUAL_TTS}{url}"
+            elif url.startswith("http") is False:
+                data["audio_url"] = f"{ACTUAL_TTS}/{url.lstrip('/')}"
+        return JSONResponse(status_code=r.status_code, content=data)
+
+# ---------- TRANSCRIBE (STT) ----------
+@app.post("/transcribe")
+async def transcribe(file: UploadFile = File(...)):
+    raw = await file.read()
+    files = {"file": (file.filename, raw, file.content_type or "audio/wav")}
+    async with httpx.AsyncClient(timeout=120) as s:
+        r = await s.post(f"{ACTUAL_STT}/transcribe", files=files)
+        return JSONResponse(status_code=r.status_code, content=r.json())