PatchModule (layer)

Bases: Module, RasterBase

The spatial sub-module base class. Inherit from this class to create a submodule. This tutorial shows the model structure. This is also a raster layer, inherited from the 'mesa-geo.RasterLayer' class. ABSESpy extends this class, so it can: 1. place agents (by _CellAgentsContainer class.) 2. work with xarray, rasterio packages for better data I/O workflow.

Attributes:

Name	Type	Description
`cell_properties`	`set[str]`	The accessible attributes of cells stored in this layer. When a `PatchCell`'s method is decorated by `raster_attribute`, it should be appeared here as a property attribute.
`attributes`	`set[str]`	All accessible attributes from this layer, including cell_properties.
`shape2d`	`Coordinate`	Raster shape in 2D (heigh, width).
`shape3d`	`Coordinate`	Raster shape in 3D (1, heigh, width), this is for compatibility with `mg.RasterLayer` and `rasterio`.
`array_cells`	`ndarray`	Array type of the `PatchCell` stored in this module.
`coords`	`Coordinate`	Coordinate system of the raster data. This is useful when working with `xarray.DataArray`.
`random`	`ListRandom`	A random proxy by calling the cells as an `ActorsList`.

Source code in abses/patch.py

def __init__(
    self,
    model: MainModel[Any, Any],
    name: Optional[str] = None,
    cell_cls: Type[PatchCell] = PatchCell,
    **kwargs: Any,
):
    """This method copied some of the `mesa-geo.RasterLayer`'s methods."""
    Module.__init__(self, model, name=name)
    RasterBase.__init__(self, **kwargs)
    self.cell_cls = cell_cls
    logger.info("Initializing a new Model Layer...")
    logger.info(f"Using rioxarray version: {rioxarray.__version__}")

    func = np.vectorize(lambda row, col: cell_cls(self, (row, col)))
    self._cells: np.ndarray = np.fromfunction(
        func, shape=(self.height, self.width), dtype=object
    )
    self._attributes: Set[str] = set()
    self.mask: np.ndarray = np.ones(self.shape2d)

mask `property` `writable` ¶

mask

Where is not accessible.

cells `cached` `property` ¶

cells

The cells stored in this layer.

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 `cached` `property` ¶

xda

Get the xarray raster layer with spatial coordinates.

plot `cached` `property` ¶

plot

Plotting

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.

array_cells `cached` `property` ¶

array_cells

Array type of the PatchCell stored in this module.

coords `property` ¶

coords

Coordinate system of the raster data.

This is useful when working with xarray.DataArray.

random `property` ¶

random

Randomly

transform_coord ¶

transform_coord(row, col)

Transforming the row, col to the real-world coordinate.

Source code in abses/patch.py

def transform_coord(self, row: int, col: int) -> Coordinate:
    """Transforming the row, col to the real-world coordinate."""
    if self.out_of_bounds(pos=(row, col)):
        raise IndexError(f"Out of bounds: {row, col}")
    return self.transform * (col, row)

to_crs ¶

to_crs(crs, inplace=False)

Converting the raster data to a another CRS.

Source code in abses/patch.py

