updated frozenlake (b7b6b2cf) · Commits · 张泽凯 / VAGEN

vagen/env/frozenlake/service.py

+0 −194

Original line number	Diff line number	Diff line
		@@ -291,198 +291,4 @@ class FrozenLakeService(BaseService):
		self.environments.pop(env_id, None)
		self.env_configs.pop(env_id, None)

		# FrozenLake specific batch methods - can be kept as additional functionality

		No newline at end of file
		def get_maps_batch(self, env_ids: List[str]) -> Dict[Any, Optional[List[List[str]]]]:
		"""
		Get maps for multiple FrozenLake environments in parallel.

		Args:
		env_ids: A list of environment IDs

		Returns:
		A dictionary mapping each environment ID to its map, or None where maps are not available
		"""
		results = {}

		# Define worker function
		def get_map_single_env(env_id):
		try:
		if env_id not in self.environments:
		return env_id, None, f"Environment {env_id} not found"

		env = self.environments[env_id]
		if not hasattr(env, "gym_env") or not hasattr(env.gym_env, "desc"):
		return env_id, None, "Map not available for this environment"

		# Convert bytes to strings
		map_data = [[cell.decode('utf-8') for cell in row] for row in env.gym_env.desc]
		return env_id, map_data, None
		except Exception as e:
		return env_id, None, str(e)

		# Use ThreadPoolExecutor for parallel retrieval
		with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
		# Submit all retrieval tasks
		futures = {
		executor.submit(get_map_single_env, env_id): env_id
		for env_id in env_ids
		}

		# Process results as they complete
		for future in as_completed(futures):
		env_id = futures[future]
		env_id, result, error = future.result()
		if error:
		print(f"Error getting map for environment {env_id}: {error}")
		results[env_id] = None
		else:
		results[env_id] = result

		return results

		def get_player_positions_batch(self, env_ids: List[str]) -> Dict[Any, Optional[Tuple[int, int]]]:
		"""
		Get player positions for multiple FrozenLake environments in parallel.

		Args:
		env_ids: A list of environment IDs

		Returns:
		A dictionary mapping each environment ID to its player position,
		or None where positions are not available
		"""
		results = {}

		# Define worker function
		def get_position_single_env(env_id):
		try:
		if env_id not in self.environments:
		return env_id, None, f"Environment {env_id} not found"

		env = self.environments[env_id]
		if not hasattr(env, "_get_player_position"):
		return env_id, None, "Player position not available for this environment"

		position = env._get_player_position()
		# Convert to tuple of ints
		position = tuple(int(x) for x in position)
		return env_id, position, None
		except Exception as e:
		return env_id, None, str(e)

		# Use ThreadPoolExecutor for parallel retrieval
		with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
		# Submit all retrieval tasks
		futures = {
		executor.submit(get_position_single_env, env_id): env_id
		for env_id in env_ids
		}

		# Process results as they complete
		for future in as_completed(futures):
		env_id = futures[future]
		env_id, result, error = future.result()
		if error:
		print(f"Error getting player position for environment {env_id}: {error}")
		results[env_id] = None
		else:
		results[env_id] = result

		return results

		def check_success_batch(self, env_ids: List[str]) -> Dict[Any, Optional[bool]]:
		"""
		Check if agents have reached goals in multiple FrozenLake environments in parallel.

		Args:
		env_ids: A list of environment IDs

		Returns:
		A dictionary mapping each environment ID to its success state,
		or None where success check is not available
		"""
		results = {}

		# Define worker function
		def check_success_single_env(env_id):
		try:
		if env_id not in self.environments:
		return env_id, None, f"Environment {env_id} not found"

		env = self.environments[env_id]
		if not hasattr(env, "_success"):
		return env_id, None, "Success check not available for this environment"

		success = env._success()
		return env_id, success, None
		except Exception as e:
		return env_id, None, str(e)

		# Use ThreadPoolExecutor for parallel checks
		with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
		# Submit all check tasks
		futures = {
		executor.submit(check_success_single_env, env_id): env_id
		for env_id in env_ids
		}

		# Process results as they complete
		for future in as_completed(futures):
		env_id = futures[future]
		env_id, result, error = future.result()
		if error:
		print(f"Error checking success for environment {env_id}: {error}")
		results[env_id] = None
		else:
		results[env_id] = result

		return results

		def is_done_batch(self, env_ids: List[str]) -> Dict[Any, Optional[bool]]:
		"""
		Check if episodes are done in multiple FrozenLake environments in parallel.

		Args:
		env_ids: A list of environment IDs

		Returns:
		A dictionary mapping each environment ID to its done state,
		or None where done check is not available
		"""
		results = {}

		# Define worker function
		def is_done_single_env(env_id):
		try:
		if env_id not in self.environments:
		return env_id, None, f"Environment {env_id} not found"

		env = self.environments[env_id]
		if not hasattr(env, "_finished"):
		return env_id, None, "Done check not available for this environment"

		done = env._finished()
		return env_id, done, None
		except Exception as e:
		return env_id, None, str(e)

		# Use ThreadPoolExecutor for parallel checks
		with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
		# Submit all check tasks
		futures = {
		executor.submit(is_done_single_env, env_id): env_id
		for env_id in env_ids
		}

		# Process results as they complete
		for future in as_completed(futures):
		env_id = futures[future]
		env_id, result, error = future.result()
		if error:
		print(f"Error checking done state for environment {env_id}: {error}")
		results[env_id] = None
		else:
		results[env_id] = result

		return results
		No newline at end of file

Admin message