Geo — Raster

Raster substrate classes and utilities backed by NumPy arrays.

dissmodel.geo.raster.backend

dissmodel/geo/raster/backend.py

Vectorized engine for cellular automata on raster grids (NumPy 2D arrays).

Responsibility

Provide generic spatial operations (shift, dilate, focal_sum, snapshot) with no domain knowledge — no land-use classes, no CRS, no I/O, no project-specific constants.

Domain models (FloodRasterModel, MangroveRasterModel, …) import RasterBackend and operate on named arrays stored in self.arrays.

Minimal example

from dissmodel.geo.raster.backend import RasterBackend, DIRS_MOORE

b = RasterBackend(shape=(100, 100))
b.set("state", np.zeros((100, 100), dtype=np.int8))

state    = b.get("state").copy()          # equivalent to cell.past[attr]
contact  = b.neighbor_contact(state == 1)
for dr, dc in DIRS_MOORE:
    neighbour = RasterBackend.shift2d(state, dr, dc)
    ...
b.arrays["state"] = new_state

DIRS_MOORE = [(-1, -1), (-1, 0), (-1, 1), (0, -1), (0, 1), (1, -1), (1, 0), (1, 1)] module-attribute

DIRS_VON_NEUMANN = [(-1, 0), (0, -1), (0, 1), (1, 0)] module-attribute

RasterBackend

Storage and vectorized operations for 2D raster grids.

Replaces TerraME's forEachCell / forEachNeighbor with pure NumPy operations. The backend is shared across multiple models running in the same Environment — each model reads and writes named arrays every step.

Arrays

Stored in self.arrays as np.ndarray of shape (rows, cols). No names are reserved — domain models define their own ("uso", "alt", "solo", "state", "temperature", …).

Parameters:

Name	Type	Description	Default
shape	tuple[int, int]	Grid shape as (rows, cols).	required
nodata_value	float | int | None	Sentinel value used to mark cells outside the study extent. When provided, nodata_mask derives the extent mask automatically, so RasterMap renders those cells as transparent without any extra configuration. Default: None.	None

Examples:

>>> b = RasterBackend(shape=(10, 10))
>>> b.set("state", np.zeros((10, 10), dtype=np.int8))
>>> b.get("state").shape
(10, 10)

>>> b = RasterBackend(shape=(10, 10), nodata_value=-1)
>>> b.nodata_mask   # True = valid cell, False = outside extent

Source code in dissmodel/geo/raster/backend.py

