Skip to content

Geo — Vector

Vector substrate classes and utilities backed by GeoDataFrame.

dissmodel.geo.vector.spatial_model

dissmodel/geo/spatial_model.py

Classe base para modelos com suporte a GeoDataFrame e vizinhança.

Responsabilidade

Prover infraestrutura espacial — gdf, criação de vizinhança, acesso a vizinhos — sem impor nenhum contrato de regra de transição.

Hierarquia 
Model  (dissmodel.core)
  └── SpatialModel          ← este arquivo
        ├── CellularAutomaton  gdf + rule(idx) por célula (pull)
        └── (modelos livres)   gdf + execute() orientado a fonte (push)
Por que separar

CellularAutomaton.rule(idx) assume que cada célula calcula seu próprio novo estado de forma independente (modelo pull). Modelos orientados a FONTE — como o Hidro do BR-MANGUE — modificam vizinhos a partir da fonte (modelo push). Eles precisam da infraestrutura espacial mas não podem usar o contrato de rule().

SpatialModel fornece: - self.gdf GeoDataFrame compartilhado - create_neighborhood() constrói _neighs via libpysal ou dict - neighs_id(idx) lista de índices vizinhos (cache) - neighs(idx) GeoDataFrame dos vizinhos - neighbor_values(idx, col) array numpy dos valores vizinhos

Uso 
class MeuModelo(SpatialModel):
    def __init__(self, gdf, meu_param=1.0, **kwargs):
        super().__init__(gdf, **kwargs)
        self.meu_param = meu_param
        self.create_neighborhood()

    def execute(self):
        nivel = self.env.now() * self.meu_param
        # lógica livre — orientada a fonte, por grupo, etc.

SpatialModel

Bases: Model

Model com suporte a GeoDataFrame e vizinhança.

Subclasse de Model que acrescenta infraestrutura espacial sem impor contrato de regra de transição. Pode ser herdada diretamente por modelos com execute() livre (push/fonte) ou indiretamente via CellularAutomaton (pull/rule).

Parameters:

Name Type Description Default
gdf GeoDataFrame

Grade ou polígonos da simulação.

 required
step float

Time increment per execution step, by default 1.

 1
start_time float

Simulation start time, by default 0.

 0
end_time float

Simulation end time, by default math.inf.

 inf
name str

Optional model name, by default "".

 ''
**kwargs Any

Extra keyword arguments forwarded to :class:~dissmodel.core.Model.

 {}
Source code in dissmodel/geo/vector/spatial_model.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class SpatialModel(Model):
    """
    Model com suporte a GeoDataFrame e vizinhança.

    Subclasse de Model que acrescenta infraestrutura espacial sem impor
    contrato de regra de transição. Pode ser herdada diretamente por
    modelos com execute() livre (push/fonte) ou indiretamente via
    CellularAutomaton (pull/rule).

    Parameters
    ----------
    gdf : geopandas.GeoDataFrame
        Grade ou polígonos da simulação.
    step : float, optional
        Time increment per execution step, by default 1.
    start_time : float, optional
        Simulation start time, by default 0.
    end_time : float, optional
        Simulation end time, by default ``math.inf``.
    name : str, optional
        Optional model name, by default ``""``.
    **kwargs :
        Extra keyword arguments forwarded to :class:`~dissmodel.core.Model`.
    """

    def __init__(
        self,
        gdf: gpd.GeoDataFrame,
        step: float = 1,
        start_time: float = 0,
        end_time: float = math.inf,
        name: str = "",
        **kwargs: Any,
    ) -> None:
        self.gdf: gpd.GeoDataFrame       = gdf
        self._neighborhood_created: bool = False
        self._neighs_cache: dict         = {}
        super().__init__(
            step=step,
            start_time=start_time,
            end_time=end_time,
            name=name,
            **kwargs,
        )

    # ── vizinhança ────────────────────────────────────────────────────────────

    def create_neighborhood(
        self,
        strategy: StrategyType = Queen,
        neighbors_dict: Optional[dict | str] = None,
        **kwargs: Any,
    ) -> None:
        """
        Constrói e anexa a estrutura de vizinhança ao GeoDataFrame.

        Popula ``gdf["_neighs"]`` com a lista de índices vizinhos por célula.

        Parameters
        ----------
        strategy : type, optional
            Libpysal weight class (e.g. ``Queen``, ``Rook``),
            by default ``Queen``.
        neighbors_dict : dict or str, optional
            Precomputed ``{id: [neighbor_ids]}`` mapping or a path to a JSON
            file with the same structure. If provided, ``strategy`` is ignored.
        **kwargs :
            Extra keyword arguments forwarded to the strategy.
        """
        self.gdf = attach_neighbors(
            gdf=self.gdf,
            strategy=strategy,
            neighbors_dict=neighbors_dict,
            **kwargs,
        )
        self._neighborhood_created = True
        self._neighs_cache = self.gdf["_neighs"].to_dict()

    def neighs_id(self, idx: Any) -> list[Any]:
        """
        Return the neighbor indices for cell ``idx``.

        Parameters
        ----------
        idx : any
            Index of the cell in the GeoDataFrame.

        Returns
        -------
        list
            List of neighbor indices.
        """
        if self._neighs_cache:
            return self._neighs_cache.get(idx, [])
        return self.gdf.loc[idx, "_neighs"]

    def neighs(self, idx: Any) -> gpd.GeoDataFrame:
        """
        Return the neighboring cells of ``idx`` as a GeoDataFrame.

        Parameters
        ----------
        idx : any
            Index of the cell in the GeoDataFrame.

        Returns
        -------
        geopandas.GeoDataFrame
            GeoDataFrame containing the neighboring rows.

        Raises
        ------
        RuntimeError
            If the neighborhood has not been created yet.
        """
        if not self._neighborhood_created:
            raise RuntimeError(
                "Neighborhood has not been created yet. "
                "Call `.create_neighborhood()` first."
            )
        return self.gdf.loc[self.neighs_id(idx)]

    def neighbor_values(self, idx: Any, col: str) -> np.ndarray:
        """
        Return the values of ``col`` for all neighbors of cell ``idx``.

        Faster than ``neighs(idx)[col]`` because it skips geometry overhead.

        Parameters
        ----------
        idx : any
            Index of the cell in the GeoDataFrame.
        col : str
            Column name to retrieve.

        Returns
        -------
        numpy.ndarray
            Array of neighbor values.
        """
        return self.gdf.loc[self.neighs_id(idx), col].values

