spatial Module

Spatial simulation models with multi-deme support.

Overview

The spatial subpackage provides SpatialPopulation, SpatialConfigurator, and topology classes for multi-deme population simulations with migration.

Complete Module Reference

natal.spatial

Spatial population models, topology, and configuration.

BatchSetting

BatchSetting(values: Union[Sequence[Any], NDArray[floating[Any]], Callable[..., float]])

Deferred per-deme parameter specification.

Wraps one of three value kinds used by SpatialConfigurator to express parameters that vary across demes:

• scalar: A Python sequence (list/tuple), one element per deme. Each element can be a scalar or an array (e.g. per-deme equilibrium distributions).
• array: A 1D or 2D numpy array. 1D arrays have one element per deme (flat-index order); 2D arrays use (row, col) layout matching the topology grid and are flattened in row-major order at build time.
• spatial: A callable (flat_idx) -> float or (row, col) -> float (auto-detected by parameter count), expanded at build time.

SpatialConfigurator detects BatchSetting values in builder method calls, stores them, and expands them during build().

Initialize a BatchSetting from one of three value kinds.

Parameters:

Name	Type	Description	Default
values	Union[Sequence[Any], NDArray[floating[Any]], Callable[..., float]]	Per-deme specification — a sequence of scalars (one per deme), a 1D/2D numpy array, or a callable for spatial expansion.	required

Source code in src/natal/spatial/configurator.py

def __init__(
    self,
    values: Union[Sequence[Any], NDArray[np.floating[Any]], Callable[..., float]],
):
    """Initialize a BatchSetting from one of three value kinds.

    Args:
        values: Per-deme specification — a sequence of scalars (one per
            deme), a 1D/2D numpy array, or a callable for spatial
            expansion.
    """
    self._fn: Optional[Callable[..., float]] = None
    self._fn_param_count: Optional[int] = None
    self._values: Optional[List[Any]] = None
    self._values_array: Optional[NDArray[np.floating[Any]]] = None
    self._n_demes: Optional[int] = None

    if callable(values):
        self._kind: str = self._KIND_SPATIAL
        self._fn = values
    elif isinstance(values, np.ndarray):
        if values.ndim not in (1, 2):
            raise ValueError(
                f"BatchSetting array must be 1D or 2D, got shape {values.shape}"
            )
        self._kind = self._KIND_ARRAY
        self._values_array = np.asarray(values)
        self._n_demes = int(self._values_array.size)
    else:
        self._kind = self._KIND_SCALAR
        self._values = list(values)
        self._n_demes = len(self._values)

kind property

kind: str

str: The kind of value source ("scalar", "array", or "spatial").

expand

expand(n_demes: int, topology: Optional[GridTopology] = None) -> List[Any]

Expand to a concrete list of per-deme values.

• scalar/array: validate length, return as list (2D arrays are flattened row-major).
• spatial: call the function for each deme index. Parameter count is auto-detected: 1 param → fn(flat_idx), 2 params → fn(row, col).

Parameters:

Name	Type	Description	Default
n_demes	int	Number of demes to expand to.	required
topology	Optional[GridTopology]	Optional GridTopology required for spatial kind.	None

Returns:

Type	Description
List[Any]	List of per-deme values (scalars or arrays).

Raises:

Type	Description
ValueError	If length mismatch or spatial kind without topology.

Source code in src/natal/spatial/configurator.py

def expand(
    self,
    n_demes: int,
    topology: Optional[GridTopology] = None,
) -> List[Any]:
    """Expand to a concrete list of per-deme values.

    - **scalar/array**: validate length, return as list (2D arrays are
      flattened row-major).
    - **spatial**: call the function for each deme index.  Parameter
      count is auto-detected: 1 param → ``fn(flat_idx)``,
      2 params → ``fn(row, col)``.

    Args:
        n_demes: Number of demes to expand to.
        topology: Optional ``GridTopology`` required for spatial kind.

    Returns:
        List of per-deme values (scalars or arrays).

    Raises:
        ValueError: If length mismatch or spatial kind without topology.
    """
    if self._kind == self._KIND_SCALAR:
        if self._values is None:
            raise ValueError("BatchSetting scalar values are None")
        if len(self._values) != n_demes:
            raise ValueError(
                f"BatchSetting has {len(self._values)} values "
                f"but {n_demes} demes are required"
            )
        return list(self._values)

    elif self._kind == self._KIND_ARRAY:
        if self._values_array is None:
            raise ValueError("BatchSetting array values are None")
        if self._n_demes != n_demes:
            raise ValueError(
                f"BatchSetting array has {self._n_demes} values "
                f"but {n_demes} demes are required"
            )
        arr = self._values_array
        if arr.ndim == 2:
            if topology is not None and arr.shape != (topology.rows, topology.cols):
                raise ValueError(
                    f"BatchSetting 2D array shape {arr.shape} does not match "
                    f"topology shape ({topology.rows}, {topology.cols})"
                )
            return arr.ravel(order="C").tolist()
        return arr.tolist()

    elif self._kind == self._KIND_SPATIAL:
        if topology is None:
            raise ValueError(
                "Spatial BatchSetting requires topology for expansion."
            )
        fn = cast(Callable[..., float], self._fn)
        # Auto-detect parameter count: 2 → (row, col), else (flat_idx).
        if self._fn_param_count is None:
            import inspect
            try:
                sig = inspect.signature(fn)
                self._fn_param_count = len(sig.parameters)
            except (ValueError, TypeError):
                self._fn_param_count = 1
        if self._fn_param_count >= 2:
            return [
                float(fn(*topology.from_index(i)))
                for i in range(n_demes)
            ]
        return [float(fn(i)) for i in range(n_demes)]

    raise ValueError(f"Unknown kind: {self._kind}")

first_value

first_value() -> Any

Return a single concrete element for template-builder delegation.

SpatialConfigurator holds a single-deme template builder internally. When a parameter is wrapped in batch_setting (a per-deme list), the template builder still needs one scalar value to proceed through setup() → … → build(). This method provides that value — typically the first element of the list or array.

For spatial kind (lazy callable), returns None because the value cannot be resolved without topology expansion at build time.

Returns:

Type	Description
Any	The first element for scalar/array kinds, or None for spatial kind.

Source code in src/natal/spatial/configurator.py

def first_value(self) -> Any:
    """Return a single concrete element for template-builder delegation.

    ``SpatialConfigurator`` holds a single-deme template builder internally.
    When a parameter is wrapped in ``batch_setting`` (a per-deme list),
    the template builder still needs one scalar value to proceed through
    ``setup() → … → build()``. This method provides that value —
    typically the first element of the list or array.

    For spatial kind (lazy callable), returns ``None`` because the value
    cannot be resolved without topology expansion at build time.

    Returns:
        The first element for scalar/array kinds, or ``None`` for spatial kind.
    """
    if self._kind == self._KIND_SCALAR:
        return self._values[0] if self._values else None
    elif self._kind == self._KIND_ARRAY:
        if self._values_array is not None and self._values_array.size > 0:
            flat = self._values_array.ravel(order="C")
            val = flat[0]
            return val.item() if hasattr(val, 'item') else val
        return None
    return None  # spatial kind: deferred until expand() has topology

SpatialConfigurator

SpatialConfigurator(species: Species, n_demes: int, topology: Optional[GridTopology] = None, *, pop_type: Literal['age_structured', 'discrete_generation'] = 'age_structured')

Fluent builder for SpatialPopulation.

Wraps a single-deme Configurator as a template. All chainable configuration methods delegate to the template and return self.

Spatial-specific parameters (topology, migration, adjacency) are stored directly and forwarded to SpatialPopulation at build time.

Initialize the spatial configurator.

Creates a single-deme template Configurator internally and stores spatial parameters (topology, migration) for later use during build().

Parameters:

Name	Type	Description	Default
species	Species	Genetic architecture shared by all demes.	required
n_demes	int	Number of demes in the spatial layout.	required
topology	Optional[GridTopology]	Optional grid topology for migration routing.	None
pop_type	Literal['age_structured', 'discrete_generation']	Population model type — "age_structured" (default) or "discrete_generation".	'age_structured'

Raises:

Type	Description
ValueError	If n_demes is less than 1.

Source code in src/natal/spatial/configurator.py

def __init__(
    self,
    species: Species,
    n_demes: int,
    topology: Optional[GridTopology] = None,
    *,
    pop_type: Literal["age_structured", "discrete_generation"] = "age_structured",
):
    """Initialize the spatial configurator.

    Creates a single-deme template ``Configurator`` internally and
    stores spatial parameters (topology, migration) for later use
    during ``build()``.

    Args:
        species: Genetic architecture shared by all demes.
        n_demes: Number of demes in the spatial layout.
        topology: Optional grid topology for migration routing.
        pop_type: Population model type — ``"age_structured"``
            (default) or ``"discrete_generation"``.

    Raises:
        ValueError: If ``n_demes`` is less than 1.
    """
    if n_demes < 1:
        raise ValueError(f"n_demes must be >= 1, got {n_demes}")

    self._species = species
    self._n_demes = n_demes
    self._topology = topology
    self._pop_type: Literal["age_structured", "discrete_generation"] = pop_type

    # Observation groups (optional, applied at build time).
    self._observation_groups: Optional[GroupsInput] = None
    self._observation_collapse_age: bool = False

    # Create the template configurator (new path).
    if pop_type == "age_structured":
        self._template: DiscreteConfigurator | AgeStructuredConfigurator = \
            Configurator.for_age_structured(species)
    else:
        self._template: DiscreteConfigurator | AgeStructuredConfigurator = \
            Configurator.for_discrete(species)

    # Accumulated batch settings: param_name -> BatchSetting.
    self._batch_settings: Dict[str, BatchSetting] = {}

    # Replay log: list of (method_name, kwargs_with_batch_settings).
    self._replay_log: List[tuple[str, Dict[str, Any]]] = []

    # Spatial migration parameters.
    self._migration_kernel: Optional[NDArray[np.float64]] = None
    self._migration_kernel_batch: Optional[BatchSetting] = None
    self._migration_rate: float = 0.0
    self._migration_strategy: Literal["auto", "adjacency", "kernel", "hybrid"] = "auto"
    self._migration_adjacency: Optional[object] = None
    self._kernel_bank: Optional[Sequence[NDArray[np.float64]]] = None
    self._deme_kernel_ids: Optional[NDArray[np.int64]] = None
    self._kernel_include_center: bool = False
    self._adjust_migration_on_edge: bool = False

    # Parameter registry (mirrors PopulationBuilderBase._param_values).
    self._param_values: dict[str, object] = {}
    self._spatial_name: str = "SpatialPopulation"

setup

setup(name: str = 'SpatialPopulation', stochastic: bool = True, continuous_sampling: bool = False, fixed_egg_count: bool = False) -> SpatialConfigurator

Configure basic population settings.

Parameters:

Name	Type	Description	Default
name	str	Human-readable population name.	'SpatialPopulation'
stochastic	bool	Whether to use stochastic sampling.	True
continuous_sampling	bool	If True, use Dirichlet sampling.	False
fixed_egg_count	bool	If True, egg count is fixed.	False

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def setup(
    self,
    name: str = "SpatialPopulation",
    stochastic: bool = True,
    continuous_sampling: bool = False,
    fixed_egg_count: bool = False,
) -> SpatialConfigurator:
    """Configure basic population settings.

    Args:
        name: Human-readable population name.
        stochastic: Whether to use stochastic sampling.
        continuous_sampling: If True, use Dirichlet sampling.
        fixed_egg_count: If True, egg count is fixed.

    Returns:
        Self for chaining.
    """
    self._spatial_name = name
    self._replay_log.append(("setup", {
        "name": name,
        "stochastic": stochastic,
        "continuous_sampling": continuous_sampling,
        "fixed_egg_count": fixed_egg_count,
    }))
    self._template.setup(
        name=name,
        stochastic=stochastic,
        continuous_sampling=continuous_sampling,
        fixed_egg_count=fixed_egg_count,
    )
    return self

age_structure

age_structure(n_ages: int = 8, new_adult_age: int = 2, generation_time: Optional[float] = None, equilibrium_distribution: Optional[Union[List[float], NDArray[float64]]] = None) -> SpatialConfigurator

Configure age structure (age-structured models only).

Parameters:

Name	Type	Description	Default
n_ages	int	Number of age classes.	8
new_adult_age	int	Age at which individuals become adults.	2
generation_time	Optional[float]	Optional pre-computed generation time.	None
equilibrium_distribution	Optional[Union[List[float], NDArray[float64]]]	Optional equilibrium distribution.	None

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def age_structure(
    self,
    n_ages: int = 8,
    new_adult_age: int = 2,
    generation_time: Optional[float] = None,
    equilibrium_distribution: Optional[Union[List[float], NDArray[np.float64]]] = None,
) -> SpatialConfigurator:
    """Configure age structure (age-structured models only).

    Args:
        n_ages: Number of age classes.
        new_adult_age: Age at which individuals become adults.
        generation_time: Optional pre-computed generation time.
        equilibrium_distribution: Optional equilibrium distribution.

    Returns:
        Self for chaining.
    """
    if self._pop_type != "age_structured":
        raise TypeError("age_structure() is only valid for age_structured pop_type")
    return self._detect_and_delegate(
        "age_structure",
        {
            "n_ages": n_ages,
            "new_adult_age": new_adult_age,
            "generation_time": generation_time,
            "equilibrium_distribution": equilibrium_distribution,
        },
    )

