Spaces:
Sleeping
Sleeping
File size: 2,700 Bytes
cb922bf | 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 | """
Asynchronous wrapper for CloudResourceEnv.
This module provides an async interface to the synchronous CloudResourceEnv,
enabling async execution patterns while preserving all environment behavior.
"""
from typing import Dict, Tuple
import numpy as np
from env.environment import CloudResourceEnv
class AsyncEnvWrapper:
"""
Asynchronous wrapper for CloudResourceEnv.
This wrapper provides async methods (reset, step, close) that call the
synchronous CloudResourceEnv methods. It preserves all return values and
maintains the same interface contract as the wrapped environment.
The wrapper does not modify environment logic or introduce actual async I/O.
It simply provides an async interface structure that allows future async
enhancements without interface changes.
"""
def __init__(self, env: CloudResourceEnv):
"""
Initialize the async wrapper.
Args:
env: CloudResourceEnv instance to wrap
"""
self.env = env
async def reset(self) -> np.ndarray:
"""
Async wrapper for env.reset().
Resets the environment to initial state and returns the initial observation.
This method calls the synchronous env.reset() and preserves its return value.
Returns:
Initial observation as numpy array [cpu_util, memory_util, request_rate, allocated_resources]
"""
return self.env.reset()
async def step(self, action: int) -> Tuple[np.ndarray, float, bool, Dict]:
"""
Async wrapper for env.step(action).
Executes one time step with the given action and returns the standard RL tuple.
This method calls the synchronous env.step(action) and preserves all return values.
Args:
action: Action to execute (0: decrease, 1: maintain, 2: increase)
Returns:
observation: Current state after action as numpy array
reward: Reward signal from reward calculator
done: Whether episode has terminated
info: Additional diagnostic information dictionary
Raises:
ValueError: If action is not in valid range {0, 1, 2}
RuntimeError: If step is called after episode termination
"""
return self.env.step(action)
async def close(self):
"""
Async cleanup method.
Performs any necessary cleanup operations. For CloudResourceEnv,
no cleanup is required, so this is a no-op method. This method
never raises exceptions to ensure clean shutdown.
"""
pass
|