Experimental Features

# Main experimental package
from mesa import experimental

# Submodules
from mesa.experimental import devs
from mesa.experimental import mesa_signals  
from mesa.experimental import meta_agents
from mesa.experimental import continuous_space

# Visualization (experimental)
from mesa.visualization import SolaraViz, JupyterViz

from mesa.experimental.devs import DEVSModel, DEVSAgent, Event, EventScheduler

class Event:
    """
    Discrete event with timing and payload information.
    
    Events represent state changes that occur at specific simulation times,
    enabling more efficient simulation of systems with sparse activity.
    """
    
    def __init__(self, time: float, event_type: str, data: dict = None):
        """
        Initialize an event.
        
        Parameters:
            time: When the event should occur (simulation time)
            event_type: Type identifier for the event
            data: Additional event payload data
        """
        ...
    
    @property
    def time(self) -> float:
        """Event occurrence time."""
        ...
    
    @property
    def event_type(self) -> str:
        """Event type identifier."""
        ...
    
    @property
    def data(self) -> dict:
        """Event data payload."""
        ...

class EventScheduler:
    """
    Priority queue-based event scheduler for DEVS simulation.
    
    Manages the event queue and provides efficient scheduling
    and execution of discrete events.
    """
    
    def __init__(self):
        """Initialize empty event scheduler."""
        ...
    
    def schedule_event(self, event: Event):
        """
        Schedule an event for future execution.
        
        Parameters:
            event: Event to schedule
        """
        ...
    
    def schedule_in(self, delay: float, event_type: str, data: dict = None) -> Event:
        """
        Schedule an event after a delay from current time.
        
        Parameters:
            delay: Time delay from current simulation time
            event_type: Type of event to schedule
            data: Event data payload
        
        Returns:
            Scheduled Event object
        """
        ...
    
    def get_next_event(self) -> Event | None:
        """
        Get and remove the next scheduled event.
        
        Returns:
            Next event to process, or None if queue is empty
        """
        ...
    
    def peek_next_event(self) -> Event | None:
        """
        Peek at next event without removing it.
        
        Returns:
            Next event or None if queue is empty
        """
        ...
    
    @property
    def current_time(self) -> float:
        """Current simulation time."""
        ...

class DEVSAgent(Agent):
    """
    Agent class for discrete event simulation models.
    
    Extends basic Agent with event handling capabilities and
    lifecycle management for event-driven simulation.
    """
    
    def __init__(self, model: 'DEVSModel'):
        """
        Initialize DEVS agent.
        
        Parameters:
            model: The DEVS model containing this agent
        """
        super().__init__(model)
        ...
    
    def handle_event(self, event: Event):
        """
        Process an incoming event.
        
        This method should be overridden by subclasses to define
        agent-specific event handling behavior.
        
        Parameters:
            event: Event to process
        """
        ...
    
    def schedule_event(self, delay: float, event_type: str, data: dict = None) -> Event:
        """
        Schedule an event for this agent.
        
        Parameters:
            delay: Time delay from current time
            event_type: Type of event to schedule
            data: Event data payload
        
        Returns:
            Scheduled Event object
        """
        return self.model.scheduler.schedule_in(delay, event_type, data)
    
    def on_activate(self):
        """Called when agent is activated in the simulation."""
        ...
    
    def on_deactivate(self):
        """Called when agent is deactivated/removed."""
        ...

class DEVSModel(Model):
    """
    Model class for discrete event simulation.
    
    Manages event-driven simulation with precise timing and
    efficient execution of sparse events.
    """
    
    def __init__(self, **kwargs):
        """
        Initialize DEVS model.
        
        Parameters:
            **kwargs: Model configuration parameters
        """
        super().__init__(**kwargs)
        self.scheduler = EventScheduler()
        ...
    
    def step(self):
        """
        Execute one simulation step (process next event).
        
        In DEVS, a step processes the next scheduled event
        rather than advancing by a fixed time increment.
        """
        event = self.scheduler.get_next_event()
        if event:
            self._process_event(event)
        else:
            self.running = False  # No more events
    
    def _process_event(self, event: Event):
        """
        Process a discrete event.
        
        Parameters:
            event: Event to process
        """
        # Update simulation time
        self.scheduler.current_time = event.time
        
        # Handle model-level events
        if hasattr(self, f'handle_{event.event_type}'):
            handler = getattr(self, f'handle_{event.event_type}')
            handler(event)
    
    def run_until(self, end_time: float):
        """
        Run simulation until specified time.
        
        Parameters:
            end_time: Simulation time to run until
        """
        while (self.running and 
               self.scheduler.peek_next_event() and 
               self.scheduler.peek_next_event().time <= end_time):
            self.step()
    
    @property
    def current_time(self) -> float:
        """Current simulation time."""
        return self.scheduler.current_time

