Skip to content

Bases: BaseModule, RasterLayer

Base class for managing raster-based spatial modules in ABSESpy.

Inherits from both Module and RasterLayer to provide comprehensive spatial data management. Extends mesa-geo's RasterLayer with additional capabilities for: - Agent placement and management - Integration with xarray/rasterio for data I/O - Dynamic attribute handling - Spatial operations and analysis

Attributes:

Name Type Description
cell_properties set[str]

Set of accessible cell attributes (decorated by @raster_attribute).

attributes set[str]

All accessible attributes including cell_properties.

shape2d Coordinate

Raster dimensions as (height, width).

shape3d Coordinate

Raster dimensions as (1, height, width) for rasterio compatibility.

array_cells ndarray

NumPy array of PatchCell objects.

coords Coordinate

Coordinate system dictionary with 'x' and 'y' arrays.

random ListRandom

Random selection proxy for cells.

mask ndarray

Boolean array indicating accessible cells.

cells_lst ActorsList[PatchCell]

ActorsList containing all cells.

plot ActorsList[PatchCell]

Visualization interface for the module.

This constructor automatically determines the appropriate creation method based on the provided parameters. It supports creating a PatchModule from: - Resolution and shape - Copying an existing layer - xarray DataArray - Vector file or GeoDataFrame - Raster file

Parameters:

Name Type Description Default
model MainModelProtocol

Parent model instance.

required
name Optional[str]

Module identifier. Defaults to lowercase class name.

None
cell_cls Type[PatchCell]

Class to use for creating cells. Defaults to PatchCell.

PatchCell
shape Optional[Coordinate]

Array shape (height, width) for creating a new module.

None
resolution Number | None | Tuple[Number, Number]

Spatial resolution when creating coordinates.

1
source_layer Optional[PatchModule]

Existing PatchModule to copy.

None
xda Optional[DataArray]

xarray DataArray containing raster data.

None
attr_name Optional[str]

Attribute name for the loaded raster data.

None
apply_raster bool

Whether to apply raster data to cells.

False
masked bool

Whether to use mask from the data.

True
vector_file Optional[Union[str, GeoDataFrame]]

Path to vector file or GeoDataFrame.

None
raster_file Optional[str]

Path to raster file.

None
band int

Band number to use from raster file.

1
crs Optional[Union[CRS, str]]

Coordinate Reference System.

None
total_bounds Optional[List[float]]

Spatial bounds [minx, miny, maxx, maxy].

None
width Optional[int]

Width of the raster in cells.

None
height Optional[int]

Height of the raster in cells.

None
**kwargs Any

Additional arguments passed to RasterLayer initialization.

