Spaces:
Sleeping
Sleeping
File size: 3,821 Bytes
8c0b3c9 | 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 98 99 100 101 102 103 | import numpy as np
from ai.vector_env_legacy import VectorGameState
from gymnasium import spaces
from stable_baselines3.common.vec_env import VecEnv
class VectorEnvAdapter(VecEnv):
"""
Wraps the LEGACY Numba-accelerated VectorGameState (320 dim).
"""
metadata = {"render_modes": ["rgb_array"]}
def __init__(self, num_envs, observation_space_dim=320, action_space=None):
self.num_envs = num_envs
self.game_state = VectorGameState(num_envs)
# Observation Space size - Flexible Legacy
obs_dim = observation_space_dim
self.observation_space = spaces.Box(low=0, high=1, shape=(obs_dim,), dtype=np.float32)
if action_space is None:
action_space = spaces.Discrete(1000)
self.action_space = action_space
self.actions = None
self.render_mode = None
# Track previous scores for delta-based rewards (Same logic is fine)
self.prev_scores = np.zeros(num_envs, dtype=np.int32)
def reset(self):
self.game_state.reset()
self.prev_scores.fill(0)
return self.game_state.get_observations()
def step_async(self, actions):
self.actions = actions
def step_wait(self):
actions_int32 = self.actions.astype(np.int32)
# Legacy step doesn't support opponent simulation internally usually?
# Checked vector_env_legacy.py: step_vectorized DOES exist.
# But looking at legacy file content:
# It calls batch_apply_action.
# It does NOT call step_opponent_vectorized.
# So legacy environment is "Solitaire" only?
# That means Opponent Score never increases?
# If so, comparing against Random Opponent logic inside New Env is unfair.
# But wait, if Legacy Model was trained in Solitaire, it expects Solitaire.
# If I want to compare "Performance", I should use the same conditions.
# However, the user wants to compare "Checkpoints".
# If legacy checkpoint was trained for "Reach 10 points fast", then benchmark is "Average Turns to 10".
self.game_state.step(actions_int32)
obs = self.game_state.get_observations()
# Rewards (Same logic as modern adapter to ensure fair comparison of metrics?)
current_scores = self.game_state.batch_scores
delta_scores = current_scores - self.prev_scores
rewards = delta_scores.astype(np.float32)
rewards -= 0.001
dones = current_scores >= 10
win_mask = dones & (delta_scores > 0)
rewards[win_mask] += 5.0
self.prev_scores = current_scores.copy()
if np.any(dones):
reset_indices = np.where(dones)[0]
self.game_state.reset(list(reset_indices))
self.prev_scores[reset_indices] = 0
obs = self.game_state.get_observations()
infos = []
for i in range(self.num_envs):
if dones[i]:
infos.append({"terminal_observation": obs[i], "episode": {"r": rewards[i], "l": 10}})
else:
infos.append({})
else:
infos = [{} for _ in range(self.num_envs)]
return obs, rewards, dones, infos
def close(self):
pass
def get_attr(self, attr_name, indices=None):
return []
def set_attr(self, attr_name, value, indices=None):
pass
def env_method(self, method_name, *method_args, **method_kwargs):
return []
def env_is_wrapped(self, wrapper_class, indices=None):
return [False] * self.num_envs
def action_masks(self):
# Legacy env has no masks, return all True
return np.ones((self.num_envs, 1000), dtype=bool)
|