Agent Manager API¶

# opencda_marl/core/agent_manager.py
class MARLAgentManager:
    """
    Manages all vehicles and their agents in MARL scenarios.

    Features:
    - Processes TrafficManager spawn events
    - Creates vehicles with configured agent types
    - Coordinates multi-agent simulation steps
    - Handles vehicle lifecycle and cleanup
    - Integrates with collision detection system
    """

    def __init__(self, config: Dict[str, Any], state: Dict[str, Any],
                 world: carla.World, cav_world):
        """
        Initialize Agent Manager.

        Parameters
        ----------
        config : dict
            Agent configuration including agent_type and behavior settings
        state : dict
            Shared simulation state (traffic events, map data)
        world : carla.World
            CARLA world instance
        cav_world : CavWorld
            CAV world instance for vehicle management
        """

def step_simulation(self, external_actions: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """
    Execute one simulation step for all agents.

    Process:
    1. Pull spawn events from traffic manager
    2. Spawn new vehicles with configured agents
    3. Update all vehicle states and run control steps
    4. Apply vehicle controls and handle collisions
    5. Clean up destroyed vehicles

    Parameters
    ----------
    external_actions : dict, optional
        External actions for agents (e.g., from MARL algorithms)
        Format: {agent_id: target_speed} or {agent_id: {'target_speed': value}}

    Returns
    -------
    dict
        Step statistics including:
        - step_count: Current simulation step
        - spawn: Spawn attempt statistics
        - control: Vehicle control statistics
        - active_agents: List of active agent IDs
        - total_actors: Total number of active vehicles

    Example
    -------
    # Autonomous mode
    stats = agent_manager.step_simulation()

    # With external control
    external_actions = {
        'agent_001': 25.0,  # Target speed in km/h
        'agent_002': {'target_speed': 30.0}
    }
    stats = agent_manager.step_simulation(external_actions)
    """

def get_agent_info(self) -> Dict[str, Any]:
    """
    Get comprehensive agent system information.

    Returns
    -------
    dict
        Agent system statistics:
        - total_vehicles: Number of spawned vehicles
        - total_agents: Number of active agents (equals total_vehicles in MARL)
        - vehicles_without_agents: Should be 0 in MARL mode
        - agent_coverage_ratio: Should be 1.0 in MARL mode
        - collision_enabled_vehicles: Vehicles with collision detection
        - collision_sensor_coverage: Collision detection coverage ratio

    Example
    -------
    info = agent_manager.get_agent_info()
    print(f"Active agents: {info['total_agents']}")
    print(f"Coverage: {info['agent_coverage_ratio']*100:.1f}%")
    """

def get_collision_debug_info(self) -> Dict[str, Any]:
    """
    Get detailed collision system debug information.

    Returns
    -------
    dict
        Collision system debug info:
        - collision_marked_for_destruction: Vehicles marked for collision removal
        - vehicle_adapters: Per-vehicle adapter information
    """

def reset_episode(self) -> None:
    """
    Reset agent manager for new episode.

    Resets step counter while preserving active vehicles.
    Used for multi-episode training scenarios.
    """

def cleanup(self) -> None:
    """
    Clean up all vehicles and adapters.

    Called at simulation end to properly destroy all vehicles
    and clean up resources.
    """

# opencda_marl/core/agents/agent_factory.py
class AgentFactory:
    """
    Factory class for creating different agent types.

    Supported Agent Types:
    - behavior: Standard OpenCDA BehaviorAgent (fallback)
    - vanilla: Enhanced collision avoidance agent
    - rule_based: 3-stage rule-based intersection agent
    - marl: MARL agent with RL speed control
    """

    @staticmethod
    def create_agent(agent_type: str, vehicle: carla.Vehicle, 
                    carla_map: carla.Map, behavior_config: Dict[str, Any],
                    **kwargs) -> BehaviorAgent:
        """Create agent based on specified type."""

    @staticmethod
    def get_available_types() -> List[str]:
        """Get all available agent types."""
        return ["vanilla", "rule_based", "marl", "behavior"]

    @staticmethod
    def get_baseline_types() -> List[str]:
        """Get baseline agent types for benchmarking."""
        return ["behavior", "vanilla", "rule_based"]

# Example of runtime agent creation
def create_agent_for_vehicle(vehicle, config):
    agent_type = config.get("agent_type", "behavior")
    behavior_config = config.get("agent_behavior", {})

    # Create agent through factory
    agent = AgentFactory.create_agent(
        agent_type=agent_type,
        vehicle=vehicle,
        carla_map=carla_map,
        behavior_config=behavior_config,
        **config.get("agent_kwargs", {})
    )

    return agent

# Initialize agent manager with multiple agent types
config = {
    "agent_type": "vanilla",
    "agent_behavior": {
        "max_speed": 35,
        "safety_time": 3.0,
        "emergency_param": 0.4
    },
    "vanilla": {
        "intersection_safety_multiplier": 2.0,
        "multi_vehicle_ttc": True,
        "max_vehicles_to_track": 5
    }
}

agent_manager = MARLAgentManager(
    traffic_manager=traffic_manager,
    config=config,
    cav_world=cav_world
)

# Multi-agent simulation with external actions
def run_multi_agent_simulation():
    for step in range(1000):
        # Get external actions from RL algorithms
        external_actions = {}

        # Get current observations
        agent_info = agent_manager.get_agent_info()
        active_agents = agent_info.get('active_agents', [])

        # Generate actions for each agent
        for agent_id in active_agents:
            # Example: RL policy action
            observation = get_agent_observation(agent_id)
            action = rl_policy.get_action(observation)
            external_actions[agent_id] = action

        # Execute simulation step
        step_stats = agent_manager.step_simulation(external_actions)

        # Log statistics
        print(f"Step {step}: {step_stats['total_actors']} active vehicles")
        if step_stats['control']['collision_destroyed'] > 0:
            print(f"Collisions: {step_stats['control']['collision_destroyed']}")

        # World step
        world.tick()

# Monitor agent performance during simulation
def monitor_agent_performance(agent_manager):
    info = agent_manager.get_agent_info()
    collision_info = agent_manager.get_collision_debug_info()

    print("=== Agent System Status ===")
    print(f"Active Vehicles: {info['total_vehicles']}")
    print(f"Active Agents: {info['total_agents']}")
    print(f"Agent Coverage: {info['agent_coverage_ratio']*100:.1f}%")
    print(f"Collision Detection Coverage: {info['collision_sensor_coverage']*100:.1f}%")

    print("\n=== Collision System ===")
    print(f"Vehicles Marked for Destruction: {collision_info['collision_marked_for_destruction']}")

    return info

# Multi-episode training with agent manager
def run_training_episodes(num_episodes=100):
    for episode in range(num_episodes):
        print(f"\nEpisode {episode + 1}/{num_episodes}")

        # Reset for new episode
        agent_manager.reset_episode()

        episode_data = []
        for step in range(300):  # 5-minute episodes
            # Collect observations
            observations = collect_observations(agent_manager)

            # Get actions from RL algorithms
            actions = get_rl_actions(observations)

            # Execute step
            step_stats = agent_manager.step_simulation(actions)

            # Store experience
            episode_data.append({
                'observations': observations,
                'actions': actions,
                'step_stats': step_stats
            })

            # Check episode termination
            if should_terminate_episode(step_stats):
                break

        # Process episode data for training
        process_episode_data(episode_data)

        # Episode cleanup (vehicles auto-cleaned between episodes)
        print(f"Episode completed with {len(episode_data)} steps")

# Agent manager integration with MARL environment
class MARLEnvironment:
    def __init__(self, scenario_manager):
        self.scenario_manager = scenario_manager
        self.agent_manager = scenario_manager.agent_manager

    def step(self, actions):
        # Pass actions to agent manager
        step_stats = self.agent_manager.step_simulation(actions)

        # Get observations from active agents
        observations = self._get_observations()
        rewards = self._calculate_rewards(step_stats)
        dones = self._check_termination(step_stats)

        return observations, rewards, dones, step_stats

# Benchmark integration for performance testing
def run_agent_benchmark(agent_types, scenarios):
    results = {}

    for agent_type in agent_types:
        for scenario in scenarios:
            # Configure agent manager with specific type
            config = load_config(scenario)
            config['agent_type'] = agent_type

            # Run test
            agent_manager = MARLAgentManager(traffic_manager, config, cav_world)
            performance = run_test_scenario(agent_manager, scenario)

            results[f"{agent_type}_{scenario}"] = performance

    return results