Skip to content

hooks Module

Domain-Specific Language for simulation hooks.

Overview

The hooks subpackage provides the core components for the declarative hook system.

Complete Module Reference

natal.hooks

Hook subsystem public API.

CompiledEventHooks 

CompiledEventHooks()

Per-event hook container used by lifecycle wrappers.

Holds one combined callable per event (first / early / late / finish) plus a registry (HookProgram) for CSR dispatch. This is a pure container — code generation and lifecycle wrapper compilation live in natal.engine.lifecycle_wrappers.

Slots

first / early / late / finish Combined hook callables for each event. registry HookProgram — may be filtered for mixed events.

Initialise all event hooks to no-op and registry to None.

Source code in src/natal/hooks/compile/container.py 
63
64
65
66
67
68
69
70
def __init__(self) -> None:
    """Initialise all event hooks to no-op and registry to None."""
    self.first = _noop_hook
    self.early = _noop_hook
    self.late = _noop_hook
    self.finish = _noop_hook
    self.registry = None
    self._event_hooks = dict.fromkeys(EVENT_NAMES, _noop_hook)
get_hook 
get_hook(event_name: str) -> HookCallable

Return the combined callable for event_name.

Source code in src/natal/hooks/compile/container.py 
72
73
74
def get_hook(self, event_name: str) -> HookCallable:
    """Return the combined callable for *event_name*."""
    return self._event_hooks.get(event_name, _noop_hook)
set_hook 
set_hook(event_name: str, hook_fn: HookCallable) -> None

Set the combined callable for event_name (both dict and attr).

Source code in src/natal/hooks/compile/container.py 
76
77
78
79
def set_hook(self, event_name: str, hook_fn: HookCallable) -> None:
    """Set the combined callable for *event_name* (both dict and attr)."""
    self._event_hooks[event_name] = hook_fn
    setattr(self, event_name, hook_fn)

Op

Factory helpers for building declarative operations.

The methods here only build data objects and do not touch population state. Compilation happens later in compile_declarative_hook.

scale staticmethod 
scale(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', factor: float = 1.0, when: Optional[str] = None) -> HookOp

Create a scaling operation that multiplies counts by a factor.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector ("*" for all, specific genotype, or list)

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector ("*" for all, specific age, range, or list)

 '*'
sex Literal['female', 'male', 'both']

Sex selector ("female", "male", or "both")

 'both'
factor float

Scaling factor (e.g., 0.5 halves the count, 2.0 doubles it)

 1.0
when Optional[str]

Optional condition expression (e.g., "tick >= 100")

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@staticmethod
def scale(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    factor: float = 1.0,
    when: Optional[str] = None,
) -> HookOp:
    """Create a scaling operation that multiplies counts by a factor.

    Args:
        genotypes: Genotype selector ("*" for all, specific genotype, or list)
        ages: Age selector ("*" for all, specific age, range, or list)
        sex: Sex selector ("female", "male", or "both")
        factor: Scaling factor (e.g., 0.5 halves the count, 2.0 doubles it)
        when: Optional condition expression (e.g., "tick >= 100")

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.SCALE, genotypes, ages, sex, factor, when)
set_count staticmethod 
set_count(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', value: float = 0.0, when: Optional[str] = None) -> HookOp

Create an operation that sets counts to a specific value.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
value float

Target count value (individuals will be added/removed to match)

 0.0
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@staticmethod
def set_count(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    value: float = 0.0,
    when: Optional[str] = None,
) -> HookOp:
    """Create an operation that sets counts to a specific value.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        value: Target count value (individuals will be added/removed to match)
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.SET, genotypes, ages, sex, value, when)
add staticmethod 
add(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', delta: float = 0.0, when: Optional[str] = None) -> HookOp

Create an operation that adds a fixed number of individuals.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
delta float

Number of individuals to add (can be negative to remove)

 0.0
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@staticmethod
def add(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    delta: float = 0.0,
    when: Optional[str] = None,
) -> HookOp:
    """Create an operation that adds a fixed number of individuals.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        delta: Number of individuals to add (can be negative to remove)
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.ADD, genotypes, ages, sex, delta, when)
subtract staticmethod 
subtract(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', delta: float = 0.0, when: Optional[str] = None) -> HookOp

Create an operation that subtracts a fixed number of individuals.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
delta float

Number of individuals to subtract

 0.0
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@staticmethod
def subtract(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    delta: float = 0.0,
    when: Optional[str] = None,
) -> HookOp:
    """Create an operation that subtracts a fixed number of individuals.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        delta: Number of individuals to subtract
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.SUBTRACT, genotypes, ages, sex, delta, when)
kill staticmethod 
kill(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', prob: float = 0.0, when: Optional[str] = None) -> HookOp

Create a probabilistic killing operation.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
prob float

Probability of killing each selected individual (0.0 to 1.0)

 0.0
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation

Raises:

Type Description
ValueError

If probability is not in [0, 1]
Source code in src/natal/hooks/entry/declarative.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@staticmethod
def kill(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    prob: float = 0.0,
    when: Optional[str] = None,
) -> HookOp:
    """Create a probabilistic killing operation.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        prob: Probability of killing each selected individual (0.0 to 1.0)
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation

    Raises:
        ValueError: If probability is not in [0, 1]
    """
    if not 0.0 <= prob <= 1.0:
        raise ValueError(f"prob must be in [0, 1], got {prob}")
    return HookOp(OpType.KILL, genotypes, ages, sex, prob, when)
sample staticmethod 
sample(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', size: int = 0, when: Optional[str] = None) -> HookOp

Create a sampling operation that selects individuals without replacement.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
size int

Number of individuals to sample

 0
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@staticmethod
def sample(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    size: int = 0,
    when: Optional[str] = None,
) -> HookOp:
    """Create a sampling operation that selects individuals without replacement.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        size: Number of individuals to sample
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.SAMPLE, genotypes, ages, sex, float(size), when)
stop_if_zero staticmethod 
stop_if_zero(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', when: Optional[str] = None) -> HookOp

Create an operation that stops the simulation if selected count reaches zero.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
@staticmethod
def stop_if_zero(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    when: Optional[str] = None,
) -> HookOp:
    """Create an operation that stops the simulation if selected count reaches zero.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.STOP_IF_ZERO, genotypes, ages, sex, 0.0, when)
stop_if_below staticmethod 
stop_if_below(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', threshold: float = 1.0, when: Optional[str] = None) -> HookOp

Create an operation that stops the simulation if count falls below threshold.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
threshold float

Minimum count threshold

 1.0
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
@staticmethod
def stop_if_below(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    threshold: float = 1.0,
    when: Optional[str] = None,
) -> HookOp:
    """Create an operation that stops the simulation if count falls below threshold.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        threshold: Minimum count threshold
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.STOP_IF_BELOW, genotypes, ages, sex, float(threshold), when)
stop_if_above staticmethod 
stop_if_above(genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', threshold: float = 1000000.0, when: Optional[str] = None) -> HookOp

Create an operation that stops the simulation if count exceeds threshold.

Parameters:

Name Type Description Default
genotypes Union[str, List[str], Literal['*']]

Genotype selector

 '*'
ages Union[int, List[int], range, Literal['*']]

Age selector

 '*'
sex Literal['female', 'male', 'both']

Sex selector

 'both'
threshold float

Maximum count threshold

 1000000.0
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
@staticmethod
def stop_if_above(
    genotypes: Union[str, List[str], Literal["*"]] = "*",
    ages: Union[int, List[int], range, Literal["*"]] = "*",
    sex: Literal["female", "male", "both"] = "both",
    threshold: float = 1_000_000.0,
    when: Optional[str] = None,
) -> HookOp:
    """Create an operation that stops the simulation if count exceeds threshold.

    Args:
        genotypes: Genotype selector
        ages: Age selector
        sex: Sex selector
        threshold: Maximum count threshold
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.STOP_IF_ABOVE, genotypes, ages, sex, float(threshold), when)
stop_if_extinction staticmethod 
stop_if_extinction(when: Optional[str] = None) -> HookOp

Create an operation that stops the simulation if total population goes extinct.

Parameters:

Name Type Description Default
when Optional[str]

Optional condition expression

 None

Returns:

Name Type Description
HookOp HookOp

Operation descriptor for compilation
Source code in src/natal/hooks/entry/declarative.py 
255
256
257
258
259
260
261
262
263
264
265
@staticmethod
def stop_if_extinction(when: Optional[str] = None) -> HookOp:
    """Create an operation that stops the simulation if total population goes extinct.

    Args:
        when: Optional condition expression

    Returns:
        HookOp: Operation descriptor for compilation
    """
    return HookOp(OpType.STOP_IF_EXTINCTION, "*", "*", "both", 0.0, when)

HookExecutor 

HookExecutor(registry: HookProgram, hooks_by_event: Dict[int, List[CompiledHookDescriptor]])

Python-layer coordinator for hook execution (Numba-disabled only).

When Numba is disabled, all hook types — CSR plans, njit functions, Python wrappers — must run through a single sequential path. This class holds descriptors sorted by priority and dispatches each one in order.

Initialise with a pre-built registry and descriptor map.

Parameters:

Name Type Description Default
registry HookProgram

HookProgram for CSR operations (may be empty).

 required
hooks_by_event Dict[int, List[CompiledHookDescriptor]]

Descriptors grouped by event_id and sorted by priority. Built by from_compiled_hooks.

 required
Source code in src/natal/hooks/runtime/fallback.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    registry: HookProgram,
    hooks_by_event: Dict[int, List[CompiledHookDescriptor]],
) -> None:
    """Initialise with a pre-built registry and descriptor map.

    Args:
        registry: HookProgram for CSR operations (may be empty).
        hooks_by_event: Descriptors grouped by event_id and sorted
            by priority.  Built by ``from_compiled_hooks``.
    """
    self.registry = registry
    self.hooks_by_event = hooks_by_event
