Source code for modelly.space.cells.collection

"""Collection class for managing and querying groups of cells.

The CellCollection class provides a consistent interface for operating on multiple
cells, supporting:
- Filtering and selecting cells based on conditions
- Random cell and agent selection
- Access to contained agents
- Group operations

This is useful for implementing area effects, zones, or any operation that needs
to work with multiple cells as a unit. The collection handles efficient iteration
and agent access across cells. The class is used throughout the cell space
implementation to represent neighborhoods, selections, and other cell groupings.
"""

from __future__ import annotations

import itertools
import warnings
from collections.abc import Callable, Iterable, Mapping
from functools import cached_property
from random import Random
from typing import TYPE_CHECKING, TypeVar

from modelly.core.agentset._utils import _INF, _validate_at_most

if TYPE_CHECKING:
    from modelly.space.cells.agents import CellAgent
    from modelly.space.cells.cell import Cell

T = TypeVar("T", bound="Cell")

RAISES = object()


def _derived_random(cells: Mapping[Cell, list[CellAgent]]) -> Random | None:
    """The generator of the first cell that has one, else the first agent's model's."""
    for cell, agents in cells.items():
        if cell.random is not None:
            return cell.random
        if agents:
            return agents[0].model.random
    return None


[docs] class CellCollection[T: Cell]: """An immutable collection of cells. Attributes: cells (List[Cell]): The list of cells this collection represents agents (List[CellAgent]) : List of agents occupying the cells in this collection random (Random) : The random number generator Notes: With `random=None` the collection takes the generator of its first cell that has one, else the model generator of its first agent. A `UserWarning` is issued only when there is neither; pass `random=self.random` from a `Model` or `Agent` to avoid it. """ def __init__( self, cells: Mapping[T, list[CellAgent]] | Iterable[T], random: Random | None = None, ) -> None: """Initialize a CellCollection. Args: cells: cells to add to the collection random: a seeded random number generator; None takes the cells' or their agents' model generator. """ if isinstance(cells, dict): self._cells = cells else: self._cells = {cell: cell._agents for cell in cells} # Get capacity from first cell if collection is not empty self._capacity: int | None = ( next(iter(self._cells.keys())).capacity if self._cells else None ) if random is None: random = _derived_random(self._cells) if random is None: warnings.warn( "Random number generator not specified, this can make models non-reproducible. Please pass a random number generator explicitly", UserWarning, stacklevel=2, ) random = Random() # noqa: MDL001 (no cell or agent has a seeded generator) self.random = random def __iter__(self): # noqa return iter(self._cells) def __getitem__(self, key: T) -> Iterable[CellAgent]: # noqa return self._cells[key] # @cached_property def __len__(self) -> int: # noqa return len(self._cells) def __repr__(self): # noqa return f"CellCollection({self._cells})" @cached_property def cells(self) -> list[T]: # noqa return list(self._cells.keys()) @property def agents(self) -> Iterable[CellAgent]: # noqa return itertools.chain.from_iterable(self._cells.values()) @property def empties(self) -> CellCollection[T]: """The cells of this collection that hold no agent, as a new collection. Reads the live occupancy lists directly, so it costs one truth test per cell rather than a property call per cell. """ return CellCollection( {cell: agents for cell, agents in self._cells.items() if not agents}, random=self.random, )
[docs] def select_random_cell(self, default=RAISES) -> T | None: """Select a random cell from the collection. Args: default: Value to return if the collection is empty. If not provided, raises LookupError. Returns: T: A random cell, or the default value if provided and collection is empty. Raises: LookupError: If collection is empty and no default is provided. """ if not self._cells: if default is RAISES: raise LookupError("Cannot select random cell from empty collection") return default return self.random.choice(self.cells)
[docs] def select_random_agent(self, default=RAISES) -> CellAgent | None: """Select a random agent from the collection. Args: default: Value to return if the collection is empty. If not provided, raises LookupError. Returns: CellAgent: A random agent, or the default value if provided and collection is empty. Raises: LookupError: If collection is empty and no default is provided. """ agents = list(self.agents) if not agents: if default is RAISES: raise LookupError("Cannot select random agent from empty collection") return default return self.random.choice(agents)
[docs] def select( self, filter_func: Callable[[T], bool] | None = None, at_most: int | float = float("inf"), ): """Select cells based on filter function. Args: filter_func: filter function at_most: The maximum amount of cells to select. Defaults to infinity. - If an integer, at most the first number of matching cells is selected. - If a float between 0 and 1, at most that fraction of original number of cells Returns: CellCollection Raises: TypeError: If at_most is not an integer or float, or is a boolean. ValueError: If at_most is a negative integer, or a float not in the range (0.0, 1.0]. """ if at_most == _INF: if filter_func is None: return self # One pass straight into the new collection's mapping. kept = {cell: cell._agents for cell in self._cells if filter_func(cell)} return CellCollection(kept, random=self.random) limit = _validate_at_most(at_most, len(self._cells)) matches = ( iter(self._cells) if filter_func is None else (cell for cell in self._cells if filter_func(cell)) ) # Stop filtering as soon as ``limit`` cells matched. kept = {cell: cell._agents for cell in itertools.islice(matches, limit)} return CellCollection(kept, random=self.random)