Many small changes to get typing to work between files/classes. This …

…adds default classes for territory and community.

virtual_ecosystem/models/animal/animal_cohorts.py

@@ -1,10 +1,4 @@
-"""The ''animal'' module provides animal module functionality.
-
-Notes:
-- assume each grid = 1 km2
-- assume each tick = 1 day (28800s)
-- damuth ~ 4.23*mass**(-3/4) indiv / km2
-"""
+"""The ''animal'' module provides animal module functionality."""
 
 from __future__ import annotations
 
@@ -36,6 +30,7 @@ def __init__(
         mass: float,
         age: float,
         individuals: int,
+        territory: Territory,
         constants: AnimalConsts = AnimalConsts(),
     ) -> None:
         if age < 0:
@@ -57,6 +52,8 @@ def __init__(
         """The age of the animal cohort [days]."""
         self.individuals = individuals
         """The number of individuals in this cohort."""
+        self.territory = territory
+        """The territory of animal communities occupied by the cohort."""
         self.constants = constants
         """Animal constants."""
         self.damuth_density: int = sf.damuths_law(
@@ -81,8 +78,6 @@ def __init__(
         """The identification of useable food resources."""
         self.territory_size = sf.territory_size(self.functional_group.adult_mass)
         """The size in hectares of the animal cohorts territory."""
-        self.territory: Territory | None = None
-        """The AnimalTerritory object associated with the cohort."""
         # TODO - In future this should be parameterised using a constants dataclass, but
         # this hasn't yet been implemented for the animal model
         self.decay_fraction_excrement: float = self.constants.decay_fraction_excrement
@@ -267,25 +262,32 @@ def die_individual(self, number_dead: int, carcass_pool: DecayPool) -> None:
         ) * carcass_mass
         carcass_pool.decomposed_energy += self.decay_fraction_carcasses * carcass_mass
 
-    def update_carcass_pool(self, carcass_mass: float, carcass_pool: DecayPool) -> None:
-        """Updates the carcass pool based on consumed mass and predator's efficiency.
+    def update_carcass_pool(
+        self, carcass_mass: float, carcass_pools: Sequence[DecayPool]
+    ) -> None:
+        """Updates the carcass pools based on consumed mass and predator's efficiency.
 
         Args:
             carcass_mass: The total mass consumed from the prey cohort.
-            carcass_pool: The pool to which remains of eaten individuals are delivered.
+            carcass_pools: The pools to which remains of eaten individuals are
+            delivered.
         """
-
-        # Update the carcass pool with the remainder
-        carcass_pool.scavengeable_energy += (
-            1 - self.decay_fraction_carcasses
-        ) * carcass_mass
-        carcass_pool.decomposed_energy += self.decay_fraction_carcasses * carcass_mass
+        number_carcass_pools = len(carcass_pools)
+        carcass_mass_per_pool = carcass_mass / number_carcass_pools
+
+        for carcass_pool in carcass_pools:
+            # Update the carcass pool with the remainder
+            carcass_pool.scavengeable_energy += (
+                1 - self.decay_fraction_carcasses
+            ) * carcass_mass_per_pool
+            carcass_pool.decomposed_energy += (
+                self.decay_fraction_carcasses * carcass_mass_per_pool
+            )
 
     def get_eaten(
         self,
         potential_consumed_mass: float,
         predator: Consumer,
-        carcass_pool: DecayPool,
     ) -> float:
         """Removes individuals according to mass demands of a predation event.
 
@@ -324,8 +326,13 @@ def get_eaten(
         # Update the number of individuals in the prey cohort
         self.individuals -= actual_individuals_killed
 
+        # Find the intersection of prey and predator territories
+        intersection_carcass_pools = self.territory.find_intersecting_carcass_pools(
+            predator.territory
+        )
+
         # Update the carcass pool with carcass mass
-        self.update_carcass_pool(carcass_mass, carcass_pool)
+        self.update_carcass_pool(carcass_mass, intersection_carcass_pools)
 
         return actual_mass_consumed
 
@@ -576,7 +583,6 @@ def delta_mass_predation(
         self,
         animal_list: Sequence[Consumer],
         excrement_pools: Sequence[DecayPool],
-        carcass_pool: CarcassPool,
     ) -> float:
         """This method handles mass assimilation from predation.
 
@@ -588,7 +594,6 @@ def delta_mass_predation(
             animal_list: A sequence of animal cohorts that can be consumed by the
                          predator.
             excrement_pools: The pools representing the excrement in the territory.
-            carcass_pool: A pool representing the animal carcasses in the grid cell.
 
         Returns:
             The change in mass experienced by the predator.
@@ -602,9 +607,7 @@ def delta_mass_predation(
                 animal_list, prey_cohort
             )
             # Call get_eaten on the prey cohort to update its mass and individuals
-            actual_consumed_mass = prey_cohort.get_eaten(
-                consumed_mass, self, carcass_pool
-            )
+            actual_consumed_mass = prey_cohort.get_eaten(consumed_mass, self)
             # Update total mass gained by the predator
             total_consumed_mass += actual_consumed_mass
 
@@ -671,15 +674,13 @@ def forage_cohort(
         plant_list: Sequence[Resource],
         animal_list: Sequence[Consumer],
         excrement_pools: Sequence[DecayPool],
-        carcass_pool: CarcassPool,
     ) -> None:
         """This function handles selection of resources from a list for consumption.
 
         Args:
             plant_list: A sequence of plant resources available for herbivory.
             animal_list: A sequence of animal cohorts available for predation.
             excrement_pools: A pool representing the excrement in the grid cell.
-            carcass_pool: A pool representing the carcasses in the grid cell.
 
         Return:
             A float value of the net change in consumer mass due to foraging.
@@ -698,9 +699,7 @@ def forage_cohort(
         # Carnivore diet
         elif self.functional_group.diet == DietType.CARNIVORE and animal_list:
             # Calculate the mass gained from predation
-            consumed_mass = self.delta_mass_predation(
-                animal_list, excrement_pools, carcass_pool
-            )
+            consumed_mass = self.delta_mass_predation(animal_list, excrement_pools)
             # Update the predator's mass with the total gained mass
             self.eat(consumed_mass)
 

virtual_ecosystem/models/animal/animal_communities.py

@@ -1,10 +1,4 @@
-"""The ''animal'' module provides animal module functionality.
-
-Notes:
-- assume each grid = 1 km2
-- assume each tick = 1 day (28800s)
-- damuth ~ 4.23*mass**(-3/4) indiv / km2
-"""
+"""The ''animal'' module provides animal module functionality."""
 
 from __future__ import annotations
 
@@ -27,6 +21,12 @@
     get_functional_group_by_name,
 )
 from virtual_ecosystem.models.animal.plant_resources import PlantResources
+from virtual_ecosystem.models.animal.protocols import (
+    Consumer,
+    DecayPool,
+    Resource,
+    Territory,
+)
 from virtual_ecosystem.models.animal.scaling_functions import damuths_law
 
 
@@ -159,6 +159,7 @@ def populate_community(self) -> None:
                 functional_group.adult_mass,
                 0.0,
                 individuals,
+                DefaultTerritory(),
                 self.constants,
             )
             # add the cohort to the community
@@ -288,6 +289,7 @@ def birth(self, parent_cohort: AnimalCohort) -> None:
             parent_cohort.functional_group.birth_mass,
             0.0,
             number_offspring,
+            DefaultTerritory(),
             self.constants,
         )
 
@@ -340,14 +342,12 @@ def forage_community(self) -> None:
             prey_list = consumer_cohort.territory.get_prey(consumer_cohort)
             plant_list = consumer_cohort.territory.get_plant_resources()
             excrement_list = consumer_cohort.territory.get_excrement_pools()
-            # carcass_list = consumer_cohort.territory.get_carcass_pools()
 
             # Initiate foraging for the consumer cohort with the prepared resources
             consumer_cohort.forage_cohort(
                 plant_list=plant_list,
                 animal_list=prey_list,
                 excrement_pools=excrement_list,
-                carcass_pool=self.carcass_pool,
             )
 
             # Check if the cohort has been depleted to zero individuals post-foraging
@@ -488,6 +488,7 @@ def metamorphose(self, larval_cohort: AnimalCohort) -> None:
             adult_functional_group.birth_mass,
             0.0,
             larval_cohort.individuals,
+            DefaultTerritory(),
             self.constants,
         )
 