from mesa.experimental.devs import DEVSModel, DEVSAgent, Event
import random

class Customer(DEVSAgent):
    """Customer agent that generates service requests."""
    
    def __init__(self, model, arrival_rate=1.0):
        super().__init__(model)
        self.arrival_rate = arrival_rate
        self.service_time = 0
        
    def on_activate(self):
        """Schedule first arrival."""
        inter_arrival_time = random.expovariate(self.arrival_rate)
        self.schedule_event(inter_arrival_time, "arrival", {"customer_id": self.unique_id})
    
    def handle_event(self, event):
        if event.event_type == "arrival":
            # Process service request
            self.model.handle_customer_arrival(self)
            
            # Schedule next arrival
            inter_arrival_time = random.expovariate(self.arrival_rate)
            self.schedule_event(inter_arrival_time, "arrival", {"customer_id": self.unique_id})
        
        elif event.event_type == "service_complete":
            # Customer leaves system
            self.service_time = event.data.get("service_duration", 0)
            self.model.log_service_completion(self)

class Server(DEVSAgent):
    """Server agent that processes customer requests."""
    
    def __init__(self, model, service_rate=2.0):
        super().__init__(model)
        self.service_rate = service_rate
        self.busy = False
        self.queue = []
        self.customers_served = 0
    
    def handle_event(self, event):
        if event.event_type == "service_request":
            customer = event.data["customer"]
            
            if not self.busy:
                # Start service immediately
                self._start_service(customer)
            else:
                # Add to queue
                self.queue.append(customer)
        
        elif event.event_type == "service_complete":
            self.busy = False
            self.customers_served += 1
            
            # Notify customer
            customer = event.data["customer"] 
            customer_event = Event(
                self.model.current_time, 
                "service_complete",
                {"service_duration": event.data["service_duration"]}
            )
            customer.handle_event(customer_event)
            
            # Serve next customer if any
            if self.queue:
                next_customer = self.queue.pop(0)
                self._start_service(next_customer)
    
    def _start_service(self, customer):
        """Start serving a customer."""
        self.busy = True
        service_duration = random.expovariate(self.service_rate)
        
        self.schedule_event(service_duration, "service_complete", {
            "customer": customer,
            "service_duration": service_duration
        })

class QueuingModel(DEVSModel):
    """Queueing system simulation using DEVS."""
    
    def __init__(self, n_customers=5, n_servers=2, arrival_rate=1.0, service_rate=2.0):
        super().__init__()
        
        self.arrival_rate = arrival_rate
        self.service_rate = service_rate
        
        # Create servers
        self.servers = [Server(self, service_rate) for _ in range(n_servers)]
        for server in self.servers:
            self.register_agent(server)
        
        # Create customers
        for i in range(n_customers):
            customer = Customer(self, arrival_rate)
            self.register_agent(customer)
            customer.on_activate()
        
        # Statistics tracking
        self.total_arrivals = 0
        self.total_completions = 0
        self.service_times = []
        
        self.running = True
    
    def handle_customer_arrival(self, customer):
        """Handle customer arrival - assign to server."""
        self.total_arrivals += 1
        
        # Find server with shortest queue
        best_server = min(self.servers, key=lambda s: len(s.queue) if s.busy else 0)
        
        # Send service request to server
        request_event = Event(
            self.current_time,
            "service_request", 
            {"customer": customer}
        )
        best_server.handle_event(request_event)
    
    def log_service_completion(self, customer):
        """Log completed service."""
        self.total_completions += 1
        self.service_times.append(customer.service_time)
    
    def get_statistics(self):
        """Get simulation statistics."""
        return {
            "total_arrivals": self.total_arrivals,
            "total_completions": self.total_completions,
            "avg_service_time": sum(self.service_times) / len(self.service_times) if self.service_times else 0,
            "server_utilization": [s.customers_served / self.current_time * self.service_rate 
                                 for s in self.servers]
        }