from_compiled_hooks staticmethod 
from_compiled_hooks(registry: HookProgram, compiled_hooks: List[CompiledHookDescriptor]) -> HookExecutor

Group descriptors by event_id and sort by priority.

Descriptors without a recognised event_id or without any execution payload are silently skipped.

Source code in src/natal/hooks/runtime/fallback.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@staticmethod
def from_compiled_hooks(
    registry: HookProgram,
    compiled_hooks: List[CompiledHookDescriptor],
) -> HookExecutor:
    """Group descriptors by event_id and sort by priority.

    Descriptors without a recognised event_id or without any
    execution payload are silently skipped.
    """
    hooks_by_event: Dict[int, List[CompiledHookDescriptor]] = defaultdict(list)
    for desc in compiled_hooks:
        event_id = EVENT_ID_MAP.get(desc.event)
        if event_id is not None:
            if desc.plan is not None or desc.njit_fn is not None or desc.py_wrapper is not None:
                hooks_by_event[event_id].append(desc)

    for event_id in hooks_by_event:
        hooks_by_event[event_id].sort(key=lambda x: x.priority)

    return HookExecutor(registry, dict(hooks_by_event))
execute_event 
execute_event(event_id: int, population: BasePopulation[Any], tick: int, deme_id: int = 0) -> int

Run all hooks for event_id in priority order.

For each descriptor, in priority order:

  1. CSR plan — unpacks arrays and calls execute_csr_event_arrays with a single-hook wrapper. Aborts on RESULT_STOP.
  2. njit function — calls desc.njit_fn(state, config, deme_id).
  3. Python wrapper — calls with population (1-param) or (state, config, deme_id) (3-param). Only allowed when Numba is disabled.

Returns:

Type Description
int

RESULT_CONTINUE or RESULT_STOP.
Source code in src/natal/hooks/runtime/fallback.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def execute_event(
    self,
    event_id: int,
    population: BasePopulation[Any],
    tick: int,
    deme_id: int = 0,
) -> int:
    """Run all hooks for *event_id* in priority order.

    For each descriptor, in priority order:

    1. CSR plan — unpacks arrays and calls ``execute_csr_event_arrays``
       with a single-hook wrapper.  Aborts on ``RESULT_STOP``.
    2. njit function — calls ``desc.njit_fn(state, config, deme_id)``.
    3. Python wrapper — calls with population (1-param) or
       ``(state, config, deme_id)`` (3-param).  Only allowed when
       Numba is disabled.

    Returns:
        ``RESULT_CONTINUE`` or ``RESULT_STOP``.
    """
    if event_id < 0 or event_id >= NUM_EVENTS:
        return RESULT_CONTINUE

    ind_count = population.state.individual_count

    # Resolve runtime state flags.
    sperm_store = getattr(population.state, "sperm_storage", None)
    has_sperm_storage = sperm_store is not None and sperm_store.size > 0
    if not has_sperm_storage:
        sperm_store = np.zeros((0, 0, 0), dtype=np.float64)
    assert sperm_store is not None
    stochastic = bool(getattr(getattr(population, "_config", None), "stochastic", False))
    continuous_sampling = bool(
        getattr(getattr(population, "_config", None), "continuous_sampling", False)
    )

    from natal.numba.utils import NUMBA_ENABLED

    for desc in self.hooks_by_event.get(event_id, []):
        if not deme_selector_matches(desc.deme_selector, deme_id):
            continue

        # CSR: declarative Op.*
        if desc.plan is not None:
            result = execute_csr_event_arrays(
                n_events=np.int32(1),
                n_hooks=np.int32(1),
                hook_offsets=np.array([0, 1], dtype=np.int32),
                n_ops_list=np.array([desc.plan.n_ops], dtype=np.int32),
                op_offsets=np.array([0, desc.plan.n_ops], dtype=np.int32),
                op_types_data=desc.plan.op_types,
                zidx_offsets_data=desc.plan.zidx_offsets,
                zidx_data=desc.plan.zidx_data,
                age_offsets_data=desc.plan.age_offsets,
                age_data=desc.plan.age_data,
                sex_masks_data=desc.plan.sex_masks.ravel(),
                params_data=desc.plan.params,
                condition_offsets_data=desc.plan.condition_offsets,
                condition_types_data=desc.plan.condition_types,
                condition_params_data=desc.plan.condition_params,
                deme_selector_types=np.array([0], dtype=np.int32),
                deme_selector_offsets=np.array([0, 0], dtype=np.int32),
                deme_selector_data=np.array([], dtype=np.int32),
                event_id=0,
                individual_count=ind_count,
                sperm_storage=sperm_store,
                has_sperm_storage=has_sperm_storage,
                tick=tick,
                stochastic=stochastic,
                continuous_sampling=continuous_sampling,
                deme_id=deme_id,
            )
            if result == RESULT_STOP:
                return RESULT_STOP

        # njit: compiled custom/selector hook.
        if desc.njit_fn is not None:
            try:
                result = desc.njit_fn(population.state, population.config, deme_id)
                if result == RESULT_STOP:
                    return RESULT_STOP
            except Exception as e:
                raise RuntimeError(f"Error in njit hook '{desc.name}': {e}") from e

        # Python wrapper (Numba off only).
        if desc.py_wrapper is not None and desc.njit_fn is None:
            if NUMBA_ENABLED:
                raise RuntimeError(
                    f"Python py_wrapper hook '{desc.name}' is not allowed "
                    "when Numba is enabled."
                )
            try:
                import inspect

                sig = inspect.signature(desc.py_wrapper)
                params = list(sig.parameters.values())
                if len(params) == 1:
                    desc.py_wrapper(population)
                elif len(params) == 3:
                    desc.py_wrapper(population.state, population.config, deme_id)
                else:
                    raise TypeError(
                        f"py_wrapper hook '{desc.name}' has {len(params)} "
                        "parameters; must have 1 (population) or 3 "
                        "(state, config, deme_id)."
                    )
            except Exception as e:
                raise RuntimeError(f"Error in py_wrapper hook '{desc.name}': {e}") from e

    return RESULT_CONTINUE
