rllib/utils/pre_checks/env.py
def check_multiagent_environments(env: "MultiAgentEnv") -> None:
    """Checking for common errors in RLlib MultiAgentEnvs.

    Args:
        env: The env to be checked.

    """
    from ray.rllib.env import MultiAgentEnv

    if not isinstance(env, MultiAgentEnv):
        raise ValueError("The passed env is not a MultiAgentEnv.")
    elif not (
        hasattr(env, "observation_space")
        and hasattr(env, "action_space")
        and hasattr(env, "_agent_ids")
        and hasattr(env, "_spaces_in_preferred_format")
    ):
        if log_once("ma_env_super_ctor_called"):
            logger.warning(
                f"Your MultiAgentEnv {env} does not have some or all of the needed "
                "base-class attributes! Make sure you call `super().__init__` from "
                "within your MutiAgentEnv's constructor. "
                "This will raise an error in the future."
            )
        return

    reset_obs = env.reset()
    sampled_obs = env.observation_space_sample()
    _check_if_element_multi_agent_dict(env, reset_obs, "reset()")
    _check_if_element_multi_agent_dict(
        env, sampled_obs, "env.observation_space_sample()"
    )

    try:
        env.observation_space_contains(reset_obs)
    except Exception as e:
        raise ValueError(
            "Your observation_space_contains function has some error "
        ) from e

    if not env.observation_space_contains(reset_obs):
        error = (
            _not_contained_error("env.reset", "observation")
            + f"\n\n reset_obs: {reset_obs}\n\n env.observation_space_sample():"
            f" {sampled_obs}\n\n "
        )
        raise ValueError(error)

    if not env.observation_space_contains(sampled_obs):
        error = (
            _not_contained_error("observation_space_sample", "observation")
            + f"\n\n env.observation_space_sample():"
            f" {sampled_obs}\n\n "
        )
        raise ValueError(error)

    sampled_action = env.action_space_sample()
    _check_if_element_multi_agent_dict(env, sampled_action, "action_space_sample")
    try:
        env.action_space_contains(sampled_action)
    except Exception as e:
        raise ValueError("Your action_space_contains function has some error ") from e

    if not env.action_space_contains(sampled_action):
        error = (
            _not_contained_error("action_space_sample", "action")
            + "\n\n sampled_action {sampled_action}\n\n"
        )
        raise ValueError(error)

    next_obs, reward, done, info = env.step(sampled_action)
    _check_if_element_multi_agent_dict(env, next_obs, "step, next_obs")
    _check_if_element_multi_agent_dict(env, reward, "step, reward")
    _check_if_element_multi_agent_dict(env, done, "step, done")
    _check_if_element_multi_agent_dict(env, info, "step, info")
    _check_reward(
        {"dummy_env_id": reward}, base_env=True, agent_ids=env.get_agent_ids()
    )
    _check_done({"dummy_env_id": done}, base_env=True, agent_ids=env.get_agent_ids())
    _check_info({"dummy_env_id": info}, base_env=True, agent_ids=env.get_agent_ids())
    if not env.observation_space_contains(next_obs):
        error = (
            _not_contained_error("env.step(sampled_action)", "observation")
            + ":\n\n next_obs: {next_obs} \n\n sampled_obs: {sampled_obs}"
        )
        raise ValueError(error)
Checking for common errors in RLlib MultiAgentEnvs.

    Args:
        env: The env to be checked.

    
Checking for common errors in RLlib MultiAgentEnvs.
Checking for common errors in RLlib MultiAgentEnvs .
def check_multiagent_environments(env: "MultiAgentEnv") -> None:
    from ray.rllib.env import MultiAgentEnv
    if not isinstance(env, MultiAgentEnv):
        raise ValueError("The passed env is not a MultiAgentEnv.")
    elif not (
        hasattr(env, "observation_space")
        and hasattr(env, "action_space")
        and hasattr(env, "_agent_ids")
        and hasattr(env, "_spaces_in_preferred_format")
    ):
        if log_once("ma_env_super_ctor_called"):
            logger.warning(
                f"Your MultiAgentEnv {env} does not have some or all of the needed "
                "base-class attributes! Make sure you call `super().__init__` from "
                "within your MutiAgentEnv's constructor. "
                "This will raise an error in the future."
            )
        return
    reset_obs = env.reset()
    sampled_obs = env.observation_space_sample()
    _check_if_element_multi_agent_dict(env, reset_obs, "reset()")
    _check_if_element_multi_agent_dict(
        env, sampled_obs, "env.observation_space_sample()"
    )
    try:
        env.observation_space_contains(reset_obs)
    except Exception as e:
        raise ValueError(
            "Your observation_space_contains function has some error "
        ) from e
    if not env.observation_space_contains(reset_obs):
        error = (
            _not_contained_error("env.reset", "observation")
            + f"\n\n reset_obs: {reset_obs}\n\n env.observation_space_sample():"
            f" {sampled_obs}\n\n "
        )
        raise ValueError(error)
    if not env.observation_space_contains(sampled_obs):
        error = (
            _not_contained_error("observation_space_sample", "observation")
            + f"\n\n env.observation_space_sample():"
            f" {sampled_obs}\n\n "
        )
        raise ValueError(error)
    sampled_action = env.action_space_sample()
    _check_if_element_multi_agent_dict(env, sampled_action, "action_space_sample")
    try:
        env.action_space_contains(sampled_action)
    except Exception as e:
        raise ValueError("Your action_space_contains function has some error ") from e
    if not env.action_space_contains(sampled_action):
        error = (
            _not_contained_error("action_space_sample", "action")
            + "\n\n sampled_action {sampled_action}\n\n"
        )
        raise ValueError(error)
    next_obs, reward, done, info = env.step(sampled_action)
    _check_if_element_multi_agent_dict(env, next_obs, "step, next_obs")
    _check_if_element_multi_agent_dict(env, reward, "step, reward")
    _check_if_element_multi_agent_dict(env, done, "step, done")
    _check_if_element_multi_agent_dict(env, info, "step, info")
    _check_reward(
        {"dummy_env_id": reward}, base_env=True, agent_ids=env.get_agent_ids()
    )
    _check_done({"dummy_env_id": done}, base_env=True, agent_ids=env.get_agent_ids())
    _check_info({"dummy_env_id": info}, base_env=True, agent_ids=env.get_agent_ids())
    if not env.observation_space_contains(next_obs):
        error = (
            _not_contained_error("env.step(sampled_action)", "observation")
            + ":\n\n next_obs: {next_obs} \n\n sampled_obs: {sampled_obs}"
        )
        raise ValueError(error)