def to_crs(self, crs, inplace=False) -> Optional[PatchModule]:
    """Converting the raster data to a another CRS."""
    super()._to_crs_check(crs)
    layer = self if inplace else copy.copy(self)

    src_crs = rasterio.crs.CRS.from_user_input(layer.crs)
    dst_crs = rasterio.crs.CRS.from_user_input(crs)
    if not layer.crs.is_exact_same(crs):
        transform, _, _ = calculate_default_transform(
            src_crs,
            dst_crs,
            self.width,
            self.height,
            *layer.total_bounds,
        )
        layer.total_bounds = [
            *transform_bounds(src_crs, dst_crs, *layer.total_bounds)
        ]
        layer.crs = crs
        layer.transform = transform

    return None if inplace else layer

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`	2D numpy.ndarray data of the variable.

Source code in abses/patch.py

def dynamic_var(
    self,
    attr_name: str,
    dtype: Literal["numpy", "xarray", "rasterio"] = "numpy",
) -> np.ndarray:
    """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)
    if dtype == "rasterio":
        return self.get_rasterio(attr_name, update=False)
    raise ValueError(f"Unknown expected dtype {dtype}.")

get_rasterio ¶

get_rasterio(attr_name=None, update=True)

Gets the Rasterio raster layer corresponding to the attribute. Save to a temporary rasterio memory file.

Parameters:

Name	Type	Description	Default
`attr_name`	`Optional[str]`	The attribute name for creating the rasterio file.	`None`

Returns:

Type	Description
`MemoryFile`	The rasterio tmp memory file of raster.

Source code in abses/patch.py

def get_rasterio(
    self,
    attr_name: Optional[str] = None,
    update: bool = True,
) -> rasterio.MemoryFile:
    """Gets the Rasterio raster layer corresponding to the attribute. Save to a temporary rasterio memory file.

    Parameters:
        attr_name:
            The attribute name for creating the rasterio file.

    Returns:
        The rasterio tmp memory file of raster.
    """
    if attr_name is None:
        data = np.ones(self.shape2d)
    else:
        data = self.get_raster(attr_name=attr_name, update=update)
    # 如果获取到的是2维，重整为3维
    if len(data.shape) != 3:
        data = data.reshape(self.shape3d)
    with rasterio.MemoryFile() as mem_file:
        with mem_file.open(
            driver="GTiff",
            height=data.shape[1],
            width=data.shape[2],
            count=data.shape[0],  # number of bands
            dtype=str(data.dtype),
            crs=self.crs,
            transform=self.transform,
        ) as dataset:
            dataset.write(data)
        # Open the dataset again for reading and return
        return mem_file.open()

get_xarray ¶

get_xarray(attr_name=None, update=True)

Get the xarray raster layer with spatial coordinates.

Parameters:

Name	Type	Description	Default
`attr_name`	`Optional[str]`	The attribute to retrieve. If None (by default), return all available attributes (3D DataArray). Otherwise, 2D DataArray of the chosen attribute.	`None`

Returns:

Type	Description
`DataArray`	Xarray.DataArray data with spatial coordinates of the chosen attribute.

Source code in abses/patch.py

def get_xarray(
    self,
    attr_name: Optional[str] = None,
    update: bool = True,
) -> xr.DataArray:
    """Get the xarray raster layer with spatial coordinates.

    Parameters:
        attr_name:
            The attribute to retrieve. If None (by default),
            return all available attributes (3D DataArray).
            Otherwise, 2D DataArray of the chosen attribute.

    Returns:
        Xarray.DataArray data with spatial coordinates of the chosen attribute.
    """
    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)

Select cells from this layer. Also has a shortcut alias for this method: .sel.

Parameters:

Name	Type	Description	Default
`where`	`Optional[CellFilter]`	The condition to select cells. If None (by default), select all cells. If a string, select cells by the attribute name. If a numpy.ndarray, select cells by the mask array. If a Shapely Geometry, select cells by the intersection with the geometry.	`None`

Raises:

Type	Description
`TypeError`	If the input type is not supported.

Returns:

Type	Description
`ActorsList[PatchCell]`	An `ActorsList` with all selected cells stored.

Source code in abses/patch.py

def select(
    self,
    where: Optional[CellFilter] = None,
) -> ActorsList[PatchCell]:
    """Select cells from this layer.
    Also has a shortcut alias for this method: `.sel`.

    Parameters:
        where:
            The condition to select cells.
            If None (by default), select all cells.
            If a string, select cells by the attribute name.
            If a numpy.ndarray, select cells by the mask array.
            If a Shapely Geometry, select cells by the intersection with the geometry.

    Raises:
        TypeError:
            If the input type is not supported.

    Returns:
        An `ActorsList` with all selected cells stored.
    """
    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_ = np.nan_to_num(mask_, nan=0.0).astype(bool) & self.mask
    return ActorsList(self.model, self.array_cells[mask_])

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/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, *args, **kwargs)

coord_iter ¶

coord_iter()

An iterator that returns coordinates as well as cell contents.

Source code in abses/patch.py

def coord_iter(self) -> Iterator[tuple[Coordinate, PatchCell]]:
    """
    An iterator that returns coordinates as well as cell contents.
    """
    return np.ndenumerate(self.array_cells)

apply_raster ¶

apply_raster(data, attr_name, **kwargs)

Apply raster data to the cells.

Parameters:

Name	Type	Description	Default
`data`	`Raster`	np.ndarray data: 2D numpy array with shape (1, height, width). xr.DataArray data: xarray DataArray with spatial coordinates. xr.Dataset data: xarray Dataset with spatial coordinates.	required
`attr_name`	`Optional[str]`	Name of the attribute to be added to the cells. If None, a random name will be generated. Default is None.	required
`**kwargs`	`Any`	cover_crs: Whether to cover the crs of the input data. If False, it assumes the input data has crs info. If True, it will cover the crs of the input data by the crs of this layer. Default is False. resampling_method: The resampling method when re-projecting the input data. Default is "nearest". flipud: Whether to flip the input data upside down. Set to True when the input data is not in the same direction as the raster layer. Default is False.	`{}`

Source code in abses/patch.py

def apply_raster(
    self, data: Raster, attr_name: Optional[str], **kwargs: Any
) -> None:
    """Apply raster data to the cells.

    Parameters:
        data:
            np.ndarray data: 2D numpy array with shape (1, height, width).
            xr.DataArray data: xarray DataArray with spatial coordinates.
            xr.Dataset data: xarray Dataset with spatial coordinates.
        attr_name:
            Name of the attribute to be added to the cells.
            If None, a random name will be generated.
            Default is None.
        **kwargs:
            cover_crs:
                Whether to cover the crs of the input data.
                If False, it assumes the input data has crs info.
                If True, it will cover the crs of the input data by the crs of this layer.
                Default is False.
            resampling_method:
                The [resampling method](https://rasterio.readthedocs.io/en/stable/api/rasterio.enums.html#rasterio.enums.Resampling)
                when re-projecting the input data.
                Default is "nearest".
            flipud:
                Whether to flip the input data upside down.
                Set to True when the input data is not in the same direction as the raster layer.
                Default is False.
    """
    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/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/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_neighborhood `cached` ¶

get_neighborhood(pos, moore, include_center=False, radius=1, annular=False, return_mask=False)

Getting neighboring positions of the given coordinate.

Parameters:

Name	Type	Description	Default
`pos`	`Coordinate`	The coordinate 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`
`return_mask`	`bool`	Whether to return a mask array. If True, return a mask array of the neighboring locations. Otherwise, return a `ActorsList` of neighboring cells. Default is False.	`False`