{}
Source code in abses/space/patch.py
def __init__(
    self,
    model: MainModelProtocol,
    name: Optional[str] = None,
    cell_cls: Type[PatchCell] = PatchCell,
    *,
    # Resolution-based creation parameters
    shape: Optional[Coordinate] = None,
    resolution: Number | None | Tuple[Number, Number] = 1,
    # Layer copy parameters
    source_layer: Optional[PatchModule] = None,
    # Xarray-based creation parameters
    xda: Optional[xr.DataArray] = None,
    attr_name: Optional[str] = None,
    apply_raster: bool = False,
    masked: bool = True,
    # Vector-based creation parameters
    vector_file: Optional[Union[str, gpd.GeoDataFrame]] = None,
    # Raster file-based creation parameters
    raster_file: Optional[str] = None,
    band: int = 1,
    # Common parameters
    crs: Optional[Union[pyproj.CRS, str]] = None,
    total_bounds: Optional[List[float]] = None,
    width: Optional[int] = None,
    height: Optional[int] = None,
    **kwargs: Any,
):
    """Initializes a new PatchModule instance with a unified API.

    This constructor automatically determines the appropriate creation method based on the
    provided parameters. It supports creating a PatchModule from:
    - Resolution and shape
    - Copying an existing layer
    - xarray DataArray
    - Vector file or GeoDataFrame
    - Raster file

    Args:
        model: Parent model instance.
        name: Module identifier. Defaults to lowercase class name.
        cell_cls: Class to use for creating cells. Defaults to PatchCell.

        # Resolution-based creation parameters
        shape: Array shape (height, width) for creating a new module.
        resolution: Spatial resolution when creating coordinates.

        # Layer copy parameters
        source_layer: Existing PatchModule to copy.

        # Xarray-based creation parameters
        xda: xarray DataArray containing raster data.
        attr_name: Attribute name for the loaded raster data.
        apply_raster: Whether to apply raster data to cells.
        masked: Whether to use mask from the data.

        # Vector-based creation parameters
        vector_file: Path to vector file or GeoDataFrame.

        # Raster file-based creation parameters
        raster_file: Path to raster file.
        band: Band number to use from raster file.

        # Common parameters
        crs: Coordinate Reference System.
        total_bounds: Spatial bounds [minx, miny, maxx, maxy].
        width: Width of the raster in cells.
        height: Height of the raster in cells.
        **kwargs: Additional arguments passed to RasterLayer initialization.
    """
    # Initialize BaseModule
    BaseModule.__init__(self, model, name=name)

    # Normalize CRS if provided
    if crs is not None:
        crs = crs

    # Determine creation method based on provided parameters
    if raster_file is not None:
        # Create from raster file
        xda = rioxarray.open_rasterio(raster_file, masked=masked, **kwargs)
        xda = xda.sel(band=band)

        # Update parameters for xarray-based creation
        width = xda.rio.width
        height = xda.rio.height
        crs = xda.rio.crs
        total_bounds = xda.rio.bounds()

        # Apply raster data if attr_name is provided (backward compatibility)
        # In 0.7.x, providing attr_name automatically applied the raster data
        if attr_name is not None:
            apply_raster = True

    elif vector_file is not None:
        # Create from vector file
        if resolution is None:
            raise ValueError(
                "Resolution must be provided when creating from vector file"
            )

        # Convert to GeoDataFrame if needed
        if isinstance(vector_file, str):
            gdf = gpd.read_file(vector_file)
        elif isinstance(vector_file, gpd.GeoDataFrame):
            gdf = vector_file
        else:
            raise TypeError(f"Unsupported vector type: {type(vector_file)}")

        # Set attribute name if not provided
        if attr_name is None:
            gdf, attr_name = gdf.reset_index(), "index"

        # Convert resolution to tuple if needed
        if isinstance(resolution, (int, float)):
            resolution = (resolution, resolution)

        # Create xarray from vector
        xda = make_geocube(gdf, measurements=[attr_name], resolution=resolution)[
            attr_name
        ]

        # Update parameters for xarray-based creation
        width = xda.rio.width
        height = xda.rio.height
        crs = xda.rio.crs
        total_bounds = xda.rio.bounds()

        # Apply raster data if attr_name is provided (backward compatibility)
        if attr_name is not None:
            apply_raster = True

    elif xda is not None:
        # Create from xarray DataArray
        # Flip data if y-axis is ascending
        if xda.y[0].item() < xda.y[-1].item():
            xda.data = np.flipud(xda.data)

        # Update parameters
        width = xda.rio.width
        height = xda.rio.height
        crs = xda.rio.crs
        total_bounds = xda.rio.bounds()

        # Apply raster data if attr_name is provided (backward compatibility)
        if attr_name is not None:
            apply_raster = True

    elif source_layer is not None:
        # Copy from existing layer
        if not isinstance(source_layer, PatchModule):
            raise TypeError(f"{source_layer} is not a valid PatchModule.")

        # Copy parameters from source layer
        width = source_layer.width
        height = source_layer.height
        crs = source_layer.crs
        total_bounds = source_layer.total_bounds

    elif shape is not None:
        # Create from resolution and shape
        assert isinstance(resolution, (int, float))
        assert width is None and height is None
        height, width = shape
        if crs is None:
            crs = DEFAULT_CRS  # Already normalized
        total_bounds = [0, 0, width * resolution, height * resolution]

    # Ensure required parameters are provided
    if width is None or height is None or crs is None or total_bounds is None:
        raise ValueError(
            "Insufficient parameters provided. Must provide either: "
            "1) shape and resolution, 2) source_layer, 3) xda, "
            "4) vector_file and resolution, or 5) raster_file"
        )

    # Remove PatchModule-specific parameters that shouldn't be passed to RasterLayer
    # Even though they are explicit parameters, we remove them from kwargs as a safeguard
    for key in ["apply_raster", "attr_name"]:
        kwargs.pop(key, None)

    # Initialize RasterLayer with the determined parameters
    RasterLayer.__init__(
        self,
        model=model,
        width=width,
        height=height,
        crs=crs,  # Now using normalized CRS
        total_bounds=total_bounds,
        cell_cls=cell_cls,
        **kwargs,
    )

    logger.info("Initializing a new Model Layer...")
    self._mask: np.ndarray = np.ones(self.shape2d).astype(bool)

    # Apply mask if provided
    if masked and xda is not None:
        self.mask = xda.notnull().to_numpy()

    # Apply raster data if requested
    if apply_raster and xda is not None and attr_name is not None:
        self.apply_raster(xda.to_numpy(), attr_name=attr_name)