# Run DEVS simulation
model = QueuingModel(n_customers=10, n_servers=3, arrival_rate=0.8, service_rate=1.5)
model.run_until(100.0)  # Run for 100 time units

stats = model.get_statistics()
print(f"Simulation completed at time {model.current_time}")
print(f"Arrivals: {stats['total_arrivals']}, Completions: {stats['total_completions']}")
print(f"Average service time: {stats['avg_service_time']:.2f}")
print(f"Server utilizations: {[f'{u:.2%}' for u in stats['server_utilization']]}")

from mesa.experimental.mesa_signals import ObservableAgentSet, Signal, Observer

class Signal:
    """
    Signal system for reactive communication between model components.
    
    Enables decoupled communication through publisher-subscriber patterns
    and reactive updates based on state changes.
    """
    
    def __init__(self, name: str):
        """
        Initialize signal.
        
        Parameters:
            name: Signal identifier
        """
        ...
    
    def connect(self, handler: callable, sender=None):
        """
        Connect a handler to this signal.
        
        Parameters:
            handler: Function to call when signal is emitted
            sender: Optional specific sender to listen for
        """
        ...
    
    def disconnect(self, handler: callable, sender=None):
        """
        Disconnect a handler from this signal.
        
        Parameters:
            handler: Handler function to disconnect
            sender: Optional specific sender
        """
        ...
    
    def emit(self, sender=None, **kwargs):
        """
        Emit the signal to all connected handlers.
        
        Parameters:
            sender: Object emitting the signal
            **kwargs: Signal data payload
        """
        ...

class ObservableAgentSet(AgentSet):
    """
    AgentSet with reactive capabilities and change notifications.
    
    Extends standard AgentSet with signal emission for collection
    changes, enabling reactive responses to agent additions/removals.
    """
    
    def __init__(self, agents=None, random=None):
        """
        Initialize observable agent set.
        
        Parameters:
            agents: Initial agents to include
            random: Random number generator
        """
        super().__init__(agents, random)
        
        # Signals for collection changes
        self.agent_added = Signal("agent_added")
        self.agent_removed = Signal("agent_removed") 
        self.collection_changed = Signal("collection_changed")
    
    def add(self, agent):
        """Add agent and emit signals."""
        super().add(agent)
        self.agent_added.emit(sender=self, agent=agent)
        self.collection_changed.emit(sender=self, change_type="add", agent=agent)
    
    def remove(self, agent):
        """Remove agent and emit signals."""
        super().remove(agent)
        self.agent_removed.emit(sender=self, agent=agent)
        self.collection_changed.emit(sender=self, change_type="remove", agent=agent)

class Observer:
    """
    Base class for reactive observers that respond to signals.
    
    Provides infrastructure for objects that need to react to
    changes in model state or agent collections.
    """
    
    def __init__(self, name: str = None):
        """
        Initialize observer.
        
        Parameters:
            name: Optional observer identifier
        """
        self.name = name
        self._connections = []
    
    def observe(self, signal: Signal, handler: callable, sender=None):
        """
        Start observing a signal.
        
        Parameters:
            signal: Signal to observe
            handler: Method to call when signal emits
            sender: Optional specific sender to observe
        """
        signal.connect(handler, sender)
        self._connections.append((signal, handler, sender))
    
    def stop_observing(self, signal: Signal = None):
        """
        Stop observing signals.
        
        Parameters:
            signal: Specific signal to stop observing (None for all)
        """
        if signal:
            connections_to_remove = [c for c in self._connections if c[0] == signal]
            for conn in connections_to_remove:
                signal.disconnect(conn[1], conn[2])
                self._connections.remove(conn)
        else:
            # Disconnect all
            for signal, handler, sender in self._connections:
                signal.disconnect(handler, sender)
            self._connections.clear()

from mesa import Agent, Model
from mesa.experimental.mesa_signals import ObservableAgentSet, Signal, Observer