class RasterBackend:
    """
    Storage and vectorized operations for 2D raster grids.

    Replaces TerraME's ``forEachCell`` / ``forEachNeighbor`` with pure NumPy
    operations. The backend is shared across multiple models running in the
    same ``Environment`` — each model reads and writes named arrays every step.

    Arrays
    ------
    Stored in ``self.arrays`` as ``np.ndarray`` of shape ``(rows, cols)``.
    No names are reserved — domain models define their own
    (``"uso"``, ``"alt"``, ``"solo"``, ``"state"``, ``"temperature"``, …).

    Parameters
    ----------
    shape : tuple[int, int]
        Grid shape as ``(rows, cols)``.
    nodata_value : float | int | None
        Sentinel value used to mark cells outside the study extent.
        When provided, ``nodata_mask`` derives the extent mask automatically,
        so ``RasterMap`` renders those cells as transparent without any extra
        configuration. Default: ``None``.

    Examples
    --------
    >>> b = RasterBackend(shape=(10, 10))
    >>> b.set("state", np.zeros((10, 10), dtype=np.int8))
    >>> b.get("state").shape
    (10, 10)

    >>> b = RasterBackend(shape=(10, 10), nodata_value=-1)
    >>> b.nodata_mask   # True = valid cell, False = outside extent
    """

    def __init__(
        self,
        shape: tuple[int, int],
        nodata_value: float | int | None = None,
        transform: Any = None,
        crs: Any = None,
    ) -> None:
        self.shape        = shape
        self.arrays: dict[str, np.ndarray] = {}
        self.nodata_value = nodata_value   # sentinel for out-of-extent cells

        self.transform    = transform
        self.crs          = crs

    # ── extent mask ───────────────────────────────────────────────────────────

    @property
    def nodata_mask(self) -> np.ndarray | None:
        """
        Boolean mask: ``True`` = valid cell, ``False`` = outside extent / nodata.

        Derived in priority order:
        1. ``arrays["mask"]``  — explicit mask band (dissluc / coastal convention:
                                 non-zero = valid).
        2. ``nodata_value``    — applied over the first available array.
        3. ``None``            — no information; ``RasterMap`` skips auto-masking.

        Used by ``RasterMap`` (``auto_mask=True``) to render out-of-extent pixels
        as transparent without any per-project configuration.
        """
        if "mask" in self.arrays:
            return self.arrays["mask"] != 0

        if self.nodata_value is not None and self.arrays:
            first = next(iter(self.arrays.values()))
            return first != self.nodata_value

        return None

    # ── read / write ──────────────────────────────────────────────────────────

    def set(self, name: str, array: np.ndarray) -> None:
        """Store a copy of ``array`` under ``name``."""
        self.arrays[name] = np.asarray(array).copy()

    def get(self, name: str) -> np.ndarray:
        """
        Return a direct reference to the named array.

        Use ``.copy()`` to obtain a snapshot equivalent to TerraME's ``.past``.

        Raises
        ------
        KeyError
            If ``name`` is not in ``self.arrays``.
        """
        return self.arrays[name]

    def snapshot(self) -> dict[str, np.ndarray]:
        """
        Return a deep copy of all arrays — equivalent to TerraME's ``.past`` mechanism.

        Typical usage::

            past     = backend.snapshot()
            uso_past = past["uso"]   # state at the beginning of the step

        Returns
        -------
        dict[str, np.ndarray]
            Dictionary mapping array names to independent copies.
        """
        return {k: v.copy() for k, v in self.arrays.items()}

    def rename_band(self, old: str, new: str) -> None:
        """
        Rename an array in-place. No-op if ``old`` does not exist.

        Parameters
        ----------
        old : str
            Current band name.
        new : str
            Target band name.
        """
        if old in self.arrays:
            self.arrays[new] = self.arrays.pop(old)

    # ── xarray interoperability ───────────────────────────────────────────────

    def to_xarray(self, time: int | None = None):
        """
        Convert the backend to an ``xr.Dataset``.

        Each array in ``self.arrays`` becomes a ``DataVariable`` with
        dimensions ``(y, x)``. If ``time`` is given, a scalar ``time``
        coordinate is added — useful when assembling multi-step outputs.

        Spatial coordinates are derived from ``self.transform`` when available
        (rasterio Affine), following the Pangeo convention of cell centres.
        When ``transform`` is ``None`` (e.g. backends rasterized from vector),
        integer pixel indices are used instead.

        CRS is stored as a ``spatial_ref`` coordinate (CF convention) when
        ``self.crs`` is set and ``pyproj`` is available.

        Parameters
        ----------
        time : int | None
            Optional simulation step to attach as a scalar ``time`` coordinate.

        Returns
        -------
        xr.Dataset

        Raises
        ------
        ImportError
            If ``xarray`` is not installed.

        Examples
        --------
        >>> ds = backend.to_xarray()
        >>> ds["uso"].dims
        ('y', 'x')

        >>> ds = backend.to_xarray(time=42)
        >>> ds.coords["time"].item()
        42
        """
        try:
            import xarray as xr
        except ImportError:
            raise ImportError(
                "xarray is required for RasterBackend.to_xarray(). "
                "Install it with: pip install xarray"
            )

        rows, cols = self.shape

        # spatial coordinates — cell centres from Affine transform or pixel indices
        if self.transform is not None:
            try:
                # rasterio Affine: transform * (col + 0.5, row + 0.5) = centre
                xs = np.array([self.transform.c + (c + 0.5) * self.transform.a
                                for c in range(cols)])
                ys = np.array([self.transform.f + (r + 0.5) * self.transform.e
                                for r in range(rows)])
            except AttributeError:
                # transform present but not rasterio Affine — fall back to indices
                xs = np.arange(cols, dtype=float)
                ys = np.arange(rows, dtype=float)
        else:
            xs = np.arange(cols, dtype=float)
            ys = np.arange(rows, dtype=float)

        coords: dict = {"y": ys, "x": xs}

        if time is not None:
            coords["time"] = time

        # CRS as spatial_ref coordinate (CF / rioxarray convention)
        if self.crs is not None:
            try:
                from pyproj import CRS as ProjCRS
                crs_obj = ProjCRS.from_user_input(self.crs)
                coords["spatial_ref"] = xr.DataArray(
                    0,
                    attrs={
                        "crs_wkt":       crs_obj.to_wkt(),
                        "grid_mapping":  "spatial_ref",
                    },
                )
            except Exception:
                # pyproj unavailable or CRS unresolvable — skip spatial_ref
                pass

        data_vars = {}
        for name, arr in self.arrays.items():
            attrs: dict = {}
            if self.nodata_value is not None:
                attrs["_FillValue"] = self.nodata_value
            if self.crs is not None and "spatial_ref" in coords:
                attrs["grid_mapping"] = "spatial_ref"

            data_vars[name] = xr.DataArray(
                arr.copy(),
                dims=["y", "x"],
                coords={"y": ys, "x": xs},
                attrs=attrs,
            )

        ds = xr.Dataset(data_vars, coords=coords)
        ds.attrs["Conventions"] = "CF-1.8"
        return ds

    @classmethod
    def from_xarray(cls, ds, nodata_value: float | int | None = None) -> "RasterBackend":
        """
        Build a ``RasterBackend`` from an ``xr.Dataset`` or ``xr.DataArray``.

        All variables with exactly two dimensions ``(y, x)`` (in any order)
        are imported as arrays. Variables with other dimensionality
        (e.g. ``spatial_ref`` scalars) are silently skipped.

        CRS is recovered from the ``spatial_ref`` coordinate (CF convention)
        when present and ``pyproj`` is available.

        Parameters
        ----------
        ds : xr.Dataset | xr.DataArray
            Source dataset. A ``DataArray`` is wrapped into a single-variable
            ``Dataset`` using ``da.name`` (falling back to ``"data"``).
        nodata_value : float | int | None
            Forwarded to the new backend's ``nodata_value``. Default: ``None``.

        Returns
        -------
        RasterBackend

        Raises
        ------
        ImportError
            If ``xarray`` is not installed.
        ValueError
            If ``ds`` contains no variables with ``(y, x)`` dimensions.

        Examples
        --------
        >>> backend2 = RasterBackend.from_xarray(ds)
        >>> np.array_equal(backend2.get("uso"), backend.get("uso"))
        True
        """
        try:
            import xarray as xr
        except ImportError:
            raise ImportError(
                "xarray is required for RasterBackend.from_xarray(). "
                "Install it with: pip install xarray"
            )

        # normalise DataArray → Dataset
        if isinstance(ds, xr.DataArray):
            name = ds.name or "data"
            ds = ds.to_dataset(name=name)

        # collect 2D (y, x) variables — skip scalars and non-spatial vars
        spatial_vars = {
            name: var
            for name, var in ds.data_vars.items()
            if set(var.dims) >= {"y", "x"} and var.ndim == 2
        }

        if not spatial_vars:
            raise ValueError(
                "No 2D (y, x) variables found in the Dataset. "
                "Ensure dimensions are named 'y' and 'x'."
            )

        # infer shape from first variable
        first_var = next(iter(spatial_vars.values()))
        y_idx = first_var.dims.index("y")
        x_idx = first_var.dims.index("x")
        rows = first_var.shape[y_idx]
        cols = first_var.shape[x_idx]

        # recover transform from y/x coords if they look like spatial coords
        transform = None
        try:
            import rasterio.transform
            ys = ds.coords["y"].values
            xs = ds.coords["x"].values
            if len(ys) >= 2 and len(xs) >= 2:
                # reconstruct Affine from cell-centre coordinates
                res_y = float(ys[1] - ys[0])   # negative for north-up
                res_x = float(xs[1] - xs[0])
                origin_x = float(xs[0]) - res_x / 2
                origin_y = float(ys[0]) - res_y / 2
                transform = rasterio.transform.from_origin(
                    origin_x, origin_y - res_y * (rows - 1), res_x, abs(res_y)
                ) if res_y < 0 else rasterio.transform.Affine(
                    res_x, 0, origin_x, 0, res_y, origin_y
                )
        except Exception:
            pass  # transform recovery is best-effort

        # recover CRS from spatial_ref coordinate (CF convention)
        crs = None
        if "spatial_ref" in ds.coords:
            try:
                from pyproj import CRS as ProjCRS
                wkt = ds.coords["spatial_ref"].attrs.get("crs_wkt", "")
                if wkt:
                    crs = ProjCRS.from_wkt(wkt)
            except Exception:
                pass

        backend = cls(
            shape=(rows, cols),
            nodata_value=nodata_value,
            transform=transform,
            crs=crs,
        )

        for name, var in spatial_vars.items():
            # transpose to (y, x) canonical order before storing
            arr = var.transpose("y", "x").values
            backend.arrays[name] = arr.copy()

        return backend

    # ── spatial operations ────────────────────────────────────────────────────

    @staticmethod
    def shift2d(arr: np.ndarray, dr: int, dc: int) -> np.ndarray:
        """
        Shift ``arr`` by ``(dr, dc)`` rows/columns without wrap-around.
        Edges are filled with zero.

        Parameters
        ----------
        arr : np.ndarray
        dr : int
            Row offset (positive = down, negative = up).
        dc : int
            Column offset (positive = right, negative = left).

        Returns
        -------
        np.ndarray
            Shifted array of the same shape as ``arr``.
        """
        rows, cols = arr.shape
        out = np.zeros_like(arr)
        rs  = slice(max(0, -dr), min(rows, rows - dr))
        rd  = slice(max(0,  dr), min(rows, rows + dr))
        cs_ = slice(max(0, -dc), min(cols, cols - dc))
        cd  = slice(max(0,  dc), min(cols, cols + dc))
        out[rd, cd] = arr[rs, cs_]
        return out

    def band_names(self) -> list[str]:
        """Return the names of all arrays currently stored in the backend."""
        return list(self.arrays.keys())

    @staticmethod
    def neighbor_contact(
        condition: np.ndarray,
        neighborhood: list[tuple[int, int]] | None = None,
    ) -> np.ndarray:
        """
        Return a boolean mask where each cell has at least one neighbour
        satisfying ``condition``.

        Parameters
        ----------
        condition : np.ndarray
        neighborhood : list[tuple[int, int]] | None
            ``None`` uses Moore neighbourhood via ``binary_dilation``.

        Returns
        -------
        np.ndarray
            Boolean array.
        """
        if neighborhood is None:
            return binary_dilation(condition.astype(bool), structure=np.ones((3, 3)))
        result = np.zeros_like(condition, dtype=bool)
        for dr, dc in neighborhood:
            result |= RasterBackend.shift2d(condition.astype(np.int8), dr, dc) > 0
        return result

    def focal_sum(
        self,
        name: str,
        neighborhood: list[tuple[int, int]] = DIRS_MOORE,
    ) -> np.ndarray:
        """
        Focal sum: for each cell, sum the values of ``name`` across its neighbours.
        The cell itself is not included.

        Parameters
        ----------
        name : str
        neighborhood : list[tuple[int, int]]
            Default: ``DIRS_MOORE``.

        Returns
        -------
        np.ndarray
        """
        arr    = self.arrays[name]
        result = np.zeros_like(arr, dtype=float)
        for dr, dc in neighborhood:
            result += self.shift2d(arr, dr, dc)
        return result

    def focal_sum_mask(
        self,
        mask: np.ndarray,
        neighborhood: list[tuple[int, int]] = DIRS_MOORE,
    ) -> np.ndarray:
        """
        Count neighbours where ``mask`` is ``True``.

        Parameters
        ----------
        mask : np.ndarray
        neighborhood : list[tuple[int, int]]
            Default: ``DIRS_MOORE``.

        Returns
        -------
        np.ndarray
            Integer array with per-cell neighbour counts.
        """
        result = np.zeros(self.shape, dtype=int)
        m = mask.astype(np.int8)
        for dr, dc in neighborhood:
            result += self.shift2d(m, dr, dc)
        return result

    # ── utilities ─────────────────────────────────────────────────────────────

    def __repr__(self) -> str:
        bands = ", ".join(
            f"{k}:{v.dtype}[{v.shape}]" for k, v in self.arrays.items()
        )
        return f"RasterBackend(shape={self.shape}, arrays=[{bands}])"

