Spaces:
Sleeping
Sleeping
File size: 2,493 Bytes
585cd37 | 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 | """Fish Farm Environment Client.
WebSocket-based client for interacting with the Fish Farm OpenEnv server.
Provides async and sync interfaces for reset/step/state operations.
"""
from typing import Any, Dict
from openenv.core import EnvClient
from openenv.core.client_types import StepResult
from openenv.core.env_server.types import State
from .models import FarmAction, FarmObservation
class FishFarmEnv(EnvClient[FarmAction, FarmObservation, State]):
"""
Client for the Fish Farm OpenEnv Environment.
Maintains a persistent WebSocket connection for efficient multi-step
interactions with a Nile Tilapia RAS aquaculture simulation.
Example (async):
>>> async with FishFarmEnv(base_url="http://localhost:8000") as env:
... result = await env.reset(task_id="feeding_basics")
... while not result.done:
... action = FarmAction(feeding_rate=0.4, aeration_rate=0.6)
... result = await env.step(action)
Example (sync):
>>> env = FishFarmEnv(base_url="http://localhost:8000").sync()
>>> with env:
... result = env.reset(task_id="feeding_basics")
... result = env.step(FarmAction(feeding_rate=0.5))
Example (from HuggingFace):
>>> env = await FishFarmEnv.from_env("rahul24raj/fish-farm-env")
"""
def _step_payload(self, action: FarmAction) -> Dict[str, Any]:
"""Convert FarmAction to JSON payload for step message."""
return {
"feeding_rate": action.feeding_rate,
"aeration_rate": action.aeration_rate,
"heater_setting": action.heater_setting,
"water_exchange_rate": action.water_exchange_rate,
"harvest_decision": action.harvest_decision,
"treatment": action.treatment,
}
def _parse_result(self, payload: Dict[str, Any]) -> StepResult[FarmObservation]:
"""Parse server response into StepResult[FarmObservation]."""
obs_data = payload.get("observation", {})
observation = FarmObservation(**obs_data)
return StepResult(
observation=observation,
reward=payload.get("reward"),
done=payload.get("done", False),
)
def _parse_state(self, payload: Dict[str, Any]) -> State:
"""Parse server response into State object."""
return State(
episode_id=payload.get("episode_id"),
step_count=payload.get("step_count", 0),
)
|