| """Screen Companion — Gradio Server frontend. |
| |
| Serves the custom HTML landing page and proxies /api/analyze requests to |
| the Modal backend. Run with SCREEN_COMPANION_MOCK=1 for local testing. |
| """ |
| import os |
| from pathlib import Path |
|
|
| import httpx |
| from fastapi.responses import HTMLResponse |
| from fastapi.staticfiles import StaticFiles |
| from gradio import Server |
| from pydantic import BaseModel |
|
|
| app = Server() |
|
|
| BACKEND_URL = os.environ.get( |
| "SCREEN_COMPANION_API_URL", |
| "https://rafalbogusdxc--screen-companion-api.modal.run", |
| ).rstrip("/") |
|
|
| MOCK = os.environ.get("SCREEN_COMPANION_MOCK", "").strip().lower() in ("1", "true", "yes") |
|
|
| STATIC_DIR = Path(__file__).parent / "static" |
|
|
| http = httpx.AsyncClient(base_url=BACKEND_URL, timeout=120, follow_redirects=True) |
|
|
|
|
| class AnalyzeRequest(BaseModel): |
| image_b64: str |
| memories: str = "" |
|
|
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def homepage(): |
| return (Path(__file__).parent / "index.html").read_text() |
|
|
|
|
| app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") |
|
|
|
|
| @app.post("/api/analyze") |
| async def proxy_analyze(req: AnalyzeRequest): |
| if MOCK: |
| import random |
| observations = [ |
| "user editing code in IDE", |
| "browser with documentation open", |
| "terminal with test output", |
| "same as before", |
| "user reading article", |
| ] |
| msgs = [ |
| ("Nice progress! That code looks clean.", "happy"), |
| ("Hmm, looks like you're deep in thought...", "neutral"), |
| ("You've been at this a while - remember to stretch!", "happy"), |
| ("Ooh, interesting problem you're working on!", "excited"), |
| ("Errors happen to everyone. You'll figure it out!", "sad"), |
| ("That's a great approach, keep going!", "excited"), |
| ("I see you're reading docs - smart move!", "happy"), |
| ("Take a deep breath, you've got this!", "neutral"), |
| (None, "neutral"), |
| (None, "happy"), |
| (None, "neutral"), |
| ] |
| msg, emotion = random.choice(msgs) |
| obs = random.choice(observations) |
| return {"message": msg, "emotion": emotion, "observation": obs} |
|
|
| try: |
| r = await http.post("/analyze-screen", json={ |
| "image_b64": req.image_b64, |
| "memories": req.memories, |
| }) |
| r.raise_for_status() |
| return r.json() |
| except httpx.ConnectError: |
| return {"message": "Waking up... one moment.", "emotion": "neutral", "observation": "backend cold starting"} |
| except httpx.ReadTimeout: |
| return {"message": "Hmm, lost my train of thought.", "emotion": "neutral", "observation": "response timed out"} |
| except Exception as e: |
| print(f"[screen-companion] proxy error: {type(e).__name__}: {e}") |
| return {"message": None, "emotion": "neutral", "observation": f"error: {type(e).__name__}"} |
|
|
|
|
| @app.get("/api/health") |
| async def proxy_health(): |
| if MOCK: |
| return {"status": "mock", "model": "SmolVLM-2B (mock mode)"} |
| try: |
| r = await http.get("/health", timeout=10) |
| r.raise_for_status() |
| return r.json() |
| except Exception: |
| return {"status": "unreachable", "url": BACKEND_URL} |
|
|
|
|
| if __name__ == "__main__": |
| app.launch(server_name="0.0.0.0", server_port=7860) |
|
|