initial_state

initial_state(individual_count: Any, sperm_storage: Optional[Any] = None) -> SpatialConfigurator

Configure the initial population state.

Parameters:

Name	Type	Description	Default
individual_count	Any	Initial abundance mapping.	required
sperm_storage	Optional[Any]	Optional initial sperm storage (age-structured only).	None

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def initial_state(
    self,
    individual_count: Any,
    sperm_storage: Optional[Any] = None,
) -> SpatialConfigurator:
    """Configure the initial population state.

    Args:
        individual_count: Initial abundance mapping.
        sperm_storage: Optional initial sperm storage (age-structured only).

    Returns:
        Self for chaining.
    """
    kwargs: Dict[str, Any] = {"individual_count": individual_count}
    if sperm_storage is not None:
        kwargs["sperm_storage"] = sperm_storage
    return self._detect_and_delegate("initial_state", kwargs)

survival

survival(female_age_based_survival: Optional[Any] = None, male_age_based_survival: Optional[Any] = None, generation_time: Optional[float] = None, equilibrium_distribution: Optional[Any] = None, female_age0_survival: Optional[float] = None, male_age0_survival: Optional[float] = None) -> SpatialConfigurator

Configure survival rates.

Parameters:

Name	Type	Description	Default
female_age_based_survival	Optional[Any]	Per-age female survival (age-structured).	None
male_age_based_survival	Optional[Any]	Per-age male survival (age-structured).	None
generation_time	Optional[float]	Optional generation time override.	None
equilibrium_distribution	Optional[Any]	Optional equilibrium distribution.	None
female_age0_survival	Optional[float]	Female age-0 survival (discrete-generation).	None
male_age0_survival	Optional[float]	Male age-0 survival (discrete-generation).	None

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def survival(
    self,
    # Age-structured params
    female_age_based_survival: Optional[Any] = None,
    male_age_based_survival: Optional[Any] = None,
    generation_time: Optional[float] = None,
    equilibrium_distribution: Optional[Any] = None,
    # Discrete-generation params
    female_age0_survival: Optional[float] = None,
    male_age0_survival: Optional[float] = None,
) -> SpatialConfigurator:
    """Configure survival rates.

    Args:
        female_age_based_survival: Per-age female survival (age-structured).
        male_age_based_survival: Per-age male survival (age-structured).
        generation_time: Optional generation time override.
        equilibrium_distribution: Optional equilibrium distribution.
        female_age0_survival: Female age-0 survival (discrete-generation).
        male_age0_survival: Male age-0 survival (discrete-generation).

    Returns:
        Self for chaining.
    """
    if self._pop_type == "age_structured":
        return self._detect_and_delegate(
            "survival",
            {
                "female_age_based_survival": female_age_based_survival,
                "male_age_based_survival": male_age_based_survival,
                "generation_time": generation_time,
                "equilibrium_distribution": equilibrium_distribution,
            },
        )
    else:
        return self._detect_and_delegate(
            "survival",
            {
                "female_age0_survival": female_age0_survival,
                "male_age0_survival": male_age0_survival,
            },
        )

reproduction

reproduction(eggs_per_female: Union[float, BatchSetting] = 50.0, sex_ratio: Union[float, BatchSetting] = 0.5, fixed_egg_count: bool = False, female_age_based_mating_rate: Optional[Any] = None, male_age_based_mating_rate: Optional[Any] = None, age_based_reproduction_rate: Optional[Any] = None, female_age_based_fertility: Optional[Any] = None, sperm_displacement_rate: float = 0.05, female_adult_mating_rate: float = 1.0, male_adult_mating_rate: float = 1.0) -> SpatialConfigurator

Configure reproduction and mating parameters.

Parameters:

Name	Type	Description	Default
eggs_per_female	Union[float, BatchSetting]	Expected offspring per adult female. Accepts BatchSetting.	50.0
sex_ratio	Union[float, BatchSetting]	Proportion of female offspring.	0.5
fixed_egg_count	bool	If True, egg count is deterministic.	False
female_age_based_mating_rate	Optional[Any]	Female mating rates (age-structured).	None
male_age_based_mating_rate	Optional[Any]	Male mating rates (age-structured).	None
age_based_reproduction_rate	Optional[Any]	Reproduction participation rates.	None
female_age_based_fertility	Optional[Any]	Fertility weights.	None
sperm_displacement_rate	float	Rate of sperm displacement (age-structured).	0.05
female_adult_mating_rate	float	Adult female mating rate (discrete-generation).	1.0
male_adult_mating_rate	float	Adult male mating rate (discrete-generation).	1.0

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def reproduction(
    self,
    # Shared params (accept BatchSetting for per-deme variation)
    eggs_per_female: Union[float, BatchSetting] = 50.0,
    sex_ratio: Union[float, BatchSetting] = 0.5,
    fixed_egg_count: bool = False,
    # Age-structured params
    female_age_based_mating_rate: Optional[Any] = None,
    male_age_based_mating_rate: Optional[Any] = None,
    age_based_reproduction_rate: Optional[Any] = None,
    female_age_based_fertility: Optional[Any] = None,
    sperm_displacement_rate: float = 0.05,
    # Discrete-generation params
    female_adult_mating_rate: float = 1.0,
    male_adult_mating_rate: float = 1.0,
) -> SpatialConfigurator:
    """Configure reproduction and mating parameters.

    Args:
        eggs_per_female: Expected offspring per adult female. Accepts ``BatchSetting``.
        sex_ratio: Proportion of female offspring.
        fixed_egg_count: If True, egg count is deterministic.
        female_age_based_mating_rate: Female mating rates (age-structured).
        male_age_based_mating_rate: Male mating rates (age-structured).
        age_based_reproduction_rate: Reproduction participation rates.
        female_age_based_fertility: Fertility weights.
        sperm_displacement_rate: Rate of sperm displacement (age-structured).
        female_adult_mating_rate: Adult female mating rate (discrete-generation).
        male_adult_mating_rate: Adult male mating rate (discrete-generation).

    Returns:
        Self for chaining.
    """
    if self._pop_type == "age_structured":
        return self._detect_and_delegate(
            "reproduction",
            {
                "female_age_based_mating_rate": female_age_based_mating_rate,
                "male_age_based_mating_rate": male_age_based_mating_rate,
                "age_based_reproduction_rate": age_based_reproduction_rate,
                "female_age_based_fertility": female_age_based_fertility,
                "eggs_per_female": eggs_per_female,
                "fixed_egg_count": fixed_egg_count,
                "sex_ratio": sex_ratio,
                "sperm_displacement_rate": sperm_displacement_rate,
            },
        )
    else:
        return self._detect_and_delegate(
            "reproduction",
            {
                "eggs_per_female": eggs_per_female,
                "sex_ratio": sex_ratio,
                "female_adult_mating_rate": female_adult_mating_rate,
                "male_adult_mating_rate": male_adult_mating_rate,
            },
        )

competition

competition(competition_strength: float = 5.0, juvenile_growth_mode: Union[int, str, BatchSetting] = 'logistic', low_density_growth_rate: Union[float, BatchSetting] = 6.0, age_1_carrying_capacity: Union[int, None, BatchSetting] = None, old_juvenile_carrying_capacity: Union[int, None, BatchSetting] = None, expected_num_new_adult_females: Union[int, None, BatchSetting] = None, equilibrium_distribution: Optional[Union[List[float], NDArray[float64], BatchSetting]] = None, carrying_capacity: Union[int, None, BatchSetting] = None) -> SpatialConfigurator

Configure competition and density-dependence.

Parameters:

Name	Type	Description	Default
competition_strength	float	Relative competition factor for age-1 juveniles (age-structured only).	5.0
juvenile_growth_mode	Union[int, str, BatchSetting]	Growth model identifier. Accepts BatchSetting.	'logistic'
low_density_growth_rate	Union[float, BatchSetting]	Growth rate at low density. Accepts BatchSetting.	6.0
age_1_carrying_capacity	Union[int, None, BatchSetting]	Carrying capacity at age=1 (age-structured). Accepts BatchSetting.	None
old_juvenile_carrying_capacity	Union[int, None, BatchSetting]	Alias for age_1_carrying_capacity.	None
expected_num_new_adult_females	Union[int, None, BatchSetting]	Equilibrium adult females. Accepts BatchSetting.	None
equilibrium_distribution	Optional[Union[List[float], NDArray[float64], BatchSetting]]	Optional equilibrium distribution.	None
carrying_capacity	Union[int, None, BatchSetting]	Carrying capacity (discrete-generation). Accepts BatchSetting.	None

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def competition(
    self,
    # Age-structured params
    competition_strength: float = 5.0,
    juvenile_growth_mode: Union[int, str, BatchSetting] = "logistic",
    low_density_growth_rate: Union[float, BatchSetting] = 6.0,
    age_1_carrying_capacity: Union[int, None, BatchSetting] = None,
    old_juvenile_carrying_capacity: Union[int, None, BatchSetting] = None,
    expected_num_new_adult_females: Union[int, None, BatchSetting] = None,
    equilibrium_distribution: Optional[Union[List[float], NDArray[np.float64], BatchSetting]] = None,
    # Discrete-generation params
    carrying_capacity: Union[int, None, BatchSetting] = None,
) -> SpatialConfigurator:
    """Configure competition and density-dependence.

    Args:
        competition_strength: Relative competition factor for age-1 juveniles
            (age-structured only).
        juvenile_growth_mode: Growth model identifier. Accepts ``BatchSetting``.
        low_density_growth_rate: Growth rate at low density. Accepts ``BatchSetting``.
        age_1_carrying_capacity: Carrying capacity at age=1 (age-structured).
            Accepts ``BatchSetting``.
        old_juvenile_carrying_capacity: Alias for ``age_1_carrying_capacity``.
        expected_num_new_adult_females: Equilibrium adult females. Accepts ``BatchSetting``.
        equilibrium_distribution: Optional equilibrium distribution.
        carrying_capacity: Carrying capacity (discrete-generation). Accepts ``BatchSetting``.

    Returns:
        Self for chaining.
    """
    if self._pop_type == "age_structured":
        # Alias carrying_capacity / old_juvenile_carrying_capacity → age_1_carrying_capacity
        resolved_cc = age_1_carrying_capacity
        if resolved_cc is None:
            resolved_cc = old_juvenile_carrying_capacity
        if resolved_cc is None:
            resolved_cc = carrying_capacity

        return self._detect_and_delegate(
            "competition",
            {
                "competition_strength": competition_strength,
                "juvenile_growth_mode": juvenile_growth_mode,
                "low_density_growth_rate": low_density_growth_rate,
                "age_1_carrying_capacity": resolved_cc,
                "expected_num_new_adult_females": expected_num_new_adult_females,
                "equilibrium_distribution": equilibrium_distribution,
            },
        )
    else:
        return self._detect_and_delegate(
            "competition",
            {
                "juvenile_growth_mode": juvenile_growth_mode,
                "low_density_growth_rate": low_density_growth_rate,
                "carrying_capacity": carrying_capacity,
            },
        )

presets

presets(*preset_list: GeneticPreset) -> SpatialConfigurator

Add gene-drive presets (applied during build).

Each positional argument may be a BatchSetting of preset objects, allowing different demes to receive different presets.

Parameters:

Name	Type	Description	Default
*preset_list	GeneticPreset	One or more preset objects, or BatchSetting instances wrapping per-deme preset values.	()

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def presets(self, *preset_list: GeneticPreset) -> SpatialConfigurator:
    """Add gene-drive presets (applied during build).

    Each positional argument may be a ``BatchSetting`` of preset objects,
    allowing different demes to receive different presets.

    Args:
        *preset_list: One or more preset objects, or ``BatchSetting``
            instances wrapping per-deme preset values.

    Returns:
        Self for chaining.
    """
    # Detect BatchSetting in positional args.
    concrete_args: list[object] = []
    for i, item in enumerate(preset_list):
        if isinstance(item, BatchSetting):
            self._batch_settings[f"_preset_{i}"] = item
            first = item.first_value()
            if first is not None:
                concrete_args.append(first)
        else:
            concrete_args.append(item)

    self._replay_log.append(("presets", {"preset_list": preset_list}))
    # concrete_args contains GeneticPreset instances resolved from potential
    # BatchSetting wrappers; cast needed because first_value() returns object.
    self._template.presets(*cast('list[GeneticPreset]', concrete_args))
    return self

