Update app.py

app.py

@@ -1,233 +1,73 @@
-# app.py — Brain (Milestone 1: streaming + bus + telemetry)
-import asyncio, json, os, time, uuid
-from datetime import datetime, timezone
-from typing import AsyncGenerator, Dict, List, Optional
-
-import httpx
-from fastapi import FastAPI, Form, UploadFile, File, Request
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import StreamingResponse, JSONResponse
-
-APP_START = time.time()
-
-# ---- Config -----------------------------------------------------------------
-ACTUAL_TTS = os.getenv("ACTUAL_TTS_URL", "https://Percy3822-ActualTTS.hf.space")
-ACTUAL_STT = os.getenv("ACTUAL_STT_URL", "https://Percy3822-ActualSTT.hf.space")
-PY_AI      = os.getenv("PYTHON_AI_URL",   "https://Percy3822-Python-ai.hf.space")
-FILES_DIR  = os.getenv("FILES_DIR", "/home/user/files")
-LOG_DIR    = os.path.join(FILES_DIR, "logs")
-os.makedirs(LOG_DIR, exist_ok=True)
-
-# ---- App --------------------------------------------------------------------
-app = FastAPI(title="Brain Streaming")
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# ---- Simple pub/sub for SSE -------------------------------------------------
-class Bus:
-    def __init__(self):
-        self.subs: List[asyncio.Queue] = []
-
-    def sub(self) -> asyncio.Queue:
-        q = asyncio.Queue()
-        self.subs.append(q)
-        return q
-
-    async def pub(self, event: Dict):
-        # Ensure an iso timestamp and monotonic ms
-        event.setdefault("ts_iso", datetime.now(timezone.utc).isoformat())
-        event.setdefault("ts_ms", int(time.time() * 1000))
-        # log to jsonl
-        with open(os.path.join(LOG_DIR, "events.jsonl"), "a", encoding="utf-8") as f:
-            f.write(json.dumps(event, ensure_ascii=False) + "\n")
-        # fan out
-        for q in list(self.subs):
-            try:
-                q.put_nowait(event)
-            except asyncio.QueueFull:
-                pass
-
-BUS = Bus()
-
-def _sse_encode(ev: Dict) -> bytes:
-    return f"data: {json.dumps(ev, ensure_ascii=False)}\n\n".encode("utf-8")
-
-async def _sse_stream(q: asyncio.Queue) -> AsyncGenerator[bytes, None]:
-    # Heartbeat to keep connections alive
+import os
+import time
+import json
+from pathlib import Path
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+
+# ------------------
+# Setup paths
+# ------------------
+BASE_DIR = Path(__file__).parent
+FILES_DIR = BASE_DIR / "files"
+LOG_DIR = FILES_DIR / "logs"
+
+FILES_DIR.mkdir(parents=True, exist_ok=True)
+LOG_DIR.mkdir(parents=True, exist_ok=True)
+
+# ------------------
+# Safe logger
+# ------------------
+def _safe_append_jsonl(path: Path, event: dict):
+    """Safely append JSON event to a file without crashing."""
     try:
-        while True:
-            try:
-                ev = await asyncio.wait_for(q.get(), timeout=15)
-                yield _sse_encode(ev)
-            except asyncio.TimeoutError:
-                yield b": hb\n\n"
-    except asyncio.CancelledError:
-        return
-
-@app.get("/events")
-async def events():
-    q = BUS.sub()
-    return StreamingResponse(_sse_stream(q), media_type="text/event-stream")
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(event, ensure_ascii=False) + "\n")
+    except Exception as e:
+        print(f"[WARN] Could not log event: {e}")
 
-# ---- Health + warmup --------------------------------------------------------
-@app.post("/warmup")
-async def warmup():
-    out = {}
-    async with httpx.AsyncClient(timeout=15) as client:
-        try:
-            r = await client.get(f"{ACTUAL_TTS}/health")
-            out["tts"] = ("ok" if r.status_code == 200 else f"status {r.status_code}")
-        except Exception as e:
-            out["tts"] = f"err: {e!r}"
-        try:
-            r = await client.get(f"{ACTUAL_STT}/health")
-            out["stt"] = ("ok" if r.status_code == 200 else f"status {r.status_code}")
-        except Exception as e:
-            out["stt"] = f"err: {e!r}"
-        try:
-            r = await client.post(f"{PY_AI}/code_help", json={"utterance":"ping"})
-            out["python_ai"] = "ok" if r.status_code in (200,422) else f"status {r.status_code}"
-        except Exception as e:
-            out["python_ai"] = f"err: {e!r}"
-    return {"ok": True, "notes": out}
+# ------------------
+# FastAPI app
+# ------------------
+app = FastAPI(title="Brain", version="0.1")
 
 @app.get("/health")
 async def health():