cells_lst cached property

cells_lst

The cells stored in this layer.

mask property writable

mask

Where is not accessible.

cell_properties property

cell_properties

The accessible attributes of cells stored in this layer. All PatchCell methods decorated by raster_attribute should be appeared here.

xda property

xda

Get the xarray raster layer with spatial coordinates.

attributes property

attributes

All accessible attributes from this layer.

shape2d property

shape2d

Raster shape in 2D (height, width). This is useful when working with 2d numpy.array.

shape3d property

shape3d

Raster shape in 3D (1, heigh, width). This is useful when working with rasterio band.

cells cached property

cells

The cells stored in this layer.

array_cells cached property

array_cells

Array of cells stored in this module.

Returns a 2D numpy array with dtype object containing PatchCell.

coords property

coords

Coordinate system of the raster data.

This is useful when working with xarray.DataArray.

agents property

agents

Return a list of all agents in the module.

random property

random

Randomly

transform_coord

transform_coord(row, col)

Converts grid indices to real-world coordinates.

Parameters:

Name Type Description Default
row int

Grid row index.

required
col int

Grid column index.

required

Returns:

Type Description
Coordinate

Tuple of (x, y) real-world coordinates.

Raises:

Type Description
IndexError

If indices are out of bounds.

Source code in abses/space/patch.py
def transform_coord(self, row: int, col: int) -> Coordinate:
    """Converts grid indices to real-world coordinates.

    Args:
        row: Grid row index.
        col: Grid column index.

    Returns:
        Tuple of (x, y) real-world coordinates.

    Raises:
        IndexError: If indices are out of bounds.
    """
    if self.indices_out_of_bounds(pos=(row, col)):
        raise IndexError(f"Out of bounds: {row, col}")
    return self.transform * (col, row)

dynamic_var

dynamic_var(attr_name, dtype='numpy')

Update and get dynamic variable.

Parameters:

Name Type Description Default
attr_name str

The dynamic variable to retrieve.

required

Returns:

Type Description
ndarray | DataArray

2D numpy.ndarray data of the variable.

Source code in abses/space/patch.py
def dynamic_var(
    self,
    attr_name: str,
    dtype: Literal["numpy", "xarray"] = "numpy",
) -> np.ndarray | xr.DataArray:
    """Update and get dynamic variable.

    Parameters:
        attr_name:
            The dynamic variable to retrieve.

    Returns:
        2D numpy.ndarray data of the variable.
    """
    # 获取动态变量,及其附加属性
    array = super().dynamic_var(attr_name)
    assert isinstance(array, (np.ndarray, xr.DataArray, xr.Dataset))
    kwargs = super().dynamic_variables[attr_name].attrs
    # 将矩阵转换为三维,并更新空间数据
    self.apply_raster(array, attr_name=attr_name, **kwargs)
    if dtype == "numpy":
        return self.get_raster(attr_name, update=False)
    if dtype == "xarray":
        return self.get_xarray(attr_name, update=False)
    raise ValueError(f"Unknown expected dtype {dtype}.")

get_xarray

get_xarray(attr_name=None, update=True)