fitness

fitness(viability: Optional[Any] = None, fecundity: Optional[Any] = None, sexual_selection: Optional[Any] = None, zygote_viability: Optional[Any] = None, mode: str = 'replace') -> SpatialConfigurator

Configure fitness values (applied after presets).

Parameters:

Name	Type	Description	Default
viability	Optional[Any]	Genotype selectors to viability fitness values.	None
fecundity	Optional[Any]	Genotype selectors to fecundity fitness values.	None
sexual_selection	Optional[Any]	Mating preference mapping.	None
zygote_viability	Optional[Any]	Genotype selectors to zygote viability values.	None
mode	str	"replace" (default) or "multiply".	'replace'

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def fitness(
    self,
    viability: Optional[Any] = None,
    fecundity: Optional[Any] = None,
    sexual_selection: Optional[Any] = None,
    zygote_viability: Optional[Any] = None,
    mode: str = "replace",
) -> SpatialConfigurator:
    """Configure fitness values (applied after presets).

    Args:
        viability: Genotype selectors to viability fitness values.
        fecundity: Genotype selectors to fecundity fitness values.
        sexual_selection: Mating preference mapping.
        zygote_viability: Genotype selectors to zygote viability values.
        mode: ``"replace"`` (default) or ``"multiply"``.

    Returns:
        Self for chaining.
    """
    return self._detect_and_delegate(
        "fitness",
        {
            "viability": viability,
            "fecundity": fecundity,
            "sexual_selection": sexual_selection,
            "zygote_viability": zygote_viability,
            "mode": mode,
        },
    )

hooks

hooks(*hook_items: _HookItem) -> SpatialConfigurator

Register lifecycle hooks.

Parameters:

Name	Type	Description	Default
*hook_items	_HookItem	Functions decorated with @hook or hook mappings.	()

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def hooks(self, *hook_items: _HookItem) -> SpatialConfigurator:
    """Register lifecycle hooks.

    Args:
        *hook_items: Functions decorated with ``@hook`` or hook mappings.

    Returns:
        Self for chaining.
    """
    self._replay_log.append(("hooks", {"hook_items": hook_items}))
    self._template.hooks(*hook_items)
    return self

modifiers

modifiers(gamete_modifiers: Optional[List[Tuple[int, Optional[str], Callable[..., object]]]] = None, zygote_modifiers: Optional[List[Tuple[int, Optional[str], Callable[..., object]]]] = None) -> SpatialConfigurator

Configure custom modifier functions.

Parameters:

Name	Type	Description	Default
gamete_modifiers	Optional[List[Tuple[int, Optional[str], Callable[..., object]]]]	Modifiers for gamete production.	None
zygote_modifiers	Optional[List[Tuple[int, Optional[str], Callable[..., object]]]]	Modifiers for zygote formation.	None

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def modifiers(
    self,
    gamete_modifiers: Optional[List[Tuple[int, Optional[str], Callable[..., object]]]] = None,
    zygote_modifiers: Optional[List[Tuple[int, Optional[str], Callable[..., object]]]] = None,
) -> SpatialConfigurator:
    """Configure custom modifier functions.

    Args:
        gamete_modifiers: Modifiers for gamete production.
        zygote_modifiers: Modifiers for zygote formation.

    Returns:
        Self for chaining.
    """
    return self._detect_and_delegate(
        "modifiers",
        {
            "gamete_modifiers": gamete_modifiers,
            "zygote_modifiers": zygote_modifiers,
        },
    )

with_observation

with_observation(groups: GroupsInput, *, collapse_age: bool = False) -> SpatialConfigurator

Register observation groups for compressed history recording.

The groups are compiled into a binary mask at build time and passed to the spatial simulation kernel on each run() call. Once set, history records store aggregated observation data instead of raw flattened state across all demes.

Parameters:

Name	Type	Description	Default
groups	GroupsInput	Observation groups (dict of name -> spec, list of specs, or None for one-group-per-genotype).	required
collapse_age	bool	Whether to collapse the age axis in exports.	False

Returns:

Name	Type	Description
SpatialConfigurator	SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def with_observation(
    self,
    groups: GroupsInput,
    *,
    collapse_age: bool = False,
) -> SpatialConfigurator:
    """Register observation groups for compressed history recording.

    The groups are compiled into a binary mask at build time and passed
    to the spatial simulation kernel on each ``run()`` call. Once set,
    history records store aggregated observation data instead of raw
    flattened state across all demes.

    Args:
        groups: Observation groups (dict of name -> spec, list of specs,
            or None for one-group-per-genotype).
        collapse_age: Whether to collapse the age axis in exports.

    Returns:
        SpatialConfigurator: Self for chaining.
    """
    self._observation_groups = groups
    self._observation_collapse_age = collapse_age
    return self

migration

migration(kernel: Optional[NDArray[float64]] = None, migration_rate: float = 0.0, strategy: Literal['auto', 'adjacency', 'kernel', 'hybrid'] = 'auto', adjacency: Optional[object] = None, kernel_bank: Optional[Sequence[NDArray[float64]]] = None, deme_kernel_ids: Optional[NDArray[int64]] = None, kernel_include_center: bool = False, adjust_migration_on_edge: bool = False) -> SpatialConfigurator

Configure spatial migration parameters.

Parameters:

Name	Type	Description	Default
kernel	Optional[NDArray[float64]]	Odd-shaped 2D migration kernel.	None
migration_rate	float	Fraction of each deme that migrates.	0.0
strategy	Literal['auto', 'adjacency', 'kernel', 'hybrid']	Migration strategy ("auto", "adjacency", "kernel", "hybrid").	'auto'
adjacency	Optional[object]	Explicit adjacency matrix.	None
kernel_bank	Optional[Sequence[NDArray[float64]]]	Optional heterogeneous kernel bank.	None
deme_kernel_ids	Optional[NDArray[int64]]	Per-deme kernel ids into kernel_bank.	None
kernel_include_center	bool	Whether kernel includes center cell.	False
adjust_migration_on_edge	bool	Whether to adjust migration rates on boundaries. When False (default), boundary demes migrate less due to fewer valid neighbors. When True, all demes have the same total migration rate regardless of position.	False

Returns:

Type	Description
SpatialConfigurator	Self for chaining.

Source code in src/natal/spatial/configurator.py

def migration(
    self,
    kernel: Optional[NDArray[np.float64]] = None,
    migration_rate: float = 0.0,
    strategy: Literal["auto", "adjacency", "kernel", "hybrid"] = "auto",
    adjacency: Optional[object] = None,
    kernel_bank: Optional[Sequence[NDArray[np.float64]]] = None,
    deme_kernel_ids: Optional[NDArray[np.int64]] = None,
    kernel_include_center: bool = False,
    adjust_migration_on_edge: bool = False,
) -> SpatialConfigurator:
    """Configure spatial migration parameters.

    Args:
        kernel: Odd-shaped 2D migration kernel.
        migration_rate: Fraction of each deme that migrates.
        strategy: Migration strategy (``"auto"``, ``"adjacency"``,
            ``"kernel"``, ``"hybrid"``).
        adjacency: Explicit adjacency matrix.
        kernel_bank: Optional heterogeneous kernel bank.
        deme_kernel_ids: Per-deme kernel ids into ``kernel_bank``.
        kernel_include_center: Whether kernel includes center cell.
        adjust_migration_on_edge: Whether to adjust migration rates on
            boundaries. When False (default), boundary demes migrate less
            due to fewer valid neighbors. When True, all demes have the
            same total migration rate regardless of position.

    Returns:
        Self for chaining.
    """
    if isinstance(kernel, BatchSetting):
        if kernel_bank is not None or self._kernel_bank is not None:
            raise ValueError(
                "Cannot use batch_setting for kernel when kernel_bank "
                "is also provided. Use one or the other."
            )
        if deme_kernel_ids is not None or self._deme_kernel_ids is not None:
            raise ValueError(
                "Cannot use batch_setting for kernel when deme_kernel_ids "
                "is also provided — indices would conflict."
            )
        self._migration_kernel_batch = kernel
    elif kernel is not None:
        self._migration_kernel = np.asarray(kernel, dtype=np.float64)
    self._migration_rate = float(migration_rate)
    self._migration_strategy = strategy
    if adjacency is not None:
        self._migration_adjacency = adjacency
    if kernel_bank is not None:
        self._kernel_bank = kernel_bank
    if deme_kernel_ids is not None:
        self._deme_kernel_ids = np.asarray(deme_kernel_ids, dtype=np.int64)
    self._kernel_include_center = bool(kernel_include_center)
    self._adjust_migration_on_edge = bool(adjust_migration_on_edge)
    self._param_values["migration.migration_rate"] = float(migration_rate)
    return self

get_params

get_params() -> dict[str, object]

Return all registered parameter values.

Merges spatial-specific params with values read from the template config.

Source code in src/natal/spatial/configurator.py

def get_params(self) -> dict[str, object]:
    """Return all registered parameter values.

    Merges spatial-specific params with values read from the template config.
    """
    from natal.utils.parameters import ALL_PARAMETERS

    params = dict(self._param_values)
    # Read scalar config values through ParamDescriptor registry
    for key, desc in ALL_PARAMETERS.items():
        if desc.config_field is None or desc.is_tensor:
            continue
        field: object = getattr(self._template.config, desc.config_field, None)
        if field is None:
            continue
        val: object
        if desc.config_path and isinstance(field, np.ndarray):
            val = cast(object, field[desc.config_path])
        elif isinstance(field, np.ndarray) and field.ndim == 0:
            val = cast(object, field[()])
        else:
            val = field  # pyright: ignore[reportUnknownVariableType] — getattr returns unknowable types
        params[key] = val
    return params

get_param

get_param(domain: str, name: str) -> object | None

Look up a single registered parameter value.

Source code in src/natal/spatial/configurator.py

def get_param(self, domain: str, name: str) -> object | None:
    """Look up a single registered parameter value."""
    return self.get_params().get(f"{domain}.{name}")

build

build() -> SpatialPopulation

Build and return the configured SpatialPopulation.

Two paths, chosen automatically based on whether any batch_setting() values were used during the chain:

• Homogeneous (no batch_setting): build ONE template deme, clone N-1 times via _clone_deme(). All demes share the same config object — maximum memory efficiency, zero redundant work.

• Heterogeneous (any batch_setting): expand batch values, group demes by parameter signature, build one template per group. Within a group, demes share a config. Across groups, heavy ndarrays (genotype maps, fitness tensors) are shared via _replace() when possible.

Returns:

Type	Description
SpatialPopulation	A SpatialPopulation with all demes initialised.

Source code in src/natal/spatial/configurator.py

def build(self) -> SpatialPopulation:
    """Build and return the configured ``SpatialPopulation``.

    Two paths, chosen automatically based on whether any ``batch_setting()``
    values were used during the chain:

    - **Homogeneous** (no ``batch_setting``): build ONE template deme,
      clone N-1 times via ``_clone_deme()``.  All demes share the same
      config object — maximum memory efficiency, zero redundant work.

    - **Heterogeneous** (any ``batch_setting``): expand batch values,
      group demes by parameter signature, build one template per group.
      Within a group, demes share a config.  Across groups, heavy
      ndarrays (genotype maps, fitness tensors) are shared via
      ``_replace()`` when possible.

    Returns:
        A ``SpatialPopulation`` with all demes initialised.
    """
    if not self._batch_settings:
        return self._build_homogeneous()
    return self._build_heterogeneous()

for_population staticmethod

for_population(spatial_pop: SpatialPopulation, *, deme: int | None = None) -> Configurator | _SpatialUpdate

Create a Configurator wired to one deme, or a multi-deme updater.

This is the runtime-update counterpart of SpatialConfigurator() (build-time). It returns an object whose chainable methods modify the existing spatial population's configs:

• for_population(pop, deme=3) → a single-deme Configurator (clone-on-write, same as pop.update_deme(3)).
• for_population(pop) → a multi-deme updater whose chainable methods apply the change to every deme.

Examples::

# Single deme
SpatialConfigurator.for_population(pop, deme=2).competition(
    carrying_capacity=800,
)

# All demes
SpatialConfigurator.for_population(pop).competition(
    carrying_capacity=800,
)

# All demes with per-deme variation
from natal.spatial.configurator import batch_setting
SpatialConfigurator.for_population(pop).competition(
    carrying_capacity=batch_setting([100, 200, 300, 400]),
)

Source code in src/natal/spatial/configurator.py