class ReactiveAgent(Agent):
    """Agent that emits signals for state changes."""
    
    def __init__(self, model, wealth=100):
        super().__init__(model)
        self.wealth = wealth
        self._previous_wealth = wealth
        
        # Signals for state changes
        self.wealth_changed = Signal("wealth_changed")
        self.became_wealthy = Signal("became_wealthy")
        self.became_poor = Signal("became_poor")
    
    def step(self):
        # Simple wealth dynamics
        change = self.random.randint(-10, 15)
        self.wealth += change
        self.wealth = max(0, self.wealth)
        
        # Emit wealth change signal
        if self.wealth != self._previous_wealth:
            self.wealth_changed.emit(
                sender=self, 
                old_wealth=self._previous_wealth,
                new_wealth=self.wealth,
                change=change
            )
            
            # Emit status change signals
            if self._previous_wealth < 200 and self.wealth >= 200:
                self.became_wealthy.emit(sender=self, wealth=self.wealth)
            elif self._previous_wealth >= 50 and self.wealth < 50:
                self.became_poor.emit(sender=self, wealth=self.wealth)
        
        self._previous_wealth = self.wealth

class WealthTracker(Observer):
    """Observer that tracks wealth statistics."""
    
    def __init__(self):
        super().__init__("WealthTracker")
        self.wealth_changes = []
        self.wealthy_agents = set()
        self.poor_agents = set()
    
    def on_wealth_change(self, sender, old_wealth, new_wealth, change, **kwargs):
        """Handle wealth change events."""
        self.wealth_changes.append({
            'agent_id': sender.unique_id,
            'old_wealth': old_wealth,
            'new_wealth': new_wealth,
            'change': change,
            'timestamp': sender.model.steps
        })
    
    def on_became_wealthy(self, sender, wealth, **kwargs):
        """Handle agents becoming wealthy."""
        self.wealthy_agents.add(sender.unique_id)
        print(f"Agent {sender.unique_id} became wealthy with ${wealth}")
    
    def on_became_poor(self, sender, wealth, **kwargs):
        """Handle agents becoming poor."""
        self.poor_agents.add(sender.unique_id)
        print(f"Agent {sender.unique_id} became poor with ${wealth}")
    
    def get_statistics(self):
        """Get tracked statistics."""
        return {
            'total_changes': len(self.wealth_changes),
            'wealthy_count': len(self.wealthy_agents),
            'poor_count': len(self.poor_agents),
            'avg_change': sum(c['change'] for c in self.wealth_changes) / len(self.wealth_changes) 
                         if self.wealth_changes else 0
        }

class ReactiveModel(Model):
    """Model using reactive programming patterns."""
    
    def __init__(self, n_agents=50):
        super().__init__()
        
        # Use observable agent set
        self._agents = ObservableAgentSet()
        
        # Create wealth tracker
        self.wealth_tracker = WealthTracker()
        
        # Market signals
        self.market_crash = Signal("market_crash")
        self.market_boom = Signal("market_boom")
        
        # Connect market event handlers
        self.market_crash.connect(self.handle_market_crash)
        self.market_boom.connect(self.handle_market_boom)
        
        # Create agents and observe their signals
        for i in range(n_agents):
            agent = ReactiveAgent(self, wealth=self.random.randint(50, 200))
            
            # Connect agent signals to tracker
            self.wealth_tracker.observe(agent.wealth_changed, self.wealth_tracker.on_wealth_change)
            self.wealth_tracker.observe(agent.became_wealthy, self.wealth_tracker.on_became_wealthy)
            self.wealth_tracker.observe(agent.became_poor, self.wealth_tracker.on_became_poor)
        
        # Observe agent collection changes
        self._agents.agent_added.connect(self.on_agent_added)
        self._agents.agent_removed.connect(self.on_agent_removed)
        
        self.running = True
    
    @property
    def agents(self):
        """Override agents property to return observable set."""
        return self._agents
    
    def register_agent(self, agent):
        """Override to use observable agent set."""
        self._agents.add(agent)
    
    def deregister_agent(self, agent):
        """Override to use observable agent set."""
        self._agents.discard(agent)
    
    def on_agent_added(self, sender, agent, **kwargs):
        """Handle new agents being added."""
        print(f"New agent {agent.unique_id} joined the simulation")
    
    def on_agent_removed(self, sender, agent, **kwargs):
        """Handle agents being removed."""
        print(f"Agent {agent.unique_id} left the simulation")
    
    def handle_market_crash(self, sender, severity=0.5, **kwargs):
        """Handle market crash events."""
        print(f"Market crash! Severity: {severity}")
        for agent in self.agents:
            loss = int(agent.wealth * severity)
            agent.wealth = max(0, agent.wealth - loss)
    
    def handle_market_boom(self, sender, multiplier=1.5, **kwargs):
        """Handle market boom events."""
        print(f"Market boom! Multiplier: {multiplier}")
        for agent in self.agents:
            agent.wealth = int(agent.wealth * multiplier)
    
    def step(self):
        self.agents.shuffle_do("step")
        
        # Random market events
        if self.random.random() < 0.05:  # 5% chance
            if self.random.random() < 0.3:  # 30% of events are crashes
                self.market_crash.emit(sender=self, severity=self.random.uniform(0.2, 0.8))
            else:  # 70% are booms
                self.market_boom.emit(sender=self, multiplier=self.random.uniform(1.1, 1.8))

