| from typing import Dict |
| from openenv.core import EnvClient |
| from openenv.core.client_types import StepResult |
| from openenv.core.env_server.types import State |
|
|
| from .models import BatteryAction, BatteryObservation |
|
|
| class BatteryEnv(EnvClient[BatteryAction, BatteryObservation, State]): |
| """ |
| Client for the Battery Environment. |
| """ |
| def _step_payload(self, action: BatteryAction) -> Dict: |
| return { |
| "market_choice": action.market_choice, |
| "p_fraction": action.p_fraction, |
| } |
|
|
| def _parse_result(self, payload: Dict) -> StepResult[BatteryObservation]: |
| obs_data = payload.get("observation", {}) |
| observation = BatteryObservation( |
| energy_price=obs_data.get("energy_price", 0.0), |
| fcr_price=obs_data.get("fcr_price", 0.0), |
| soc=obs_data.get("soc", 0.0), |
| reward=payload.get("reward"), |
| done=payload.get("done", False), |
| metadata=obs_data.get("metadata", {}), |
| ) |
| return StepResult( |
| observation=observation, |
| reward=payload.get("reward"), |
| done=payload.get("done", False), |
| ) |
|
|
| def _parse_state(self, payload: Dict) -> State: |
| return State( |
| episode_id=payload.get("episode_id"), |
| step_count=payload.get("step_count", 0), |
| ) |
|
|