@staticmethod
def for_population(
    spatial_pop: SpatialPopulation,
    *,
    deme: int | None = None,
) -> Configurator | _SpatialUpdate:
    """Create a ``Configurator`` wired to one deme, or a multi-deme updater.

    This is the runtime-update counterpart of ``SpatialConfigurator()``
    (build-time).  It returns an object whose chainable methods modify
    the existing spatial population's configs:

    - ``for_population(pop, deme=3)`` → a single-deme ``Configurator``
      (clone-on-write, same as ``pop.update_deme(3)``).
    - ``for_population(pop)`` → a multi-deme updater whose chainable
      methods apply the change to every deme.

    Examples::

        # Single deme
        SpatialConfigurator.for_population(pop, deme=2).competition(
            carrying_capacity=800,
        )

        # All demes
        SpatialConfigurator.for_population(pop).competition(
            carrying_capacity=800,
        )

        # All demes with per-deme variation
        from natal.spatial.configurator import batch_setting
        SpatialConfigurator.for_population(pop).competition(
            carrying_capacity=batch_setting([100, 200, 300, 400]),
        )
    """
    if deme is not None:
        return spatial_pop.update_deme(deme)
    return _SpatialUpdate(spatial_pop)

SpatialPopulation

SpatialPopulation(demes: Sequence[DemePopulation], *, topology: Optional[GridTopology] = None, adjacency: Optional[object] = None, migration_kernel: Optional[NDArray[float64]] = None, migration_strategy: Literal['auto', 'adjacency', 'kernel', 'hybrid'] = 'auto', kernel_bank: Optional[Sequence[NDArray[float64]]] = None, deme_kernel_ids: Optional[NDArray[int64]] = None, kernel_include_center: bool = False, migration_rate: float | NDArray[float64] | Sequence[float] = 0.0, adjust_migration_on_edge: bool = False, name: str = 'SpatialPopulation')

Spatial container composed of per-deme population objects.

This class models spatial structure via composition: every deme is one already-initialized BasePopulation subclass instance.

Attributes:

Name	Type	Description
name	str	Human-readable name for the spatial container.
demes	Sequence[DemePopulation]	Immutable view of managed demes.
n_demes	int	Number of demes in the spatial system.
species	object	Shared species object used by all demes.
topology	GridTopology | None	Spatial topology used by the landscape.
adjacency	NDArray[float64]	Outbound migration matrix between demes when migration mode is "adjacency".
migration_strategy	Literal['auto', 'adjacency', 'kernel', 'hybrid']	Strategy selector for migration backend. "hybrid" is reserved for future mixed routing and currently follows "auto" runtime behavior.
migration_mode	Literal['adjacency', 'kernel']	Active migration backend.
migration_kernel	NDArray[float64] | None	Migration kernel used when migration_mode is "kernel".
kernel_bank	tuple[NDArray[float64], ...] | None	Optional bank of per-pattern kernels reserved for future per-deme kernel routing.
deme_kernel_ids	NDArray[int64] | None	Optional per-deme kernel id mapping into kernel_bank reserved for future mixed routing.
migration_rate	float	Fraction of each deme that participates in migration on each tick.
tick	int	Current shared simulation tick across all demes.

Initialize a spatial population container from existing demes.

Parameters:

Name	Type	Description	Default
demes	Sequence[DemePopulation]	Sequence of already-initialized deme populations.	required
topology	Optional[GridTopology]	Optional grid topology used to derive adjacency when adjacency is not provided.	None
adjacency	Optional[object]	Optional explicit migration matrix with shape (n_demes, n_demes). Supports dense ndarray, CSR tuple (indptr, indices, data), or sparse-like objects exposing toarray().	None
migration_kernel	Optional[NDArray[float64]]	Optional odd-shaped 2D kernel used for topology- aware migration. When provided, topology is required and migration runs in kernel mode.	None
migration_strategy	Literal['auto', 'adjacency', 'kernel', 'hybrid']	Backend selection policy. "auto" keeps existing behavior (kernel when migration_kernel is set, otherwise adjacency). "hybrid" is accepted as a forward- compatible alias of "auto" for now.	'auto'
kernel_bank	Optional[Sequence[NDArray[float64]]]	Optional kernel bank reserved for future per-deme heterogeneous-kernel routing.	None
deme_kernel_ids	Optional[NDArray[int64]]	Optional per-deme kernel id array reserved for future heterogeneous-kernel routing.	None
kernel_include_center	bool	Whether kernel migration includes the kernel center as an outbound target for the source deme.	False
migration_rate	float | NDArray[float64] | Sequence[float]	Fraction of each deme that migrates each tick. Scalar applies only to adult ages (>= new_adult_age from config); juvenile ages default to 0. Array indexed by age.	0.0
adjust_migration_on_edge	bool	Whether to adjust migration rates on boundaries. When False (default), boundary demes migrate less due to fewer valid neighbors. When True, all demes have the same total migration rate regardless of position.	False
name	str	Human-readable container name.	'SpatialPopulation'

Raises:

Type	Description
ValueError	If demes is empty, demes do not share the same species object, topology size does not match the number of demes, migration strategy is invalid, adjacency input is invalid, migration kernel is invalid, or deme ticks do not match.

Source code in src/natal/spatial/population.py

def __init__(
    self,
    demes: Sequence[DemePopulation],
    *,
    topology: Optional[GridTopology] = None,
    adjacency: Optional[object] = None,
    migration_kernel: Optional[NDArray[np.float64]] = None,
    migration_strategy: Literal["auto", "adjacency", "kernel", "hybrid"] = "auto",
    kernel_bank: Optional[Sequence[NDArray[np.float64]]] = None,
    deme_kernel_ids: Optional[NDArray[np.int64]] = None,
    kernel_include_center: bool = False,
    migration_rate: float | NDArray[np.float64] | Sequence[float] = 0.0,
    adjust_migration_on_edge: bool = False,
    name: str = "SpatialPopulation",
) -> None:
    """Initialize a spatial population container from existing demes.

    Args:
        demes: Sequence of already-initialized deme populations.
        topology: Optional grid topology used to derive adjacency when
            ``adjacency`` is not provided.
        adjacency: Optional explicit migration matrix with shape
            ``(n_demes, n_demes)``. Supports dense ``ndarray``, CSR tuple
            ``(indptr, indices, data)``, or sparse-like objects exposing
            ``toarray()``.
        migration_kernel: Optional odd-shaped 2D kernel used for topology-
            aware migration. When provided, ``topology`` is required and
            migration runs in kernel mode.
        migration_strategy: Backend selection policy. ``"auto"`` keeps
            existing behavior (kernel when ``migration_kernel`` is set,
            otherwise adjacency). ``"hybrid"`` is accepted as a forward-
            compatible alias of ``"auto"`` for now.
        kernel_bank: Optional kernel bank reserved for future per-deme
            heterogeneous-kernel routing.
        deme_kernel_ids: Optional per-deme kernel id array reserved for
            future heterogeneous-kernel routing.
        kernel_include_center: Whether kernel migration includes the kernel
            center as an outbound target for the source deme.
        migration_rate: Fraction of each deme that migrates each tick.
            Scalar applies only to adult ages (>= new_adult_age from
            config); juvenile ages default to 0.  Array indexed by age.
        adjust_migration_on_edge: Whether to adjust migration rates on
            boundaries. When False (default), boundary demes migrate less
            due to fewer valid neighbors. When True, all demes have the
            same total migration rate regardless of position.
        name: Human-readable container name.

    Raises:
        ValueError: If ``demes`` is empty, demes do not share the same
            species object, topology size does not match the number of
            demes, migration strategy is invalid, adjacency input is
            invalid, migration kernel is invalid, or deme ticks do not
            match.
    """
    if not demes:
        raise ValueError("demes must contain at least one BasePopulation instance")

    # Keep a stable list internally; public accessor returns an immutable
    # tuple view to prevent accidental external mutation.
    self._demes: List[DemePopulation] = list(demes)

    # Warn if any deme has per-deme record_observation set — spatial
    # population uses its own container-level observation and the per-deme
    # setting will be silently ignored during spatial run() / run_tick().
    for deme_idx, deme in enumerate(self._demes):
        if getattr(deme, "record_observation", None) is not None:
            warnings.warn(
                f"deme[{deme_idx}] has record_observation set, but SpatialPopulation "
                "uses its own observation mask from the container level. "
                "The per-deme setting will be ignored during spatial run(). "
                "Use spatial.record_observation or spatial.set_observations() instead.",
                stacklevel=2,
            )
            break

    # Spatial container expects all demes to share one Species object so
    # genotype indexing and config semantics are globally consistent.
    first_species = self._demes[0].species
    for idx, deme in enumerate(self._demes[1:], start=1):
        if deme.species is not first_species:
            raise ValueError(
                f"deme[{idx}] species does not match deme[0]; all demes must share the same Species object"
            )

    n_demes = len(self._demes)
    if topology is not None and topology.n_demes != n_demes:
        raise ValueError(
            f"topology.n_demes ({topology.n_demes}) must match number of demes ({n_demes})"
        )

    if migration_strategy not in {"auto", "adjacency", "kernel", "hybrid"}:
        raise ValueError(
            "migration_strategy must be one of: auto, adjacency, kernel, hybrid"
        )

    # Resolve strategy-level policy into one concrete backend mode.
    if migration_strategy == "adjacency":
        migration_mode: Literal["adjacency", "kernel"] = "adjacency"
    elif migration_strategy == "kernel":
        migration_mode = "kernel"
    else:
        # ``auto`` and ``hybrid`` currently share runtime behavior.
        # TODO(spatial-migration/hybrid-dispatch): Implement true hybrid
        # backend selection.
        # Scope:
        # - Add runtime branch policy for mixed routing (not alias to auto).
        # - Allow per-run/per-tick decision between adjacency and kernel.
        # Definition of done:
        # - `migration_strategy="hybrid"` yields behavior distinguishable
        #   from `auto` in at least one tested scenario.
        migration_mode = "kernel" if migration_kernel is not None else "adjacency"

    if migration_mode == "kernel":
        if topology is None:
            raise ValueError("topology is required when migration_kernel is provided")
        has_heterogeneous_kernels = kernel_bank is not None and deme_kernel_ids is not None
        if migration_kernel is None and not has_heterogeneous_kernels:
            raise ValueError(
                "migration_kernel is required in kernel mode unless kernel_bank and "
                "deme_kernel_ids are both provided"
            )
        if migration_kernel is not None:
            # Kernels are centered on one source cell; odd dimensions are
            # required so a unique center index exists.
            migration_kernel = np.asarray(migration_kernel, dtype=np.float64)
            if (
                migration_kernel.ndim != 2
                or migration_kernel.shape[0] % 2 == 0
                or migration_kernel.shape[1] % 2 == 0
            ):
                raise ValueError("migration_kernel must be a 2D array with odd dimensions")

    if adjacency is None:
        # Default adjacency:
        # - no topology: identity matrix (no migration unless diagonal used)
        # - with topology: topology-derived neighborhood matrix
        if topology is None:
            adjacency = np.eye(n_demes, dtype=np.float64)
        else:
            adjacency = build_adjacency_matrix(topology)

    adjacency_dense = _coerce_adjacency_dense(adjacency, n_demes=n_demes)

    normalized_kernel_bank: tuple[NDArray[np.float64], ...] | None = None
    if kernel_bank is not None:
        if len(kernel_bank) == 0:
            raise ValueError("kernel_bank must not be empty when provided")
        kernels: List[NDArray[np.float64]] = []
        for kernel_idx, kernel_value in enumerate(kernel_bank):
            kernel_arr = np.asarray(kernel_value, dtype=np.float64)
            if (
                kernel_arr.ndim != 2
                or kernel_arr.shape[0] % 2 == 0
                or kernel_arr.shape[1] % 2 == 0
            ):
                raise ValueError(
                    "kernel_bank entries must be 2D arrays with odd dimensions "
                    f"(invalid at index {kernel_idx})"
                )
            kernels.append(kernel_arr)
        normalized_kernel_bank = tuple(kernels)

    normalized_deme_kernel_ids: NDArray[np.int64] | None = None
    if deme_kernel_ids is not None:
        if normalized_kernel_bank is None:
            raise ValueError("deme_kernel_ids requires kernel_bank to be provided")
        normalized_deme_kernel_ids = np.asarray(deme_kernel_ids, dtype=np.int64)
        if normalized_deme_kernel_ids.shape != (n_demes,):
            raise ValueError(
                "deme_kernel_ids shape mismatch: expected "
                f"({n_demes},), got {normalized_deme_kernel_ids.shape}"
            )
        for deme_idx in range(n_demes):
            kernel_id = int(normalized_deme_kernel_ids[deme_idx])
            if kernel_id < 0 or kernel_id >= len(normalized_kernel_bank):
                raise ValueError(
                    f"deme_kernel_ids[{deme_idx}]={kernel_id} out of range for kernel_bank size "
                    f"{len(normalized_kernel_bank)}"
                )

    # Spatial hooks are local-to-deme by design, so container-level hooks
    # must always be rebuilt from all demes.
    self._hooks = self._compile_spatial_hooks_from_demes()

    # Heterogeneous kernels use kernel mode directly (no dense pre-build).
    # The kernel migration function selects per-deme kernels on-the-fly.
    if migration_mode == "adjacency" and normalized_kernel_bank is not None and normalized_deme_kernel_ids is not None:
        migration_mode = "kernel"

    self._name = name
    self._topology = topology
    self._adjacency = adjacency_dense
    self._migration_strategy: Literal["auto", "adjacency", "kernel", "hybrid"] = migration_strategy
    self._migration_mode: Literal["adjacency", "kernel"] = migration_mode
    self._migration_kernel = migration_kernel
    self._kernel_bank = normalized_kernel_bank
    self._deme_kernel_ids = normalized_deme_kernel_ids
    self._kernel_include_center = bool(kernel_include_center)
    self._migration_mode_code = 0 if migration_mode == "adjacency" else 1
    ind = self._demes[0].state.individual_count
    n_ages = int(ind.shape[1]) if ind.ndim == 3 else 1
    adult_start_age = self._adult_start_age(n_ages)
    self._migration_rate = _normalize_migration_rate(migration_rate, n_ages, adult_start_age)
    self._adjust_migration_on_edge = bool(adjust_migration_on_edge)
    # Spatial container and all demes share one logical tick counter.
    self._tick = int(self._demes[0].tick)

    # Pre-built parameter bundles for kernel dispatch.
    self._spatial_topo = SpatialTopology(
        rows=0 if topology is None else int(topology.rows),
        cols=0 if topology is None else int(topology.cols),
        wrap=False if topology is None else bool(topology.wrap),
    )
    self._migration_params = MigrationParams(
        kernel=self._migration_kernel_array(),
        include_center=bool(kernel_include_center),
        rate=self._migration_rate,
        adjust_on_edge=bool(adjust_migration_on_edge),
        adjacency=adjacency_dense,
        mode_code=0 if migration_mode == "adjacency" else 1,
    )

    # Observation-based history recording.
    self._observation: Optional[Observation] = None  # type: ignore[valid-type]
    self._observation_mask: Optional[NDArray[np.float64]] = None

    # Compact observation metadata (populated when record_observation is set).
    self._compact_meta: Optional[CompactMeta] = None
    self._deme_modes: Dict[int, Tuple[str, List[int]]] = {}

    # Evolution history as (tick, flattened_array) tuples.
    self._history: List[Tuple[int, NDArray[np.float64]]] = []

    # History config
    self.max_history: int = 5000  # Default rolling window size

    for idx, deme in enumerate(self._demes[1:], start=1):
        if int(deme.tick) != self._tick:
            raise ValueError(
                f"deme[{idx}] tick ({deme.tick}) does not match deme[0] tick ({self._tick})"
            )