nodata_mask property

Boolean mask: True = valid cell, False = outside extent / nodata.

Derived in priority order: 1. arrays["mask"] — explicit mask band (dissluc / coastal convention: non-zero = valid). 2. nodata_value — applied over the first available array. 3. None — no information; RasterMap skips auto-masking.

Used by RasterMap (auto_mask=True) to render out-of-extent pixels as transparent without any per-project configuration.

band_names()

Return the names of all arrays currently stored in the backend.

Source code in dissmodel/geo/raster/backend.py

def band_names(self) -> list[str]:
    """Return the names of all arrays currently stored in the backend."""
    return list(self.arrays.keys())

focal_sum(name, neighborhood=DIRS_MOORE)

Focal sum: for each cell, sum the values of name across its neighbours. The cell itself is not included.

Parameters:

Name	Type	Description	Default
name	str		required
neighborhood	list[tuple[int, int]]	Default: DIRS_MOORE.	DIRS_MOORE

Returns:

Type	Description
ndarray

Source code in dissmodel/geo/raster/backend.py

def focal_sum(
    self,
    name: str,
    neighborhood: list[tuple[int, int]] = DIRS_MOORE,
) -> np.ndarray:
    """
    Focal sum: for each cell, sum the values of ``name`` across its neighbours.
    The cell itself is not included.

    Parameters
    ----------
    name : str
    neighborhood : list[tuple[int, int]]
        Default: ``DIRS_MOORE``.

    Returns
    -------
    np.ndarray
    """
    arr    = self.arrays[name]
    result = np.zeros_like(arr, dtype=float)
    for dr, dc in neighborhood:
        result += self.shift2d(arr, dr, dc)
    return result

focal_sum_mask(mask, neighborhood=DIRS_MOORE)

Count neighbours where mask is True.

Parameters:

Name	Type	Description	Default
mask	ndarray		required
neighborhood	list[tuple[int, int]]	Default: DIRS_MOORE.	DIRS_MOORE

Returns:

Type	Description
ndarray	Integer array with per-cell neighbour counts.

Source code in dissmodel/geo/raster/backend.py

def focal_sum_mask(
    self,
    mask: np.ndarray,
    neighborhood: list[tuple[int, int]] = DIRS_MOORE,
) -> np.ndarray:
    """
    Count neighbours where ``mask`` is ``True``.

    Parameters
    ----------
    mask : np.ndarray
    neighborhood : list[tuple[int, int]]
        Default: ``DIRS_MOORE``.

    Returns
    -------
    np.ndarray
        Integer array with per-cell neighbour counts.
    """
    result = np.zeros(self.shape, dtype=int)
    m = mask.astype(np.int8)
    for dr, dc in neighborhood:
        result += self.shift2d(m, dr, dc)
    return result

from_xarray(ds, nodata_value=None) classmethod

Build a RasterBackend from an xr.Dataset or xr.DataArray.

All variables with exactly two dimensions (y, x) (in any order) are imported as arrays. Variables with other dimensionality (e.g. spatial_ref scalars) are silently skipped.

CRS is recovered from the spatial_ref coordinate (CF convention) when present and pyproj is available.

Parameters:

Name	Type	Description	Default
ds	Dataset | DataArray	Source dataset. A DataArray is wrapped into a single-variable Dataset using da.name (falling back to "data").	required
nodata_value	float | int | None	Forwarded to the new backend's nodata_value. Default: None.	None

Returns:

Type	Description
RasterBackend

Raises:

Type	Description
ImportError	If xarray is not installed.
ValueError	If ds contains no variables with (y, x) dimensions.

Examples:

>>> backend2 = RasterBackend.from_xarray(ds)
>>> np.array_equal(backend2.get("uso"), backend.get("uso"))
True

Source code in dissmodel/geo/raster/backend.py