# Run reactive simulation
model = ReactiveModel(n_agents=20)
for i in range(50):
    model.step()

# Get wealth tracking statistics
stats = model.wealth_tracker.get_statistics()
print(f"\nWealth tracking results:")
print(f"Total wealth changes: {stats['total_changes']}")
print(f"Agents who became wealthy: {stats['wealthy_count']}")
print(f"Agents who became poor: {stats['poor_count']}")
print(f"Average wealth change: ${stats['avg_change']:.2f}")

from mesa.experimental.continuous_space import ContinuousSpace3D, SpatialIndex

class ContinuousSpace3D:
    """
    Three-dimensional continuous space with enhanced spatial queries.
    
    Extends 2D continuous space to support 3D positioning and
    spatial relationships with optimized neighbor searching.
    """
    
    def __init__(self, x_max: float, y_max: float, z_max: float,
                 torus: bool = False, x_min: float = 0, 
                 y_min: float = 0, z_min: float = 0):
        """
        Initialize 3D continuous space.
        
        Parameters:
            x_max: Maximum x coordinate
            y_max: Maximum y coordinate  
            z_max: Maximum z coordinate
            torus: Whether space wraps around at boundaries
            x_min: Minimum x coordinate
            y_min: Minimum y coordinate
            z_min: Minimum z coordinate
        """
        ...
    
    def place_agent(self, agent, pos: tuple[float, float, float]):
        """
        Place agent at 3D coordinates.
        
        Parameters:
            agent: Agent to place
            pos: (x, y, z) coordinate tuple
        """
        ...
    
    def move_agent(self, agent, pos: tuple[float, float, float]):
        """Move agent to new 3D coordinates."""
        ...
    
    def get_neighbors(self, pos: tuple[float, float, float], radius: float,
                     include_center: bool = False) -> list[Agent]:
        """
        Get agents within 3D radius of position.
        
        Parameters:
            pos: Center position (x, y, z)
            radius: Search radius
            include_center: Whether to include agents at exact center
        """
        ...
    
    def get_distance_3d(self, pos_1: tuple[float, float, float], 
                       pos_2: tuple[float, float, float]) -> float:
        """Get 3D Euclidean distance between positions."""
        ...

class SpatialIndex:
    """
    Spatial index for efficient neighbor queries in continuous spaces.
    
    Provides optimized spatial data structures for fast proximity
    searches in large continuous spaces.
    """
    
    def __init__(self, bounds: tuple[float, float, float, float]):
        """
        Initialize spatial index.
        
        Parameters:
            bounds: (x_min, y_min, x_max, y_max) bounding rectangle
        """
        ...
    
    def insert(self, agent, pos: tuple[float, float]):
        """Insert agent at position into spatial index."""
        ...
    
    def remove(self, agent):
        """Remove agent from spatial index."""
        ...
    
    def query_radius(self, pos: tuple[float, float], radius: float) -> list[Agent]:
        """Query agents within radius efficiently."""
        ...
    
    def query_rectangle(self, bounds: tuple[float, float, float, float]) -> list[Agent]:
        """Query agents within rectangular bounds."""
        ...