name property

name: str

str: Human-readable name for the spatial container.

demes property

demes: Sequence[DemePopulation]

Sequence[DemePopulation]: Immutable view of all managed demes.

n_demes property

n_demes: int

int: Number of demes in the spatial system.

species property

species: Species

Species: Shared species object used by all demes.

adjacency property

adjacency: NDArray[float64]

NDArray[np.float64]: Outbound migration matrix between demes.

topology property

topology: GridTopology | None

GridTopology | None: Landscape topology used by the spatial model.

migration_mode property

migration_mode: Literal['adjacency', 'kernel']

Literal["adjacency", "kernel"]: Active migration backend.

migration_strategy property

migration_strategy: Literal['auto', 'adjacency', 'kernel', 'hybrid']

Literal["auto", "adjacency", "kernel", "hybrid"]: Strategy policy.

migration_kernel property

migration_kernel: NDArray[float64] | None

NDArray[np.float64] | None: Kernel used by topology-aware migration.

kernel_bank property

kernel_bank: tuple[NDArray[float64], ...] | None

tuple[NDArray[np.float64], ...] | None: Reserved heterogeneous kernels.

deme_kernel_ids property

deme_kernel_ids: NDArray[int64] | None

NDArray[np.int64] | None: Reserved per-deme kernel ids.

migration_rate property writable

migration_rate: NDArray[float64]

NDArray[np.float64]: Age-specific migration rate per tick.

adjust_migration_on_edge property

adjust_migration_on_edge: bool

bool: Whether to adjust migration rates on boundaries.

tick property

tick: int

int: Shared simulation tick across all demes.

history property

history: List[Tuple[int, NDArray[float64]]]

A list of recorded historical states as (tick, flattened_array) tuples.

Each flattened array contains the full stacked spatial state: [tick, ind_all.ravel(), sperm_all.ravel()].

record_observation property writable

record_observation: Optional[Observation]

The compiled Observation used for observation-mode history.

hooks property

hooks: LifecycleWrappers

LifecycleWrappers: Compiled hooks and lifecycle loop functions.

builder classmethod

builder(species: Species, n_demes: int, topology: Optional[GridTopology] = None, *, pop_type: Literal['age_structured', 'discrete_generation'] = 'age_structured') -> SpatialConfigurator

Create a SpatialConfigurator for fluent spatial population construction.

Parameters:

Name	Type	Description	Default
species	Species	Genetic architecture shared by all demes.	required
n_demes	int	Number of demes in the spatial layout.	required
topology	Optional[GridTopology]	Optional grid topology for migration.	None
pop_type	Literal['age_structured', 'discrete_generation']	"age_structured" (default) or "discrete_generation".	'age_structured'

Returns:

Type	Description
SpatialConfigurator	A SpatialConfigurator instance ready for chaining.

Examples:

>>> pop = SpatialPopulation.builder(species, n_demes=100) \
...     .setup(name="demo") \
...     .initial_state(...) \
...     .competition(carrying_capacity=batch_setting([...])) \
...     .build()

Source code in src/natal/spatial/population.py

@classmethod
def builder(
    cls,
    species: Species,
    n_demes: int,
    topology: Optional[GridTopology] = None,
    *,
    pop_type: Literal["age_structured", "discrete_generation"] = "age_structured",
) -> SpatialConfigurator:
    """Create a ``SpatialConfigurator`` for fluent spatial population construction.

    Args:
        species: Genetic architecture shared by all demes.
        n_demes: Number of demes in the spatial layout.
        topology: Optional grid topology for migration.
        pop_type: ``"age_structured"`` (default) or ``"discrete_generation"``.

    Returns:
        A ``SpatialConfigurator`` instance ready for chaining.

    Examples:
        >>> pop = SpatialPopulation.builder(species, n_demes=100) \\
        ...     .setup(name="demo") \\
        ...     .initial_state(...) \\
        ...     .competition(carrying_capacity=batch_setting([...])) \\
        ...     .build()
    """
    from natal.spatial.configurator import SpatialConfigurator
    return SpatialConfigurator(
        species=species,
        n_demes=n_demes,
        topology=topology,
        pop_type=pop_type,
    )

deme

deme(idx: int) -> DemePopulation

Return one deme by positional index.

Parameters:

Name	Type	Description	Default
idx	int	Zero-based deme index.	required

Returns:

Type	Description
DemePopulation	The deme population at idx.

Source code in src/natal/spatial/population.py

def deme(self, idx: int) -> DemePopulation:
    """Return one deme by positional index.

    Args:
        idx: Zero-based deme index.

    Returns:
        The deme population at ``idx``.
    """
    return self._demes[idx]

update

update(deme: int | None = None) -> _SpatialUpdate

Return an updater for modifying this population's config.

Supports both scalar and batch_setting values in chain calls, matching the SpatialConfigurator API::

# Modify all demes simultaneously
pop.update().competition(carrying_capacity=5000)

# Modify a specific deme (auto-detaches shared config)
pop.update(deme=3).competition(carrying_capacity=8000)

# Batch per-deme modification (same API as build time)
from natal.spatial.configurator import batch_setting
pop.update().competition(
    carrying_capacity=batch_setting([100, 200, 300, 400])
)

Source code in src/natal/spatial/population.py

def update(self, deme: int | None = None) -> _SpatialUpdate:
    """Return an updater for modifying this population's config.

    Supports both scalar and ``batch_setting`` values in chain calls,
    matching the ``SpatialConfigurator`` API::

        # Modify all demes simultaneously
        pop.update().competition(carrying_capacity=5000)

        # Modify a specific deme (auto-detaches shared config)
        pop.update(deme=3).competition(carrying_capacity=8000)

        # Batch per-deme modification (same API as build time)
        from natal.spatial.configurator import batch_setting
        pop.update().competition(
            carrying_capacity=batch_setting([100, 200, 300, 400])
        )
    """
    return _SpatialUpdate(self, deme=deme)

update_deme

update_deme(demi: int) -> Configurator

Get a Configurator for a specific deme with clone-on-write.

Source code in src/natal/spatial/population.py

def update_deme(self, demi: int) -> Configurator:
    """Get a Configurator for a specific deme with clone-on-write."""
    from natal.configurator import Configurator

    target = self._demes[demi]
    config = target.config

    k_array = config.carrying_capacity
    shared_count = sum(
        1 for d in self._demes if d.config.carrying_capacity is k_array
    )
    if shared_count > 1:
        private = config._replace(
            carrying_capacity=config.carrying_capacity.copy(),
            eggs_per_female=config.eggs_per_female.copy(),
            sex_ratio=config.sex_ratio.copy(),
            sperm_displacement_rate=config.sperm_displacement_rate.copy(),
            low_density_growth_rate=config.low_density_growth_rate.copy(),
            juvenile_growth_mode=config.juvenile_growth_mode.copy(),
            expected_competition_strength=config.expected_competition_strength.copy(),
            expected_survival_rate=config.expected_survival_rate.copy(),
            generation_time=config.generation_time.copy(),
        )
        object.__setattr__(target, '_config', private)
        config = private

    return Configurator.for_population(target)

get_history

get_history() -> NDArray[np.float64]

Return the recorded history as a stacked 2D array.

Each row is (tick, ind_all.ravel(), sperm_all.ravel()). Returns a zero-row array if no history has been recorded.

Source code in src/natal/spatial/population.py

def get_history(self) -> NDArray[np.float64]:
    """Return the recorded history as a stacked 2D array.

    Each row is ``(tick, ind_all.ravel(), sperm_all.ravel())``.
    Returns a zero-row array if no history has been recorded.
    """
    if len(self._history) == 0:
        return np.zeros((0, 0), dtype=np.float64)
    return np.array([rec[1] for rec in self._history], dtype=np.float64)

clear_history

clear_history() -> None

Clear all recorded history.

Source code in src/natal/spatial/population.py

def clear_history(self) -> None:
    """Clear all recorded history."""
    self._history.clear()

set_observations

set_observations(groups: Any, *, collapse_age: bool = False) -> None

Register observation groups and immediately compile the binary mask.

Similar to BasePopulation.set_observations but also builds the per-group deme selector from optional "deme" keys in each spec.

Parameters:

Name	Type	Description	Default
groups	Any	Observation groups (dict of name -> spec, list of specs, or None for one-group-per-genotype).	required
collapse_age	bool	Whether to collapse the age axis during projection.	False

Source code in src/natal/spatial/population.py

def set_observations(
    self,
    groups: Any,
    *,
    collapse_age: bool = False,
) -> None:
    """Register observation groups and immediately compile the binary mask.

    Similar to ``BasePopulation.set_observations`` but also builds the
    per-group deme selector from optional ``"deme"`` keys in each spec.

    Args:
        groups: Observation groups (dict of name -> spec, list of specs,
            or None for one-group-per-genotype).
        collapse_age: Whether to collapse the age axis during projection.
    """
    from natal.output.observation import ObservationFilter

    ref_deme = self._demes[0]
    obs_filter = ObservationFilter(ref_deme.index_registry)
    self._observation = obs_filter.build_filter(
        diploid_genotypes=ref_deme.species,
        groups=groups,
        collapse_age=bool(collapse_age),
    )
    state = ref_deme.state
    ind = state.individual_count
    self._observation_mask = self._observation.build_mask(
        n_sexes=ind.shape[0],
        n_ages=ind.shape[1] if ind.ndim == 3 else 1,
        n_ztypes=ind.shape[-1],
    )
    self._rebuild_compact_meta()

set_hook

set_hook(event_name: str, func: Callable[..., None], hook_id: Optional[int] = None, hook_name: Optional[str] = None, compile: bool = True, deme_selector: Optional[DemeSelector] = None) -> None

Register an event hook for selected demes.

If the function carries @hook(deme=...) metadata and no explicit deme_selector is given, the metadata value is used automatically — you don't need to repeat the selector.

Parameters:

Name	Type	Description	Default
event_name	str	Event name (must exist in ALLOWED_EVENTS).	required
func	Callable[..., None]	Callback function.	required
hook_id	Optional[int]	Numeric execution priority (optional, auto-assigned if omitted).	None
hook_name	Optional[str]	Optional human-readable name for debugging.	None
compile	bool	Whether to try compiling @hook-decorated functions.	True
deme_selector	Optional[DemeSelector]	Optional deme selector. If omitted, reads from @hook metadata automatically.	None