Creates an xarray DataArray representation with spatial coordinates.

Parameters:

Name Type Description Default
attr_name Optional[str]

Attribute to retrieve. If None, returns all attributes.

None
update bool

If True, updates dynamic variables before retrieval.

True

Returns:

Type Description
DataArray

xarray.DataArray with spatial coordinates and CRS information.

Source code in abses/space/patch.py
def get_xarray(
    self,
    attr_name: Optional[str] = None,
    update: bool = True,
) -> xr.DataArray:
    """Creates an xarray DataArray representation with spatial coordinates.

    Args:
        attr_name: Attribute to retrieve. If None, returns all attributes.
        update: If True, updates dynamic variables before retrieval.

    Returns:
        xarray.DataArray with spatial coordinates and CRS information.
    """
    data = self.get_raster(attr_name=attr_name, update=update)
    if attr_name:
        name = attr_name
        data = data.reshape(self.shape2d)
        coords = self.coords
    else:
        coords = {"variable": list(self.attributes)}
        coords |= self.coords
        name = self.name
    return xr.DataArray(
        data=data,
        name=name,
        coords=coords,
    ).rio.write_crs(self.crs)

select

select(where=None)

Selects cells based on specified criteria.

Parameters:

Name Type Description Default
where Optional[CellFilter | dict[str, Any]]

Selection filter. Can be: - None: Select all cells - str: Select by attribute name - numpy.ndarray: Boolean mask array - Shapely.Geometry: Select cells intersecting geometry - dict: Select by attribute-value pairs, e.g., {"state": 3}

None

Returns:

Type Description
ActorsList[PatchCell]

ActorsList containing selected cells.

Raises:

Type Description
TypeError

If where parameter is of unsupported type.

Example

Select cells with elevation > 100

high_cells = module.select(module.get_raster("elevation") > 100)

Select cells within polygon

cells = module.select(polygon)

Select cells by attribute dict

burned = module.select({"state": 3})

Source code in abses/space/patch.py
def select(
    self,
    where: Optional[CellFilter | dict[str, Any]] = None,
) -> ActorsList[PatchCell]:
    """Selects cells based on specified criteria.

    Args:
        where: Selection filter. Can be:
            - None: Select all cells
            - str: Select by attribute name
            - numpy.ndarray: Boolean mask array
            - Shapely.Geometry: Select cells intersecting geometry
            - dict: Select by attribute-value pairs, e.g., {"state": 3}

    Returns:
        ActorsList containing selected cells.

    Raises:
        TypeError: If where parameter is of unsupported type.

    Example:
        >>> # Select cells with elevation > 100
        >>> high_cells = module.select(module.get_raster("elevation") > 100)
        >>> # Select cells within polygon
        >>> cells = module.select(polygon)
        >>> # Select cells by attribute dict
        >>> burned = module.select({"state": 3})
    """
    # Handle dictionary filters (common use case)
    if isinstance(where, dict):
        # Delegate to cells_lst.select which supports dict filters
        return self.cells_lst.select(where)

    # Handle other filter types
    if isinstance(where, Geometry):
        mask_ = self._select_by_geometry(geometry=where)
    elif isinstance(where, (np.ndarray, str, xr.DataArray)) or where is None:
        mask_ = self._attr_or_array(where).reshape(self.shape2d)
    else:
        raise TypeError(f"{type(where)} is not supported for selecting cells.")
    # mask_ is expected to be boolean here
    mask_bool = mask_.astype(bool)
    return ActorsList(self.model, self.array_cells[mask_bool])

apply

apply(ufunc, *args, **kwargs)

Apply a function to array cells.

Parameters:

Name Type Description Default
ufunc Callable[..., Any]

A function to apply.

required
*args Any

Positional arguments to pass to the function.

()
**kwargs Any

Keyword arguments to pass to the function.

{}

Returns:

Type Description
ndarray

The result of the function applied to the array cells.

