File size: 1,833 Bytes
0eb0abc
b0c701c
 
0eb0abc
b0c701c
 
7a8a0f0
18fd5bc
 
0eb0abc
 
18fd5bc
 
46c2f98
18fd5bc
0eb0abc
 
 
b0c701c
0eb0abc
7a8a0f0
0eb0abc
7a8a0f0
 
 
0eb0abc
7a8a0f0
18fd5bc
b0c701c
 
 
 
 
 
 
 
 
 
 
46c2f98
b0c701c
 
808bf4d
 
 
 
 
 
 
 
 
46c2f98
 
 
 
 
 
 
 
 
18fd5bc
0eb0abc
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
import os
from pathlib import Path

from dotenv import load_dotenv
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from openenv.core import create_app
import uvicorn

load_dotenv()

from environment.actions import ContextCorruptionAction, EpisodeObservation
from environment.env import ContextCorruptionEnv
from environment.model_inference import InferenceRequest, model_status, run_inference

_difficulty_env = os.getenv("DIFFICULTY")
_difficulty = int(_difficulty_env) if _difficulty_env else None
_max_sessions = int(os.getenv("MAX_CONCURRENT_ENVS", "64"))
_frontend_dir = Path(__file__).parent.parent / "frontend"

app = create_app(
    env=lambda: ContextCorruptionEnv(difficulty=_difficulty),
    action_cls=ContextCorruptionAction,
    observation_cls=EpisodeObservation,
    env_name="ContextCorruption-Env",
    max_concurrent_envs=_max_sessions,
)

if _frontend_dir.exists():
    app.mount("/static", StaticFiles(directory=_frontend_dir), name="static")


@app.get("/", include_in_schema=False)
def frontend():
    index_path = _frontend_dir / "index.html"
    if index_path.exists():
        return FileResponse(index_path)
    return api_status()


@app.get("/api/status")
def api_status():
    return {
        "name": "ContextCorruption-Env",
        "status": "running",
        "openenv_endpoints": ["/health", "/metadata", "/schema", "/reset", "/step", "/state"],
        "model_endpoints": ["/model/status", "/model/infer"],
        "docs": "/docs",
    }


@app.get("/model/status")
def get_model_status():
    return model_status()


@app.post("/model/infer")
def infer_with_trained_model(request: InferenceRequest):
    return run_inference(request.observation)

if __name__ == "__main__":
    uvicorn.run("environment.server:app", host="0.0.0.0", port=7860, reload=False)