Note

This API is a spatial convenience entrypoint. deme_selector is interpreted here to choose target demes, then forwarded hooks are registered on selected demes with panmictic selector semantics. Compiler-level selector fields are transport-only metadata.

Source code in src/natal/spatial/population.py

def set_hook(
    self,
    event_name: str,
    func: Callable[..., None],
    hook_id: Optional[int] = None,
    hook_name: Optional[str] = None,
    compile: bool = True,
    deme_selector: Optional[DemeSelector] = None,
) -> None:
    """Register an event hook for selected demes.

    If the function carries ``@hook(deme=...)`` metadata and no explicit
    ``deme_selector`` is given, the metadata value is used automatically
    — you don't need to repeat the selector.

    Args:
        event_name: Event name (must exist in ALLOWED_EVENTS).
        func: Callback function.
        hook_id: Numeric execution priority (optional, auto-assigned if omitted).
        hook_name: Optional human-readable name for debugging.
        compile: Whether to try compiling @hook-decorated functions.
        deme_selector: Optional deme selector.  If omitted, reads from
            ``@hook`` metadata automatically.

    Note:
        This API is a spatial convenience entrypoint. ``deme_selector`` is
        interpreted here to choose target demes, then forwarded hooks are
        registered on selected demes with panmictic selector semantics.
        Compiler-level selector fields are transport-only metadata.
    """
    # Auto-read deme from @hook metadata when no explicit selector given.
    if deme_selector is None:
        meta = getattr(func, 'meta', None)
        if meta is not None:
            demean_meta = meta.get('deme_selector')
            if demean_meta is not None and demean_meta != "*":
                deme_selector = demean_meta

    # Spatial-level selector handling: select target demes here and avoid
    # passing non-wildcard selectors into BasePopulation.
    for deme_id, deme in enumerate(self._demes):
        if deme_selector is not None and not self._selector_matches_deme(deme_selector, deme_id):
            continue
        deme.set_hook(event_name, func, hook_id, hook_name, compile, None)

    # Rebuild aggregate hooks once after all per-deme mutations.
    self._hooks = self._compile_spatial_hooks_from_demes()

remove_hook

remove_hook(event_name: str, hook_id: int) -> bool

Remove a specific hook from all demes.

Parameters:

Name	Type	Description	Default
event_name	str	Event name.	required
hook_id	int	Hook ID.	required

Returns:

Type	Description
bool	True if removed successfully from all demes, otherwise False.

Note

Hook removal follows the same consistency rule as registration: mutate each deme first, then rebuild the aggregate compiled hooks.

Source code in src/natal/spatial/population.py

def remove_hook(self, event_name: str, hook_id: int) -> bool:
    """Remove a specific hook from all demes.

    Args:
        event_name: Event name.
        hook_id: Hook ID.

    Returns:
        True if removed successfully from all demes, otherwise False.

    Note:
        Hook removal follows the same consistency rule as registration:
        mutate each deme first, then rebuild the aggregate compiled hooks.
    """
    success = True
    for deme in self._demes:
        if not deme.remove_hook(event_name, hook_id):
            success = False

    # Keep aggregate compiled hooks synchronized with deme-local state.
    self._hooks = self._compile_spatial_hooks_from_demes()

    return success

trigger_event

trigger_event(event_name: str, deme_id: int = 0) -> int

Trigger an event and execute all registered hooks for a specific deme.

Parameters:

Name	Type	Description	Default
event_name	str	Event name to trigger.	required
deme_id	int	Deme ID (default: 0).	0

Returns:

Name	Type	Description
int	int	RESULT_CONTINUE (0) to continue, RESULT_STOP (1) to stop.

Source code in src/natal/spatial/population.py

def trigger_event(self, event_name: str, deme_id: int = 0) -> int:
    """Trigger an event and execute all registered hooks for a specific deme.

    Args:
        event_name: Event name to trigger.
        deme_id: Deme ID (default: 0).

    Returns:
        int: RESULT_CONTINUE (0) to continue, RESULT_STOP (1) to stop.
    """
    if 0 <= deme_id < self.n_demes:
        return self._demes[deme_id].trigger_event(event_name, deme_id)
    return 0  # RESULT_CONTINUE

get_total_count

get_total_count() -> int

Return the total count across all demes.

Source code in src/natal/spatial/population.py

def get_total_count(self) -> int:
    """Return the total count across all demes."""
    return int(sum(deme.get_total_count() for deme in self._demes))

get_female_count

get_female_count() -> int

Return the total female count across all demes.

Source code in src/natal/spatial/population.py

def get_female_count(self) -> int:
    """Return the total female count across all demes."""
    return int(sum(deme.get_female_count() for deme in self._demes))

get_male_count

get_male_count() -> int

Return the total male count across all demes.

Source code in src/natal/spatial/population.py

def get_male_count(self) -> int:
    """Return the total male count across all demes."""
    return int(sum(deme.get_male_count() for deme in self._demes))

reset

reset() -> None

Reset all demes and synchronize the container tick.

This resets each underlying deme using its own reset logic and then updates the spatial container tick to match the demes.

Source code in src/natal/spatial/population.py

def reset(self) -> None:
    """Reset all demes and synchronize the container tick.

    This resets each underlying deme using its own reset logic and then
    updates the spatial container tick to match the demes.
    """
    for deme in self._demes:
        deme.reset()
    self._tick = int(self._demes[0].tick)

aggregate_individual_count

aggregate_individual_count() -> NDArray[np.float64]

Return the total individual-count tensor summed over all demes.

Source code in src/natal/spatial/population.py

def aggregate_individual_count(self) -> NDArray[np.float64]:
    """Return the total individual-count tensor summed over all demes."""
    return np.sum(
        np.stack([deme.state.individual_count for deme in self._demes], axis=0),
        axis=0,
    )

aggregate_state

aggregate_state() -> PopulationState

Build one aggregate state for global summaries across all demes.

Source code in src/natal/spatial/population.py

def aggregate_state(self) -> PopulationState:
    """Build one aggregate state for global summaries across all demes."""
    ind_all, sperm_all = self._stack_deme_state_arrays()
    return PopulationState(
        n_tick=int(self._tick),
        individual_count=np.sum(ind_all, axis=0),
        sperm_storage=np.sum(sperm_all, axis=0),
    )

compute_allele_frequencies

compute_allele_frequencies() -> dict[str, float]

Compute allele frequencies from the aggregate multi-deme state.

Source code in src/natal/spatial/population.py

def compute_allele_frequencies(self) -> dict[str, float]:
    """Compute allele frequencies from the aggregate multi-deme state."""
    allele_counts: dict[str, float] = {}
    locus_totals: dict[str, float] = {}
    genotype_counts = self.aggregate_individual_count().sum(axis=(0, 1))
    registry = self._demes[0].registry

    for chromosome in self.species.chromosomes:
        for locus in chromosome.loci:
            locus_totals[locus.name] = 0.0
            for gene in locus.alleles:
                allele_counts[gene.name] = 0.0

    for z_idx, (genotype, _slab) in enumerate(registry.index_to_ztype):
        count = genotype_counts[z_idx]
        if count <= 0:
            continue
        for chromosome in self.species.chromosomes:
            for locus in chromosome.loci:
                mat, pat = genotype.get_alleles_at_locus(locus)
                for allele in (mat, pat):
                    if allele is not None:
                        allele_counts[allele.name] += float(count)
                        locus_totals[locus.name] += float(count)

    frequencies: dict[str, float] = {}
    for allele_name, count in allele_counts.items():
        gene = self.species.gene_index.get(allele_name)
        if gene is None:
            frequencies[allele_name] = 0.0
            continue
        total = locus_totals[gene.locus.name]
        frequencies[allele_name] = 0.0 if total <= 0.0 else count / total
    return frequencies

migration_row

migration_row(source_idx: int) -> NDArray[np.float64]

Return normalized outbound migration weights for one source deme.

Parameters:

Name	Type	Description	Default
source_idx	int	Source deme index.	required

Returns:

Type	Description
NDArray[float64]	A dense float64 vector of length n_demes with outbound weights
NDArray[float64]	from source_idx.

Source code in src/natal/spatial/population.py

def migration_row(self, source_idx: int) -> NDArray[np.float64]:
    """Return normalized outbound migration weights for one source deme.

    Args:
        source_idx: Source deme index.

    Returns:
        A dense float64 vector of length ``n_demes`` with outbound weights
        from ``source_idx``.
    """
    if self._migration_mode == "adjacency":
        weights = self._adjacency[source_idx].astype(np.float64, copy=True)
        total = float(weights.sum())
        if total > 0.0:
            weights /= total
        return weights

    assert self._topology is not None, "topology is required for kernel migration"

    # Select kernel (single or from bank).
    if self._deme_kernel_ids is not None and self._kernel_bank is not None:
        kernel = self._kernel_bank[int(self._deme_kernel_ids[source_idx])]
    else:
        assert self._migration_kernel is not None, "migration_kernel required"
        kernel = self._migration_kernel

    weights = np.zeros(self.n_demes, dtype=np.float64)
    src_coord = self._topology.from_index(source_idx)
    kr = kernel.shape[0] // 2
    kc = kernel.shape[1] // 2

    for row in range(kernel.shape[0]):
        for col in range(kernel.shape[1]):
            if not self._kernel_include_center and row == kr and col == kc:
                continue
            weight = float(kernel[row, col])
            if weight <= 0.0:
                continue
            mapped = self._topology.normalize_coord(
                src_coord[0] + row - kr,
                src_coord[1] + col - kc,
            )
            if mapped is None:
                continue
            weights[self._topology.to_index(mapped)] += weight

    total = float(weights.sum())
    if total > 0.0:
        weights /= total
    return weights

get_compiled_hooks

get_compiled_hooks(event: Optional[str] = None) -> list[Any]

Get compiled hook descriptors, optionally filtered by event.

Parameters:

Name	Type	Description	Default
event	Optional[str]	Optional event name to filter by.	None

Returns:

Type	Description
list[Any]	List of CompiledHookDescriptor sorted by priority.

Source code in src/natal/spatial/population.py

def get_compiled_hooks(self, event: Optional[str] = None) -> list[Any]:
    """Get compiled hook descriptors, optionally filtered by event.

    Args:
        event: Optional event name to filter by.

    Returns:
        List of ``CompiledHookDescriptor`` sorted by priority.
    """
    hooks = self._collect_effective_compiled_hooks()
    if event is not None:
        hooks = [h for h in hooks if h.event == event]
    return sorted(hooks, key=lambda h: h.priority)

run_tick

run_tick() -> SpatialPopulation

Run one spatial tick via the spatial kernel.

Returns:

Type	Description
SpatialPopulation	This spatial population instance after in-place state update.

Raises:

Type	Description
RuntimeError	If any deme has already finished.

Source code in src/natal/spatial/population.py

def run_tick(self) -> SpatialPopulation:
    """Run one spatial tick via the spatial kernel.

    Returns:
        This spatial population instance after in-place state update.

    Raises:
        RuntimeError: If any deme has already finished.
    """
    self._ensure_demes_runnable(context="run spatial tick")

    if self._should_use_python_dispatch():
        # Hook-aware fallback: preserve per-deme local hook semantics.
        was_stopped = self._run_python_dispatch_tick()
    else:
        # Global Numba path: run spatial kernel for one full spatial tick.
        was_stopped = self._run_codegen_wrapper_tick()
    if was_stopped:
        self._mark_all_demes_stopped()
    return self

run

run(n_steps: int, record_every: int = 1, finish: bool = False, clear_history_on_start: bool = False) -> SpatialPopulation

Run multiple spatial ticks via the spatial kernel.

Parameters:

Name	Type	Description	Default
n_steps	int	Number of ticks to execute.	required
record_every	int	History recording interval forwarded to the compiled spatial kernel.	1
finish	bool	Whether to mark all demes finished when the run completes without an early stop event.	False
clear_history_on_start	bool	Whether to clear existing history before appending new snapshots.	False

Returns:

Type	Description
SpatialPopulation	This spatial population instance after in-place state update.

Raises:

Type	Description
ValueError	If n_steps is negative.
RuntimeError	If any deme has already finished.

Source code in src/natal/spatial/population.py