create_neighborhood(strategy=Queen, neighbors_dict=None, **kwargs)

Constrói e anexa a estrutura de vizinhança ao GeoDataFrame.

Popula gdf["_neighs"] com a lista de índices vizinhos por célula.

Parameters:

Name Type Description Default
strategy type

Libpysal weight class (e.g. Queen, Rook), by default Queen.

 Queen
neighbors_dict dict or str

Precomputed {id: [neighbor_ids]} mapping or a path to a JSON file with the same structure. If provided, strategy is ignored.

 None
**kwargs Any

Extra keyword arguments forwarded to the strategy.

 {}
Source code in dissmodel/geo/vector/spatial_model.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def create_neighborhood(
    self,
    strategy: StrategyType = Queen,
    neighbors_dict: Optional[dict | str] = None,
    **kwargs: Any,
) -> None:
    """
    Constrói e anexa a estrutura de vizinhança ao GeoDataFrame.

    Popula ``gdf["_neighs"]`` com a lista de índices vizinhos por célula.

    Parameters
    ----------
    strategy : type, optional
        Libpysal weight class (e.g. ``Queen``, ``Rook``),
        by default ``Queen``.
    neighbors_dict : dict or str, optional
        Precomputed ``{id: [neighbor_ids]}`` mapping or a path to a JSON
        file with the same structure. If provided, ``strategy`` is ignored.
    **kwargs :
        Extra keyword arguments forwarded to the strategy.
    """
    self.gdf = attach_neighbors(
        gdf=self.gdf,
        strategy=strategy,
        neighbors_dict=neighbors_dict,
        **kwargs,
    )
    self._neighborhood_created = True
    self._neighs_cache = self.gdf["_neighs"].to_dict()

neighbor_values(idx, col)

Return the values of col for all neighbors of cell idx.

Faster than neighs(idx)[col] because it skips geometry overhead.

Parameters:

Name Type Description Default
idx any

Index of the cell in the GeoDataFrame.

 required
col str

Column name to retrieve.

 required

Returns:

Type Description
ndarray

Array of neighbor values.
Source code in dissmodel/geo/vector/spatial_model.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def neighbor_values(self, idx: Any, col: str) -> np.ndarray:
    """
    Return the values of ``col`` for all neighbors of cell ``idx``.

    Faster than ``neighs(idx)[col]`` because it skips geometry overhead.

    Parameters
    ----------
    idx : any
        Index of the cell in the GeoDataFrame.
    col : str
        Column name to retrieve.

    Returns
    -------
    numpy.ndarray
        Array of neighbor values.
    """
    return self.gdf.loc[self.neighs_id(idx), col].values

neighs(idx)

Return the neighboring cells of idx as a GeoDataFrame.

Parameters:

Name Type Description Default
idx any

Index of the cell in the GeoDataFrame.

 required

