Spaces:
Paused
Paused
File size: 3,334 Bytes
1070765 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """
FastAPI application for the WatchDog Environment.
Endpoints:
- GET /: Play UI (Gradio)
- POST /reset: Reset the environment
- POST /step: Execute an action
- GET /state: Get current environment state
- GET /schema: Get action/observation schemas
- WS /ws: WebSocket endpoint for persistent sessions
Usage:
uvicorn server.app:app --reload --host 0.0.0.0 --port 8000
"""
import logging
logging.basicConfig(level=logging.INFO, format="%(name)s: %(message)s")
from pathlib import Path
# Load ../.env (openenv_hack/.env) before any imports that use env vars
def _load_env() -> None:
import logging
import os
_log = logging.getLogger("watchdog_env")
_env_path = Path(__file__).resolve().parent.parent.parent / ".env"
if not _env_path.is_file():
_log.info("[app] No .env at %s, skipping", _env_path)
return
try:
from dotenv import load_dotenv
load_dotenv(_env_path, override=True) # override so .env takes precedence
_log.info("[app] Loaded .env from %s", _env_path)
_log.info("[app] WATCHDOG_LLM_BACKEND=%s, GEMINI_API_KEY=%s",
os.environ.get("WATCHDOG_LLM_BACKEND"), "set" if os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") else "NOT SET")
return
except ImportError:
pass
# Fallback: parse .env manually when python-dotenv not installed
for line in _env_path.read_text().encode("utf-8", errors="replace").decode().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, _, value = line.partition("=")
key, value = key.strip(), value.strip().strip("'\"")
if key:
os.environ[key] = value
_load_env()
from fastapi import FastAPI
import gradio as gr
from openenv.core.env_server.http_server import create_app
from models import MultiTurnAction, MultiTurnObservation
from .watchdog_environment import WatchDogMultiTurnEnvironment
from .ui import build_ui, UI_CSS, UI_THEME
# Ensure plugins are registered (Avalon, Cicero)
try:
import plugins # noqa: F401
except ImportError:
import watchdog_env.plugins # noqa: F401
app = create_app(
WatchDogMultiTurnEnvironment,
MultiTurnAction,
MultiTurnObservation,
env_name="watchdog_env",
max_concurrent_envs=4,
)
# Mount Gradio play UI at root (theme/css passed to launch per Gradio 6.0)
gradio_app = build_ui()
app = gr.mount_gradio_app(app, gradio_app, path="/", theme=UI_THEME, css=UI_CSS)
@app.get("/api")
def api_info():
"""API info (UI is at /)."""
return {
"message": "WatchDog OpenEnv Environment API",
"play_ui": "/",
"endpoints": {
"health": "GET /health",
"schema": "GET /schema",
"reset": "POST /reset",
"step": "POST /step",
"state": "GET /state",
"docs": "GET /docs",
"ws": "WS /ws",
},
}
def main(host: str = "0.0.0.0", port: int = 8000) -> None:
import uvicorn
uvicorn.run(app, host=host, port=port)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=8000)
args = parser.parse_args()
main(port=args.port)
|