@classmethod
def from_xarray(cls, ds, nodata_value: float | int | None = None) -> "RasterBackend":
    """
    Build a ``RasterBackend`` from an ``xr.Dataset`` or ``xr.DataArray``.

    All variables with exactly two dimensions ``(y, x)`` (in any order)
    are imported as arrays. Variables with other dimensionality
    (e.g. ``spatial_ref`` scalars) are silently skipped.

    CRS is recovered from the ``spatial_ref`` coordinate (CF convention)
    when present and ``pyproj`` is available.

    Parameters
    ----------
    ds : xr.Dataset | xr.DataArray
        Source dataset. A ``DataArray`` is wrapped into a single-variable
        ``Dataset`` using ``da.name`` (falling back to ``"data"``).
    nodata_value : float | int | None
        Forwarded to the new backend's ``nodata_value``. Default: ``None``.

    Returns
    -------
    RasterBackend

    Raises
    ------
    ImportError
        If ``xarray`` is not installed.
    ValueError
        If ``ds`` contains no variables with ``(y, x)`` dimensions.

    Examples
    --------
    >>> backend2 = RasterBackend.from_xarray(ds)
    >>> np.array_equal(backend2.get("uso"), backend.get("uso"))
    True
    """
    try:
        import xarray as xr
    except ImportError:
        raise ImportError(
            "xarray is required for RasterBackend.from_xarray(). "
            "Install it with: pip install xarray"
        )

    # normalise DataArray → Dataset
    if isinstance(ds, xr.DataArray):
        name = ds.name or "data"
        ds = ds.to_dataset(name=name)

    # collect 2D (y, x) variables — skip scalars and non-spatial vars
    spatial_vars = {
        name: var
        for name, var in ds.data_vars.items()
        if set(var.dims) >= {"y", "x"} and var.ndim == 2
    }

    if not spatial_vars:
        raise ValueError(
            "No 2D (y, x) variables found in the Dataset. "
            "Ensure dimensions are named 'y' and 'x'."
        )

    # infer shape from first variable
    first_var = next(iter(spatial_vars.values()))
    y_idx = first_var.dims.index("y")
    x_idx = first_var.dims.index("x")
    rows = first_var.shape[y_idx]
    cols = first_var.shape[x_idx]

    # recover transform from y/x coords if they look like spatial coords
    transform = None
    try:
        import rasterio.transform
        ys = ds.coords["y"].values
        xs = ds.coords["x"].values
        if len(ys) >= 2 and len(xs) >= 2:
            # reconstruct Affine from cell-centre coordinates
            res_y = float(ys[1] - ys[0])   # negative for north-up
            res_x = float(xs[1] - xs[0])
            origin_x = float(xs[0]) - res_x / 2
            origin_y = float(ys[0]) - res_y / 2
            transform = rasterio.transform.from_origin(
                origin_x, origin_y - res_y * (rows - 1), res_x, abs(res_y)
            ) if res_y < 0 else rasterio.transform.Affine(
                res_x, 0, origin_x, 0, res_y, origin_y
            )
    except Exception:
        pass  # transform recovery is best-effort

    # recover CRS from spatial_ref coordinate (CF convention)
    crs = None
    if "spatial_ref" in ds.coords:
        try:
            from pyproj import CRS as ProjCRS
            wkt = ds.coords["spatial_ref"].attrs.get("crs_wkt", "")
            if wkt:
                crs = ProjCRS.from_wkt(wkt)
        except Exception:
            pass

    backend = cls(
        shape=(rows, cols),
        nodata_value=nodata_value,
        transform=transform,
        crs=crs,
    )

    for name, var in spatial_vars.items():
        # transpose to (y, x) canonical order before storing
        arr = var.transpose("y", "x").values
        backend.arrays[name] = arr.copy()

    return backend

get(name)

Return a direct reference to the named array.

Use .copy() to obtain a snapshot equivalent to TerraME's .past.

Raises:

Type	Description
KeyError	If name is not in self.arrays.

Source code in dissmodel/geo/raster/backend.py

def get(self, name: str) -> np.ndarray:
    """
    Return a direct reference to the named array.

    Use ``.copy()`` to obtain a snapshot equivalent to TerraME's ``.past``.

    Raises
    ------
    KeyError
        If ``name`` is not in ``self.arrays``.
    """
    return self.arrays[name]

neighbor_contact(condition, neighborhood=None) staticmethod

Return a boolean mask where each cell has at least one neighbour satisfying condition.

Parameters:

Name	Type	Description	Default
condition	ndarray		required
neighborhood	list[tuple[int, int]] | None	None uses Moore neighbourhood via binary_dilation.	None

Returns:

Type	Description
ndarray	Boolean array.

Source code in dissmodel/geo/raster/backend.py

@staticmethod
def neighbor_contact(
    condition: np.ndarray,
    neighborhood: list[tuple[int, int]] | None = None,
) -> np.ndarray:
    """
    Return a boolean mask where each cell has at least one neighbour
    satisfying ``condition``.

    Parameters
    ----------
    condition : np.ndarray
    neighborhood : list[tuple[int, int]] | None
        ``None`` uses Moore neighbourhood via ``binary_dilation``.

    Returns
    -------
    np.ndarray
        Boolean array.
    """
    if neighborhood is None:
        return binary_dilation(condition.astype(bool), structure=np.ones((3, 3)))
    result = np.zeros_like(condition, dtype=bool)
    for dr, dc in neighborhood:
        result |= RasterBackend.shift2d(condition.astype(np.int8), dr, dc) > 0
    return result

rename_band(old, new)

Rename an array in-place. No-op if old does not exist.

Parameters:

Name	Type	Description	Default
old	str	Current band name.	required
new	str	Target band name.	required

Source code in dissmodel/geo/raster/backend.py

def rename_band(self, old: str, new: str) -> None:
    """
    Rename an array in-place. No-op if ``old`` does not exist.

    Parameters
    ----------
    old : str
        Current band name.
    new : str
        Target band name.
    """
    if old in self.arrays:
        self.arrays[new] = self.arrays.pop(old)

set(name, array)

Store a copy of array under name.

Source code in dissmodel/geo/raster/backend.py

def set(self, name: str, array: np.ndarray) -> None:
    """Store a copy of ``array`` under ``name``."""
    self.arrays[name] = np.asarray(array).copy()

shift2d(arr, dr, dc) staticmethod

Shift arr by (dr, dc) rows/columns without wrap-around. Edges are filled with zero.

Parameters:

Name	Type	Description	Default
arr	ndarray		required
dr	int	Row offset (positive = down, negative = up).	required
dc	int	Column offset (positive = right, negative = left).	required

Returns:

Type	Description
ndarray	Shifted array of the same shape as arr.

Source code in dissmodel/geo/raster/backend.py

@staticmethod
def shift2d(arr: np.ndarray, dr: int, dc: int) -> np.ndarray:
    """
    Shift ``arr`` by ``(dr, dc)`` rows/columns without wrap-around.
    Edges are filled with zero.

    Parameters
    ----------
    arr : np.ndarray
    dr : int
        Row offset (positive = down, negative = up).
    dc : int
        Column offset (positive = right, negative = left).

    Returns
    -------
    np.ndarray
        Shifted array of the same shape as ``arr``.
    """
    rows, cols = arr.shape
    out = np.zeros_like(arr)
    rs  = slice(max(0, -dr), min(rows, rows - dr))
    rd  = slice(max(0,  dr), min(rows, rows + dr))
    cs_ = slice(max(0, -dc), min(cols, cols - dc))
    cd  = slice(max(0,  dc), min(cols, cols + dc))
    out[rd, cd] = arr[rs, cs_]
    return out