get_hooks_for_event 
get_hooks_for_event(event_id: int) -> List[CompiledHookDescriptor]

Return descriptors for event_id, sorted by priority.

Source code in src/natal/hooks/runtime/fallback.py 
193
194
195
def get_hooks_for_event(self, event_id: int) -> List[CompiledHookDescriptor]:
    """Return descriptors for *event_id*, sorted by priority."""
    return self.hooks_by_event.get(event_id, [])

CompiledHookDescriptor dataclass 

CompiledHookDescriptor(name: str, event: str, priority: int = 0, deme_selector: DemeSelector = '*', plan: Optional[CompiledHookPlan] = None, selectors: Dict[str, ndarray] = _empty_selector_map(), static_arrays: Tuple[ndarray, ...] = tuple(), meta: Dict[str, int] = _empty_meta_map(), njit_fn: Optional[Callable[..., object]] = None, py_wrapper: Optional[Callable[..., object]] = None, ops: Optional[List[HookOp]] = None)

Unified descriptor for all hook modes.

Exactly one of plan, njit_fn, or py_wrapper is typically used as the primary execution payload for a descriptor.

CompiledHookPlan dataclass 

CompiledHookPlan(n_ops: int, op_types: ndarray, zidx_offsets: ndarray, zidx_data: ndarray, age_offsets: ndarray, age_data: ndarray, sex_masks: ndarray, params: ndarray, condition_offsets: ndarray, condition_types: ndarray, condition_params: ndarray)

Compiled declarative plan with CSR-style flattened arrays.

Variable-length fields (genotypes/ages/conditions) are represented via *_offsets + *_data to keep kernel inputs contiguous and compact.

to_tuple 
to_tuple() -> Tuple[object, ...]

Convert this plan to a flat tuple for HDF5 / array storage.

Returns:

Type Description
Tuple[object, ...]

A tuple of all fields in declaration order.
Source code in src/natal/hooks/types.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def to_tuple(self) -> Tuple[object, ...]:
    """Convert this plan to a flat tuple for HDF5 / array storage.

    Returns:
        A tuple of all fields in declaration order.
    """
    return (
        self.n_ops,
        self.op_types,
        self.zidx_offsets,
        self.zidx_data,
        self.age_offsets,
        self.age_data,
        self.sex_masks,
        self.params,
        self.condition_offsets,
        self.condition_types,
        self.condition_params,
    )

HookOp dataclass 

HookOp(op_type: OpType, genotypes: Union[str, List[str], Literal['*']] = '*', ages: Union[int, List[int], range, Literal['*']] = '*', sex: Literal['female', 'male', 'both'] = 'both', param: float = 1.0, condition: Optional[str] = None)

Single declarative operation before compilation.

Fields in this class can still be symbolic (for example genotype labels). The compiler resolves all symbolic fields into concrete integer arrays.

HookProgram

Bases: NamedTuple

Event-grouped plain-data CSR representation for declarative hooks.

OpType

Bases: IntEnum

Operation opcodes consumed by the runtime kernel.

We intentionally keep integer values stable because these values are serialized into CompiledHookPlan.op_types and interpreted in the executor hot-loop.

compile_combined_hook 

compile_combined_hook(njit_fns: List[HookCallable], deme_selectors: Optional[List[DemeSelector]] = None) -> HookCallable

Combine multiple njit hooks into a single generated njit function.

Generates Python source for a new module containing a function that calls each hook in sequence. Each call is wrapped with an optional if deme_id == X guard for spatial simulations.

The source is written to .numba_cache/ so Numba's cache=True survives process restarts — this is not possible with runtime-composed closures.

Parameters:

Name Type Description Default
njit_fns List[HookCallable]

Njit-compiled hook functions to call in order.

 required
deme_selectors Optional[List[DemeSelector]]

Per-function deme target. None or "*" (default) means no guard. int, list, range produce a guard for that specific deme or set.

 None

Returns:

Type Description
HookCallable

An @njit_switch(cache=True) function with signature
HookCallable

(state, config=None, deme_id=-1) -> int. Returns the
HookCallable