from mesa.experimental.meta_agents import MetaAgent, AgentGroup, HierarchicalAgent

class MetaAgent(Agent):
    """
    Agent that can contain and manage other agents.
    
    Enables hierarchical agent structures and composite behaviors
    for complex multi-level agent-based models.
    """
    
    def __init__(self, model, sub_agents=None):
        """
        Initialize meta-agent.
        
        Parameters:
            model: The model instance
            sub_agents: Initial sub-agents to contain
        """
        super().__init__(model)
        self.sub_agents = AgentSet(sub_agents or [])
        ...
    
    def add_sub_agent(self, agent):
        """Add a sub-agent to this meta-agent."""
        ...
    
    def remove_sub_agent(self, agent):
        """Remove a sub-agent from this meta-agent."""
        ...
    
    def step(self):
        """Execute meta-agent behavior and coordinate sub-agents."""
        # Meta-agent logic
        self._execute_meta_behavior()
        
        # Coordinate sub-agent actions
        self.sub_agents.shuffle_do("step")
    
    def _execute_meta_behavior(self):
        """Execute meta-level behavior - override in subclasses."""
        pass

class AgentGroup(MetaAgent):
    """
    Group of agents that act collectively.
    
    Provides coordination mechanisms for groups of agents
    to achieve collective goals and behaviors.
    """
    
    def __init__(self, model, agents=None, coordination_strategy="consensus"):
        """
        Initialize agent group.
        
        Parameters:
            model: Model instance
            agents: Agents to include in group
            coordination_strategy: How to coordinate group decisions
        """
        super().__init__(model, agents)
        self.coordination_strategy = coordination_strategy
        self.group_goals = []
        ...
    
    def add_group_goal(self, goal):
        """Add a collective goal for the group."""
        ...
    
    def coordinate_action(self, action_type: str, **kwargs):
        """Coordinate a collective action among group members."""
        if self.coordination_strategy == "consensus":
            return self._consensus_coordination(action_type, **kwargs)
        elif self.coordination_strategy == "leader":
            return self._leader_coordination(action_type, **kwargs)
        # Additional coordination strategies...
    
    def _consensus_coordination(self, action_type: str, **kwargs):
        """Coordinate through consensus among group members."""
        ...
    
    def _leader_coordination(self, action_type: str, **kwargs):
        """Coordinate through designated leader."""
        ...

from mesa.visualization import SolaraViz, JupyterViz, make_space_altair

class SolaraViz:
    """
    Browser-based visualization using Solara framework.
    
    Provides interactive, modern web-based visualization for Mesa models
    with real-time updates and user controls.
    """
    
    def __init__(self, 
                 model_cls,
                 model_params=None,
                 measures=None,
                 name="Mesa Model",
                 space_drawer=None,
                 agent_portrayal=None):
        """
        Initialize Solara visualization.
        
        Parameters:
            model_cls: Mesa model class to visualize
            model_params: Dictionary of model parameters with UI controls
            measures: List of measures to plot over time
            name: Display name for the model
            space_drawer: Function to draw spatial elements
            agent_portrayal: Function to define agent appearance
        """
        ...
    
    def launch(self, port=8521):
        """
        Launch the visualization server.
        
        Parameters:
            port: Port number for the web server
        """
        ...

class JupyterViz:
    """
    Jupyter notebook visualization component.
    
    Provides inline visualization widgets for Jupyter notebooks
    with interactive controls and real-time model execution.
    """
    
    def __init__(self, 
                 model_cls,
                 model_params=None,
                 measures=None,
                 space_drawer=None,
                 agent_portrayal=None):
        """
        Initialize Jupyter visualization.
        
        Parameters:
            model_cls: Mesa model class
            model_params: Parameter controls
            measures: Measures to display
            space_drawer: Spatial visualization function
            agent_portrayal: Agent appearance function
        """
        ...
    
    def show(self):
        """Display the visualization widget in Jupyter."""
        ...

def make_space_altair(space_drawer):
    """
    Create Altair-based space visualization.
    
    Parameters:
        space_drawer: Function that returns visualization data
    
    Returns:
        Altair chart specification for spatial visualization
    """
    ...