@@ -514,3 +515,68 @@ def metamorphose_community(self) -> None:
                 and (cohort.mass_current >= cohort.functional_group.adult_mass)
             ):
                 self.metamorphose(cohort)
+
+
+class DefaultCommunity(AnimalCommunity):
+    """A default community that represents an empty or non-functional state."""
+
+    def __init__(self) -> None:
+        self.functional_groups: tuple[FunctionalGroup, ...] = ()
+        self.data: Data = self.data
+        self.community_key: int = -1
+        self.neighbouring_keys: list[int] = []
+        self.constants: AnimalConsts = AnimalConsts()
+        self.carcass_pool: CarcassPool = CarcassPool(10000.0, 0.0)
+        """A pool for animal carcasses within the community."""
+        self.excrement_pool: ExcrementPool = ExcrementPool(10000.0, 0.0)
+        """A pool for excrement within the community."""
+        self.plant_community: PlantResources = PlantResources(
+            data=self.data,
+            cell_id=self.community_key,
+            constants=self.constants,
+        )
+
+    def collect_prey(
+        self, consumer_cohort: AnimalCohort
+    ) -> MutableSequence[AnimalCohort]:
+        """Default method."""
+        return []
+
+    def get_community_by_key(self, key: int) -> AnimalCommunity:
+        """Default method."""
+        return self
+
+
+class DefaultTerritory(Territory):
+    """A default territory that represents an empty or non-functional state."""
+
+    def __init__(self) -> None:
+        """Default method."""
+        self.grid_cell_keys: list[int] = []
+        self._get_community_by_key = lambda key: DefaultCommunity()
+
+    def update_territory(self, consumer_cohort: Consumer) -> None:
+        """Default method."""
+        pass
+
+    def get_prey(self, consumer_cohort: Consumer) -> MutableSequence[Consumer]:
+        """Default method."""
+        return []
+
+    def get_plant_resources(self) -> MutableSequence[Resource]:
+        """Default method."""
+        return []
+
+    def get_excrement_pools(self) -> MutableSequence[DecayPool]:
+        """Default method."""
+        return []
+
+    def get_carcass_pools(self) -> MutableSequence[DecayPool]:
+        """Default method."""
+        return []
+
+    def find_intersecting_carcass_pools(
+        self, animal_territory: Territory
+    ) -> MutableSequence[DecayPool]:
+        """Default method."""
+        return []