first non-zero result (RESULT_STOP), or 0.
Source code in src/natal/hooks/compile/codegen.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def compile_combined_hook(
    njit_fns: List[HookCallable],
    deme_selectors: Optional[List[DemeSelector]] = None,
) -> HookCallable:
    """Combine multiple njit hooks into a single generated njit function.

    Generates Python source for a new module containing a function that
    calls each hook in sequence.  Each call is wrapped with an optional
    ``if deme_id == X`` guard for spatial simulations.

    The source is written to ``.numba_cache/`` so Numba's ``cache=True``
    survives process restarts — this is not possible with runtime-composed
    closures.

    Args:
        njit_fns: Njit-compiled hook functions to call in order.
        deme_selectors: Per-function deme target.  ``None`` or ``"*"``
            (default) means no guard.  ``int``, ``list``, ``range``
            produce a guard for that specific deme or set.

    Returns:
        An ``@njit_switch(cache=True)`` function with signature
        ``(state, config=None, deme_id=-1) -> int``.  Returns the
        first non-zero result (``RESULT_STOP``), or 0.
    """
    if len(njit_fns) == 0:
        return noop_hook

    # Normalize to list so pyright can track the type (not Optional).
    ds_list: List[DemeSelector] = deme_selectors if deme_selectors is not None else []
    needs_guard = any(ds != "*" for ds in ds_list)

    # Single-hook, no guard — return the function directly (optimisation).
    if not needs_guard and len(njit_fns) == 1:
        return njit_fns[0]

    # ---- Build stable cache key ----
    if needs_guard:
        combined_parts = ["combined_guarded"]
        for fn, ds in zip(njit_fns, ds_list):
            combined_parts.append(stable_callable_identity(fn))
            combined_parts.append(str(ds))
    else:
        combined_parts = ["combined"] + [
            stable_callable_identity(fn) for fn in njit_fns
        ]
    key = hash_key(combined_parts)
    fn_name = f"_combined_hook_{key}"
    module_stem = f"combined_hook_{key}"
    placeholder_names = [f"_FN_{i}" for i in range(len(njit_fns))]

    # ---- Build module-level placeholder declarations ----
    # Each _FN_i = lambda … is a type-stub that gets overridden via setattr.
    fn_decl_lines: List[str] = []
    for pname in placeholder_names:
        fn_decl_lines.append(
            f"{pname}: Callable[..., int] = "
            "lambda _s, _c=None, _d=-1: 0  # type: ignore[assignment]"
        )

    # ---- Build schedule body from njit entry template ----
    # Reuse the same njit entry fragment that unified dispatch uses.
    # Guard conditions computed identically to compile_unified_event_hook.
    njit_tmpl = _read_hook_template("unified_hook_njit_entry.tmpl.py")
    schedule_entries: List[str] = []

    for pname, ds in zip(placeholder_names, ds_list):
        if ds == "*":
            guard_cond = "True"
        elif isinstance(ds, int):
            guard_cond = f"deme_id == {int(ds)}"
        elif isinstance(ds, range):
            guard_cond = f"{ds.start} <= deme_id < {ds.stop}"
        else:
            items = ", ".join(str(int(x)) for x in ds)
            guard_cond = f"deme_id in ({items})"
        schedule_entries.append(
            njit_tmpl.replace("PLACEHOLDER_NJIT_FN_NAME", pname).replace(
                "PLACEHOLDER_NJIT_GUARD_CONDITION", guard_cond
            )
        )

    # ---- Assemble module from template ----
    template = _read_hook_template("combined_hook.tmpl.py")
    source = (
        template.replace("_combined_hook_TEMPLATE", fn_name)
        .replace("# PLACEHOLDER_FN_DECLARATIONS", "\n".join(fn_decl_lines))
        .replace("# PLACEHOLDER_SCHEDULE_BODY", "\n".join(schedule_entries))
    )

    # ---- Write, load, wire up globals ----
    # Module is written to .numba_cache/ so Numba's cache=True survives
    # restarts.  setattr overrides each placeholder with the real njit
    # function before returning the generated callable.
    module_path = write_codegen_module(module_stem, source)
    module = load_codegen_module(module_stem, module_path)

    for placeholder, fn in zip(placeholder_names, njit_fns):
        setattr(module, placeholder, fn)

    return getattr(module, fn_name)

compile_declarative_hook 

compile_declarative_hook(ops: List[HookOp], pop: BasePopulation[Any], event: str, priority: int = 0, deme_selector: DemeSelector = '*', name: str = 'declarative_hook') -> CompiledHookDescriptor

Compile declarative ops into a CompiledHookDescriptor.

The compiler packs all per-op fields into parallel arrays. Offsets arrays (*_offsets) define CSR spans for variable-length selector/condition data and avoid Python object usage in runtime engine.

Source code in src/natal/hooks/entry/declarative.py 
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
def compile_declarative_hook(
    ops: List[HookOp],
    pop: BasePopulation[Any],
    event: str,
    priority: int = 0,
    deme_selector: DemeSelector = "*",
    name: str = "declarative_hook",
) -> CompiledHookDescriptor:
    """Compile declarative ops into a ``CompiledHookDescriptor``.

    The compiler packs all per-op fields into parallel arrays. Offsets arrays
    (``*_offsets``) define CSR spans for variable-length selector/condition
    data and avoid Python object usage in runtime engine.
    """
    # Get population configuration and registry for resolving genotype/age indices
    index_registry = pop.index_registry
    species = index_registry.index_to_genotype[0].species if index_registry.index_to_genotype else pop.species
    n_ztypes = index_registry.n_ztypes
    n_ages = pop.config.n_ages

    # Initialize data structures for storing compiled hook operations
    # These will be packed into parallel arrays for efficient runtime execution

    # 1. Operation type stream - stores the operation code for each hook
    op_types_list: List[int] = []

    # 2. Genotype selection data (CSR format)
    # zidx_offsets: CSR offsets defining genotype index ranges for each operation
    # zidx_data: Flattened list of all genotype indices across all operations
    zidx_offsets: List[int] = [0]  # Start with offset 0 for the first operation
    zidx_data_list: List[int] = []

    # 3. Age selection data (CSR format)
    # age_offsets: CSR offsets defining age index ranges for each operation
    # age_data: Flattened list of all age indices across all operations
    age_offsets: List[int] = [0]  # Start with offset 0 for the first operation
    age_data_list: List[int] = []

    # 4. Sex selection and operation parameters
    # sex_masks: Boolean masks for male/female selection (2D array: [op][sex])
    # params: Numeric parameters for each operation (e.g., fitness values)
    sex_masks_list: List[NDArray[np.bool_]] = []
    params_list: List[float] = []

    # 5. Condition expression data (CSR format)
    # condition_offsets: CSR offsets defining condition token ranges for each operation
    # condition_types: Flattened list of condition operation types
    # condition_params: Flattened list of condition parameters
    condition_offsets: List[int] = [0]  # Start with offset 0 for the first operation
    condition_types_list: List[int] = []
    condition_params_list: List[int] = []

    # Process each hook operation and compile it into the packed arrays
    for op in ops:
        # 1) Operation type - convert enum to integer for efficient runtime lookup
        op_types_list.append(int(op.op_type))

        # 2) Genotype span - resolve genotype selectors to actual genotype indices
        # Examples: "A1|A1" -> [0], "*" -> [0, 1, 2, ..., n_genotypes-1]
        zidx_array = _resolve_genotypes(op.genotypes, index_registry, species, n_ztypes)
        zidx_data_list.extend(zidx_array.tolist())
        zidx_offsets.append(len(zidx_data_list))  # Record end offset for this operation

        # 3) Age span - resolve age selectors to actual age indices
        # Examples: "0-5" -> [0, 1, 2, 3, 4, 5], "*" -> [0, 1, ..., n_ages-1]
        age_array = _resolve_ages(op.ages, n_ages)
        age_data_list.extend(age_array.tolist())
        age_offsets.append(len(age_data_list))  # Record end offset for this operation

        # 4) Sex mask + numeric parameter
        # Convert sex selector to boolean mask [male_selected, female_selected]
        sex_masks_list.append(_resolve_sex(op.sex))
        params_list.append(float(op.param))  # Convert parameter to float

        # 5) Compiled condition token span
        # Parse condition expression into RPN (Reverse Polish Notation) tokens
        cond_types, cond_params = _parse_condition(op.condition)
        condition_types_list.extend(cond_types.tolist())
        condition_params_list.extend(cond_params.tolist())
        condition_offsets.append(len(condition_types_list))  # Record end offset

    # Create the compiled execution plan with all packed arrays
    plan = CompiledHookPlan(
        n_ops=len(ops),  # Total number of operations

        # Operation type stream - each element is an integer operation code
        op_types=np.array(op_types_list, dtype=np.int32),

        # Genotype selection data in CSR format
        # zidx_offsets[i] to zidx_offsets[i+1] defines genotype indices for operation i
        zidx_offsets=np.array(zidx_offsets, dtype=np.int32),
        zidx_data=np.array(zidx_data_list, dtype=np.int32) if zidx_data_list else np.array([], dtype=np.int32),

        # Age selection data in CSR format
        # age_offsets[i] to age_offsets[i+1] defines age indices for operation i
        age_offsets=np.array(age_offsets, dtype=np.int32),
        age_data=np.array(age_data_list, dtype=np.int32) if age_data_list else np.array([], dtype=np.int32),

        # Sex selection masks - 2D boolean array [n_ops x 2]
        # Each row: [male_selected, female_selected]
        sex_masks=np.vstack(sex_masks_list) if sex_masks_list else np.zeros((0, 2), dtype=np.bool_),

        # Operation parameters - numeric values for each operation
        params=np.array(params_list, dtype=np.float64),

        # Condition expression data in CSR format
        # condition_offsets[i] to condition_offsets[i+1] defines condition tokens for operation i
        condition_offsets=np.array(condition_offsets, dtype=np.int32),
        condition_types=np.array(condition_types_list, dtype=np.int32),
        condition_params=np.array(condition_params_list, dtype=np.int32),
    )

    # Return the complete hook descriptor with metadata
    return CompiledHookDescriptor(
        name=name,                    # Human-readable name for debugging
        event=event,                  # Simulation event when this hook triggers
        priority=priority,            # Execution priority (higher = earlier)
        deme_selector=deme_selector, # Which demes this hook applies to
        plan=plan,                    # Compiled execution plan
        meta={"n_ztypes": index_registry.n_ztypes, "n_ages": n_ages},  # Population metadata
        ops=ops,                     # Original operations for reference/debugging
    )