def check_multiagent_environments ( env : "MultiAgentEnv" ) -> None : from ray . rllib . env import MultiAgentEnv if not isinstance ( env , MultiAgentEnv ) : raise ValueError ( "The passed env is not a MultiAgentEnv." ) elif not ( hasattr ( env , "observation_space" ) and hasattr ( env , "action_space" ) and hasattr ( env , "_agent_ids" ) and hasattr ( env , "_spaces_in_preferred_format" ) ) : if log_once ( "ma_env_super_ctor_called" ) : logger . warning ( f"Your MultiAgentEnv {env} does not have some or all of the needed " "base-class attributes! Make sure you call `super().__init__` from " "within your MutiAgentEnv's constructor. " "This will raise an error in the future." ) return reset_obs = env . reset ( ) sampled_obs = env . observation_space_sample ( ) _check_if_element_multi_agent_dict ( env , reset_obs , "reset()" ) _check_if_element_multi_agent_dict ( env , sampled_obs , "env.observation_space_sample()" ) try : env . observation_space_contains ( reset_obs ) except Exception as e : raise ValueError ( "Your observation_space_contains function has some error " ) from e if not env . observation_space_contains ( reset_obs ) : error = ( _not_contained_error ( "env.reset" , "observation" ) + f"\n\n reset_obs: {reset_obs}\n\n env.observation_space_sample():" f" {sampled_obs}\n\n " ) raise ValueError ( error ) if not env . observation_space_contains ( sampled_obs ) : error = ( _not_contained_error ( "observation_space_sample" , "observation" ) + f"\n\n env.observation_space_sample():" f" {sampled_obs}\n\n " ) raise ValueError ( error ) sampled_action = env . action_space_sample ( ) _check_if_element_multi_agent_dict ( env , sampled_action , "action_space_sample" ) try : env . action_space_contains ( sampled_action ) except Exception as e : raise ValueError ( "Your action_space_contains function has some error " ) from e if not env . action_space_contains ( sampled_action ) : error = ( _not_contained_error ( "action_space_sample" , "action" ) + "\n\n sampled_action {sampled_action}\n\n" ) raise ValueError ( error ) next_obs , reward , done , info = env . step ( sampled_action ) _check_if_element_multi_agent_dict ( env , next_obs , "step, next_obs" ) _check_if_element_multi_agent_dict ( env , reward , "step, reward" ) _check_if_element_multi_agent_dict ( env , done , "step, done" ) _check_if_element_multi_agent_dict ( env , info , "step, info" ) _check_reward ( { "dummy_env_id" : reward } , base_env = True , agent_ids = env . get_agent_ids ( ) ) _check_done ( { "dummy_env_id" : done } , base_env = True , agent_ids = env . get_agent_ids ( ) ) _check_info ( { "dummy_env_id" : info } , base_env = True , agent_ids = env . get_agent_ids ( ) ) if not env . observation_space_contains ( next_obs ) : error = ( _not_contained_error ( "env.step(sampled_action)" , "observation" ) + ":\n\n next_obs: {next_obs} \n\n sampled_obs: {sampled_obs}" ) raise ValueError ( error )
Checking for common errors in RLlib MultiAgentEnvs.
Checking for common errors in RLlib MultiAgentEnvs .
"""Checking for common errors in RLlib MultiAgentEnvs.

    Args:
        env: The env to be checked.

    """
env
"MultiAgentEnv"
The env to be checked.
The env to be checked .