from mesa import Agent, Model
from mesa.space import MultiGrid
from mesa.datacollection import DataCollector
from mesa.visualization import SolaraViz
import altair as alt

class VisualizationAgent(Agent):
    def __init__(self, model, agent_type="normal"):
        super().__init__(model)
        self.agent_type = agent_type
        self.energy = 100
        self.size = self.random.uniform(0.5, 1.5)
    
    def step(self):
        # Simple movement
        neighbors = self.model.grid.get_neighborhood(self.pos, moore=True)
        new_pos = self.random.choice(neighbors)
        self.model.grid.move_agent(self, new_pos)
        
        # Energy dynamics
        self.energy += self.random.randint(-5, 10)
        self.energy = max(0, min(200, self.energy))

class VisualizationModel(Model):
    def __init__(self, n_agents=50, width=20, height=20, agent_types=None):
        super().__init__()
        
        self.grid = MultiGrid(width, height, torus=True)
        
        self.datacollector = DataCollector(
            model_reporters={
                "Total Agents": lambda m: len(m.agents),
                "Average Energy": lambda m: sum(a.energy for a in m.agents) / len(m.agents),
                "High Energy Agents": lambda m: len([a for a in m.agents if a.energy > 150])
            }
        )
        
        agent_types = agent_types or ["normal", "special", "rare"]
        
        # Create agents
        for i in range(n_agents):
            agent_type = self.random.choice(agent_types)
            agent = VisualizationAgent(self, agent_type=agent_type)
            
            # Place randomly
            x = self.random.randrange(width)
            y = self.random.randrange(height)
            self.grid.place_agent(agent, (x, y))
        
        self.running = True
    
    def step(self):
        self.datacollector.collect(self)
        self.agents.shuffle_do("step")

# Agent portrayal for visualization
def agent_portrayal(agent):
    """Define how agents appear in visualization."""
    size = agent.size * 20  # Scale for visibility
    
    # Color by type and energy
    if agent.agent_type == "special":
        color = "blue"
    elif agent.agent_type == "rare":
        color = "red" 
    else:
        color = "green"
    
    # Adjust opacity by energy
    opacity = agent.energy / 200.0
    
    return {
        "color": color,
        "size": size,
        "opacity": opacity,
        "shape": "circle"
    }

# Model parameters for UI controls
model_params = {
    "n_agents": {
        "type": "SliderInt",
        "value": 50,
        "label": "Number of Agents",
        "min": 10,
        "max": 200,
        "step": 10
    },
    "width": {
        "type": "SliderInt", 
        "value": 20,
        "label": "Grid Width",
        "min": 10,
        "max": 50
    },
    "height": {
        "type": "SliderInt",
        "value": 20, 
        "label": "Grid Height",
        "min": 10,
        "max": 50
    }
}

# Measures to plot
measures = ["Total Agents", "Average Energy", "High Energy Agents"]

# Create and launch visualization
viz = SolaraViz(
    model_cls=VisualizationModel,
    model_params=model_params,
    measures=measures,
    name="Agent Energy Dynamics",
    agent_portrayal=agent_portrayal
)

# Launch in browser (uncomment to run)
# viz.launch(port=8521)

# Or for Jupyter notebooks
jupyter_viz = JupyterViz(
    model_cls=VisualizationModel,
    model_params=model_params,
    measures=measures,
    agent_portrayal=agent_portrayal
)

# Display in notebook (uncomment to run)
# jupyter_viz.show()

# Pin Mesa version when using experimental features
# requirements.txt:
# Mesa==3.2.0  # Specific version for API stability

# Graceful fallbacks for missing experimental features
try:
    from mesa.experimental.devs import DEVSModel
    HAS_DEVS = True
except ImportError:
    HAS_DEVS = False
    DEVSModel = None

class AdaptiveModel(Model):
    """Model that adapts to available experimental features."""
    
    def __init__(self, use_devs=True, **kwargs):
        if use_devs and HAS_DEVS:
            # Use DEVS if available
            self._use_devs_features()
        else:
            # Fall back to standard Mesa
            super().__init__(**kwargs)
    
    def _use_devs_features(self):
        """Initialize DEVS-specific features."""
        # DEVS initialization code
        pass