|
| 1 | +# SPDX-License-Identifier: BSD-3-Clause |
| 2 | + |
| 3 | +"""Client for the Pelican SVG environment.""" |
| 4 | + |
| 5 | +from __future__ import annotations |
| 6 | + |
| 7 | +from typing import Any, Dict |
| 8 | + |
| 9 | +from openenv.core.client_types import StepResult |
| 10 | +from openenv.core.env_client import EnvClient |
| 11 | + |
| 12 | +from .models import PelicanSvgAction, PelicanSvgObservation, PelicanSvgState |
| 13 | + |
| 14 | + |
| 15 | +class PelicanSvgEnv( |
| 16 | + EnvClient[PelicanSvgAction, PelicanSvgObservation, PelicanSvgState] |
| 17 | +): |
| 18 | + """Connects to a running Pelican SVG environment server. |
| 19 | +
|
| 20 | + Examples: |
| 21 | +
|
| 22 | + ```python |
| 23 | + with PelicanSvgEnv(base_url="http://localhost:8000") as env: |
| 24 | + observation = env.reset().observation |
| 25 | + reply = my_model(observation.prompt) |
| 26 | + result = env.step(PelicanSvgAction(response=reply)) |
| 27 | + print(result.reward, result.observation.feedback) |
| 28 | + ``` |
| 29 | + """ |
| 30 | + |
| 31 | + def _step_payload(self, action: PelicanSvgAction) -> Dict[str, Any]: |
| 32 | + """Convert an action into the JSON body of a step request.""" |
| 33 | + return {"response": action.response} |
| 34 | + |
| 35 | + def _parse_result( |
| 36 | + self, payload: Dict[str, Any] |
| 37 | + ) -> StepResult[PelicanSvgObservation]: |
| 38 | + """Parse a server response into a typed step result. |
| 39 | +
|
| 40 | + The server hoists `reward` and `done` onto the response envelope and |
| 41 | + drops them from the serialised observation, so they are read from the |
| 42 | + envelope first and only then from the observation body. |
| 43 | + """ |
| 44 | + data = payload.get("observation", {}) |
| 45 | + reward = payload.get("reward", data.get("reward")) |
| 46 | + done = payload.get("done", data.get("done", False)) |
| 47 | + observation = PelicanSvgObservation( |
| 48 | + prompt=data.get("prompt", ""), |
| 49 | + task_id=data.get("task_id", ""), |
| 50 | + subject=data.get("subject", ""), |
| 51 | + vehicle=data.get("vehicle", ""), |
| 52 | + expected_wheels=data.get("expected_wheels", 2), |
| 53 | + held_out=data.get("held_out", True), |
| 54 | + feedback=data.get("feedback", ""), |
| 55 | + gate_passed=data.get("gate_passed", False), |
| 56 | + structure_score=data.get("structure_score", 0.0), |
| 57 | + semantic_score=data.get("semantic_score", 0.0), |
| 58 | + judged=data.get("judged", False), |
| 59 | + violations=data.get("violations", []), |
| 60 | + breakdown=data.get("breakdown", {}), |
| 61 | + image_png_base64=data.get("image_png_base64"), |
| 62 | + done=bool(done), |
| 63 | + reward=reward, |
| 64 | + metadata=payload.get("metadata", data.get("metadata", {})), |
| 65 | + ) |
| 66 | + return StepResult( |
| 67 | + observation=observation, |
| 68 | + reward=observation.reward, |
| 69 | + done=observation.done, |
| 70 | + ) |
| 71 | + |
| 72 | + def _parse_state(self, payload: Dict[str, Any]) -> PelicanSvgState: |
| 73 | + """Parse a response from the state endpoint into a typed state.""" |
| 74 | + return PelicanSvgState( |
| 75 | + episode_id=payload.get("episode_id"), |
| 76 | + step_count=payload.get("step_count", 0), |
| 77 | + task_id=payload.get("task_id", ""), |
| 78 | + submitted=payload.get("submitted", False), |
| 79 | + ) |
0 commit comments