Returns:

Type Description
GeoDataFrame

GeoDataFrame containing the neighboring rows.

Raises:

Type Description
RuntimeError

If the neighborhood has not been created yet.
Source code in dissmodel/geo/vector/spatial_model.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def neighs(self, idx: Any) -> gpd.GeoDataFrame:
    """
    Return the neighboring cells of ``idx`` as a GeoDataFrame.

    Parameters
    ----------
    idx : any
        Index of the cell in the GeoDataFrame.

    Returns
    -------
    geopandas.GeoDataFrame
        GeoDataFrame containing the neighboring rows.

    Raises
    ------
    RuntimeError
        If the neighborhood has not been created yet.
    """
    if not self._neighborhood_created:
        raise RuntimeError(
            "Neighborhood has not been created yet. "
            "Call `.create_neighborhood()` first."
        )
    return self.gdf.loc[self.neighs_id(idx)]

neighs_id(idx)

Return the neighbor indices for cell idx.

Parameters:

Name Type Description Default
idx any

Index of the cell in the GeoDataFrame.

 required

Returns:

Type Description
list

List of neighbor indices.
Source code in dissmodel/geo/vector/spatial_model.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def neighs_id(self, idx: Any) -> list[Any]:
    """
    Return the neighbor indices for cell ``idx``.

    Parameters
    ----------
    idx : any
        Index of the cell in the GeoDataFrame.

    Returns
    -------
    list
        List of neighbor indices.
    """
    if self._neighs_cache:
        return self._neighs_cache.get(idx, [])
    return self.gdf.loc[idx, "_neighs"]

dissmodel.geo.vector.sync_model

dissmodel/geo/vector/sync_model.py

SyncSpatialModel — SpatialModel with automatic _past snapshot semantics.

This module provides the snapshot mechanism equivalent to TerraME's cs:synchronize(), as a reusable base class for any vector 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 value of every column in land_use_types to a <col>_past column in the GeoDataFrame. Models can read gdf["f_past"] to access the state at the beginning of the current step, regardless of changes made during execution.

Usage

Subclass SyncSpatialModel instead of SpatialModel and declare self.land_use_types in setup():