-    up = time.time() - APP_START
-    return {"ok": True, "uptime_sec": round(up,2)}
-
-# ---- Telemetry intake -------------------------------------------------------
-@app.post("/telemetry")
-async def telemetry(payload: Dict):
-    # payload example: { kind, cpu, mem, active_app, net, meter, etc. }
-    await BUS.pub({"type":"telemetry", "payload":payload})
-    return {"ok": True}
-
-# ---- Streaming NLM (demo) ---------------------------------------------------
-@app.post("/stream/nlm")
-async def stream_nlm(body: Dict):
-    """
-    Body:
-      { "id": "...(optional)", "prompt": "text to answer", "answer": "(optional: if you want me to stream this string)" }
-    For now we stream the provided 'answer' (or a canned one) token-by-token to let you wire the UI.
-    """
-    req_id = body.get("id") or str(uuid.uuid4())
-    prompt = body.get("prompt", "")
-    answer = body.get("answer") or f"Ok! I received your prompt: {prompt[:100]}"
-
-    async def gen():
-        start = time.time()
-        await BUS.pub({"type":"nlm_start", "id":req_id, "prompt":prompt})
-        buf = []
-        for token in answer.split(" "):
-            buf.append(token)
-            chunk = " ".join(buf)
-            ev = {"type":"nlm_token", "id":req_id, "delta":token, "text":chunk}
-            yield _sse_encode(ev)
-            await asyncio.sleep(0.03)  # tiny delay so UI shows a stream
-        dur = time.time() - start
-        await BUS.pub({"type":"nlm_done", "id":req_id, "duration_ms":int(dur*1000)})
-        yield _sse_encode({"type":"nlm_done", "id":req_id})
-    return StreamingResponse(gen(), media_type="text/event-stream")
-
-# ---- Streaming TTS orchestration (status stream) ----------------------------
-@app.post("/stream/tts")
-async def stream_tts(body: Dict):
-    """
-    Body: { "text": "...", "voice": "en_US-amy-medium", "description": "optional" }
-    Streams status events; ends with {type: "tts_audio", audio_url: "..."} when ready.
-    """
-    req_id = str(uuid.uuid4())
-    text = body.get("text","").strip()
-    voice = body.get("voice")
-    desc  = body.get("description")
-
-    async def gen():
-        start = time.time()
-        yield _sse_encode({"type":"tts_status","id":req_id,"phase":"queued"})
-        await BUS.pub({"type":"tts_start","id":req_id,"text":text,"voice":voice})
-
-        try:
-            async with httpx.AsyncClient(timeout=120) as client:
-                yield _sse_encode({"type":"tts_status","id":req_id,"phase":"synthesizing"})
-                r = await client.post(f"{ACTUAL_TTS}/speak", json={"text":text, **({"voice":voice} if voice else {}), **({"description":desc} if desc else {})})
-                r.raise_for_status()
-                data = r.json()
-                if not data.get("ok"):
-                    raise RuntimeError(data.get("error") or "TTS failed")
-                audio_url = data.get("audio_url")
-                yield _sse_encode({"type":"tts_audio","id":req_id,"audio_url":audio_url})
-                dur = time.time() - start
-                await BUS.pub({"type":"tts_done","id":req_id,"audio_url":audio_url,"duration_ms":int(dur*1000)})
-                yield _sse_encode({"type":"tts_done","id":req_id})
-        except Exception as e:
-            err = {"type":"error","scope":"tts","id":req_id,"msg":repr(e)}
-            await BUS.pub(err)
-            yield _sse_encode(err)
-
-    return StreamingResponse(gen(), media_type="text/event-stream")
-
-# ---- Streaming STT orchestration (status stream) ----------------------------
-@app.post("/stream/stt")
-async def stream_stt(file: UploadFile = File(...), beam_size: int = 1, vad_filter: bool = True):
-    """
-    Uploads one file, streams status, ends with text result.
-    """
-    req_id = str(uuid.uuid4())
-
-    async def gen():
-        start = time.time()
-        yield _sse_encode({"type":"stt_status","id":req_id,"phase":"received","filename":file.filename})
-        await BUS.pub({"type":"stt_start","id":req_id,"filename":file.filename})
-
-        try:
-            # forward to ActualSTT
-            files = {"file": (file.filename, await file.read(), file.content_type or "audio/wav")}
-            params = {"beam_size":beam_size, "vad_filter":"true" if vad_filter else "false"}
-            async with httpx.AsyncClient(timeout=120) as client:
-                yield _sse_encode({"type":"stt_status","id":req_id,"phase":"processing"})
-                r = await client.post(f"{ACTUAL_STT}/transcribe", params=params, files=files)
-                r.raise_for_status()
-                data = r.json()
-                text = data.get("text","")
-                lang = data.get("language","")
-                dur_a = data.get("duration",0)
-                yield _sse_encode({"type":"stt_text","id":req_id,"text":text,"language":lang,"audio_duration":dur_a})
-                dur = time.time() - start
-                await BUS.pub({"type":"stt_done","id":req_id,"text":text,"duration_ms":int(dur*1000)})
-                yield _sse_encode({"type":"stt_done","id":req_id})
-        except Exception as e:
-            err = {"type":"error","scope":"stt","id":req_id,"msg":repr(e)}
-            await BUS.pub(err)
-            yield _sse_encode(err)
-
-    return StreamingResponse(gen(), media_type="text/event-stream")
-
-# ---- Existing compatibility endpoints (non-stream) --------------------------
-@app.post("/speak")
-async def speak(body: Dict):
-    text = body.get("text","")
-    voice = body.get("voice")
-    desc  = body.get("description")
-    async with httpx.AsyncClient(timeout=120) as client:
-        r = await client.post(f"{ACTUAL_TTS}/speak", json={"text":text, **({"voice":voice} if voice else {}), **({"description":desc} if desc else {})})
-        return JSONResponse(r.json(), status_code=r.status_code)
-
-@app.post("/transcribe")
-async def transcribe(file: UploadFile = File(...), beam_size: int = 1, vad_filter: bool = True):
-    files = {"file": (file.filename, await file.read(), file.content_type or "audio/wav")}
-    params = {"beam_size":beam_size, "vad_filter":"true" if vad_filter else "false"}
-    async with httpx.AsyncClient(timeout=120) as client:
-        r = await client.post(f"{ACTUAL_STT}/transcribe", params=params, files=files)
-        return JSONResponse(r.json(), status_code=r.status_code)
-
-# (Optional) passthrough to python_ai/code_help so your tests still work
-@app.post("/code_help")
-async def code_help(body: Dict):
-    async with httpx.AsyncClient(timeout=60) as client:
-        r = await client.post(f"{PY_AI}/code_help", json=body)
-        return JSONResponse(r.json(), status_code=r.status_code)
+    return {
+        "ok": True,
+        "service": "brain",
+        "time": time.time(),
+        "files_dir": str(FILES_DIR),
+        "logs": str(LOG_DIR),
+    }
+
+@app.post("/process")
+async def process(request: Request):
+    payload = await request.json()
+
+    event = {
+        "ts": time.time(),
+        "type": "process",
+        "data": payload,
+    }
+    _safe_append_jsonl(LOG_DIR / "events.jsonl", event)
+
+    # Echo back
+    return JSONResponse({"ok": True, "received": payload})
+
+@app.get("/stream/logs")
+async def stream_logs():
+    """Stream logs line-by-line to client (SSE style)."""
+
+    async def event_stream():
+        path = LOG_DIR / "events.jsonl"
+        if not path.exists():
+            yield "data: {}\n\n"
+            return
+
+        with open(path, "r", encoding="utf-8") as f:
+            for line in f:
+                yield f"data: {line.strip()}\n\n"
+
+    return StreamingResponse(event_stream(), media_type="text/event-stream")