hook 

hook(event: Optional[str] = None, selectors: Optional[Dict[str, Any]] = None, priority: int = 0, custom: bool = False, deme: DemeSelector = '*', mode: str = 'auto') -> Callable[[Callable[..., Any]], DecoratedHookFn]

Decorator for all supported hook authoring styles.

The decorated function gains a .register(pop) method that compiles and registers a CompiledHookDescriptor against a population instance.

Hook type auto-detection (evaluated at .register() time):

  • selectors= is set → Selector hook (compile_selector_hook)
  • custom=True or function has required parameters → Custom hook (njit or Python wrapper)
  • Otherwise → Declarative hook (function returns List[HookOp], compiled via compile_declarative_hook)

For custom and selector hooks, Numba compilation is automatic — you do not need to stack @njit. When Numba is enabled the function is wrapped with njit_switch automatically; when Numba is disabled a pure-Python wrapper is used.

In spatial simulations the deme_id parameter receives the current deme index, enabling one hook function to serve all demes with per-deme branching.

Parameters:

Name Type Description Default
event Optional[str]

Hook event name ("first", "early", "late", "finish").

 None
selectors Optional[Dict[str, Any]]

Symbolic selectors for selector-mode hooks.

 None
priority int

Execution priority — lower values run first.

 0
custom bool

If True, treat as custom hook regardless of signature.

 False
deme DemeSelector

Target deme(s). "*" (default) means all demes. Accepts int, list, tuple, or range.

 '*'
mode str

Selector passing style. "auto" (default) auto-detects from the function signature. "expand" passes each selector as a separate keyword argument. "aggregate" packs all selectors into a single namedtuple argument.

 'auto'

Returns:

Type Description
Callable[[Callable[..., Any]], DecoratedHookFn]

A decorator that transforms a function into a DecoratedHookFn
Callable[[Callable[..., Any]], DecoratedHookFn]

with .register(pop) capability.

Raises:

Type Description
ValueError

If mode is not one of "auto", "expand", "aggregate".

Examples:

Declarative hook (returns ops):

    @hook(event="early", priority=0)
    def cull_juveniles():
        return [Op.scale(ages=[0,1], factor=0.9)]

Custom njit hook:

    @hook(event="first", priority=1)
    def release_males(state, config, deme_id=-1):
        state.individual_count[1, 2, 0] += 100
        return 0

Selector hook:

    @hook(event="late", selectors={"target": "AA"})
    def count_homozygotes(state, config, target, deme_id=-1):
        ...