def run(
    self,
    n_steps: int,
    record_every: int = 1,
    finish: bool = False,
    clear_history_on_start: bool = False,
) -> SpatialPopulation:
    """Run multiple spatial ticks via the spatial kernel.

    Args:
        n_steps: Number of ticks to execute.
        record_every: History recording interval forwarded to the compiled
            spatial kernel.
        finish: Whether to mark all demes finished when the run completes
            without an early stop event.
        clear_history_on_start: Whether to clear existing history before
            appending new snapshots.

    Returns:
        This spatial population instance after in-place state update.

    Raises:
        ValueError: If ``n_steps`` is negative.
        RuntimeError: If any deme has already finished.
    """
    if n_steps < 0:
        raise ValueError("n_steps must be >= 0")

    self._ensure_demes_runnable(context="run spatial simulation")

    if clear_history_on_start:
        self.clear_history()

    if self._should_use_python_dispatch():
        # Hook-aware fallback: keep local hook timeline semantics.
        was_stopped = False
        if record_every > 0 and (self._tick % record_every == 0):
            self._record_snapshot()
        for _ in range(n_steps):
            if self._run_python_dispatch_tick():
                was_stopped = True
                break
            if record_every > 0 and (self._tick % record_every == 0):
                self._record_snapshot()
    else:
        # Global Numba path: run batched spatial kernel.
        was_stopped = self._run_codegen_wrapper_steps(
            n_steps,
            record_every=int(record_every),
            clear_history_on_start=clear_history_on_start,
        )
    if bool(was_stopped):
        self._mark_all_demes_stopped()
    elif finish:
        for deme in self._demes:
            deme.finish_simulation()

    return self

GridTopology dataclass

GridTopology(rows: int, cols: int, wrap: bool = False, COS_OPPOSITE_ANGLE: float = 0.0)

Base topology over a 2D grid of demes.

Attributes:

Name	Type	Description
rows	int	Number of grid rows.
cols	int	Number of grid columns.
wrap	bool	Whether coordinates use periodic boundary conditions. When True, coordinates that move beyond one edge are wrapped to the opposite edge by modular arithmetic. When False, out-of-bounds coordinates are discarded.

Examples:

With wrap=False, out-of-bounds coordinates are rejected::

grid = SquareGrid(rows=3, cols=3, wrap=False)
grid.normalize_coord(-1, 1) is None

With wrap=True, opposite edges are connected::

grid = SquareGrid(rows=3, cols=3, wrap=True)
grid.normalize_coord(-1, 1) == (2, 1)
grid.normalize_coord(1, -1) == (1, 2)

n_demes property

n_demes: int

int: Total number of demes in the grid.

to_index

to_index(coord: Coord) -> int

Convert one grid coordinate to a flattened deme index.

Parameters:

Name	Type	Description	Default
coord	Coord	Grid coordinate as (row, col).	required

Returns:

Type	Description
int	Flattened deme index in row-major order.

Source code in src/natal/spatial/topology.py

def to_index(self, coord: Coord) -> int:
    """Convert one grid coordinate to a flattened deme index.

    Args:
        coord: Grid coordinate as ``(row, col)``.

    Returns:
        Flattened deme index in row-major order.
    """
    row, col = coord
    return row * self.cols + col

from_index

from_index(index: int) -> Coord

Convert one flattened deme index to a grid coordinate.

Parameters:

Name	Type	Description	Default
index	int	Flattened deme index in row-major order.	required

Returns:

Type	Description
Coord	Grid coordinate as (row, col).

Source code in src/natal/spatial/topology.py

def from_index(self, index: int) -> Coord:
    """Convert one flattened deme index to a grid coordinate.

    Args:
        index: Flattened deme index in row-major order.

    Returns:
        Grid coordinate as ``(row, col)``.
    """
    row = index // self.cols
    col = index % self.cols
    return (row, col)

normalize_coord

normalize_coord(row: int, col: int) -> Coord | None

Normalize one coordinate according to the topology boundary rule.

Parameters:

Name	Type	Description	Default
row	int	Candidate row index.	required
col	int	Candidate column index.	required

Returns:

Type	Description
Coord | None	A valid grid coordinate if the location is in bounds or can be
Coord | None	wrapped. Returns None when the location lies outside a
Coord | None	non-wrapping topology. With wrap=True, the result is computed as
Coord | None	(row % rows, col % cols), so opposite edges are connected.

Source code in src/natal/spatial/topology.py

def normalize_coord(self, row: int, col: int) -> Coord | None:
    """Normalize one coordinate according to the topology boundary rule.

    Args:
        row: Candidate row index.
        col: Candidate column index.

    Returns:
        A valid grid coordinate if the location is in bounds or can be
        wrapped. Returns ``None`` when the location lies outside a
        non-wrapping topology. With ``wrap=True``, the result is computed as
        ``(row % rows, col % cols)``, so opposite edges are connected.
    """
    if self.wrap:
        return (row % self.rows, col % self.cols)
    if row < 0 or row >= self.rows or col < 0 or col >= self.cols:
        return None
    return (row, col)

neighbor_coords

neighbor_coords(coord: Coord) -> List[Coord]

Return neighboring grid coordinates for one source coordinate.

Parameters:

Name	Type	Description	Default
coord	Coord	Source grid coordinate.	required

Returns:

Type	Description
List[Coord]	Neighbor coordinates in the topology-specific neighborhood order.

Raises:

Type	Description
NotImplementedError	If a subclass does not implement neighbor generation.

Source code in src/natal/spatial/topology.py

def neighbor_coords(self, coord: Coord) -> List[Coord]:
    """Return neighboring grid coordinates for one source coordinate.

    Args:
        coord: Source grid coordinate.

    Returns:
        Neighbor coordinates in the topology-specific neighborhood order.

    Raises:
        NotImplementedError: If a subclass does not implement neighbor
            generation.
    """
    raise NotImplementedError

to_xy

to_xy(coord: Coord) -> Tuple[float, float]

Map one grid coordinate to a geometric embedding coordinate.

Parameters:

Name	Type	Description	Default
coord	Coord	Source grid coordinate.	required

Returns:

Type	Description
Tuple[float, float]	Cartesian coordinate used for geometry-aware utilities.

Source code in src/natal/spatial/topology.py

def to_xy(self, coord: Coord) -> Tuple[float, float]:
    """Map one grid coordinate to a geometric embedding coordinate.

    Args:
        coord: Source grid coordinate.

    Returns:
        Cartesian coordinate used for geometry-aware utilities.
    """
    row, col = coord
    return (float(col), float(row))

neighbor_vectors

neighbor_vectors(coord: Coord) -> List[Tuple[float, float]]

Return geometric displacement vectors from one coord to neighbors.

Parameters:

Name	Type	Description	Default
coord	Coord	Source grid coordinate.	required

Returns:

Type	Description
List[Tuple[float, float]]	Displacement vectors from coord to each neighboring coordinate
List[Tuple[float, float]]	in embedding space.

Source code in src/natal/spatial/topology.py

def neighbor_vectors(self, coord: Coord) -> List[Tuple[float, float]]:
    """Return geometric displacement vectors from one coord to neighbors.

    Args:
        coord: Source grid coordinate.

    Returns:
        Displacement vectors from ``coord`` to each neighboring coordinate
        in embedding space.
    """
    x0, y0 = self.to_xy(coord)
    vectors: List[Tuple[float, float]] = []
    for n_coord in self.neighbor_coords(coord):
        x1, y1 = self.to_xy(n_coord)
        vectors.append((x1 - x0, y1 - y0))
    return vectors

offset_dist_sq

offset_dist_sq(dr: NDArray[float64], dc: NDArray[float64]) -> NDArray[np.float64]

Squared distance between grid coords offset by (dr, dc).

Uses the law of cosines with _COS_OPPOSITE_ANGLE::

dist² = dr² + dc² - 2·dr·dc·cos(θ)

For square grids cos(90°) = 0 → Cartesian distance. Subclasses override the class attribute _COS_OPPOSITE_ANGLE to change the metric (e.g. hex grids set cos(120°) = -0.5).

Source code in src/natal/spatial/topology.py

def offset_dist_sq(self, dr: NDArray[np.float64], dc: NDArray[np.float64]) -> NDArray[np.float64]:
    """Squared distance between grid coords offset by ``(dr, dc)``.

    Uses the law of cosines with ``_COS_OPPOSITE_ANGLE``::

        dist² = dr² + dc² - 2·dr·dc·cos(θ)

    For square grids cos(90°) = 0 → Cartesian distance. Subclasses
    override the class attribute ``_COS_OPPOSITE_ANGLE`` to change the
    metric (e.g. hex grids set cos(120°) = -0.5).
    """
    return dr**2 + dc**2 - 2.0 * self.COS_OPPOSITE_ANGLE * dr * dc

neighbors

neighbors(index: int) -> List[int]

Return neighboring deme indices for one flattened deme index.

Parameters:

Name	Type	Description	Default
index	int	Source deme index in row-major order.	required

Returns:

Type	Description
List[int]	Neighbor deme indices in the same order as neighbor_coords.

Source code in src/natal/spatial/topology.py

def neighbors(self, index: int) -> List[int]:
    """Return neighboring deme indices for one flattened deme index.

    Args:
        index: Source deme index in row-major order.

    Returns:
        Neighbor deme indices in the same order as ``neighbor_coords``.
    """
    return [self.to_index(coord) for coord in self.neighbor_coords(self.from_index(index))]

HeterogeneousKernelParams

Bases: NamedTuple

Per-kernel offset tables for heterogeneous kernel routing.

Only populated when kernel_bank and deme_kernel_ids are both set on the spatial population.

Attributes:

Name	Type	Description
deme_kernel_ids	NDArray[int64]	(n_demes,) int64 — kernel index per source deme.
d_row	NDArray[int64]	(n_kernels, max_nnz) int64 — row offsets per kernel.
d_col	NDArray[int64]	(n_kernels, max_nnz) int64 — column offsets per kernel.
weights	NDArray[float64]	(n_kernels, max_nnz) float64 — per-offset weights.
nnzs	NDArray[int64]	(n_kernels,) int64 — number of valid entries per kernel.
total_sums	NDArray[float64]	(n_kernels,) float64 — sum of all weights per kernel.
max_nnz	int	Maximum valid entries across all kernels.

HexGrid dataclass

HexGrid(rows: int, cols: int, wrap: bool = False, COS_OPPOSITE_ANGLE: float = -0.5)

Bases: GridTopology

Hex grid using parallelogram coordinates with pointy-top geometric embedding.

This implementation uses a parallelogram grid where each cell has six neighbors in a hexagonal pattern. The grid is defined using two basis vectors that form a 60-degree angle, creating a natural hexagonal topology.

Examples:

On a non-wrapping grid, a corner cell loses out-of-bounds neighbors::

grid = HexGrid(rows=3, cols=4, wrap=False)
len(grid.neighbor_coords((0, 0))) < 6

On a wrapping grid, the same corner cell keeps six neighbors because out-of-bounds coordinates are folded back into the opposite edge::

wrapped = HexGrid(rows=3, cols=4, wrap=True)
len(wrapped.neighbor_coords((0, 0))) == 6

to_xy

to_xy(coord: Coord) -> Tuple[float, float]

Map parallelogram coordinate to pointy-top Cartesian coordinates.

Positive x points right, positive y points downward. With side length = 1, adjacent hex centers are distance 1 apart.

Source code in src/natal/spatial/topology.py

def to_xy(self, coord: Coord) -> Tuple[float, float]:
    """Map parallelogram coordinate to pointy-top Cartesian coordinates.

    Positive x points right, positive y points downward.
    With side length = 1, adjacent hex centers are distance 1 apart.
    """
    i, j = coord
    # Parallelogram basis vectors for pointy-top hexes:
    # moving +i shifts x by 1; moving +j shifts x by 1/2 and y by sqrt(3)/2.
    x = i + 0.5 * j
    y = self._SQRT3_OVER_2 * j
    return (float(x), float(y))

neighbor_direction_vectors

neighbor_direction_vectors() -> Tuple[Tuple[float, float], ...]

Canonical geometric neighbor vectors for pointy-top hexes.

Includes the right-down vector (1/2, sqrt(3)/2).

Source code in src/natal/spatial/topology.py

def neighbor_direction_vectors(self) -> Tuple[Tuple[float, float], ...]:
    """Canonical geometric neighbor vectors for pointy-top hexes.

    Includes the right-down vector (1/2, sqrt(3)/2).
    """
    s = self._SQRT3_OVER_2
    return (
        (1.0, 0.0),
        (0.5, s),
        (-0.5, s),
        (-1.0, 0.0),
        (-0.5, -s),
        (0.5, -s),
    )

neighbor_coords

neighbor_coords(coord: Coord) -> List[Coord]

Return neighboring coordinates for one hex-grid location.

Parameters:

Name	Type	Description	Default
coord	Coord	Source grid coordinate in parallelogram form.	required

Returns:

Type	Description
List[Coord]	Neighbor coordinates after boundary normalization.

Examples:

A center cell has six neighbors. Under periodic boundaries, edge cells also keep six neighbors because wrapped coordinates are folded back into the valid row/col range::

HexGrid(rows=4, cols=4, wrap=False).neighbor_coords((1, 1))
HexGrid(rows=4, cols=4, wrap=True).neighbor_coords((0, 0))

Source code in src/natal/spatial/topology.py