snapshot()

Return a deep copy of all arrays — equivalent to TerraME's .past mechanism.

Typical usage::

past     = backend.snapshot()
uso_past = past["uso"]   # state at the beginning of the step

Returns:

Type	Description
dict[str, ndarray]	Dictionary mapping array names to independent copies.

Source code in dissmodel/geo/raster/backend.py

def snapshot(self) -> dict[str, np.ndarray]:
    """
    Return a deep copy of all arrays — equivalent to TerraME's ``.past`` mechanism.

    Typical usage::

        past     = backend.snapshot()
        uso_past = past["uso"]   # state at the beginning of the step

    Returns
    -------
    dict[str, np.ndarray]
        Dictionary mapping array names to independent copies.
    """
    return {k: v.copy() for k, v in self.arrays.items()}

to_xarray(time=None)

Convert the backend to an xr.Dataset.

Each array in self.arrays becomes a DataVariable with dimensions (y, x). If time is given, a scalar time coordinate is added — useful when assembling multi-step outputs.

Spatial coordinates are derived from self.transform when available (rasterio Affine), following the Pangeo convention of cell centres. When transform is None (e.g. backends rasterized from vector), integer pixel indices are used instead.

CRS is stored as a spatial_ref coordinate (CF convention) when self.crs is set and pyproj is available.

Parameters:

Name	Type	Description	Default
time	int | None	Optional simulation step to attach as a scalar time coordinate.	None

Returns:

Type	Description
Dataset

Raises:

Type	Description
ImportError	If xarray is not installed.

Examples:

>>> ds = backend.to_xarray()
>>> ds["uso"].dims
('y', 'x')

>>> ds = backend.to_xarray(time=42)
>>> ds.coords["time"].item()
42

Source code in dissmodel/geo/raster/backend.py

def to_xarray(self, time: int | None = None):
    """
    Convert the backend to an ``xr.Dataset``.

    Each array in ``self.arrays`` becomes a ``DataVariable`` with
    dimensions ``(y, x)``. If ``time`` is given, a scalar ``time``
    coordinate is added — useful when assembling multi-step outputs.

    Spatial coordinates are derived from ``self.transform`` when available
    (rasterio Affine), following the Pangeo convention of cell centres.
    When ``transform`` is ``None`` (e.g. backends rasterized from vector),
    integer pixel indices are used instead.

    CRS is stored as a ``spatial_ref`` coordinate (CF convention) when
    ``self.crs`` is set and ``pyproj`` is available.

    Parameters
    ----------
    time : int | None
        Optional simulation step to attach as a scalar ``time`` coordinate.

    Returns
    -------
    xr.Dataset

    Raises
    ------
    ImportError
        If ``xarray`` is not installed.

    Examples
    --------
    >>> ds = backend.to_xarray()
    >>> ds["uso"].dims
    ('y', 'x')

    >>> ds = backend.to_xarray(time=42)
    >>> ds.coords["time"].item()
    42
    """
    try:
        import xarray as xr
    except ImportError:
        raise ImportError(
            "xarray is required for RasterBackend.to_xarray(). "
            "Install it with: pip install xarray"
        )

    rows, cols = self.shape

    # spatial coordinates — cell centres from Affine transform or pixel indices
    if self.transform is not None:
        try:
            # rasterio Affine: transform * (col + 0.5, row + 0.5) = centre
            xs = np.array([self.transform.c + (c + 0.5) * self.transform.a
                            for c in range(cols)])
            ys = np.array([self.transform.f + (r + 0.5) * self.transform.e
                            for r in range(rows)])
        except AttributeError:
            # transform present but not rasterio Affine — fall back to indices
            xs = np.arange(cols, dtype=float)
            ys = np.arange(rows, dtype=float)
    else:
        xs = np.arange(cols, dtype=float)
        ys = np.arange(rows, dtype=float)

    coords: dict = {"y": ys, "x": xs}

    if time is not None:
        coords["time"] = time

    # CRS as spatial_ref coordinate (CF / rioxarray convention)
    if self.crs is not None:
        try:
            from pyproj import CRS as ProjCRS
            crs_obj = ProjCRS.from_user_input(self.crs)
            coords["spatial_ref"] = xr.DataArray(
                0,
                attrs={
                    "crs_wkt":       crs_obj.to_wkt(),
                    "grid_mapping":  "spatial_ref",
                },
            )
        except Exception:
            # pyproj unavailable or CRS unresolvable — skip spatial_ref
            pass

    data_vars = {}
    for name, arr in self.arrays.items():
        attrs: dict = {}
        if self.nodata_value is not None:
            attrs["_FillValue"] = self.nodata_value
        if self.crs is not None and "spatial_ref" in coords:
            attrs["grid_mapping"] = "spatial_ref"

        data_vars[name] = xr.DataArray(
            arr.copy(),
            dims=["y", "x"],
            coords={"y": ys, "x": xs},
            attrs=attrs,
        )

    ds = xr.Dataset(data_vars, coords=coords)
    ds.attrs["Conventions"] = "CF-1.8"
    return ds

dissmodel.geo.raster.raster_model

dissmodel/geo/raster/model.py

Base class for models backed by RasterBackend (NumPy 2D arrays).

Analogous to SpatialModel for the raster substrate — provides infrastructure without imposing a transition rule contract.

Class hierarchy

Model  (dissmodel.core)
  ├── SpatialModel     GeoDataFrame + Queen/Rook neighbourhood  (vector)
  └── RasterModel      RasterBackend + shift2d                  (raster)  ← this file
        ├── FloodRasterModel
        └── MangroveRasterModel

Usage

class MyRasterModel(RasterModel):
    def setup(self, backend, my_param=1.0):
        super().setup(backend)
        self.my_param = my_param

    def execute(self):
        uso = self.backend.get("uso").copy()
        ...
        self.backend.arrays["uso"] = new_uso

RasterModel

Bases: Model

Model backed by a RasterBackend.

Subclass of Model that adds raster infrastructure without imposing a transition rule contract. Can be subclassed directly by any model that operates on NumPy 2D arrays.

Parameters (setup)

backend : RasterBackend Backend shared across all models in the same Environment.

Attributes available in subclasses

backend : RasterBackend The shared array store. shape : tuple[int, int] Grid shape (rows, cols) — shortcut for self.backend.shape. shift : callable Shortcut for RasterBackend.shift2d (static method). dirs : list[tuple[int, int]] DIRS_MOORE — the 8 directions of the Moore neighbourhood.

Examples:

>>> class HeatDiffusion(RasterModel):
...     def execute(self):
...         temp = self.backend.get("temp").copy()
...         for dr, dc in self.dirs:
...             temp += 0.1 * self.shift(temp, dr, dc)
...         self.backend.arrays["temp"] = temp

