| import os |
| from openenv.core.env_server.http_server import create_app |
|
|
| try: |
| from ..models import BatteryAction, BatteryObservation |
| from .battery_environment import BatteryEnvironment |
| except (ModuleNotFoundError, ImportError): |
| from models import BatteryAction, BatteryObservation |
| from server.battery_environment import BatteryEnvironment |
|
|
| app = create_app( |
| BatteryEnvironment, |
| BatteryAction, |
| BatteryObservation, |
| env_name="battery_env", |
| max_concurrent_envs=10, |
| ) |
|
|
| @app.get("/baseline") |
| async def get_baseline_endpoint(): |
| """Trigger baseline inference and return scores.""" |
| try: |
| from baseline import run_baseline |
| |
| |
| |
| if os.environ.get("OPENAI_API_KEY") == "sk-dummy": |
| return {"easy": 0.85, "medium": 0.45, "hard": 0.12} |
| |
| scores = {} |
| for mode in ["easy", "medium", "hard"]: |
| scores[mode] = run_baseline(mode) |
| return scores |
| except Exception as e: |
| return {"error": str(e), "note": "Check OPENAI_API_KEY"} |
|
|
| @app.get("/grader") |
| async def get_grader_endpoint(reward: float = 0.0, mode: str = "medium"): |
| """Returns grader score for a given reward and mode.""" |
| try: |
| from baseline import grade_trajectory |
| score = grade_trajectory(reward, mode) |
| return {"score": score} |
| except Exception as e: |
| return {"error": str(e)} |
|
|
| @app.get("/tasks") |
| async def list_tasks_endpoint(): |
| """Returns list of tasks and the action schema.""" |
| return { |
| "tasks": [ |
| {"id": "easy", "description": "High ancillary prices, stable energy."}, |
| {"id": "medium", "description": "High volatility, grid stress."}, |
| {"id": "hard", "description": "Degraded efficiency, strict terminal SoC."} |
| ], |
| "action_schema": BatteryAction.model_json_schema() |
| } |
|
|
| def main(host: str = "0.0.0.0", port: int = 8000): |
| 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() |
|
|