Source code in abses/space/patch.py
def apply(self, ufunc: Callable[..., Any], *args: Any, **kwargs: Any) -> np.ndarray:
    """Apply a function to array cells.

    Parameters:
        ufunc:
            A function to apply.
        *args:
            Positional arguments to pass to the function.
        **kwargs:
            Keyword arguments to pass to the function.

    Returns:
        The result of the function applied to the array cells.
    """
    func = functools.partial(ufunc, *args, **kwargs)
    return np.vectorize(func)(self.array_cells)

coord_iter

coord_iter()

Iterate over coordinates and cells with precise typing.

Source code in abses/space/patch.py
def coord_iter(self) -> Iterator[tuple[Coordinate, PatchCell]]:
    """Iterate over coordinates and cells with precise typing."""
    arr = self.array_cells
    height, width = arr.shape
    for i in range(height):
        for j in range(width):
            yield (i, j), arr[i, j]

apply_raster

apply_raster(data, attr_name=None, **kwargs)

Applies raster data to cells as attributes.

Parameters:

Name Type Description Default
data Raster

Input raster data. Can be: - numpy.ndarray: 2D array matching module shape - xarray.DataArray: With spatial coordinates - xarray.Dataset: With named variables

required
attr_name Optional[str]

Name for the new attribute. Required for xarray.Dataset.

None
**kwargs Any

Additional options: cover_crs: Whether to override input data CRS resampling_method: Method for resampling ("nearest", etc.) flipud: Whether to flip data vertically

{}

Raises:

Type Description
ValueError

If attr_name not provided for Dataset input.

ValueError

If data shape doesn't match module shape.

Example

Apply elevation data

module.apply_raster(elevation_array, attr_name="elevation")

Apply data from xarray

module.apply_raster(xda, resampling_method="bilinear")

Source code in abses/space/patch.py
def apply_raster(
    self, data: Raster, attr_name: Optional[str] = None, **kwargs: Any
) -> None:
    """Applies raster data to cells as attributes.

    Args:
        data: Input raster data. Can be:
            - numpy.ndarray: 2D array matching module shape
            - xarray.DataArray: With spatial coordinates
            - xarray.Dataset: With named variables
        attr_name: Name for the new attribute. Required for xarray.Dataset.
        **kwargs: Additional options:
            cover_crs: Whether to override input data CRS
            resampling_method: Method for resampling ("nearest", etc.)
            flipud: Whether to flip data vertically

    Raises:
        ValueError: If attr_name not provided for Dataset input.
        ValueError: If data shape doesn't match module shape.

    Example:
        >>> # Apply elevation data
        >>> module.apply_raster(elevation_array, attr_name="elevation")
        >>> # Apply data from xarray
        >>> module.apply_raster(xda, resampling_method="bilinear")
    """
    if isinstance(data, np.ndarray):
        self._add_attribute(data, attr_name, **kwargs)
    elif isinstance(data, xr.DataArray):
        self._add_dataarray(data, attr_name, **kwargs)
    elif isinstance(data, xr.Dataset):
        if attr_name is None:
            raise ValueError("Attribute name is required for xr.Dataset.")
        dataarray = data[attr_name]
        self._add_dataarray(dataarray, attr_name, **kwargs)

get_raster

get_raster(attr_name=None, update=True)

Obtaining the Raster layer by attribute.

Parameters:

Name Type Description Default
attr_name Optional[str]

The attribute to retrieve. If None (by default), retrieve all attributes as a 3D array.

None

Returns:

Type Description
ndarray

A 3D array of attribute.

Source code in abses/space/patch.py
def get_raster(
    self,
    attr_name: Optional[str] = None,
    update: bool = True,
) -> np.ndarray:
    """Obtaining the Raster layer by attribute.

    Parameters:
        attr_name:
            The attribute to retrieve.
            If None (by default), retrieve all attributes as a 3D array.

    Returns:
        A 3D array of attribute.
    """
    if attr_name in self.dynamic_variables and update:
        return self.dynamic_var(attr_name=attr_name).reshape(self.shape3d)
    if attr_name is not None and attr_name not in self.attributes:
        raise ValueError(
            f"Attribute {attr_name} does not exist. "
            f"Choose from {self.attributes}, or set `attr_name` to `None` to retrieve all."
        )
    if attr_name is None:
        assert bool(self.attributes), "No attribute available."
        attr_names = self.attributes
    else:
        attr_names = {attr_name}
    data = []
    for name in attr_names:
        array = np.vectorize(getattr)(self.array_cells, name)
        data.append(array)
    return np.stack(data)