Source code in src/natal/hooks/entry/decorator.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
def hook(
    event: Optional[str] = None,
    selectors: Optional[Dict[str, Any]] = None,
    priority: int = 0,
    custom: bool = False,
    deme: DemeSelector = "*",
    mode: str = "auto",
) -> Callable[[Callable[..., Any]], DecoratedHookFn]:
    """Decorator for all supported hook authoring styles.

    The decorated function gains a ``.register(pop)`` method that
    compiles and registers a ``CompiledHookDescriptor`` against a
    population instance.

    **Hook type auto-detection** (evaluated at ``.register()`` time):

    * ``selectors=`` is set → **Selector hook**
      (``compile_selector_hook``)
    * ``custom=True`` or function has required parameters →
      **Custom hook** (njit or Python wrapper)
    * Otherwise → **Declarative hook** (function returns
      ``List[HookOp]``, compiled via ``compile_declarative_hook``)

    For custom and selector hooks, Numba compilation is automatic — you
    do **not** need to stack ``@njit``.  When Numba is enabled the
    function is wrapped with ``njit_switch`` automatically; when Numba
    is disabled a pure-Python wrapper is used.

    In spatial simulations the ``deme_id`` parameter receives the current
    deme index, enabling one hook function to serve all demes with
    per-deme branching.

    Args:
        event: Hook event name (``"first"``, ``"early"``, ``"late"``,
            ``"finish"``).
        selectors: Symbolic selectors for selector-mode hooks.
        priority: Execution priority — lower values run first.
        custom: If ``True``, treat as custom hook regardless of signature.
        deme: Target deme(s).  ``"*"`` (default) means all demes.
            Accepts ``int``, ``list``, ``tuple``, or ``range``.
        mode: Selector passing style.  ``"auto"`` (default) auto-detects
            from the function signature.  ``"expand"`` passes each
            selector as a separate keyword argument.  ``"aggregate"``
            packs all selectors into a single namedtuple argument.

    Returns:
        A decorator that transforms a function into a ``DecoratedHookFn``
        with ``.register(pop)`` capability.

    Raises:
        ValueError: If *mode* is not one of ``"auto"``, ``"expand"``,
            ``"aggregate"``.

    Examples:

        Declarative hook (returns ops):

            @hook(event="early", priority=0)
            def cull_juveniles():
                return [Op.scale(ages=[0,1], factor=0.9)]

        Custom njit hook:

            @hook(event="first", priority=1)
            def release_males(state, config, deme_id=-1):
                state.individual_count[1, 2, 0] += 100
                return 0

        Selector hook:

            @hook(event="late", selectors={"target": "AA"})
            def count_homozygotes(state, config, target, deme_id=-1):
                ...
    """
    if mode not in ("auto", "expand", "aggregate"):
        raise ValueError(
            f"mode must be 'auto', 'expand', or 'aggregate', got {mode!r}"
        )

    def decorator(func: Callable[..., Any]) -> DecoratedHookFn:
        """Transform *func* into a ``DecoratedHookFn`` with ``.register(pop)``."""
        hook_func = cast(DecoratedHookFn, func)
        hook_func.meta = {
            "event": event,
            "selectors": selectors or {},
            "priority": priority,
            "custom": custom,
            "deme_selector": deme,
            "mode": mode,
        }
        hook_func.compiled = None
        hook_func.event = event
        hook_func.selectors = selectors or {}
        hook_func.priority = priority
        hook_func.custom = custom
        hook_func.deme_selector = deme

        def register(
            pop: BasePopulation[Any],
            event_override: Optional[str] = None,
            deme_selector_override: Optional[DemeSelector] = None,
        ) -> CompiledHookDescriptor:
            """Compile this hook against *pop* and return a descriptor.

            Called by ``pop.set_hook()``.  Detects the hook type from
            the decorator metadata and function signature, then routes
            to the appropriate compiler.

            Args:
                pop: The population to compile against.
                event_override: Override the event name (used when
                    ``set_hook(event_name, ...)`` is called with a
                    different event than the decorator specifies).
                deme_selector_override: Override the deme selector
                    (used by SpatialPopulation to pin hooks to demes).

            Returns:
                A ``CompiledHookDescriptor`` registered on *pop*.
            """
            from ...numba.utils import NUMBA_ENABLED
            from ..types import CompiledHookDescriptor

            actual_event = event_override or event
            actual_deme_selector: DemeSelector = (
                deme if deme_selector_override is None else deme_selector_override
            )
            if actual_event is None:
                raise ValueError(
                    f"Event not specified for hook '{func.__name__}'. "
                    "Specify in decorator @hook(event='...') or call "
                    "pop.set_hook('event', hook)"
                )

            # Detect hook type from decorator metadata + function signature.
            # Priority: explicit selectors > explicit custom > has params
            # (and not a single-param pop hook) > declarative (returns ops).
            has_required_params = _has_required_parameters(func)
            is_decl_pop_hook = _is_declarative_population_hook(func)
            is_custom_or_selector = (
                custom
                or selectors is not None
                or (has_required_params and not is_decl_pop_hook)
            )

            if is_custom_or_selector:
                # ---- Selector mode ----
                if selectors is not None:
                    desc = compile_selector_hook(
                        func,
                        pop,
                        actual_event,
                        selectors,
                        priority,
                        deme_selector=actual_deme_selector,
                        mode=mode,
                    )
                else:
                    # ---- Custom hook (njit or Python fallback) ----
                    if is_njit_function(func):
                        # Already decorated with @njit — use directly.
                        desc = CompiledHookDescriptor(
                            name=func.__name__,
                            event=actual_event,
                            priority=priority,
                            deme_selector=actual_deme_selector,
                            njit_fn=func,
                            meta={
                                "n_ztypes": pop.index_registry.n_ztypes,
                                "n_ages": pop.config.n_ages,
                            },
                        )
                    else:
                        # Not @njit-decorated yet.  Wrap with njit_switch so
                        # the function can run in Numba's nopython mode.
                        # If Numba is disabled, njit_switch returns a Python
                        # callable — we detect this and use py_wrapper instead.
                        try:
                            decorated_func = njit_switch(cache=False)(func)
                            if NUMBA_ENABLED and is_njit_function(decorated_func):
                                norm_fn = _normalize_njit_fn(decorated_func)
                                desc = CompiledHookDescriptor(
                                    name=func.__name__,
                                    event=actual_event,
                                    priority=priority,
                                    deme_selector=actual_deme_selector,
                                    njit_fn=norm_fn,
                                    meta={
                                        "n_ztypes": pop.index_registry.n_ztypes,
                                        "n_ages": pop.config.n_ages,
                                    },
                                )
                            else:
                                # Numba disabled — use Python wrapper.
                                wrapped_func = _normalize_py_hook(func)
                                desc = CompiledHookDescriptor(
                                    name=func.__name__,
                                    event=actual_event,
                                    priority=priority,
                                    deme_selector=actual_deme_selector,
                                    njit_fn=None,
                                    py_wrapper=wrapped_func,
                                    meta={
                                        "n_ztypes": pop.index_registry.n_ztypes,
                                        "n_ages": pop.config.n_ages,
                                    },
                                )
                        except Exception:
                            # Fall back to Python wrapper.
                            wrapped_func = _normalize_py_hook(func)
                            desc = CompiledHookDescriptor(
                                name=func.__name__,
                                event=actual_event,
                                priority=priority,
                                deme_selector=actual_deme_selector,
                                njit_fn=None,
                                py_wrapper=wrapped_func,
                                meta={
                                    "n_ztypes": pop.index_registry.n_ztypes,
                                    "n_ages": pop.config.n_ages,
                                },
                            )
            elif is_decl_pop_hook:
                # Legacy single-parameter population hook (Python only).
                if NUMBA_ENABLED:
                    raise TypeError(
                        f"Python hook '{func.__name__}' is not allowed "
                        "when Numba is enabled.  Please convert it to "
                        "@njit or use declarative Op hooks."
                    )
                desc = CompiledHookDescriptor(
                    name=func.__name__,
                    event=actual_event,
                    priority=priority,
                    deme_selector=actual_deme_selector,
                    py_wrapper=func,
                    meta={
                        "n_ztypes": pop.index_registry.n_ztypes,
                        "n_ages": pop.config.n_ages,
                    },
                )
            else:
                # ---- Declarative hook (returns List[HookOp]) ----
                # The function is called ONCE at registration time.  Its
                # return value (a list of HookOp objects) is compiled into
                # a CSR plan.  The function itself is NOT stored or called
                # at runtime — only the compiled plan is.
                result = func()
                if isinstance(result, list):
                    result_ops = cast(List[object], result)
                    if not all(isinstance(op, HookOp) for op in result_ops):
                        raise TypeError(
                            f"Declarative hook '{func.__name__}' must "
                            "return List[HookOp], or use custom=True "
                            "for custom mode."
                        )
                    ops = cast(List[HookOp], result_ops)
                    desc = compile_declarative_hook(
                        ops,
                        pop,
                        actual_event,
                        priority,
                        deme_selector=actual_deme_selector,
                        name=func.__name__,
                    )
                else:
                    raise TypeError(
                        f"Hook '{func.__name__}' must return List[HookOp] "
                        "for declarative mode, or use custom=True for "
                        "custom mode."
                    )

            hook_func.compiled = desc  # type: ignore  # set on DecoratedHookFn proxy
            pop.register_compiled_hook(desc)
            return desc

        hook_func.register = register  # type: ignore  # set on DecoratedHookFn proxy
        return hook_func

    return decorator

compile_selector_hook 

compile_selector_hook(func: Callable[..., Any], pop: BasePopulation[Any], event: str, selectors_spec: Dict[str, SelectorSpec], priority: int = 0, deme_selector: DemeSelector = '*', mode: str = 'auto') -> CompiledHookDescriptor

Compile selector hook into njit or python descriptor.

resolved stores canonical selector arrays and is reused by both execution paths.

For selector hooks, Numba compilation depends on: - If function is @njit decorated, use it directly - Otherwise, use global NUMBA_ENABLED setting (auto-wrap if enabled)