virtual_ecosystem/models/animal/animal_territories.py

@@ -2,12 +2,13 @@
 
 from collections.abc import Callable, MutableSequence, Sequence
 
-from virtual_ecosystem.models.animal.animal_cohorts import AnimalCohort
-from virtual_ecosystem.models.animal.animal_communities import AnimalCommunity
-from virtual_ecosystem.models.animal.decay import CarcassPool, ExcrementPool
-from virtual_ecosystem.models.animal.plant_resources import PlantResources
-
-# from virtual_ecosystem.models.animal.protocols import Consumer, DecayPool, Resource
+from virtual_ecosystem.models.animal.protocols import (
+    Community,
+    Consumer,
+    DecayPool,
+    Resource,
+    Territory,
+)
 
 
 class AnimalTerritory:
@@ -32,23 +33,23 @@ class AnimalTerritory:
     def __init__(
         self,
         grid_cell_keys: list[int],
-        get_community_by_key: Callable[[int], AnimalCommunity],
+        get_community_by_key: Callable[[int], Community],
     ) -> None:
         # The constructor of the AnimalTerritory class.
         self.grid_cell_keys = grid_cell_keys
         """A list of grid cells present in the territory."""
         self.get_community_by_key = get_community_by_key
         """A list of animal communities present in the territory."""
