File size: 2,378 Bytes
cd5c208
e31582b
0695520
e31582b
cd5c208
 
 
 
e31582b
 
0695520
e31582b
0695520
 
e31582b
737f100
 
 
 
 
e31582b
692f802
0695520
 
692f802
0695520
cd5c208
 
 
03b82c2
 
 
 
cd5c208
 
 
 
 
 
 
737f100
 
 
 
 
 
 
 
 
 
cd5c208
737f100
cd5c208
 
 
63eda9e
cd5c208
63eda9e
 
 
 
 
 
 
 
cd5c208
 
 
 
 
 
 
 
 
e31582b
 
737f100
e31582b
 
0695520
e31582b
 
63eda9e
e31582b
 
 
0695520
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
"""OpenEnv FastAPI entrypoint with optional Gradio mounting."""

from __future__ import annotations

import os

from fastapi import FastAPI

try:
    from openenv.core.env_server.http_server import create_app
except Exception as exc:  # pragma: no cover
    raise ImportError(
        "openenv-core is required to run the API server. Install project dependencies first."
    ) from exc

try:
    import gradio as gr
except Exception:
    gr = None  # type: ignore[assignment]

try:
    from ..models import PythonCodeReviewAction, PythonCodeReviewObservation
    from .env import PythonCodeReviewEnvironment
except ImportError:
    from models import PythonCodeReviewAction, PythonCodeReviewObservation
    from server.env import PythonCodeReviewEnvironment


def _gradio_enabled() -> bool:
    for env_name in ("ENABLE_GRADIO_DEMO", "ENABLE_WEB_INTERFACE"):
        if str(os.getenv(env_name, "")).strip().lower() in {"1", "true", "yes", "on"}:
            return True
    return False


def _max_concurrent_envs() -> int:
    try:
        return max(int(os.getenv("OPENENV_MAX_CONCURRENT_ENVS", "2")), 1)
    except Exception:
        return 2


def build_application():
    """Compose the OpenEnv API with the Gradio demo frontend."""

    api_app = create_app(
        PythonCodeReviewEnvironment,
        PythonCodeReviewAction,
        PythonCodeReviewObservation,
        env_name="python_code_review_env",
        max_concurrent_envs=_max_concurrent_envs(),
    )
    served_app = api_app
    if gr is not None and _gradio_enabled():
        try:
            from .demo import CSS, build_demo
        except ImportError:
            from server.demo import CSS, build_demo
        served_app = gr.mount_gradio_app(
            api_app,
            build_demo(),
            path="/",
            theme=gr.themes.Soft(primary_hue="orange", secondary_hue="amber"),
            css=CSS,
        )

    wrapper_app = FastAPI(title="python_code_review_env", version="1.0.0")

    @wrapper_app.get("/health", include_in_schema=False)
    def _health() -> dict[str, str]:
        return {"status": "ok"}

    wrapper_app.mount("/", served_app)
    return wrapper_app


app = build_application()


def main(host: str = "0.0.0.0", port: int = 8000) -> None:
    import uvicorn

    uvicorn.run(app, host=host, port=port, access_log=False)


if __name__ == "__main__":
    main()