Source code in src/natal/hooks/entry/selector.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def compile_selector_hook(
    func: Callable[..., Any],
    pop: BasePopulation[Any],
    event: str,
    selectors_spec: Dict[str, SelectorSpec],
    priority: int = 0,
    deme_selector: DemeSelector = "*",
    mode: str = "auto",
) -> CompiledHookDescriptor:
    """Compile selector hook into njit or python descriptor.

    ``resolved`` stores canonical selector arrays and is reused by both
    execution paths.

    For selector hooks, Numba compilation depends on:
    - If function is @njit decorated, use it directly
    - Otherwise, use global NUMBA_ENABLED setting (auto-wrap if enabled)
    """
    index_registry = pop.registry
    species = pop.species

    resolved = {
        name: _resolve_selector_to_array(spec, index_registry, species)
        for name, spec in selectors_spec.items()
    }

    meta = {
        "n_ztypes": index_registry.n_ztypes,
        "n_ages": pop.config.n_ages,
    }

    from ...numba.utils import NUMBA_ENABLED

    is_njit_fn = is_numba_dispatcher(func)

    # Detect parameter layout based on *mode*.
    py_func = getattr(func, "py_func", func)
    sig = inspect.signature(py_func)
    pos_params = [p for p in sig.parameters.values() if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)]
    selector_key_set = set(resolved.keys())

    extra_params = [p for p in pos_params[2:] if p.name != "deme_id"]
    has_deme_id = any(p.name == "deme_id" for p in pos_params[2:])

    if mode == "aggregate":
        use_namedtuple = True
        nt_param_name = extra_params[0].name if extra_params else "sel"
    elif mode == "expand":
        use_namedtuple = False
        nt_param_name = "sel"
    else:  # "auto"
        use_namedtuple = (
            len(extra_params) == 1
            and extra_params[0].name not in selector_key_set
        )
        nt_param_name = extra_params[0].name if use_namedtuple else "sel"

    # Determine state type for the generated wrapper's type annotation.
    from natal.population.discrete_generation import DiscreteGenerationPopulation
    if isinstance(pop, DiscreteGenerationPopulation):
        state_type_name = "DiscretePopulationState"
    else:
        state_type_name = "PopulationState"

    if is_njit_fn or NUMBA_ENABLED:
        njit_fn = _compile_selector_njit_wrapper(
            func, resolved, has_deme_id, state_type_name, use_namedtuple, nt_param_name,
        )
        return CompiledHookDescriptor(
            name=func.__name__,
            event=event,
            priority=priority,
            deme_selector=deme_selector,
            selectors=resolved,
            meta=meta,
            njit_fn=njit_fn,
        )

    # Python path — generate a 3-param wrapper (state, config, deme_id)
    # matching the current hook calling convention.  HookExecutor.execute_event
    # detects the parameter count and dispatches accordingly.
    if use_namedtuple:
        _Sel = _build_namedtuple_class(list(resolved.keys()))
        runtime_values = _build_selector_njit_runtime_values(resolved)

        def py_wrapper(state: PopulationState, config: PopulationConfig | None = None, deme_id: int = -1) -> None:
            """Python-thunk selector hook: pack selectors into a namedtuple and call user function."""  # noqa: D400
            nt = _Sel(**runtime_values)
            # Pass namedtuple as keyword arg so user's parameter name
            # (nt_param_name) matches the kwarg.
            if has_deme_id:
                func(state, config, deme_id, **{nt_param_name: nt})
            else:
                func(state, config, **{nt_param_name: nt})
    else:
        def py_wrapper(state: PopulationState, config: PopulationConfig | None = None, deme_id: int = -1) -> None:
            """Python-thunk selector hook: build kwargs dict and call user function."""  # noqa: D400
            kwargs = _build_selector_python_kwargs(resolved)
            if has_deme_id:
                func(state, config, deme_id, **kwargs)
            else:
                func(state, config, **kwargs)

    return CompiledHookDescriptor(
        name=func.__name__,
        event=event,
        priority=priority,
        deme_selector=deme_selector,
        selectors=resolved,
        meta=meta,
        py_wrapper=py_wrapper,
    )

build_hook_program 

build_hook_program(program: HookProgram) -> HookProgram

Return program unchanged (forward-compat hook point).

Exists as a hook for potential schema upgrades or validation logic. Currently a no-op.

Source code in src/natal/hooks/runtime/csr_kernel.py 
689
690
691
692
693
694
695
def build_hook_program(program: HookProgram) -> HookProgram:
    """Return *program* unchanged (forward-compat hook point).

    Exists as a hook for potential schema upgrades or validation logic.
    Currently a no-op.
    """
    return program

deme_selector_matches 

deme_selector_matches(selector: DemeSelector, deme_id: int) -> bool

Return whether deme_id should execute under selector (Python path).

Supported forms: "*" (wildcard), int (single deme), or list / tuple / range (set of demes).

Source code in src/natal/hooks/runtime/csr_kernel.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def deme_selector_matches(selector: DemeSelector, deme_id: int) -> bool:
    """Return whether *deme_id* should execute under *selector* (Python path).

    Supported forms: ``"*"`` (wildcard), ``int`` (single deme), or
    ``list`` / ``tuple`` / ``range`` (set of demes).
    """
    if selector == "*":
        return True
    if isinstance(selector, int):
        return selector == deme_id
    if isinstance(selector, range):
        return deme_id in selector
    return deme_id in selector

execute_csr_event_arrays 

execute_csr_event_arrays(n_events: int | integer[Any], n_hooks: int | integer[Any], hook_offsets: ndarray, n_ops_list: ndarray, op_offsets: ndarray, op_types_data: ndarray, zidx_offsets_data: ndarray, zidx_data: ndarray, age_offsets_data: ndarray, age_data: ndarray, sex_masks_data: ndarray, params_data: ndarray, condition_offsets_data: ndarray, condition_types_data: ndarray, condition_params_data: ndarray, deme_selector_types: ndarray, deme_selector_offsets: ndarray, deme_selector_data: ndarray, event_id: int, individual_count: ndarray, sperm_storage: ndarray, has_sperm_storage: bool, tick: int, stochastic: bool, continuous_sampling: bool, deme_id: int) -> int

Execute all hooks for one event from flattened CSR arrays.

Resolves event_id to a hook range via hook_offsets, then calls _execute_single_csr_hook for each hook. The function signature mirrors HookProgram fields positionally so callers can unpack the NamedTuple directly.

Three-level CSR traversal::

event_id  →  hook_offsets[event_id]  →  hook range
hook_idx  →  op_offsets[hook_idx]    →  op range
op_idx    →  zidx/age/cond offsets   →  cell range

Returns:

Type Description
int

RESULT_CONTINUE (0) — all hooks executed normally.
int