Source code in dissmodel/geo/raster/raster_model.py

class RasterModel(Model):
    """
    Model backed by a RasterBackend.

    Subclass of ``Model`` that adds raster infrastructure without imposing
    a transition rule contract. Can be subclassed directly by any model
    that operates on NumPy 2D arrays.

    Parameters (setup)
    ------------------
    backend : RasterBackend
        Backend shared across all models in the same ``Environment``.

    Attributes available in subclasses
    ------------------------------------
    backend : RasterBackend
        The shared array store.
    shape : tuple[int, int]
        Grid shape ``(rows, cols)`` — shortcut for ``self.backend.shape``.
    shift : callable
        Shortcut for ``RasterBackend.shift2d`` (static method).
    dirs : list[tuple[int, int]]
        ``DIRS_MOORE`` — the 8 directions of the Moore neighbourhood.

    Examples
    --------
    >>> class HeatDiffusion(RasterModel):
    ...     def execute(self):
    ...         temp = self.backend.get("temp").copy()
    ...         for dr, dc in self.dirs:
    ...             temp += 0.1 * self.shift(temp, dr, dc)
    ...         self.backend.arrays["temp"] = temp
    """

    def setup(self, backend: RasterBackend) -> None:
        self.backend = backend
        self.shape   = backend.shape
        self.shift   = RasterBackend.shift2d
        self.dirs    = DIRS_MOORE

dissmodel.geo.raster.sync_model

dissmodel/geo/raster/sync_model.py

SyncRasterModel — RasterModel with automatic _past snapshot semantics.

This module provides the snapshot mechanism equivalent to TerraME's cs:synchronize(), as a reusable base class for any raster model that needs per-step state history (LUCC, fire spread, epidemic models, etc.).

How it works

At the start of each step, synchronize() copies the current NumPy array for every name in land_use_types to a <name>_past array in the RasterBackend. Models can call self.backend.get("f_past") to access the state at the beginning of the current step, regardless of changes made during execution.

Usage

Subclass SyncRasterModel instead of RasterModel and declare self.land_use_types in setup():

class MyRasterModel(SyncRasterModel):
    def setup(self, backend, ...):
        super().setup(backend)               # RasterModel setup
        self.land_use_types = ["f", "d"]     # arrays to snapshot

    def execute(self):
        f_past = self.backend.get("f_past")  # state at step start
        ...

The synchronize() method is called automatically: - once before the first execute() → snapshot of the initial state - once after each execute() → snapshot for the next step

It can also be called manually when needed.

Relationship to domain libraries

dissluc uses this class as the base for its raster LUCC components, exposing it under the domain-specific alias LUCRasterModel:

# dissluc/raster/core.py
from dissmodel.geo.raster.sync_model import SyncRasterModel as LUCRasterModel

SyncRasterModel

Bases: RasterModel

RasterModel with automatic _past snapshot semantics.

Extends :class:~dissmodel.geo.raster.model.RasterModel with a synchronize() method that copies each array listed in self.land_use_types to a <name>_past array in the :class:~dissmodel.geo.raster.backend.RasterBackend before and after every simulation step.

This is the raster analogue of :class:~dissmodel.geo.vector.sync_model.SyncSpatialModel and the Python equivalent of TerraME's cs:synchronize().

Subclass contract

Declare self.land_use_types (list of array names) in setup(). SyncRasterModel will manage all <name>_past arrays automatically. Subclasses must not create or update _past arrays manually.

Parameters:

Name	Type	Description	Default
backend	RasterBackend	Passed through to :class:~dissmodel.geo.raster.model.RasterModel.	required
**kwargs	Any	Any additional keyword arguments accepted by the parent class.	{}

Examples:

>>> class ForestRaster(SyncRasterModel):
...     def setup(self, backend, rate=0.01):
...         super().setup(backend)
...         self.land_use_types = ["forest", "defor"]
...         self.rate = rate
...
...     def execute(self):
...         forest_past = self.backend.get("forest_past")
...         gain = forest_past * self.rate
...         self.backend.arrays["forest"] = forest_past + gain

Source code in dissmodel/geo/raster/sync_model.py

class SyncRasterModel(RasterModel):
    """
    ``RasterModel`` with automatic ``_past`` snapshot semantics.

    Extends :class:`~dissmodel.geo.raster.model.RasterModel` with a
    ``synchronize()`` method that copies each array listed in
    ``self.land_use_types`` to a ``<name>_past`` array in the
    :class:`~dissmodel.geo.raster.backend.RasterBackend` before and after
    every simulation step.

    This is the raster analogue of
    :class:`~dissmodel.geo.vector.sync_model.SyncSpatialModel` and the
    Python equivalent of TerraME's ``cs:synchronize()``.

    Subclass contract
    -----------------
    Declare ``self.land_use_types`` (list of array names) in ``setup()``.
    ``SyncRasterModel`` will manage all ``<name>_past`` arrays automatically.
    Subclasses must **not** create or update ``_past`` arrays manually.

    Parameters
    ----------
    backend : RasterBackend
        Passed through to :class:`~dissmodel.geo.raster.model.RasterModel`.
    **kwargs
        Any additional keyword arguments accepted by the parent class.

    Examples
    --------
    >>> class ForestRaster(SyncRasterModel):
    ...     def setup(self, backend, rate=0.01):
    ...         super().setup(backend)
    ...         self.land_use_types = ["forest", "defor"]
    ...         self.rate = rate
    ...
    ...     def execute(self):
    ...         forest_past = self.backend.get("forest_past")
    ...         gain = forest_past * self.rate
    ...         self.backend.arrays["forest"] = forest_past + gain
    """

    def process(self) -> None:
        """
        Simulation loop with automatic snapshot management.

        Overrides :meth:`~dissmodel.core.Model.process` to insert
        :meth:`synchronize` calls before the first step and after each step.
        """
        if self.env.now() < self.start_time:
            self.hold(self.start_time - self.env.now())

        # initial snapshot — captures state at t=0 before any execution
        self.synchronize()

        while self.env.now() < self.end_time:
            self.execute()
            self.synchronize()   # update snapshot for the next step
            self.hold(self._step)

    def synchronize(self) -> None:
        """
        Copy each array in ``land_use_types`` to ``<name>_past`` in the backend.

        Equivalent to ``cs:synchronize()`` in TerraME. Called automatically
        before the first step and after each ``execute()``. Can also be
        called manually when an explicit mid-step snapshot is needed.

        Does nothing if ``land_use_types`` has not been set yet (safe to
        call before ``setup()`` completes).
        """
        if not hasattr(self, "land_use_types"):
            return
        for name in self.land_use_types:
            self.backend.set(name + "_past", self.backend.get(name).copy())

process()

Simulation loop with automatic snapshot management.

Overrides :meth:~dissmodel.core.Model.process to insert :meth:synchronize calls before the first step and after each step.

Source code in dissmodel/geo/raster/sync_model.py