Returns:

Type	Description
`ActorsList[PatchCell] \| ndarray`	An `ActorsList` of neighboring cells, or a mask array.
`ActorsList[PatchCell] \| ndarray`	Where the mask array is a boolean array with the same shape as the raster layer.
`ActorsList[PatchCell] \| ndarray`	The True value indicates the cell is in the neighborhood.

Source code in abses/patch.py

@functools.lru_cache(maxsize=1000)
def get_neighborhood(
    self,
    pos: Coordinate,
    moore: bool,
    include_center: bool = False,
    radius: int = 1,
    annular: bool = False,
    return_mask: bool = False,
) -> ActorsList[PatchCell] | np.ndarray:
    """Getting neighboring positions of the given coordinate.

    Parameters:
        pos:
            The coordinate 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.
        return_mask:
            Whether to return a mask array.
            If True, return a mask array of the neighboring locations.
            Otherwise, return a `ActorsList` of neighboring cells.
            Default is False.

    Returns:
        An `ActorsList` of neighboring cells, or a mask array.
        Where the mask array is a boolean array with the same shape as the raster layer.
        The True value indicates the cell is in the neighborhood.
    """
    mask_arr = np.zeros(self.shape2d, dtype=bool)
    mask_arr[pos[0], pos[1]] = True
    mask_arr = get_buffer(
        mask_arr, radius=radius, moor=moore, annular=annular
    )
    mask_arr[pos[0], pos[1]] = include_center
    if return_mask:
        return mask_arr
    return ActorsList(self.model, self.array_cells[mask_arr])

to_file ¶

to_file(raster_file, attr_name=None, driver='GTiff')

Writes a raster layer to a file.

Parameters:

Name	Type	Description	Default
`raster_file`	`str`	The path to the raster file to write to.	required
`attr_name`	`Optional[str]`	The name of the attribute to write to the raster. If None, all attributes are written. Default is None.	`None`
`driver`	`str`	The GDAL driver to use for writing the raster file. Default is 'GTiff'. See GDAL docs at https://gdal.org/drivers/raster/index.html.	`'GTiff'`

Source code in abses/patch.py

def to_file(
    self,
    raster_file: str,
    attr_name: Optional[str] = None,
    driver: str = "GTiff",
) -> None:
    """
    Writes a raster layer to a file.

    Parameters:
        raster_file:
            The path to the raster file to write to.
        attr_name:
            The name of the attribute to write to the raster.
            If None, all attributes are written. Default is None.
        driver:
            The GDAL driver to use for writing the raster file.
            Default is 'GTiff'.
            See GDAL docs at https://gdal.org/drivers/raster/index.html.
    """

    data = self.get_raster(attr_name)
    with rasterio.open(
        raster_file,
        "w",
        driver=driver,
        width=self.width,
        height=self.height,
        count=data.shape[0],
        dtype=data.dtype,
        crs=self.crs,
        transform=self.transform,
    ) as dataset:
        dataset.write(data)

out_of_bounds ¶

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/patch.py

def 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

PatchModule (layer)

mask property writable ¶

cells cached property ¶

cell_properties property ¶

xda cached property ¶

plot cached property ¶

attributes property ¶

shape2d property ¶

shape3d property ¶

array_cells cached property ¶

coords property ¶

random property ¶

transform_coord ¶

to_crs ¶

dynamic_var ¶

get_rasterio ¶

get_xarray ¶

select ¶

apply ¶

coord_iter ¶

apply_raster ¶

get_raster ¶

reproject ¶

get_neighborhood cached ¶

to_file ¶

out_of_bounds ¶

mask `property` `writable` ¶

cells `cached` `property` ¶

cell_properties `property` ¶

xda `cached` `property` ¶

plot `cached` `property` ¶

attributes `property` ¶

shape2d `property` ¶

shape3d `property` ¶

array_cells `cached` `property` ¶

coords `property` ¶

random `property` ¶

get_neighborhood `cached` ¶