Update app.py

app.py

@@ -1,68 +1,31 @@
-import os, json, time, uuid, io
+import os
+import json
+import time
+import asyncio
 from pathlib import Path
-from typing import Dict, Any, AsyncIterator
-from queue import Queue, Empty
+from typing import Dict, Any
 
-import httpx
-from fastapi import FastAPI, File, UploadFile, Request, Response
-from fastapi.responses import StreamingResponse, JSONResponse
+from fastapi import FastAPI, Body, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+from fastapi.staticfiles import StaticFiles
 
-# ---------- CONFIG ----------
-BASE_DIR  = Path(os.getenv("BRAIN_BASE_DIR", "/tmp/brain_app"))
+BASE_DIR  = Path(os.environ.get("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")
+app.mount("/files", StaticFiles(directory=str(FILES_DIR)), name="files")
 
-# simple fanout queues for SSE
-_log_subs, _err_subs, _telem_subs = set(), set(), set()
-
-def _append(path: Path, obj: Dict[str, Any]):
-    try:
-        with path.open("a", encoding="utf-8") as f:
-            f.write(json.dumps(obj, ensure_ascii=False) + "\n")
-    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)
-
-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")
+def append_jsonl(path: Path, obj: Dict[str, Any]):
+    obj = {"ts": time.time(), **obj}
+    with path.open("a", encoding="utf-8") as f:
+        f.write(json.dumps(obj, ensure_ascii=False) + "\n")
 
-# ---------- HEALTH ----------
 @app.get("/health")
 def health():
     return {
@@ -73,99 +36,56 @@ def health():
         "files_dir": str(FILES_DIR),
         "logs_dir": str(LOGS_DIR),
         "events_file": str(EVENTS_FILE),
+        "telemetry_file": str(TELEM_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(payload: Dict[str, Any], request: Request):
-    evt = {"ts": time.time(), "type": "process", "data": payload}
-    _append(EVENTS_FILE, evt)
-    _broadcast(_log_subs, evt)
+async def process(payload: Dict[str, Any] = Body(...)):
+    append_jsonl(EVENTS_FILE, {"type": "process", "data": payload})
     return {"ok": True, "received": payload}
 
-# ---------- ERRORS ----------
 @app.post("/log_error")
-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)
+async def log_error(payload: Dict[str, Any] = Body(...)):
+    append_jsonl(EVENTS_FILE, {"type": "error", "data": payload})
     return {"ok": True}
 
+@app.post("/telemetry")
+async def telemetry(payload: Dict[str, Any] = Body(...)):
+    append_jsonl(TELEM_FILE, {"type": "telemetry", "data": payload})
+    return {"ok": True}
+
+def sse_lines(path: Path, event_filter: str | None = None):
+    # Tail-like SSE; yields lines as they are appended
+    pos = 0
+    while True:
+        try:
+            with path.open("r", encoding="utf-8") as f:
+                f.seek(pos)
+                for line in f:
+                    pos = f.tell()
+                    try:
+                        obj = json.loads(line.strip())
+                        if event_filter and obj.get("type") != event_filter:
+                            continue
+                        yield f"data: {json.dumps(obj, ensure_ascii=False)}\n\n"
+                    except Exception:
+                        continue
+        except FileNotFoundError:
+            pass
+        time.sleep(0.5)
+
 @app.get("/stream/logs")
 def stream_logs():
-    return _sse_stream(_log_subs)
+    return StreamingResponse(sse_lines(EVENTS_FILE, None), media_type="text/event-stream")
 
 @app.get("/stream/errors")
 def stream_errors():
-    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}
+    return StreamingResponse(sse_lines(EVENTS_FILE, "error"), media_type="text/event-stream")
 
 @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)
+    return StreamingResponse(sse_lines(TELEM_FILE, "telemetry"), media_type="text/event-stream")
 
-# ---------- 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())
+@app.get("/")
+def index():
+    return {"ok": True, "service": "brain", "streams": ["/stream/logs", "/stream/telemetry"]}