def process(self) -> None:
    """
    Simulation loop with automatic snapshot management.

    Overrides :meth:`~dissmodel.core.Model.process` to insert
    :meth:`synchronize` calls before the first step and after each step.
    """
    if self.env.now() < self.start_time:
        self.hold(self.start_time - self.env.now())

    # initial snapshot — captures state at t=0 before any execution
    self.synchronize()

    while self.env.now() < self.end_time:
        self.execute()
        self.synchronize()   # update snapshot for the next step
        self.hold(self._step)

synchronize()

Copy each array in land_use_types to <name>_past in the backend.

Equivalent to cs:synchronize() in TerraME. Called automatically before the first step and after each execute(). Can also be called manually when an explicit mid-step snapshot is needed.

Does nothing if land_use_types has not been set yet (safe to call before setup() completes).

Source code in dissmodel/geo/raster/sync_model.py

def synchronize(self) -> None:
    """
    Copy each array in ``land_use_types`` to ``<name>_past`` in the backend.

    Equivalent to ``cs:synchronize()`` in TerraME. Called automatically
    before the first step and after each ``execute()``. Can also be
    called manually when an explicit mid-step snapshot is needed.

    Does nothing if ``land_use_types`` has not been set yet (safe to
    call before ``setup()`` completes).
    """
    if not hasattr(self, "land_use_types"):
        return
    for name in self.land_use_types:
        self.backend.set(name + "_past", self.backend.get(name).copy())

dissmodel.geo.raster.cellular_automaton

dissmodel/geo/raster_cellular_automaton.py

Base class for cellular automata backed by RasterBackend (NumPy 2D arrays).

Analogous to CellularAutomaton (GeoDataFrame), but for the raster substrate.

Hierarchy

Model
  ├── SpatialModel
  │     └── CellularAutomaton       rule(idx) → value       (vector, pull)
  └── RasterModel
        └── RasterCellularAutomaton rule(arrays) → arrays   (raster, vectorized)

Why a different rule() contract

CellularAutomaton.rule(idx) returns a single value for one cell — it is called once per cell per step (O(n) Python calls). This is correct for the vector substrate where neighborhood lookup is the bottleneck.

For the raster substrate, the bottleneck is the Python loop itself. RasterCellularAutomaton.rule() receives the full snapshot of all arrays and returns a dict of updated arrays — one NumPy call covers the entire grid. This is the natural pattern for NumPy-based CA.

Comparison

# vector CA — rule called n times per step
class GameOfLife(CellularAutomaton):
    def rule(self, idx):
        alive = self.neighbor_values(idx, "state").sum()
        ...
        return new_state

# raster CA — rule called once per step
class GameOfLife(RasterCellularAutomaton):
    def rule(self, arrays):
        state = arrays["state"]
        alive = backend.focal_sum_mask(state == 1)
        ...
        return {"state": new_state}

Usage

from dissmodel.geo.raster_cellular_automaton import RasterCellularAutomaton
from dissmodel.geo.raster.backend import RasterBackend
from dissmodel.core import Environment
import numpy as np

class GameOfLife(RasterCellularAutomaton):
    def rule(self, arrays):
        state     = arrays["state"]
        neighbors = self.backend.focal_sum_mask(state == 1)
        born      = (state == 0) & (neighbors == 3)
        survive   = (state == 1) & np.isin(neighbors, [2, 3])
        return {"state": np.where(born | survive, 1, 0)}

b = RasterBackend(shape=(50, 50))
b.set("state", np.random.randint(0, 2, (50, 50)))

env = Environment(start_time=1, end_time=100)
GameOfLife(backend=b)
env.run()

RasterCellularAutomaton

Bases: RasterModel, ABC

Base class for NumPy-based cellular automata.

Extends :class:~dissmodel.geo.raster.model.RasterModel with a vectorized transition rule — rule() receives all arrays as a snapshot and returns a dict of updated arrays.

Parameters:

Name	Type	Description	Default
backend	RasterBackend	Shared backend with the simulation arrays.	required
state_attr	str	Primary state array name, by default "state". Used only for introspection/logging — rule() can update any array.	required
**kwargs	Any	Extra keyword arguments forwarded to RasterModel.	{}

Examples:

>>> class MyCA(RasterCellularAutomaton):
...     def rule(self, arrays):
...         state = arrays["state"]
...         # ... NumPy operations over full grid ...
...         return {"state": new_state}

Source code in dissmodel/geo/raster/cellular_automaton.py

