"""FastAPI application for the Data Centric Env Environment.""" import sys import os # Ensure the project root is on the path regardless of how the server is launched _HERE = os.path.dirname(os.path.abspath(__file__)) _ROOT = os.path.dirname(_HERE) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) try: from openenv.core.env_server.http_server import create_app except Exception as e: raise ImportError( "openenv-core is required. Install with: pip install openenv-core" ) from e try: from ..models import DataCentricAction, DataCentricObservation from .data_centric_environment import DataCentricEnvironment except (ImportError, ModuleNotFoundError): from models import DataCentricAction, DataCentricObservation from server.data_centric_environment import DataCentricEnvironment from fastapi.responses import HTMLResponse # max_concurrent_envs=1: avoids concurrency safety check that instantiates the env # at startup (which would load sklearn and pandas, slowing HF health check). # Increase if running on a paid Space with more RAM. app = create_app( DataCentricEnvironment, DataCentricAction, DataCentricObservation, env_name="data_centric_env", max_concurrent_envs=1, ) _LANDING_HTML = """ Data-Centric AI RL Environment

🧠 Data-Centric AI Environment

• Running

An OpenEnv-compliant RL environment where an LLM learns to coordinate data-preprocessing agents — cleaning, imputing, and rebalancing datasets to improve a pre-trained classifier's accuracy, without modifying its weights.

GET /health Server health check
GET /docs Interactive API docs (Swagger)
POST /reset Start a new episode
POST /step Execute an action
WS /ws WebSocket (stateful session)
GET /state Current episode state

Quick start:

pip install git+https://huggingface.co/spaces/Aswini-Kumar/data-centric-env

from data_centric_env import DataCentricEnv, DataCentricAction
with DataCentricEnv(base_url="https://aswini-kumar-data-centric-env.hf.space").sync() as env:
    obs = env.reset(task="task_0_tutorial", seed=42)
    result = env.step(DataCentricAction(message="query_analyst"))
    print(result.observation.response)
API Docs  |  Health  |  GitHub  |  HF Space
""" @app.get("/", response_class=HTMLResponse, include_in_schema=False) @app.get("/web", response_class=HTMLResponse, include_in_schema=False) async def landing(): """Human-readable landing page for the HF Space App tab.""" return HTMLResponse(content=_LANDING_HTML) def main(host: str = "0.0.0.0", port: int = 7860): import uvicorn uvicorn.run(app, host=host, port=port) if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--host", default="0.0.0.0") parser.add_argument("--port", type=int, default=7860) args = parser.parse_args() main(host=args.host, port=args.port)