reproject

reproject(xda, resampling='nearest', **kwargs)

Reproject the xarray data to the same CRS as this layer.

Source code in abses/space/patch.py
def reproject(
    self,
    xda: xr.DataArray,
    resampling: Resampling | str = "nearest",
    **kwargs,
) -> xr.DataArray:
    """Reproject the xarray data to the same CRS as this layer."""
    if isinstance(resampling, str):
        resampling = getattr(Resampling, resampling)
    return xda.rio.reproject_match(self.xda, resampling=resampling, **kwargs)

get_neighboring_cells

get_neighboring_cells(
    pos, moore, include_center=False, radius=1
)

Gets neighboring cells around a position.

Parameters:

Name Type Description Default
pos Coordinate

Center position (x, y).

required
moore bool

If True, uses Moore neighborhood (8 neighbors). If False, uses von Neumann neighborhood (4 neighbors).

required
include_center bool

Whether to include the center cell.

False
radius int

Neighborhood radius in cells.

1

Returns:

Type Description
ActorsList[PatchCell]

ActorsList containing neighboring cells.

Example

Get Moore neighborhood with radius 2

neighbors = module.get_neighboring_cells((5,5), moore=True, radius=2)

Source code in abses/space/patch.py
def get_neighboring_cells(
    self,
    pos: Coordinate,
    moore: bool,
    include_center: bool = False,
    radius: int = 1,
) -> ActorsList[PatchCell]:
    """Gets neighboring cells around a position.

    Args:
        pos: Center position (x, y).
        moore: If True, uses Moore neighborhood (8 neighbors).
              If False, uses von Neumann neighborhood (4 neighbors).
        include_center: Whether to include the center cell.
        radius: Neighborhood radius in cells.

    Returns:
        ActorsList containing neighboring cells.

    Example:
        >>> # Get Moore neighborhood with radius 2
        >>> neighbors = module.get_neighboring_cells((5,5), moore=True, radius=2)
    """
    cells = super().get_neighboring_cells(pos, moore, include_center, radius)
    return ActorsList(self.model, cells)

get_neighboring_by_indices cached

get_neighboring_by_indices(
    indices,
    moore,
    include_center=False,
    radius=1,
    annular=False,
)

Getting neighboring positions of the given coordinate.

Parameters:

Name Type Description Default
indices Coordinate

The indices to get the neighborhood.

required
moore bool

Whether to use Moore neighborhood. If False, use Von Neumann neighborhood.

required
include_center bool

Whether to include the center cell. Default is False.

False
radius int

The radius of the neighborhood. Default is 1.

1
annular bool

Whether to use annular (ring) neighborhood. Default is False.

False

Returns:

Type Description
ActorsList[PatchCell]

An ActorsList of neighboring cells.

Source code in abses/space/patch.py
@functools.lru_cache(maxsize=1000)
def get_neighboring_by_indices(
    self,
    indices: Coordinate,
    moore: bool,
    include_center: bool = False,
    radius: int = 1,
    annular: bool = False,
) -> ActorsList[PatchCell]:
    """Getting neighboring positions of the given coordinate.

    Parameters:
        indices:
            The indices to get the neighborhood.
        moore:
            Whether to use Moore neighborhood.
            If False, use Von Neumann neighborhood.
        include_center:
            Whether to include the center cell.
            Default is False.
        radius:
            The radius of the neighborhood.
            Default is 1.
        annular:
            Whether to use annular (ring) neighborhood.
            Default is False.

    Returns:
        An `ActorsList` of neighboring cells.
    """
    row, col = indices
    mask_arr = np.zeros(self.shape2d, dtype=bool)
    mask_arr[row, col] = True
    mask_arr = get_buffer(mask_arr, radius=radius, moor=moore, annular=annular)
    mask_arr[row, col] = include_center
    return ActorsList(self.model, self.array_cells[mask_arr])

