Adapters API

Adapter classes for different environment types. Adapters bridge the gap between raw environment APIs and MOSAIC’s unified operator interface, translating observations and actions so that any agent type (LLM, RL, Human, or Random) can interact with any supported environment.

EnvironmentAdapter

class gym_gui.core.adapters.base.EnvironmentAdapter(context: AdapterContext | None = None)

Lifecycle contract for all Gymnasium environment adapters.

id

The Gymnasium environment ID (e.g., “CartPole-v1”).

Type:

str

supported_control_modes

Control modes this adapter supports.

Type:

tuple[gym_gui.core.enums.ControlMode, …]

supported_render_modes

Render modes this adapter supports.

Type:

tuple[gym_gui.core.enums.RenderMode, …]

default_render_mode

Default render mode for this adapter.

Type:

gym_gui.core.enums.RenderMode

stepping_paradigm

The RL stepping paradigm (default: SINGLE_AGENT).

Type:

gym_gui.core.enums.SteppingParadigm

property action_space: gymnasium.Space.typing.Any

apply_mouse_delta(delta_x: float, delta_y: float) → None

Apply mouse movement deltas for FPS-style control.

Subclasses override to queue or immediately apply mouse motion. The base implementation is a no-op to keep Pylance satisfied when a generic EnvironmentAdapter is referenced.

apply_wrappers(env: gymnasium.Env.(typing.Any, typing.Any)) -> gymnasium.Env.(typing.Any, typing.Any)

Hook for subclasses to apply Gymnasium wrappers before use.

bind(context: AdapterContext) → None

Bind the adapter to a runtime context after instantiation.

build_frame_reference(render_payload: Any | None, state: StepState) → str | None

Optional hook to derive an external frame reference for media pipelines.

build_render_hint(observation: ObservationT, info: Mapping[str, Any], state: StepState) → Mapping[str, Any] | None

Return lightweight render metadata for downstream consumers.

build_step_state(observation: ObservationT, info: Mapping[str, Any]) → StepState

Construct the canonical StepState for the current step.

close() → None

property context: AdapterContext | None

default_render_mode: RenderMode

elapsed_steps() → int | None

ensure_control_mode(mode: ControlMode) → None

gym_kwargs() → dict[str, Any]

Keyword arguments forwarded to gymnasium.make().

has_mouse_delta_support() → bool

Return True if the adapter supports FPS-style mouse delta control.

Subclasses that provide mouse-look controls (e.g., ViZDoom) should override this to return True.

id: str

load() → None

Instantiate underlying Gymnasium environment resources.

property logger: Logger

Return the module-level logger for backward compatibility.

property observation_space: gymnasium.Space.typing.Any

render() → Any

reset(*, seed: int | None = None, options: dict[str, Any] | None = None) → AdapterStep[ObservationT]

property settings: Any | None

property space_signature: Mapping[str, Any] | None

step(action: ActionT) → AdapterStep[ObservationT]

stepping_paradigm: SteppingParadigm = 'single_agent'

supported_control_modes: tuple[ControlMode, ...]

supported_render_modes: tuple[RenderMode, ...] = ()

supports_control_mode(mode: ControlMode) → bool

supports_render_mode(mode: RenderMode) → bool

telemetry_payload_version() → int

Version marker for downstream telemetry consumers.

property vector_metadata: Mapping[str, Any] | None

PettingZooAdapter

class gym_gui.core.adapters.pettingzoo.PettingZooAdapter(context: AdapterContext | None = None, *, config: PettingZooConfig | None = None, env_id: PettingZooEnvId | str | None = None)

Unified adapter for PettingZoo AEC and Parallel environments.

This adapter provides a consistent interface for multi-agent environments, supporting human control for turn-based games and AI control for all others.

id

Environment identifier (e.g., “chess_v6”)

supported_control_modes

Tuple of supported control modes

default_render_mode

Default rendering mode (RGB_ARRAY)

property agents: List[str]

Get list of active agent names.

build_step_state(observation: Any, info: Mapping[str, Any]) → StepState

Construct the canonical StepState for the current step.

close() → None

Close the environment.

property current_agent: str | None

Get current agent (for AEC mode).

default_render_mode: RenderMode = 'rgb_array'

get_action_mask(agent: str | None = None) → numpy.ndarray | None

Get action mask for an agent.

Parameters:

agent – Agent name (uses current agent if None)

Returns:

Action mask array or None if not available

get_action_space(agent: str | None = None)

Get action space for an agent.

Parameters:

agent – Agent name (uses current agent if None for AEC)

Returns:

Gymnasium Space

get_human_agent() → str | None

Get the agent designated for human control.

get_observation_space(agent: str | None = None)

Get observation space for an agent.

Parameters:

agent – Agent name (uses current agent if None for AEC)

Returns:

Gymnasium Space

is_done() → bool

Check if episode is complete.

is_human_controllable() → bool

Check if this environment supports human control.

