Enhance wolf-sheep with continuous energy and reactive behaviors

Update the wolf-sheep predation example to demonstrate mesa_signals and ContinuousObservable capabilities. Energy now depletes continuously over time rather than in discrete steps, and agent behaviors react automatically to energy state changes.

Changes:
- Convert Animal base class to use ContinuousObservable for energy tracking with configurable metabolic rates
- Add computed properties (is_hungry, can_reproduce) that automatically update based on energy levels
- Implement threshold-triggered events for death (energy reaches zero) and survival mode (critical hunger)
- Enhance decision logic: sheep prioritize grass when hungry, wolves hunt actively when hungry
- Add activity-based metabolic rates (movement costs more energy than resting)
- Improve GrassPatch to use Observable and react to state changes with signal handlers
- Expand data collection to track average energy levels across populations

The example now demonstrates key reactive programming patterns: observable state management, computed properties for decision making, threshold-based event triggering, and automatic dependency tracking. This makes the model more realistic (continuous dynamics) while reducing boilerplate code (declarative state management).

fix agent imports

Author: EwoutH

mesa/examples/advanced/wolf_sheep/agents.py

@@ -1,8 +1,26 @@
 from mesa.discrete_space import CellAgent, FixedAgent
+from mesa.experimental.mesa_signals import (
+    Computable,
+    Computed,
+    ContinuousObservable,
+    HasObservables,
+    Observable,
+)
 
 
-class Animal(CellAgent):
-    """The base animal class."""
+class Animal(CellAgent, HasObservables):
+    """The base animal class with reactive energy management."""
+
+    # Energy depletes continuously over time
+    energy = ContinuousObservable(
+        initial_value=8.0, rate_func=lambda value, elapsed, agent: -agent.metabolic_rate
+    )
+
+    # Computed property: animal is hungry when energy is low
+    is_hungry = Computable()
+
+    # Computed property: animal can reproduce when energy is sufficient
+    can_reproduce = Computable()
 
     def __init__(
         self, model, energy=8, p_reproduce=0.04, energy_from_food=4, cell=None
@@ -17,14 +35,40 @@ def __init__(
             cell: Cell in which the animal starts
         """
         super().__init__(model)
+
+        # Set base metabolic rate (energy loss per time unit when idle)
+        self.metabolic_rate = 0.5
+
+        # Initialize energy (triggers continuous depletion)
         self.energy = energy
         self.p_reproduce = p_reproduce
         self.energy_from_food = energy_from_food
         self.cell = cell
 
+        # Set up computed properties
+        self.is_hungry = Computed(lambda: self.energy < self.energy_from_food * 2)
+        self.can_reproduce = Computed(lambda: self.energy > self.energy_from_food * 4)
+
+        # Register threshold: die when energy reaches zero
+        self.add_threshold("energy", 0.0, self._on_energy_depleted)
+
+        # Register threshold: become critically hungry at 25% of starting energy
+        self.add_threshold("energy", energy * 0.25, self._on_critical_hunger)
+
+    def _on_energy_depleted(self, signal):
+        """Called when energy crosses zero - animal dies."""
+        if signal.direction == "down":  # Only trigger on downward crossing
+            self.remove()
+
+    def _on_critical_hunger(self, signal):
+        """Called when energy becomes critically low."""
+        if signal.direction == "down":
+            # Increase metabolic efficiency when starving (survival mode)
+            self.metabolic_rate *= 0.8
+
     def spawn_offspring(self):
         """Create offspring by splitting energy and creating new instance."""
-        self.energy /= 2
+        self.energy /= 2  # This updates the continuous observable
         self.__class__(
             self.model,
             self.energy,
@@ -35,33 +79,39 @@ def spawn_offspring(self):
 
     def feed(self):
         """Abstract method to be implemented by subclasses."""
+        raise NotImplementedError
 
     def step(self):
         """Execute one step of the animal's behavior."""
-        # Move to random neighboring cell
+        # Move to neighboring cell (uses more energy than standing still)
+        self.metabolic_rate = 1.0  # Movement costs more energy
         self.move()
 
-        self.energy -= 1
-
         # Try to feed
         self.feed()
 
-        # Handle death and reproduction
-        if self.energy < 0:
-            self.remove()
-        elif self.random.random() < self.p_reproduce:
+        # Return to resting metabolic rate
+        self.metabolic_rate = 0.5
+
+        # Reproduce if conditions are met (using computed property)
+        if self.can_reproduce and self.random.random() < self.p_reproduce:
             self.spawn_offspring()
 
 
 class Sheep(Animal):
-    """A sheep that walks around, reproduces (asexually) and gets eaten."""
+    """A sheep that walks around, reproduces and gets eaten.
+
+    Sheep prefer cells with grass and avoid wolves. They gain energy by
+    eating grass, which continuously depletes over time.
+    """
 
     def feed(self):
         """If possible, eat grass at current location."""
         grass_patch = next(
             obj for obj in self.cell.agents if isinstance(obj, GrassPatch)
         )
         if grass_patch.fully_grown:
+            # Eating gives instant energy boost
             self.energy += self.energy_from_food
             grass_patch.fully_grown = False
 
@@ -70,64 +120,82 @@ def move(self):
         cells_without_wolves = self.cell.neighborhood.select(
             lambda cell: not any(isinstance(obj, Wolf) for obj in cell.agents)
         )
-        # If all surrounding cells have wolves, stay put
+
+        # If all surrounding cells have wolves, stay put (fear overrides hunger)
         if len(cells_without_wolves) == 0:
             return
 
-        # Among safe cells, prefer those with grown grass
-        cells_with_grass = cells_without_wolves.select(
-            lambda cell: any(
-                isinstance(obj, GrassPatch) and obj.fully_grown for obj in cell.agents
+        # If critically hungry, prioritize grass over safety
+        if self.is_hungry:  # Using computed property
+            cells_with_grass = cells_without_wolves.select(
+                lambda cell: any(
+                    isinstance(obj, GrassPatch) and obj.fully_grown
+                    for obj in cell.agents
+                )
             )
-        )
-        # Move to a cell with grass if available, otherwise move to any safe cell
-        target_cells = (
-            cells_with_grass if len(cells_with_grass) > 0 else cells_without_wolves
-        )
+            # Move to grass if available, otherwise any safe cell
+            target_cells = (
+                cells_with_grass if len(cells_with_grass) > 0 else cells_without_wolves
+            )
+        else:
+            # Not hungry - just avoid wolves
+            target_cells = cells_without_wolves
+
         self.cell = target_cells.select_random_cell()
 
 
 class Wolf(Animal):
-    """A wolf that walks around, reproduces (asexually) and eats sheep."""
+    """A wolf that walks around, reproduces and eats sheep.
+
+    Wolves are more efficient predators, with higher base energy and
+    metabolic rate. They actively hunt sheep and gain substantial energy
+    from successful kills.
+    """
+
+    def __init__(
+        self, model, energy=20, p_reproduce=0.05, energy_from_food=20, cell=None
+    ):
+        """Initialize a wolf with higher energy needs than sheep."""
+        super().__init__(model, energy, p_reproduce, energy_from_food, cell)
+        # Wolves have higher metabolic rate (they're larger predators)
+        self.metabolic_rate = 1.0
 
     def feed(self):
         """If possible, eat a sheep at current location."""
         sheep = [obj for obj in self.cell.agents if isinstance(obj, Sheep)]
-        if sheep:  # If there are any sheep present
+        if sheep:  # Successful hunt
             sheep_to_eat = self.random.choice(sheep)
+            # Eating gives instant energy boost
             self.energy += self.energy_from_food
             sheep_to_eat.remove()
 
     def move(self):
         """Move to a neighboring cell, preferably one with sheep."""
-        cells_with_sheep = self.cell.neighborhood.select(
-            lambda cell: any(isinstance(obj, Sheep) for obj in cell.agents)
-        )
-        target_cells = (
-            cells_with_sheep if len(cells_with_sheep) > 0 else self.cell.neighborhood
-        )
-        self.cell = target_cells.select_random_cell()
+        # When hungry, actively hunt for sheep
+        if self.is_hungry:  # Using computed property
+            cells_with_sheep = self.cell.neighborhood.select(
+                lambda cell: any(isinstance(obj, Sheep) for obj in cell.agents)
+            )
+            target_cells = (
+                cells_with_sheep
+                if len(cells_with_sheep) > 0
+                else self.cell.neighborhood
+            )
+        else:
+            # When not hungry, wander randomly (conserve energy)
+            target_cells = self.cell.neighborhood
 
+        self.cell = target_cells.select_random_cell()
 
-class GrassPatch(FixedAgent):
-    """A patch of grass that grows at a fixed rate and can be eaten by sheep."""
 
-    @property
-    def fully_grown(self):
-        """Whether the grass patch is fully grown."""
-        return self._fully_grown
+class GrassPatch(FixedAgent, HasObservables):
+    """A patch of grass that grows at a fixed rate and can be eaten by sheep.
 
-    @fully_grown.setter
-    def fully_grown(self, value: bool) -> None:
-        """Set grass growth state and schedule regrowth if eaten."""
-        self._fully_grown = value
+    Grass growth is modeled as a continuous process with a fixed regrowth time.
+    """
 
-        if not value:  # If grass was just eaten
-            self.model.simulator.schedule_event_relative(
-                setattr,
-                self.grass_regrowth_time,
-                function_args=[self, "fully_grown", True],
-            )
+    # Observable: grass growth state
+    fully_grown = Observable()
 
     def __init__(self, model, countdown, grass_regrowth_time, cell):
         """Create a new patch of grass.
@@ -139,12 +207,25 @@ def __init__(self, model, countdown, grass_regrowth_time, cell):
             cell: Cell to which this grass patch belongs
         """
         super().__init__(model)
-        self._fully_grown = countdown == 0
+
+        self.fully_grown = countdown == 0
         self.grass_regrowth_time = grass_regrowth_time
         self.cell = cell
 
+        # Listen for when grass gets eaten, schedule regrowth
+        self.observe("fully_grown", "change", self._on_growth_change)
+
         # Schedule initial growth if not fully grown
         if not self.fully_grown:
+            self.model.simulator.schedule_event_relative(self._regrow, countdown)
+
+    def _on_growth_change(self, signal):
+        """React to grass being eaten - schedule regrowth."""
+        if signal.new is False:  # Grass was just eaten
             self.model.simulator.schedule_event_relative(
-                setattr, countdown, function_args=[self, "fully_grown", True]
+                self._regrow, self.grass_regrowth_time
             )
+
+    def _regrow(self):
+        """Regrow the grass patch."""
+        self.fully_grown = True

mesa/examples/advanced/wolf_sheep/model.py

@@ -2,6 +2,8 @@
 Wolf-Sheep Predation Model
 ================================
 
+Enhanced version with continuous energy depletion and reactive behaviors.
+
 Replication of the model found in NetLogo:
     Wilensky, U. (1997). NetLogo Wolf Sheep Predation model.
     http://ccl.northwestern.edu/netlogo/models/WolfSheepPredation.
@@ -21,11 +23,16 @@
 class WolfSheep(Model):
     """Wolf-Sheep Predation Model.
 
-    A model for simulating wolf and sheep (predator-prey) ecosystem modelling.
+    A model for simulating wolf and sheep (predator-prey) ecosystem with:
+    - Continuous energy depletion over time
+    - Reactive behaviors based on hunger levels
+    - Threshold-triggered events (death, starvation mode)
+    - Computed properties for decision making
     """
 
     description = (
-        "A model for simulating wolf and sheep (predator-prey) ecosystem modelling."
+        "A model for simulating wolf and sheep (predator-prey) ecosystem modelling "
+        "with continuous energy dynamics and reactive behaviors."
     )
 
     def __init__(
@@ -55,12 +62,13 @@ def __init__(
             wolf_gain_from_food: Energy a wolf gains from eating a sheep
             grass: Whether to have the sheep eat grass for energy
             grass_regrowth_time: How long it takes for a grass patch to regrow
-                                once it is eaten
             sheep_gain_from_food: Energy sheep gain from grass, if enabled
             seed: Random seed
             simulator: ABMSimulator instance for event scheduling
         """
         super().__init__(seed=seed)
+
+        # Initialize time-based simulator for continuous energy dynamics
         self.simulator = simulator
         self.simulator.setup(self)
 
@@ -77,19 +85,32 @@ def __init__(
             random=self.random,
         )
 
-        # Set up data collection
+        # Set up data collection (tracks observable changes automatically)
         model_reporters = {
             "Wolves": lambda m: len(m.agents_by_type[Wolf]),
             "Sheep": lambda m: len(m.agents_by_type[Sheep]),
+            "Avg Wolf Energy": lambda m: (
+                sum(w.energy for w in m.agents_by_type[Wolf])
+                / len(m.agents_by_type[Wolf])
+                if len(m.agents_by_type[Wolf]) > 0
+                else 0
+            ),
+            "Avg Sheep Energy": lambda m: (
+                sum(s.energy for s in m.agents_by_type[Sheep])
+                / len(m.agents_by_type[Sheep])
+                if len(m.agents_by_type[Sheep]) > 0
+                else 0
+            ),
         }
+
         if grass:
             model_reporters["Grass"] = lambda m: len(
                 m.agents_by_type[GrassPatch].select(lambda a: a.fully_grown)
             )
 
         self.datacollector = DataCollector(model_reporters)
 
-        # Create sheep:
+        # Create sheep with random initial energy
         Sheep.create_agents(
             self,
             initial_sheep,
@@ -98,7 +119,8 @@ def __init__(
             energy_from_food=sheep_gain_from_food,
             cell=self.random.choices(self.grid.all_cells.cells, k=initial_sheep),
         )
-        # Create Wolves:
+
+        # Create wolves with random initial energy
         Wolf.create_agents(
             self,
             initial_wolves,
@@ -123,10 +145,15 @@ def __init__(
         self.datacollector.collect(self)
 
     def step(self):
-        """Execute one step of the model."""
-        # First activate all sheep, then all wolves, both in random order
+        """Execute one step of the model.
+
+        Energy continuously depletes between steps via ContinuousObservable.
+        This step method only triggers agent decisions and actions.
+        """
+        # Activate all sheep, then all wolves, both in random order
+        # Their energy has been continuously depleting since last step
         self.agents_by_type[Sheep].shuffle_do("step")
         self.agents_by_type[Wolf].shuffle_do("step")
 
-        # Collect data
+        # Collect data (automatically captures current energy levels)
         self.datacollector.collect(self)