RESULT_STOP (1) — a hook returned STOP.
Source code in src/natal/hooks/runtime/csr_kernel.py 
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
@njit_switch(cache=True)
def execute_csr_event_arrays(
    n_events: int | np.integer[Any],
    n_hooks: int | np.integer[Any],
    hook_offsets: np.ndarray,
    n_ops_list: np.ndarray,  # pyright: ignore[reportUnusedParameter] — positional caller compatibility
    op_offsets: np.ndarray,
    op_types_data: np.ndarray,
    zidx_offsets_data: np.ndarray,
    zidx_data: np.ndarray,
    age_offsets_data: np.ndarray,
    age_data: np.ndarray,
    sex_masks_data: np.ndarray,
    params_data: np.ndarray,
    condition_offsets_data: np.ndarray,
    condition_types_data: np.ndarray,
    condition_params_data: np.ndarray,
    deme_selector_types: np.ndarray,
    deme_selector_offsets: np.ndarray,
    deme_selector_data: np.ndarray,
    event_id: int,
    individual_count: np.ndarray,
    sperm_storage: np.ndarray,
    has_sperm_storage: bool,
    tick: int,
    stochastic: bool,
    continuous_sampling: bool,
    deme_id: int,
) -> int:
    """Execute all hooks for one event from flattened CSR arrays.

    Resolves *event_id* to a hook range via ``hook_offsets``, then calls
    ``_execute_single_csr_hook`` for each hook.  The function signature
    mirrors ``HookProgram`` fields positionally so callers can unpack
    the NamedTuple directly.

    Three-level CSR traversal::

        event_id  →  hook_offsets[event_id]  →  hook range
        hook_idx  →  op_offsets[hook_idx]    →  op range
        op_idx    →  zidx/age/cond offsets   →  cell range

    Returns:
        ``RESULT_CONTINUE`` (0) — all hooks executed normally.
        ``RESULT_STOP`` (1) — a hook returned STOP.
    """
    if event_id < 0 or event_id >= n_events:
        return 0

    # hook_offsets is a prefix-sum: [event_id] is the first hook,
    # [event_id + 1] is one past the last.
    hook_start = hook_offsets[event_id]
    hook_end = hook_offsets[event_id + 1]

    for hook_idx in range(hook_start, hook_end):
        result = _execute_single_csr_hook(
            hook_idx=hook_idx,
            n_hooks=n_hooks,
            op_offsets=op_offsets,
            op_types_data=op_types_data,
            zidx_offsets_data=zidx_offsets_data,
            zidx_data=zidx_data,
            age_offsets_data=age_offsets_data,
            age_data=age_data,
            sex_masks_data=sex_masks_data,
            params_data=params_data,
            condition_offsets_data=condition_offsets_data,
            condition_types_data=condition_types_data,
            condition_params_data=condition_params_data,
            deme_selector_types=deme_selector_types,
            deme_selector_offsets=deme_selector_offsets,
            deme_selector_data=deme_selector_data,
            individual_count=individual_count,
            sperm_storage=sperm_storage,
            has_sperm_storage=has_sperm_storage,
            tick=tick,
            stochastic=stochastic,
            continuous_sampling=continuous_sampling,
            deme_id=deme_id,
        )
        if result != RESULT_CONTINUE:
            return result  # Propagate STOP immediately.

    return RESULT_CONTINUE

execute_csr_event_program 

execute_csr_event_program(program: HookProgram, event_id: int, individual_count: ndarray, tick: int) -> int

Execute one event with deterministic defaults and no sperm storage.

Convenience wrapper for quick tests or simple discrete-generation setups. For production use, prefer execute_csr_event_program_with_state which exposes the full state flags.

Source code in src/natal/hooks/runtime/csr_kernel.py 
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
@njit_switch(cache=True)
def execute_csr_event_program(
    program: HookProgram,
    event_id: int,
    individual_count: np.ndarray,
    tick: int,
) -> int:
    """Execute one event with deterministic defaults and no sperm storage.

    Convenience wrapper for quick tests or simple discrete-generation
    setups.  For production use, prefer ``execute_csr_event_program_with_state``
    which exposes the full state flags.
    """
    dummy_sperm = np.zeros((0, 0, 0), dtype=np.float64)
    return execute_csr_event_program_with_state(
        program,
        event_id,
        individual_count,
        dummy_sperm,
        tick,
        stochastic=False,
        has_sperm_storage=False,
        continuous_sampling=False,
        deme_id=0,
    )

execute_csr_event_program_with_state 

execute_csr_event_program_with_state(program: HookProgram, event_id: int, individual_count: ndarray, sperm_storage: ndarray, tick: int, stochastic: bool, has_sperm_storage: bool, continuous_sampling: bool, deme_id: int = 0) -> int

Execute one event from a HookProgram, unpacking all fields.

Primary adapter between the HookProgram NamedTuple and the flat-array interface of execute_csr_event_arrays. Lifecycle templates call this function directly.

Parameters:

Name Type Description Default
program HookProgram

Compiled HookProgram containing all declarative ops.

 required
event_id int

Which event to execute (EVENT_FIRST=0, EVENT_EARLY=1, …).

 required
individual_count ndarray

3-D array [sex, age, genotype], mutated in-place.

 required
sperm_storage ndarray

3-D array [age, genotype, gamete_male]. Pass a dummy (0,0,0) array for discrete-generation models.

 required
tick int

Current simulation tick (used for when clause evaluation).

 required
stochastic bool

Whether to use stochastic survival sampling.

 required
has_sperm_storage bool

Whether sperm_storage contains real data.

 required
continuous_sampling bool

Whether to use continuous-Dirichlet sampling.

 required
deme_id int

Deme index for spatial models (0 for panmictic).

 0

Returns:

Type Description
int

RESULT_CONTINUE or RESULT_STOP.
Source code in src/natal/hooks/runtime/csr_kernel.py 
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
@njit_switch(cache=True)
def execute_csr_event_program_with_state(
    program: HookProgram,
    event_id: int,
    individual_count: np.ndarray,
    sperm_storage: np.ndarray,
    tick: int,
    stochastic: bool,
    has_sperm_storage: bool,
    continuous_sampling: bool,
    deme_id: int = 0,
) -> int:
    """Execute one event from a ``HookProgram``, unpacking all fields.

    Primary adapter between the HookProgram NamedTuple and the flat-array
    interface of ``execute_csr_event_arrays``.  Lifecycle templates call
    this function directly.

    Args:
        program: Compiled HookProgram containing all declarative ops.
        event_id: Which event to execute (EVENT_FIRST=0, EVENT_EARLY=1, …).
        individual_count: 3-D array ``[sex, age, genotype]``, mutated in-place.
        sperm_storage: 3-D array ``[age, genotype, gamete_male]``.  Pass a
            dummy ``(0,0,0)`` array for discrete-generation models.
        tick: Current simulation tick (used for ``when`` clause evaluation).
        stochastic: Whether to use stochastic survival sampling.
        has_sperm_storage: Whether *sperm_storage* contains real data.
        continuous_sampling: Whether to use continuous-Dirichlet sampling.
        deme_id: Deme index for spatial models (0 for panmictic).

    Returns:
        ``RESULT_CONTINUE`` or ``RESULT_STOP``.
    """
    return execute_csr_event_arrays(
        n_events=program.n_events,
        n_hooks=program.n_hooks,
        hook_offsets=program.hook_offsets,
        n_ops_list=program.n_ops_list,
        op_offsets=program.op_offsets,
        op_types_data=program.op_types_data,
        zidx_offsets_data=program.zidx_offsets_data,
        zidx_data=program.zidx_data,
        age_offsets_data=program.age_offsets_data,
        age_data=program.age_data,
        sex_masks_data=program.sex_masks_data,
        params_data=program.params_data,
        condition_offsets_data=program.condition_offsets_data,
        condition_types_data=program.condition_types_data,
        condition_params_data=program.condition_params_data,
        deme_selector_types=program.deme_selector_types,
        deme_selector_offsets=program.deme_selector_offsets,
        deme_selector_data=program.deme_selector_data,
        event_id=event_id,
        individual_count=individual_count,
        sperm_storage=sperm_storage,
        has_sperm_storage=has_sperm_storage,
        tick=tick,
        stochastic=stochastic,
        continuous_sampling=continuous_sampling,
        deme_id=deme_id,
    )