indices_out_of_bounds

indices_out_of_bounds(pos)

Determines whether position is off the grid.

Parameters:

Name Type Description Default
pos Coordinate

Position to check.

required

Returns:

Type Description
bool

True if position is off the grid, False otherwise.

Source code in abses/space/patch.py
def indices_out_of_bounds(self, pos: Coordinate) -> bool:
    """
    Determines whether position is off the grid.

    Parameters:
        pos: Position to check.

    Returns:
        True if position is off the grid, False otherwise.
    """

    row, col = pos
    return row < 0 or row >= self.height or col < 0 or col >= self.width

count_agents

count_agents(agent_type=None, *, dtype='xarray')

Count the number of agents of a specific type on each cell across the entire module.

Parameters:

Name Type Description Default
agent_type Type[ActorProtocol] | None

The agent class to count (e.g., Sheep, Wolf).

None
dtype Literal['numpy', 'xarray']

Return type. Options: - "numpy": Returns 2D numpy array - "xarray": Returns xarray DataArray with spatial coordinates (default)

'xarray'

Returns:

Type Description
NDArray[int_] | DataArray

Agent counts as numpy array or xarray DataArray with shape (height, width).

Examples:

>>> # Get as xarray with spatial coordinates (default)
>>> sheep_map = grassland.count_agents(Sheep)
>>> # Now has .rio methods and coordinates
>>> sheep_map.rio.crs
>>> sheep_map.plot()
>>>
>>> # Get as numpy array
>>> sheep_map = grassland.count_agents(Sheep, dtype="numpy")
Source code in abses/space/patch.py
def count_agents(
    self,
    agent_type: Type[ActorProtocol] | None = None,
    *,
    dtype: Literal["numpy", "xarray"] = "xarray",
) -> NDArray[np.int_] | xr.DataArray:
    """
    Count the number of agents of a specific type on each cell across the entire module.

    Args:
        agent_type: The agent class to count (e.g., Sheep, Wolf).
        dtype: Return type. Options:
            - "numpy": Returns 2D numpy array
            - "xarray": Returns xarray DataArray with spatial coordinates (default)

    Returns:
        Agent counts as numpy array or xarray DataArray with shape (height, width).

    Examples:
        >>> # Get as xarray with spatial coordinates (default)
        >>> sheep_map = grassland.count_agents(Sheep)
        >>> # Now has .rio methods and coordinates
        >>> sheep_map.rio.crs
        >>> sheep_map.plot()
        >>>
        >>> # Get as numpy array
        >>> sheep_map = grassland.count_agents(Sheep, dtype="numpy")
    """
    data = self.apply(lambda cell: cell.agents.has(agent_type))

    if dtype == "xarray":
        # Convert to xarray with spatial coordinates
        xda = xr.DataArray(
            data=data,
            coords=self.coords,
            name=agent_type.__name__.lower() + "_count"
            if agent_type
            else "agent_count",
        )
        # Add spatial reference information
        xda = xda.rio.write_crs(self.crs)
        return xda

    return data

apply_agents

apply_agents(
    func=None,
    *,
    attr=None,
    default=np.nan,
    aggregator=None,
    dtype="numpy",
    name=None
)

Apply a function or attribute access to the agent(s) on each cell.

This method vectorizes over the raster cells and, for each cell, operates on its linked agents. It supports three usage modes:

1) Single-agent access (capacity=1 typical): - Provide attr (str) to fetch an attribute from the only agent. - Or provide func(agent) to compute a value from the only agent. If the cell is empty, returns default.

2) Aggregation over multiple agents per cell: - Provide aggregator(actors: ActorsList) to reduce the cell's agents into a single value (e.g., len(actors), sum(...)).

3) xarray output: - Set dtype='xarray' to get an xr.DataArray with spatial coords and CRS info baked in.

Parameters:

Name Type Description Default
func Optional[Callable[[Actor], Any]]

Function to apply to a single agent on each cell.

None
attr Optional[str]

Attribute name to fetch from a single agent on each cell.

None
default Any

Value to use when a cell has no agent (or attribute missing).

nan
aggregator Optional[Callable[[ActorsList[Actor]], Any]]