property is_parallel: bool

Check if using Parallel API (vs AEC).

load() → None

Instantiate the PettingZoo environment.

property num_agents: int

Get number of agents.

property possible_agents: List[str]

Get list of all possible agent names.

render() → Any

Render the environment.

Returns:

RGB array if render_mode is “rgb_array”, else None

reset(*, seed: int | None = None, options: dict[str, Any] | None = None) → AdapterStep[Any]

Reset the environment.

Parameters:

• seed – Optional random seed

• options – Additional reset options

Returns:

Initial step result

sample_action(agent: str | None = None) → Any

Sample a random action for an agent.

Parameters:

agent – Agent name (uses current agent if None for AEC)

Returns:

Sampled action

sample_actions() → Dict[str, Any]

Sample random actions for all active agents.

Returns:

Dict mapping agent names to sampled actions

step(action: Any) → AdapterStep[Any]

Execute action(s) in the environment.

For AEC environments, pass a single action for the current agent. For Parallel environments, pass a dict mapping agent names to actions.

Parameters:

action – Single action (AEC) or dict of actions (Parallel)

Returns:

Step result

property stepping_paradigm: SteppingParadigm

Return the stepping paradigm based on environment type.

Returns:

SIMULTANEOUS for Parallel API environments, SEQUENTIAL for AEC environments.

supported_render_modes: tuple[RenderMode, ...] = (RenderMode.RGB_ARRAY,)

ParadigmAdapter

class gym_gui.core.adapters.paradigm.ParadigmAdapter

Abstract base class for paradigm-specific stepping behavior.

ParadigmAdapter bridges between Mosaic’s GUI/orchestrator and paradigm-specific workers. It abstracts: 1. Which agents need actions at any given time 2. How to execute a step (single action vs. joint action dict) 3. How to normalize results across paradigms

Subclasses implement paradigm-specific logic: - SingleAgentAdapter: Gymnasium-style single agent - SequentialAdapter: PettingZoo AEC-style turn-based - SimultaneousAdapter: PettingZoo Parallel / RLlib POSG-style

Example

>>> adapter = get_paradigm_adapter(env)
>>> while not adapter.is_done():
...     agents = adapter.get_agents_to_act()
...     actions = {a: policy(a, adapter.get_observation(a)) for a in agents}
...     result = adapter.step(actions)

See also

• Stepping Paradigms for paradigm details

abstract property agent_ids: Sequence[str]

All agent identifiers in this environment.

For single-agent environments, returns [“agent_0”]. For multi-agent environments, returns all agent IDs.

abstract close() → None

Clean up environment resources.

abstract get_agents_to_act() → List[str]

Return agents that need actions NOW.

Returns:

List of agent IDs that require actions in the current step.

• SINGLE_AGENT: [“agent_0”]

• SEQUENTIAL: [current_agent_id] (one agent per step)

• SIMULTANEOUS: [all active agent IDs]

get_info(agent_id: str) → Dict[str, Any]

Get the info dict for a specific agent.

Default implementation returns empty dict. Subclasses should override.

get_infos(agent_ids: List[str] | None = None) → Dict[str, Dict[str, Any]]

Get info dicts for multiple agents.

Default implementation calls get_info for each agent.

abstract get_observation(agent_id: str) → Any

Get the current observation for a specific agent.

Parameters:

agent_id – The agent to get observation for.

Returns:

The observation for the specified agent.

Raises:

KeyError – If agent_id is not valid.

abstract get_observations(agent_ids: List[str] | None = None) → Dict[str, Any]

Get observations for multiple agents.

Parameters:

agent_ids – List of agent IDs. If None, returns all active agents.

Returns:

Dict mapping agent_id to observation.

abstract is_done() → bool

Check if the episode is complete for all agents.

is_sequential() → bool

Check if this is a sequential (AEC) environment.

is_simultaneous() → bool

Check if this is a simultaneous (POSG) environment.

is_single_agent() → bool

Check if this is a single-agent environment.

num_agents() → int

Return the number of agents in the environment.

abstract property paradigm: SteppingParadigm

The stepping paradigm this adapter implements.

render() → Any

Render the environment.

Default implementation returns None. Subclasses should override.

abstract reset(*, seed: int | None = None, options: Dict[str, Any] | None = None) → ParadigmStepResult

Reset the environment and return initial observations.

Parameters:

• seed – Optional random seed for reproducibility.

• options – Optional reset options dict.

Returns:

ParadigmStepResult with initial observations (rewards=0, done=False).

abstract step(actions: Dict[str, Any]) → ParadigmStepResult

Execute paradigm-appropriate step.

Parameters:

actions – Dict mapping agent_id to action. - SINGLE_AGENT: {“agent_0”: action} - SEQUENTIAL: {current_agent: action} - SIMULTANEOUS: {agent_id: action for all active agents}

Returns:

ParadigmStepResult with normalized observations, rewards, etc.