def neighbor_coords(self, coord: Coord) -> List[Coord]:
    """Return neighboring coordinates for one hex-grid location.

    Args:
        coord: Source grid coordinate in parallelogram form.

    Returns:
        Neighbor coordinates after boundary normalization.

    Examples:
        A center cell has six neighbors. Under periodic boundaries, edge
        cells also keep six neighbors because wrapped coordinates are folded
        back into the valid row/col range::

            HexGrid(rows=4, cols=4, wrap=False).neighbor_coords((1, 1))
            HexGrid(rows=4, cols=4, wrap=True).neighbor_coords((0, 0))
    """
    i, j = coord
    # Direct neighbor offsets in parallelogram coordinates
    parallelogram_offsets: Sequence[Tuple[int, int]] = (
        (1, 0),   # right
        (0, 1),   # down-right
        (-1, 1),  # down-left
        (-1, 0),  # left
        (0, -1),  # up-left
        (1, -1),  # up-right
    )
    result: List[Coord] = []
    for di, dj in parallelogram_offsets:
        # Compute neighbors directly in parallelogram space
        normalized = self.normalize_coord(i + di, j + dj)
        if normalized is not None:
            result.append(normalized)
    return result

MigrationParams

Bases: NamedTuple

Fixed migration configuration for spatial kernels.

All fields are determined at construction time and never change during a simulation.

Attributes:

Name	Type	Description
kernel	NDArray[float64]	Migration kernel weight matrix. A (1, 1) zero array when no kernel is set (adjacency mode).
include_center	bool	Whether the kernel centre contributes outbound mass.
rate	NDArray[float64]	Fraction of each deme that migrates per tick.
adjust_on_edge	bool	Whether boundary demes normalise to match internal total outbound rate.
adjacency	NDArray[float64]	Dense (n_demes, n_demes) outbound migration matrix.
mode_code	int	Backend selector (0 = adjacency, 1 = kernel).

SpatialTopology

Bases: NamedTuple

Resolved topology dimensions for kernel routing.

Built once from GridTopology at construction time. When no topology is set, all fields default to zero / False.

Attributes:

Name	Type	Description
rows	int	Number of grid rows (0 if no topology).
cols	int	Number of grid columns (0 if no topology).
wrap	bool	Whether periodic boundary conditions apply.

SquareGrid dataclass

SquareGrid(rows: int, cols: int, wrap: bool = False, COS_OPPOSITE_ANGLE: float = 0.0, neighborhood: Literal['von_neumann', 'moore'] = 'moore')

Bases: GridTopology

Square grid with Von Neumann or Moore neighborhood.

Attributes:

Name	Type	Description
neighborhood	Literal['von_neumann', 'moore']	Neighborhood rule used to enumerate adjacent demes.
wrap	bool	Inherited periodic-boundary flag. When enabled, neighbors that would cross one edge of the rectangular grid re-enter from the opposite edge.

Examples:

Compare neighbors of the corner cell (0, 0)::

grid = SquareGrid(rows=3, cols=3, neighborhood="von_neumann", wrap=False)
grid.neighbor_coords((0, 0)) == [(1, 0), (0, 1)]

wrapped = SquareGrid(rows=3, cols=3, neighborhood="von_neumann", wrap=True)
wrapped.neighbor_coords((0, 0)) == [(2, 0), (1, 0), (0, 2), (0, 1)]

neighbor_coords

neighbor_coords(coord: Coord) -> List[Coord]

Return neighboring coordinates for one square-grid location.

Parameters:

Name	Type	Description	Default
coord	Coord	Source grid coordinate.	required

Returns:

Type	Description
List[Coord]	Neighbor coordinates after applying boundary normalization.

Raises:

Type	Description
ValueError	If neighborhood is not supported.

Source code in src/natal/spatial/topology.py

def neighbor_coords(self, coord: Coord) -> List[Coord]:
    """Return neighboring coordinates for one square-grid location.

    Args:
        coord: Source grid coordinate.

    Returns:
        Neighbor coordinates after applying boundary normalization.

    Raises:
        ValueError: If ``neighborhood`` is not supported.
    """
    row, col = coord
    if self.neighborhood == "von_neumann":
        offsets: Iterable[Coord] = [(-1, 0), (1, 0), (0, -1), (0, 1)]
    elif self.neighborhood == "moore":
        offsets = [
            (-1, -1),
            (-1, 0),
            (-1, 1),
            (0, -1),
            (0, 1),
            (1, -1),
            (1, 0),
            (1, 1),
        ]
    else:
        raise ValueError(f"Unsupported neighborhood: {self.neighborhood}")

    result: List[Coord] = []
    for dr, dc in offsets:
        normalized = self.normalize_coord(row + dr, col + dc)
        if normalized is not None:
            result.append(normalized)
    return result

batch_setting

batch_setting(values: Union[Sequence[Any], NDArray[floating[Any]], Callable[..., float], BatchSetting]) -> BatchSetting

Create a BatchSetting for per-deme parameter specification.

Parameters:

Name	Type	Description	Default
values	Union[Sequence[Any], NDArray[floating[Any]], Callable[..., float], BatchSetting]	One of: - A list/tuple of scalars of length n_demes. - A 1D or 2D numpy array. 2D arrays use (row, col) layout and are flattened in row-major order. - A callable (flat_idx) -> float or (row, col) -> float (auto-detected by parameter count). - An existing BatchSetting (returned as-is).	required

Returns:

Type	Description
BatchSetting	A BatchSetting instance that SpatialConfigurator detects and
BatchSetting	expands at build time.

Source code in src/natal/spatial/configurator.py

def batch_setting(
    values: Union[Sequence[Any], NDArray[np.floating[Any]], Callable[..., float], BatchSetting],
) -> BatchSetting:
    """Create a ``BatchSetting`` for per-deme parameter specification.

    Args:
        values: One of:
            - A list/tuple of scalars of length ``n_demes``.
            - A 1D or 2D numpy array. 2D arrays use ``(row, col)`` layout and
              are flattened in row-major order.
            - A callable ``(flat_idx) -> float`` or ``(row, col) -> float``
              (auto-detected by parameter count).
            - An existing ``BatchSetting`` (returned as-is).

    Returns:
        A ``BatchSetting`` instance that ``SpatialConfigurator`` detects and
        expands at build time.
    """
    if isinstance(values, BatchSetting):
        return values
    return BatchSetting(values)

build_adjacency_matrix

build_adjacency_matrix(topology: GridTopology, include_self: bool = False, row_normalize: bool = False) -> np.ndarray

Build an adjacency matrix from one topology.

The returned matrix uses the convention A[i, j] = weight from source deme i to destination deme j.

Parameters:

Name	Type	Description	Default
topology	GridTopology	Grid topology used to define neighborhood relations.	required
include_self	bool	Whether each deme should include itself as an outbound target.	False
row_normalize	bool	Whether to normalize each non-zero row to sum to 1.	False

Returns:

Type	Description
ndarray	A dense adjacency matrix with shape (n_demes, n_demes).

Source code in src/natal/spatial/topology.py

def build_adjacency_matrix(
    topology: GridTopology,
    include_self: bool = False,
    row_normalize: bool = False,
) -> np.ndarray:
    """Build an adjacency matrix from one topology.

    The returned matrix uses the convention ``A[i, j]`` = weight from source
    deme ``i`` to destination deme ``j``.

    Args:
        topology: Grid topology used to define neighborhood relations.
        include_self: Whether each deme should include itself as an outbound
            target.
        row_normalize: Whether to normalize each non-zero row to sum to 1.

    Returns:
        A dense adjacency matrix with shape ``(n_demes, n_demes)``.
    """
    n = topology.n_demes
    adj = np.zeros((n, n), dtype=np.float64)

    for i in range(n):
        neighbors = topology.neighbors(i)
        if include_self:
            neighbors = neighbors + [i]
        for j in neighbors:
            adj[i, j] = 1.0

    if row_normalize:
        row_sum = adj.sum(axis=1, keepdims=True)
        nonzero = row_sum[:, 0] > 0.0
        adj[nonzero] = adj[nonzero] / row_sum[nonzero]

    return adj

build_gaussian_kernel

build_gaussian_kernel(topology_cls: Literal['square', 'hex'] | type[GridTopology] = 'hex', size: int = 5, sigma: float | None = None, mean_dispersal: float | None = None) -> np.ndarray

Build a normalized Gaussian migration kernel for a grid topology.

The kernel is a (size, size) matrix where each entry [r, c] is the outbound migration weight from a virtual centre cell to the grid cell at offset (r - centre, c - centre). Distances are computed via the topology's COS_OPPOSITE_ANGLE, so the correct metric is used for square grids (Cartesian) vs hex grids (oblique / law-of-cosines).

At runtime the spatial migration kernel slides this matrix over every source deme and re-normalises valid destinations at boundaries.

Parameters:

Name	Type	Description	Default
topology_cls	Literal['square', 'hex'] | type[GridTopology]	Grid topology class (e.g. SquareGrid, HexGrid) or a string shorthand "hex" / "square".	'hex'
size	int	Odd integer kernel size (default 5). Larger kernels capture longer-range dispersal but increase the non-zero offset table.	5
sigma	float | None	Gaussian width parameter. Defaults to 1.0 when neither sigma nor mean_dispersal is given.	None
mean_dispersal	float | None	Target mean dispersal distance. When provided, sigma is derived via the 2D Rayleigh mean formula:: sigma = mean_dispersal / sqrt(π / 2) Mutually exclusive with sigma.	None

Returns:

Type	Description
ndarray	Normalised (size, size) float64 kernel summing to 1.

Raises:

Type	Description
ValueError	If size is even, both sigma and mean_dispersal are given, or topology_cls is an unknown string.

Examples:

Via sigma::

kernel = build_gaussian_kernel("hex", size=11, sigma=1.5)

Via mean dispersal::

kernel = build_gaussian_kernel("hex", size=11, mean_dispersal=2.0)

Source code in src/natal/spatial/topology.py

def build_gaussian_kernel(
    topology_cls: Literal["square", "hex"] | type[GridTopology] = "hex",
    size: int = 5,
    sigma: float | None = None,
    mean_dispersal: float | None = None,
) -> np.ndarray:
    """Build a normalized Gaussian migration kernel for a grid topology.

    The kernel is a ``(size, size)`` matrix where each entry ``[r, c]`` is
    the outbound migration weight from a virtual centre cell to the grid
    cell at offset ``(r - centre, c - centre)``. Distances are computed
    via the topology's ``COS_OPPOSITE_ANGLE``, so the correct metric is
    used for square grids (Cartesian) vs hex grids (oblique /
    law-of-cosines).

    At runtime the spatial migration kernel slides this matrix over every
    source deme and re-normalises valid destinations at boundaries.

    Args:
        topology_cls: Grid topology class (e.g. ``SquareGrid``, ``HexGrid``)
            or a string shorthand ``"hex"`` / ``"square"``.
        size: Odd integer kernel size (default 5). Larger kernels capture
            longer-range dispersal but increase the non-zero offset table.
        sigma: Gaussian width parameter. Defaults to 1.0 when neither
            ``sigma`` nor ``mean_dispersal`` is given.
        mean_dispersal: Target mean dispersal distance. When provided,
            ``sigma`` is derived via the 2D Rayleigh mean formula::

                sigma = mean_dispersal / sqrt(π / 2)

            Mutually exclusive with ``sigma``.

    Returns:
        Normalised ``(size, size)`` float64 kernel summing to 1.

    Raises:
        ValueError: If ``size`` is even, both ``sigma`` and
            ``mean_dispersal`` are given, or ``topology_cls`` is an unknown
            string.

    Examples:
        Via sigma::

            kernel = build_gaussian_kernel("hex", size=11, sigma=1.5)

        Via mean dispersal::

            kernel = build_gaussian_kernel("hex", size=11, mean_dispersal=2.0)
    """
    if sigma is not None and mean_dispersal is not None:
        raise ValueError(
            "sigma and mean_dispersal are mutually exclusive; "
            "specify one or neither, not both"
        )
    if mean_dispersal is not None:
        sigma = mean_dispersal / math.sqrt(math.pi / 2.0)
    elif sigma is None:
        sigma = 1.0

    if isinstance(topology_cls, str):
        if topology_cls == "hex":
            topology_cls = HexGrid
        elif topology_cls == "square":
            topology_cls = SquareGrid
        else:
            raise ValueError(
                f"Unknown topology shorthand {topology_cls!r}, expected 'hex' or 'square'"
            )

    if size % 2 == 0:
        raise ValueError(f"kernel size must be odd, got {size}")

    center = (size - 1) / 2.0
    y_idx, x_idx = np.indices((size, size), dtype=np.float64)
    dr = y_idx - center
    dc = x_idx - center
    cos_opposite = topology_cls.COS_OPPOSITE_ANGLE
    dist_sq = dr**2 + dc**2 - 2.0 * cos_opposite * dr * dc
    kernel = np.exp(-dist_sq / (2.0 * sigma**2)).astype(np.float64, copy=False)
    kernel /= np.sum(kernel)
    return kernel