Function that reduces ActorsList of a cell to one value.

None
dtype Literal['numpy', 'xarray']

Output type. 'numpy' or 'xarray'.

'numpy'
name Optional[str]

Optional name for the xarray DataArray.

None

Returns:

Type Description
ndarray | DataArray

A 2D numpy array or xarray DataArray of computed values per cell.

Notes
  • If both aggregator and (func/attr) are provided, aggregator takes precedence and receives the entire cell ActorsList.
  • When using attr or func and multiple agents exist on a cell, the first agent is used via actors.item(default=None).
Source code in abses/space/patch.py
def apply_agents(
    self,
    func: Optional[Callable[[Actor], Any]] = None,
    *,
    attr: Optional[str] = None,
    default: Any = np.nan,
    aggregator: Optional[Callable[[ActorsList[Actor]], Any]] = None,
    dtype: Literal["numpy", "xarray"] = "numpy",
    name: Optional[str] = None,
) -> np.ndarray | xr.DataArray:
    """Apply a function or attribute access to the agent(s) on each cell.

    This method vectorizes over the raster cells and, for each cell, operates on
    its linked agents. It supports three usage modes:

    1) Single-agent access (capacity=1 typical):
       - Provide ``attr`` (str) to fetch an attribute from the only agent.
       - Or provide ``func(agent)`` to compute a value from the only agent.
       If the cell is empty, returns ``default``.

    2) Aggregation over multiple agents per cell:
       - Provide ``aggregator(actors: ActorsList)`` to reduce the cell's
         agents into a single value (e.g., len(actors), sum(...)).

    3) xarray output:
       - Set ``dtype='xarray'`` to get an ``xr.DataArray`` with spatial coords
         and CRS info baked in.

    Args:
        func: Function to apply to a single agent on each cell.
        attr: Attribute name to fetch from a single agent on each cell.
        default: Value to use when a cell has no agent (or attribute missing).
        aggregator: Function that reduces ``ActorsList`` of a cell to one value.
        dtype: Output type. ``'numpy'`` or ``'xarray'``.
        name: Optional name for the xarray DataArray.

    Returns:
        A 2D numpy array or xarray DataArray of computed values per cell.

    Notes:
        - If both ``aggregator`` and (``func``/``attr``) are provided, ``aggregator``
          takes precedence and receives the entire cell ``ActorsList``.
        - When using ``attr`` or ``func`` and multiple agents exist on a cell,
          the first agent is used via ``actors.item(default=None)``.
    """

    if aggregator is None and func is None and attr is None:
        raise ValueError("One of 'aggregator', 'func', or 'attr' must be provided.")

    def _to_float(val: Any) -> float:
        """Cast to float, returning NaN on failure."""
        try:
            return float(val)
        except Exception:
            return float("nan")

    def _cell_compute(cell: PatchCell) -> Any:
        # Aggregation path over all agents on the cell
        if aggregator is not None:
            actors_list = ActorsList(self.model, cell.agents)
            if len(actors_list) == 0:
                return default
            try:
                return _to_float(aggregator(actors_list))
            except Exception:
                logger.warning(
                    "apply_agents aggregator raised; filling NaN.",
                )
                return float("nan")

        # Single-agent path (use first/only agent if present)
        agent = cell.agents.item(default=None)
        if agent is None:
            return default
        if attr is not None:
            try:
                return _to_float(getattr(agent, attr, default))
            except Exception:
                logger.warning(
                    "apply_agents attr access raised; filling NaN. attr=%r",
                    attr,
                )
                return float("nan")
        # func provided
        if func is None:
            return default
        try:
            return _to_float(func(agent))
        except Exception:
            logger.warning(
                "apply_agents func raised; filling NaN. func=%r",
                getattr(func, "__name__", repr(func)),
            )
            return float("nan")

    # compute and force float dtype for NaN safety
    data = self.apply(_cell_compute).astype(float)

    if dtype == "xarray":
        xda = xr.DataArray(
            data=data, coords=self.coords, name=name or "applied_agents"
        )
        xda = xda.rio.write_crs(self.crs)
        return xda

    return data