class MyModel(SyncSpatialModel):
    def setup(self, gdf, ...):
        super().setup(gdf)                 # SpatialModel setup
        self.land_use_types = ["f", "d"]   # columns to snapshot

    def execute(self):
        past_f = self.gdf["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 (e.g. mid-step resets).

Relationship to domain libraries

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

# dissluc/core.py
from dissmodel.geo.vector.sync_model import SyncSpatialModel as LUCSpatialModel

SyncSpatialModel

Bases: SpatialModel

SpatialModel with automatic _past snapshot semantics.

Extends :class:~dissmodel.geo.vector.spatial_model.SpatialModel with a synchronize() method that copies each column listed in self.land_use_types to a <col>_past column before and after every simulation step.

This is the Python equivalent of TerraME's cs:synchronize() — it ensures that every model reads a consistent snapshot of the state at the beginning of the current step, even when multiple models share the same GeoDataFrame.

Subclass contract

Declare self.land_use_types (list of column names) in setup(). SyncSpatialModel will manage all <col>_past columns automatically. Subclasses must not create or update _past columns manually.

Parameters:

Name Type Description Default
gdf GeoDataFrame

Passed through to :class:~dissmodel.geo.vector.spatial_model.SpatialModel.

 required
**kwargs Any

Any additional keyword arguments accepted by the parent class.

 {}

Examples:

>>> class ForestCA(SyncSpatialModel):
...     def setup(self, gdf, rate=0.01):
...         super().setup(gdf)
...         self.land_use_types = ["forest", "defor"]
...         self.rate = rate
...
...     def execute(self):
...         # forest_past holds the state at the start of this step
...         gain = self.gdf["forest_past"] * self.rate
...         self.gdf["forest"] = self.gdf["forest_past"] + gain
Source code in dissmodel/geo/vector/sync_model.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class SyncSpatialModel(SpatialModel):
    """
    ``SpatialModel`` with automatic ``_past`` snapshot semantics.

    Extends :class:`~dissmodel.geo.vector.spatial_model.SpatialModel` with
    a ``synchronize()`` method that copies each column listed in
    ``self.land_use_types`` to a ``<col>_past`` column before and after
    every simulation step.

    This is the Python equivalent of TerraME's ``cs:synchronize()`` — it
    ensures that every model reads a consistent snapshot of the state at
    the beginning of the current step, even when multiple models share the
    same GeoDataFrame.

    Subclass contract
    -----------------
    Declare ``self.land_use_types`` (list of column names) in ``setup()``.
    ``SyncSpatialModel`` will manage all ``<col>_past`` columns automatically.
    Subclasses must **not** create or update ``_past`` columns manually.

    Parameters
    ----------
    gdf : geopandas.GeoDataFrame
        Passed through to :class:`~dissmodel.geo.vector.spatial_model.SpatialModel`.
    **kwargs
        Any additional keyword arguments accepted by the parent class.

    Examples
    --------
    >>> class ForestCA(SyncSpatialModel):
    ...     def setup(self, gdf, rate=0.01):
    ...         super().setup(gdf)
    ...         self.land_use_types = ["forest", "defor"]
    ...         self.rate = rate
    ...
    ...     def execute(self):
    ...         # forest_past holds the state at the start of this step
    ...         gain = self.gdf["forest_past"] * self.rate
    ...         self.gdf["forest"] = self.gdf["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 column in ``land_use_types`` to ``<col>_past``.

        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 col in self.land_use_types:
            self.gdf[col + "_past"] = self.gdf[col].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/vector/sync_model.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
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 column in land_use_types to <col>_past.

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/vector/sync_model.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def synchronize(self) -> None:
    """
    Copy each column in ``land_use_types`` to ``<col>_past``.

    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 col in self.land_use_types:
        self.gdf[col + "_past"] = self.gdf[col].copy()

dissmodel.geo.vector.cellular_automaton

dissmodel/geo/cellular_automaton.py

Base class for spatial cellular automata backed by a GeoDataFrame.

Extends :class:~dissmodel.geo.spatial_model.SpatialModel with a cell-by-cell transition rule loop.

For source-oriented (push) models that cannot use rule(idx), inherit :class:~dissmodel.geo.spatial_model.SpatialModel directly and implement execute() freely.

CellularAutomaton

Bases: SpatialModel, ABC

Base class for spatial cellular automata backed by a GeoDataFrame.

Extends :class:~dissmodel.geo.spatial_model.SpatialModel with neighborhood management and a cell-by-cell transition rule loop.

Parameters:

Name Type Description Default
gdf GeoDataFrame

GeoDataFrame with geometries and a state attribute.

 required
state_attr str

Column name representing the cell state, by default "state".

 'state'
step float

Time increment per execution step, by default 1.

 1
start_time float

Simulation start time, by default 0.

 0
end_time float

Simulation end time, by default math.inf.

 inf
name str

Optional model name, by default "".

 ''
dim tuple of int

Grid dimensions as (n_cols, n_rows), by default None.

 None
**kwargs Any

Extra keyword arguments forwarded to :class:~dissmodel.geo.spatial_model.SpatialModel.

 {}

Examples:

>>> class MyCA(CellularAutomaton):
...     def rule(self, idx):
...         return self.gdf.loc[idx, self.state_attr]
Source code in dissmodel/geo/vector/cellular_automaton.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class CellularAutomaton(SpatialModel, ABC):
    """
    Base class for spatial cellular automata backed by a GeoDataFrame.

    Extends :class:`~dissmodel.geo.spatial_model.SpatialModel` with
    neighborhood management and a cell-by-cell transition rule loop.

    Parameters
    ----------
    gdf : geopandas.GeoDataFrame
        GeoDataFrame with geometries and a state attribute.
    state_attr : str, optional
        Column name representing the cell state, by default ``"state"``.
    step : float, optional
        Time increment per execution step, by default 1.
    start_time : float, optional
        Simulation start time, by default 0.
    end_time : float, optional
        Simulation end time, by default ``math.inf``.
    name : str, optional
        Optional model name, by default ``""``.
    dim : tuple of int, optional
        Grid dimensions as ``(n_cols, n_rows)``, by default ``None``.
    **kwargs :
        Extra keyword arguments forwarded to
        :class:`~dissmodel.geo.spatial_model.SpatialModel`.

    Examples
    --------
    >>> class MyCA(CellularAutomaton):
    ...     def rule(self, idx):
    ...         return self.gdf.loc[idx, self.state_attr]
    """

    def __init__(
        self,
        gdf: gpd.GeoDataFrame,
        state_attr: str = "state",
        step: float = 1,
        start_time: float = 0,
        end_time: float = math.inf,
        name: str = "",
        dim: Optional[int] = None,
        **kwargs: Any,
    ) -> None:
        self.state_attr = state_attr
        self.dim        = dim
        super().__init__(
            gdf=gdf,
            step=step,
            start_time=start_time,
            end_time=end_time,
            name=name,
            **kwargs,
        )

    def initialize(self) -> None:
        """
        Set up the initial model state.

        Override in subclasses to define the starting conditions.
        """
        pass

    @abstractmethod
    def rule(self, idx: Any) -> Any:
        """
        Transition rule applied to each cell.

        Must be overridden in subclasses to define the state transition logic.

        Parameters
        ----------
        idx : any
            Index of the cell being evaluated.

        Returns
        -------
        any
            New state value for the cell.

        Raises
        ------
        NotImplementedError
            If not overridden by the subclass.
        """
        raise NotImplementedError("Subclasses must implement the transition rule.")

    def execute(self) -> None:
        """
        Execute one simulation step by applying :meth:`rule` to every cell.

        Raises
        ------
        RuntimeError
            If the neighborhood has not been created yet.

        Notes
        -----
        Because :meth:`rule` is an arbitrary Python function, the update
        cannot be vectorized automatically. Performance-critical subclasses
        should prefer :meth:`neighbor_values` over :meth:`neighs` inside
        ``rule`` to avoid geometry overhead on every lookup.
        """
        if not self._neighborhood_created:
            raise RuntimeError(
                "Neighborhood must be created before running the model. "
                "Call `.create_neighborhood()` first."
            )
        self.gdf[self.state_attr] = self.gdf.index.map(self.rule)

execute()

Execute one simulation step by applying :meth:rule to every cell.

Raises:

Type Description
RuntimeError

If the neighborhood has not been created yet.
Notes

Because :meth:rule is an arbitrary Python function, the update cannot be vectorized automatically. Performance-critical subclasses should prefer :meth:neighbor_values over :meth:neighs inside rule to avoid geometry overhead on every lookup.

Source code in dissmodel/geo/vector/cellular_automaton.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def execute(self) -> None:
    """
    Execute one simulation step by applying :meth:`rule` to every cell.

    Raises
    ------
    RuntimeError
        If the neighborhood has not been created yet.

    Notes
    -----
    Because :meth:`rule` is an arbitrary Python function, the update
    cannot be vectorized automatically. Performance-critical subclasses
    should prefer :meth:`neighbor_values` over :meth:`neighs` inside
    ``rule`` to avoid geometry overhead on every lookup.
    """
    if not self._neighborhood_created:
        raise RuntimeError(
            "Neighborhood must be created before running the model. "
            "Call `.create_neighborhood()` first."
        )
    self.gdf[self.state_attr] = self.gdf.index.map(self.rule)

initialize()

Set up the initial model state.

Override in subclasses to define the starting conditions.

Source code in dissmodel/geo/vector/cellular_automaton.py 
80
81
82
83
84
85
86
def initialize(self) -> None:
    """
    Set up the initial model state.

    Override in subclasses to define the starting conditions.
    """
    pass

rule(idx) abstractmethod

Transition rule applied to each cell.

Must be overridden in subclasses to define the state transition logic.

Parameters:

Name Type Description Default
idx any

Index of the cell being evaluated.

 required

Returns:

Type Description
any

New state value for the cell.

Raises:

Type Description
NotImplementedError

If not overridden by the subclass.
Source code in dissmodel/geo/vector/cellular_automaton.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@abstractmethod
def rule(self, idx: Any) -> Any:
    """
    Transition rule applied to each cell.

    Must be overridden in subclasses to define the state transition logic.

    Parameters
    ----------
    idx : any
        Index of the cell being evaluated.

    Returns
    -------
    any
        New state value for the cell.

    Raises
    ------
    NotImplementedError
        If not overridden by the subclass.
    """
    raise NotImplementedError("Subclasses must implement the transition rule.")

dissmodel.geo.vector.vector_grid

vector_grid(gdf=None, bounds=None, resolution=None, dimension=None, attrs=None, crs=None)

Create a regular grid of fixed-size cells.

Exactly one of the following input combinations must be provided:

  • dimension + resolution — abstract grid with no geographic location
  • bounds + resolution — grid fitted to a bounding box by cell size
  • bounds + dimension — grid fitted to a bounding box by cell count
  • gdf — bounds are extracted from the GeoDataFrame

Parameters:

Name Type Description Default
gdf GeoDataFrame

GeoDataFrame used to extract the bounding box.

 None
bounds tuple of float

Bounding box as (xmin, ymin, xmax, ymax).

 None
resolution float

Cell size in coordinate units.

 None
dimension tuple of int

Grid shape as (n_cols, n_rows).

 None
attrs dict

Extra attributes added to every cell, e.g. {'state': 0}.

 None
crs str or int

Coordinate reference system. If None, the grid is abstract (no geographic location).

 None

Returns:

Type Description
GeoDataFrame

Regular grid where each row is a cell with a Polygon geometry, indexed by 'id' in 'row-col' format.

Raises:

Type Description
ValueError

If the input combination is insufficient to define the grid.

Examples:

>>> gdf = vector_grid(dimension=(3, 3), resolution=1.0)
>>> len(gdf)
9
>>> gdf.index[0]
'0-0'
Source code in dissmodel/geo/vector/vector_grid.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def vector_grid(
    gdf: Optional[gpd.GeoDataFrame] = None,
    bounds: Optional[Bounds] = None,
    resolution: Optional[float] = None,
    dimension: Optional[Dimension] = None,
    attrs: Optional[dict[str, Any]] = None,
    crs: Optional[str | int] = None,
) -> gpd.GeoDataFrame:
    """
    Create a regular grid of fixed-size cells.

    Exactly one of the following input combinations must be provided:

    - ``dimension`` + ``resolution`` — abstract grid with no geographic location
    - ``bounds`` + ``resolution`` — grid fitted to a bounding box by cell size
    - ``bounds`` + ``dimension`` — grid fitted to a bounding box by cell count
    - ``gdf`` — bounds are extracted from the GeoDataFrame

    Parameters
    ----------
    gdf : geopandas.GeoDataFrame, optional
        GeoDataFrame used to extract the bounding box.
    bounds : tuple of float, optional
        Bounding box as ``(xmin, ymin, xmax, ymax)``.
    resolution : float, optional
        Cell size in coordinate units.
    dimension : tuple of int, optional
        Grid shape as ``(n_cols, n_rows)``.
    attrs : dict, optional
        Extra attributes added to every cell, e.g. ``{'state': 0}``.
    crs : str or int, optional
        Coordinate reference system. If ``None``, the grid is abstract
        (no geographic location).

    Returns
    -------
    geopandas.GeoDataFrame
        Regular grid where each row is a cell with a Polygon geometry,
        indexed by ``'id'`` in ``'row-col'`` format.

    Raises
    ------
    ValueError
        If the input combination is insufficient to define the grid.

    Examples
    --------
    >>> gdf = vector_grid(dimension=(3, 3), resolution=1.0)
    >>> len(gdf)
    9
    >>> gdf.index[0]
    '0-0'
    """
    attrs = attrs or {}

    resolution_x: float
    resolution_y: float
    n_cols: int
    n_rows: int
    xmin: float
    ymin: float
    xmax: float
    ymax: float

    if dimension is not None and resolution is not None and bounds is None and gdf is None:
        n_cols, n_rows = dimension
        xmin, ymin = 0.0, 0.0
        resolution_x = resolution_y = resolution
        xmax = xmin + n_cols * resolution_x
        ymax = ymin + n_rows * resolution_y

    elif bounds is not None:
        xmin, ymin, xmax, ymax = bounds
        width = xmax - xmin
        height = ymax - ymin

        if resolution is not None:
            resolution_x = resolution_y = resolution
            n_cols = int(np.ceil(width / resolution_x))
            n_rows = int(np.ceil(height / resolution_y))
        elif dimension is not None:
            n_cols, n_rows = dimension
            resolution_x = width / n_cols
            resolution_y = height / n_rows
        else:
            raise ValueError("Provide either `resolution` or `dimension`.")

    elif gdf is not None:
        return vector_grid(
            bounds=tuple(gdf.total_bounds),  # type: ignore[arg-type]
            resolution=resolution,
            dimension=dimension,
            attrs=attrs,
            crs=gdf.crs,
        )

    else:
        raise ValueError("Provide `gdf`, `bounds`, or `dimension` with `resolution`.")

    x_edges: np.ndarray = np.arange(xmin, xmax, resolution_x)
    y_edges: np.ndarray = np.arange(ymin, ymax, resolution_y)

    grid_cells = []
    ids = []
    for i, x0 in enumerate(x_edges):
        for j, y0 in enumerate(y_edges):
            grid_cells.append(box(x0, y0, x0 + resolution_x, y0 + resolution_y))
            ids.append(f"{j}-{i}")

    data: dict[str, Any] = {"geometry": grid_cells, "id": ids}
    for key, value in attrs.items():
        data[key] = [value] * len(grid_cells)

    return gpd.GeoDataFrame(data, crs=crs).set_index("id")

parse_idx(idx)

Extract row and col from an index string in 'row-col' format.

Parameters:

Name Type Description Default
idx str

Index string in 'row-col' format, e.g. '0-0', '3-4'.

 required

Returns:

Type Description
GridPos

Named tuple with fields row and col.

Examples:

>>> pos = parse_idx('3-4')
>>> pos.row
3
>>> pos.col
4
>>> row, col = parse_idx('3-4')  # tuple unpacking still works
Source code in dissmodel/geo/vector/vector_grid.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def parse_idx(idx: str) -> GridPos:
    """
    Extract row and col from an index string in ``'row-col'`` format.

    Parameters
    ----------
    idx : str
        Index string in ``'row-col'`` format, e.g. ``'0-0'``, ``'3-4'``.

    Returns
    -------
    GridPos
        Named tuple with fields ``row`` and ``col``.

    Examples
    --------
    >>> pos = parse_idx('3-4')
    >>> pos.row
    3
    >>> pos.col
    4
    >>> row, col = parse_idx('3-4')  # tuple unpacking still works
    """
    row_str, col_str = idx.split("-")
    return GridPos(row=int(row_str), col=int(col_str))

dissmodel.geo.vector.fill

FillStrategy

Bases: str, Enum

Available fill strategies for populating GeoDataFrame attributes.

Attributes:

Name Type Description
ZONAL_STATS str

Fill cells with statistics extracted from a raster.
MIN_DISTANCE str

Fill cells with the minimum distance to a target GeoDataFrame.
RANDOM_SAMPLE str

Fill cells with random samples drawn from a distribution.
PATTERN str

Fill cells using a 2-D pattern matrix.

Examples:

>>> FillStrategy.RANDOM_SAMPLE
<FillStrategy.RANDOM_SAMPLE: 'random_sample'>
>>> FillStrategy("pattern")
<FillStrategy.PATTERN: 'pattern'>
Source code in dissmodel/geo/vector/fill.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class FillStrategy(str, Enum):
    """
    Available fill strategies for populating GeoDataFrame attributes.

    Attributes
    ----------
    ZONAL_STATS : str
        Fill cells with statistics extracted from a raster.
    MIN_DISTANCE : str
        Fill cells with the minimum distance to a target GeoDataFrame.
    RANDOM_SAMPLE : str
        Fill cells with random samples drawn from a distribution.
    PATTERN : str
        Fill cells using a 2-D pattern matrix.

    Examples
    --------
    >>> FillStrategy.RANDOM_SAMPLE
    <FillStrategy.RANDOM_SAMPLE: 'random_sample'>
    >>> FillStrategy("pattern")
    <FillStrategy.PATTERN: 'pattern'>
    """

    ZONAL_STATS = "zonal_stats"
    MIN_DISTANCE = "min_distance"
    RANDOM_SAMPLE = "random_sample"
    PATTERN = "pattern"

fill(strategy, **kwargs)

Execute a fill strategy by name.

Parameters:

Name Type Description Default
strategy FillStrategy or str

Strategy to execute. Accepted values: 'pattern', 'random_sample', 'zonal_stats', 'min_distance'.

 required
**kwargs Any

Arguments forwarded to the chosen strategy function.

 {}

Returns:

Type Description
Any

Whatever the strategy function returns. Most strategies mutate the GeoDataFrame in place and return None.

Raises:

Type Description
ValueError

If strategy is not a registered :class:FillStrategy value.

Examples:

>>> fill(FillStrategy.RANDOM_SAMPLE, gdf=grid, attr="state",
...      data=[0, 1], seed=42)
>>> fill("min_distance", from_gdf=grid, to_gdf=roads,
...      attr_name="dist_road")
>>> fill(FillStrategy.PATTERN, gdf=grid, attr="zone",
...      pattern=[[1, 2], [3, 4]])
Source code in dissmodel/geo/vector/fill.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def fill(strategy: FillStrategy | str, **kwargs: Any) -> Any:
    """
    Execute a fill strategy by name.

    Parameters
    ----------
    strategy : FillStrategy or str
        Strategy to execute. Accepted values: ``'pattern'``,
        ``'random_sample'``, ``'zonal_stats'``, ``'min_distance'``.
    **kwargs :
        Arguments forwarded to the chosen strategy function.

    Returns
    -------
    Any
        Whatever the strategy function returns. Most strategies mutate the
        GeoDataFrame in place and return ``None``.

    Raises
    ------
    ValueError
        If ``strategy`` is not a registered :class:`FillStrategy` value.

    Examples
    --------
    >>> fill(FillStrategy.RANDOM_SAMPLE, gdf=grid, attr="state",
    ...      data=[0, 1], seed=42)
    >>> fill("min_distance", from_gdf=grid, to_gdf=roads,
    ...      attr_name="dist_road")
    >>> fill(FillStrategy.PATTERN, gdf=grid, attr="zone",
    ...      pattern=[[1, 2], [3, 4]])
    """
    key = FillStrategy(strategy)
    if key not in _fill_strategies:
        raise ValueError(f"Unknown strategy: {strategy!r}")
    return _fill_strategies[key](**kwargs)

register_strategy(name)

Register a fill strategy under the given name.

Parameters:

Name Type Description Default
name FillStrategy

Key under which the strategy will be registered.

 required

Returns:

Type Description
Callable

Decorator that registers and returns the decorated function.
Source code in dissmodel/geo/vector/fill.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def register_strategy(
    name: FillStrategy,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """
    Register a fill strategy under the given name.

    Parameters
    ----------
    name : FillStrategy
        Key under which the strategy will be registered.

    Returns
    -------
    Callable
        Decorator that registers and returns the decorated function.
    """
    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        _fill_strategies[name] = func
        return func
    return decorator

dissmodel.geo.vector.neighborhood

attach_neighbors(gdf, strategy=None, neighbors_dict=None, **kwargs)

Attach a neighborhood structure to a GeoDataFrame.

Adds a '_neighs' column containing the list of neighbor indices for each cell. Mutates and returns the same GeoDataFrame.

Parameters:

Name Type Description Default
gdf GeoDataFrame

GeoDataFrame whose cells will receive the neighborhood column.

 required
strategy WeightStrategy

Libpysal weight class (e.g. Queen, Rook) whose from_dataframe classmethod will be called. Ignored if neighbors_dict is provided.

 None
neighbors_dict dict or str

Precomputed neighborhood. Accepted formats:

  • dict — {index: [neighbor_indices]} mapping.
  • str — path to a JSON file with the same structure.
  • None — neighborhood will be computed via strategy.
 None
**kwargs Any

Extra keyword arguments forwarded to strategy.from_dataframe.

 {}

Returns:

Type Description
GeoDataFrame

The same gdf with the '_neighs' column added.

Raises:

Type Description
FileNotFoundError

If a string path is provided in neighbors_dict but does not exist.
ValueError

If neighbors_dict is not a dict, None, or a valid JSON path.
ValueError

If neither strategy nor neighbors_dict is provided.

Examples:

>>> from libpysal.weights import Queen
>>> gdf = attach_neighbors(gdf, strategy=Queen)
>>> gdf = attach_neighbors(gdf, neighbors_dict="neighborhood.json")
>>> gdf = attach_neighbors(gdf, strategy=Queen, ids="cell_id")
Source code in dissmodel/geo/vector/neighborhood.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def attach_neighbors(
    gdf: gpd.GeoDataFrame,
    strategy: StrategyType = None,
    neighbors_dict: Optional[dict[Any, list[Any]] | str] = None,
    **kwargs: Any,
) -> gpd.GeoDataFrame:
    """
    Attach a neighborhood structure to a GeoDataFrame.

    Adds a ``'_neighs'`` column containing the list of neighbor indices for
    each cell. Mutates and returns the same GeoDataFrame.

    Parameters
    ----------
    gdf : geopandas.GeoDataFrame
        GeoDataFrame whose cells will receive the neighborhood column.
    strategy : WeightStrategy, optional
        Libpysal weight class (e.g. ``Queen``, ``Rook``) whose
        ``from_dataframe`` classmethod will be called. Ignored if
        ``neighbors_dict`` is provided.
    neighbors_dict : dict or str, optional
        Precomputed neighborhood. Accepted formats:

        - ``dict`` — ``{index: [neighbor_indices]}`` mapping.
        - ``str`` — path to a JSON file with the same structure.
        - ``None`` — neighborhood will be computed via ``strategy``.
    **kwargs :
        Extra keyword arguments forwarded to ``strategy.from_dataframe``.

    Returns
    -------
    geopandas.GeoDataFrame
        The same ``gdf`` with the ``'_neighs'`` column added.

    Raises
    ------
    FileNotFoundError
        If a string path is provided in ``neighbors_dict`` but does not exist.
    ValueError
        If ``neighbors_dict`` is not a dict, ``None``, or a valid JSON path.
    ValueError
        If neither ``strategy`` nor ``neighbors_dict`` is provided.

    Examples
    --------
    >>> from libpysal.weights import Queen
    >>> gdf = attach_neighbors(gdf, strategy=Queen)
    >>> gdf = attach_neighbors(gdf, neighbors_dict="neighborhood.json")
    >>> gdf = attach_neighbors(gdf, strategy=Queen, ids="cell_id")
    """
    resolved = _resolve_neighbors_dict(neighbors_dict)

    w: W
    if resolved is not None:
        w = W(resolved)
    elif strategy is not None:
         if "use_index" not in kwargs:
            kwargs["use_index"] = True
         w = strategy.from_dataframe(gdf, **kwargs)
    else:
        raise ValueError("Provide either `strategy` or `neighbors_dict`.")

    gdf["_neighs"] = gdf.index.map(lambda idx: w.neighbors.get(idx, []))
    return gdf