Spaces:
Sleeping
Sleeping
File size: 3,799 Bytes
be5d908 79412de be5d908 ba6f178 be5d908 79412de be5d908 781dc5d 79412de be5d908 781dc5d 79412de 781dc5d be5d908 79412de be5d908 79412de be5d908 79412de | 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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | from __future__ import annotations
import threading
from abc import ABC, abstractmethod
from typing import Any, Dict, Tuple
from env.core import ThermalPlantEnv
class OpenEnvInterface(ABC):
"""
Abstract base class defining the contract for an OpenEnv-compliant interface.
This ensures that the API layer can interact with any environment that
adheres to this standard structure.
"""
@abstractmethod
def get_state(self) -> Dict[str, Any]:
"""Returns the full, unrounded internal state of the environment."""
raise NotImplementedError
@abstractmethod
def reset(self, task_id: str, episode_id: int) -> Dict[str, float]:
"""Resets the environment to a new initial state for a given task and episode."""
raise NotImplementedError
@abstractmethod
def step(self, action: Dict[str, float]) -> Tuple[Dict[str, float], float, bool, Dict[str, Any]]:
"""
Executes one time step in the environment.
Args:
action: A dictionary containing the agent's action.
Returns:
A tuple containing:
- observation (Dict[str, float]): The agent's observation of the current environment state.
- reward (float): The amount of reward returned after the previous action.
- done (bool): Whether the episode has ended.
- info (Dict[str, Any]): Contains auxiliary diagnostic information.
"""
raise NotImplementedError
class ConcreteOpenEnvInterface(OpenEnvInterface):
"""
Concrete implementation of the OpenEnvInterface.
This class provides a thread-safe singleton wrapper for the core thermal
plant environment. It decouples the API layer from the simulation
internals, ensuring consistent state management across sequential
evaluation tasks.
"""
_instance: "ConcreteOpenEnvInterface" | None = None
_env: ThermalPlantEnv | None = None
_lock: threading.RLock = threading.RLock()
def __new__(cls, max_steps: int | None = None) -> "ConcreteOpenEnvInterface":
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
if max_steps is None:
cls._env = ThermalPlantEnv()
else:
cls._env = ThermalPlantEnv(max_steps=int(max_steps))
elif cls._env is not None and max_steps is not None:
cls._env.max_steps = int(max_steps)
return cls._instance
def __init__(self, max_steps: int | None = None) -> None:
"""Allow non-API callers to configure the shared env without bypassing the interface."""
with self._lock:
if self._env is not None and max_steps is not None:
self._env.max_steps = int(max_steps)
def get_state(self) -> Dict[str, Any]:
"""Returns the full, unrounded internal state from the core environment."""
with self._lock:
if self._env is None:
raise RuntimeError("Environment not initialized.")
return self._env.state()
def reset(self, task_id: str, episode_id: int) -> Dict[str, float]:
"""Resets the core environment and returns the initial observation."""
with self._lock:
if self._env is None:
raise RuntimeError("Environment not initialized.")
return self._env.reset(task_id=task_id, episode_id=episode_id)
def step(self, action: Dict[str, float]) -> Tuple[Dict[str, float], float, bool, Dict[str, Any]]:
"""Performs a step in the core environment."""
with self._lock:
if self._env is None:
raise RuntimeError("Environment not initialized.")
return self._env.step(action)
|