class RasterCellularAutomaton(RasterModel, ABC):
    """
    Base class for NumPy-based cellular automata.

    Extends :class:`~dissmodel.geo.raster.model.RasterModel` with a
    vectorized transition rule — ``rule()`` receives all arrays as a
    snapshot and returns a dict of updated arrays.

    Parameters
    ----------
    backend : RasterBackend
        Shared backend with the simulation arrays.
    state_attr : str, optional
        Primary state array name, by default ``"state"``.
        Used only for introspection/logging — rule() can update any array.
    **kwargs :
        Extra keyword arguments forwarded to RasterModel.

    Examples
    --------
    >>> class MyCA(RasterCellularAutomaton):
    ...     def rule(self, arrays):
    ...         state = arrays["state"]
    ...         # ... NumPy operations over full grid ...
    ...         return {"state": new_state}
    """

    def setup(
        self,
        backend:    RasterBackend,
        state_attr: str = "state"
    ) -> None:
        super().setup(backend)
        self.state_attr = state_attr

    @abstractmethod
    def rule(self, arrays: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
        """
        Vectorized transition rule applied to the full grid.

        Receives a snapshot of all arrays (equivalent to celula.past[] in
        TerraME) and returns a dict with the arrays to update.

        Only the arrays present in the returned dict are written back —
        arrays not returned are left unchanged.

        Parameters
        ----------
        arrays : dict[str, np.ndarray]
            Snapshot of backend arrays at the start of the step.
            Modifying these arrays does NOT affect the backend — they are
            copies (equivalent to .past semantics).

        Returns
        -------
        dict[str, np.ndarray]
            Dict mapping array name → new array. Partial updates allowed.

        Examples
        --------
        >>> def rule(self, arrays):
        ...     state     = arrays["state"]           # read from snapshot
        ...     neighbors = self.backend.focal_sum_mask(state == 1)
        ...     new_state = np.where(neighbors > 3, 0, state)
        ...     return {"state": new_state}           # write back
        """
        raise NotImplementedError("Subclasses must implement rule().")

    def execute(self) -> None:
        """
        Execute one simulation step by calling rule() once over the full grid.

        Takes a snapshot of all arrays (past state), passes it to rule(),
        and writes the returned arrays back to the backend.
        """
        past    = self.backend.snapshot()   # equivale a celula.past[]
        updates = self.rule(past)
        for name, arr in updates.items():
            self.backend.arrays[name] = arr

execute()

Execute one simulation step by calling rule() once over the full grid.

Takes a snapshot of all arrays (past state), passes it to rule(), and writes the returned arrays back to the backend.

Source code in dissmodel/geo/raster/cellular_automaton.py

def execute(self) -> None:
    """
    Execute one simulation step by calling rule() once over the full grid.

    Takes a snapshot of all arrays (past state), passes it to rule(),
    and writes the returned arrays back to the backend.
    """
    past    = self.backend.snapshot()   # equivale a celula.past[]
    updates = self.rule(past)
    for name, arr in updates.items():
        self.backend.arrays[name] = arr

rule(arrays) abstractmethod

Vectorized transition rule applied to the full grid.

Receives a snapshot of all arrays (equivalent to celula.past[] in TerraME) and returns a dict with the arrays to update.

Only the arrays present in the returned dict are written back — arrays not returned are left unchanged.

Parameters:

Name	Type	Description	Default
arrays	dict[str, ndarray]	Snapshot of backend arrays at the start of the step. Modifying these arrays does NOT affect the backend — they are copies (equivalent to .past semantics).	required

Returns:

Type	Description
dict[str, ndarray]	Dict mapping array name → new array. Partial updates allowed.

Examples:

>>> def rule(self, arrays):
...     state     = arrays["state"]           # read from snapshot
...     neighbors = self.backend.focal_sum_mask(state == 1)
...     new_state = np.where(neighbors > 3, 0, state)
...     return {"state": new_state}           # write back

Source code in dissmodel/geo/raster/cellular_automaton.py

@abstractmethod
def rule(self, arrays: dict[str, np.ndarray]) -> dict[str, np.ndarray]:
    """
    Vectorized transition rule applied to the full grid.

    Receives a snapshot of all arrays (equivalent to celula.past[] in
    TerraME) and returns a dict with the arrays to update.

    Only the arrays present in the returned dict are written back —
    arrays not returned are left unchanged.

    Parameters
    ----------
    arrays : dict[str, np.ndarray]
        Snapshot of backend arrays at the start of the step.
        Modifying these arrays does NOT affect the backend — they are
        copies (equivalent to .past semantics).

    Returns
    -------
    dict[str, np.ndarray]
        Dict mapping array name → new array. Partial updates allowed.

    Examples
    --------
    >>> def rule(self, arrays):
    ...     state     = arrays["state"]           # read from snapshot
    ...     neighbors = self.backend.focal_sum_mask(state == 1)
    ...     new_state = np.where(neighbors > 3, 0, state)
    ...     return {"state": new_state}           # write back
    """
    raise NotImplementedError("Subclasses must implement rule().")

dissmodel.geo.raster.raster_grid

dissmodel/geo/raster_grid.py

Utilitário para criar RasterBackend sintético.

Análogo a vector_grid() (GeoDataFrame), mas para o substrato NumPy.

Uso

from dissmodel.geo.raster_grid import raster_grid
import numpy as np

# grade vazia com arrays zerados
b = raster_grid(rows=50, cols=50, attrs={"state": 0})

# grade com array inicial customizado
b = raster_grid(
    rows=50, cols=50,
    attrs={"state": np.random.randint(0, 2, (50, 50))}
)

raster_grid(rows, cols, attrs=None, dtype=None)

Create a RasterBackend with optional pre-filled arrays.

Analogous to :func:~dissmodel.geo.vector_grid for the raster substrate. Useful for tests, examples, and synthetic benchmarks.

Parameters:

Name	Type	Description	Default
rows	int	Number of rows in the grid.	required
cols	int	Number of columns in the grid.	required
attrs	dict	Mapping of array name → initial value. - scalar (int or float): fills the entire grid with that value. - np.ndarray of shape (rows, cols): used directly (a copy is stored). If not provided, an empty backend is returned.	None
dtype	numpy dtype	Default dtype for scalar-initialized arrays. If None, inferred from the scalar type (int → np.int32, float → np.float64).	None

Returns:

Type	Description
RasterBackend	Backend with shape (rows, cols) and the requested arrays.

Examples:

>>> b = raster_grid(10, 10, attrs={"state": 0})
>>> b.shape
(10, 10)
>>> b.get("state").shape
(10, 10)

>>> import numpy as np
>>> state = np.random.randint(0, 2, (10, 10))
>>> b = raster_grid(10, 10, attrs={"state": state})

Source code in dissmodel/geo/raster/raster_grid.py

def raster_grid(
    rows:  int,
    cols:  int,
    attrs: dict[str, AttrValue] | None = None,
    dtype: np.dtype | None             = None,
) -> RasterBackend:
    """
    Create a RasterBackend with optional pre-filled arrays.

    Analogous to :func:`~dissmodel.geo.vector_grid` for the raster
    substrate. Useful for tests, examples, and synthetic benchmarks.

    Parameters
    ----------
    rows : int
        Number of rows in the grid.
    cols : int
        Number of columns in the grid.
    attrs : dict, optional
        Mapping of array name → initial value.
        - scalar (int or float): fills the entire grid with that value.
        - np.ndarray of shape (rows, cols): used directly (a copy is stored).
        If not provided, an empty backend is returned.
    dtype : numpy dtype, optional
        Default dtype for scalar-initialized arrays. If None, inferred
        from the scalar type (int → np.int32, float → np.float64).

    Returns
    -------
    RasterBackend
        Backend with shape (rows, cols) and the requested arrays.

    Examples
    --------
    >>> b = raster_grid(10, 10, attrs={"state": 0})
    >>> b.shape
    (10, 10)
    >>> b.get("state").shape
    (10, 10)

    >>> import numpy as np
    >>> state = np.random.randint(0, 2, (10, 10))
    >>> b = raster_grid(10, 10, attrs={"state": state})
    """
    b = RasterBackend(shape=(rows, cols))

    for name, value in (attrs or {}).items():
        if isinstance(value, np.ndarray):
            if value.shape != (rows, cols):
                raise ValueError(
                    f"Array '{name}' has shape {value.shape}, "
                    f"expected ({rows}, {cols})."
                )
            b.set(name, value)
        else:
            # scalar → infer dtype
            if dtype is not None:
                arr_dtype = dtype
            elif isinstance(value, float):
                arr_dtype = np.float64
            else:
                arr_dtype = np.int32
            b.set(name, np.full((rows, cols), value, dtype=arr_dtype))

    return b

dissmodel.geo.raster.band_spec

BandSpec dataclass

Specification of a raster band in a GeoTIFF.

Attributes:

Name	Type	Description
name	str	Name used inside RasterBackend (e.g. 'uso', 'alt', 'soil').
dtype	str	NumPy dtype used to store the band.
nodata	float | int	Value representing missing data.

Source code in dissmodel/geo/raster/band_spec.py

@dataclass
class BandSpec:
    """
    Specification of a raster band in a GeoTIFF.

    Attributes
    ----------
    name : str
        Name used inside RasterBackend (e.g. 'uso', 'alt', 'soil').
    dtype : str
        NumPy dtype used to store the band.
    nodata : float | int
        Value representing missing data.
    """

    name: str
    dtype: str
    nodata: float | int