-        self.territory_prey: Sequence[AnimalCohort] = []
+        self.territory_prey: Sequence[Consumer] = []
         """A list of animal prey present in the territory."""
-        self.territory_plants: Sequence[PlantResources] = []
+        self.territory_plants: Sequence[Resource] = []
         """A list of plant resources present in the territory."""
-        self.territory_excrement: Sequence[ExcrementPool] = []
+        self.territory_excrement: Sequence[DecayPool] = []
         """A list of excrement pools present in the territory."""
-        self.territory_carcasses: Sequence[CarcassPool] = []
+        self.territory_carcasses: Sequence[DecayPool] = []
         """A list of carcass pools present in the territory."""
 
-    def update_territory(self, consumer_cohort: AnimalCohort) -> None:
+    def update_territory(self, consumer_cohort: Consumer) -> None:
         """Update territory details at initialization and after migration.
 
         Args:
@@ -61,7 +62,7 @@ def update_territory(self, consumer_cohort: AnimalCohort) -> None:
         self.territory_excrement = self.get_excrement_pools()
         self.territory_carcasses = self.get_carcass_pools()
 
-    def get_prey(self, consumer_cohort: AnimalCohort) -> Sequence[AnimalCohort]:
+    def get_prey(self, consumer_cohort: Consumer) -> MutableSequence[Consumer]:
         """Collect suitable prey from all grid cells in the territory.
 
         TODO: This is probably not the best way to go about this. Maybe alter collect
@@ -80,7 +81,7 @@ def get_prey(self, consumer_cohort: AnimalCohort) -> Sequence[AnimalCohort]:
             prey.extend(community.collect_prey(consumer_cohort))
         return prey
 
-    def get_plant_resources(self) -> MutableSequence[PlantResources]:
+    def get_plant_resources(self) -> MutableSequence[Resource]:
         """Collect plant resources from all grid cells in the territory.
 
         TODO: Update internal plant resource generation with a real link to the plant
@@ -95,7 +96,7 @@ def get_plant_resources(self) -> MutableSequence[PlantResources]:
             plant_resources.append(community.plant_community)
         return plant_resources
 
-    def get_excrement_pools(self) -> MutableSequence[ExcrementPool]:
+    def get_excrement_pools(self) -> MutableSequence[DecayPool]:
         """Combine excrement pools from all grid cells in the territory.
 
         Returns:
@@ -107,7 +108,7 @@ def get_excrement_pools(self) -> MutableSequence[ExcrementPool]:
             total_excrement.append(community.excrement_pool)
         return total_excrement
 
-    def get_carcass_pools(self) -> MutableSequence[CarcassPool]:
+    def get_carcass_pools(self) -> MutableSequence[DecayPool]:
         """Combine carcass pools from all grid cells in the territory.
 
         Returns:
@@ -119,6 +120,26 @@ def get_carcass_pools(self) -> MutableSequence[CarcassPool]:
             total_carcass.append(community.carcass_pool)
         return total_carcass
 
+    def find_intersecting_carcass_pools(
+        self, animal_territory: "Territory"
+    ) -> MutableSequence[DecayPool]:
+        """Find the carcass pools of the intersection of two territories.
+
+        Args:
+            animal_territory: Another AnimalTerritory to find the intersection with.
+
+        Returns:
+            A list of CarcassPools in the intersecting grid cells.
+        """
+        intersecting_keys = set(self.grid_cell_keys) & set(
+            animal_territory.grid_cell_keys
+        )
+        intersecting_carcass_pools = []
+        for cell_id in intersecting_keys:
+            community = self.get_community_by_key(cell_id)
+            intersecting_carcass_pools.append(community.carcass_pool)
+        return intersecting_carcass_pools
+
 
 def bfs_territory(
     centroid_key: int, target_cell_number: int, cell_nx: int, cell_ny: int