Skip to content

Abcs

abcs

Abstract base classes for flepimop2 modules.

This module provides abstract base classes (ABCs) for key modules of the flepimop2 pipeline. The ABCs defined here can also be found in their respective submodules, but are re-exported here for developer convenience.

BackendABC

Bases: ModuleBase

Abstract base class for flepimop2 file IO backends.

read(run_meta)

Read a numpy array from storage.

Parameters:

Name Type Description Default
run_meta RunMeta

Metadata about the current run.

 required

Returns:

Type Description
Float64NDArray

The numpy array read from storage.
Source code in src/flepimop2/backend/abc/__init__.py 
42
43
44
45
46
47
48
49
50
51
52
def read(self, run_meta: RunMeta) -> Float64NDArray:
    """
    Read a numpy array from storage.

    Args:
        run_meta: Metadata about the current run.

    Returns:
        The numpy array read from storage.
    """
    return self._read(run_meta)

save(data, run_meta)

Save a numpy array to storage.

Parameters:

Name Type Description Default
data Float64NDArray

The numpy array to save.

 required
run_meta RunMeta

Metadata about the current run.

 required
Source code in src/flepimop2/backend/abc/__init__.py 
32
33
34
35
36
37
38
39
40
def save(self, data: Float64NDArray, run_meta: RunMeta) -> None:
    """
    Save a numpy array to storage.

    Args:
        data: The numpy array to save.
        run_meta: Metadata about the current run.
    """
    return self._save(data, run_meta)

EngineABC

Bases: ModuleBase

Abstract class for Engines to evolve Dynamic Systems.

model_post_init(__context)

Set _runner to the no-op sentinel if not already overridden.

Concrete subclasses should call super().model_post_init(__context) first, then assign self._runner to their runner function.

Parameters:

Name Type Description Default
__context Any

Pydantic model post-init context.

 required
Source code in src/flepimop2/engine/abc/__init__.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def model_post_init(self, __context: Any, /) -> None:  # noqa: ANN401
    """
    Set `_runner` to the no-op sentinel if not already overridden.

    Concrete subclasses should call `super().model_post_init(__context)`
    first, then assign `self._runner` to their runner function.

    Args:
        __context: Pydantic model post-init context.

    """
    super().model_post_init(__context)
    if self._runner is None:
        self._runner = _no_run_func

run(system, eval_times, initial_state, params, model_state=None, **kwargs)

Run the engine with the provided system and parameters.

Parameters:

Name Type Description Default
system SystemABC

The dynamic system to be evolved.

 required
eval_times Float64NDArray

Array of time points for evaluation.

 required
initial_state dict[IdentifierString, ParameterValue]

Structured initial-state entries sampled from parameters.

 required
params Mapping[IdentifierString, ParameterValue]

Additional parameters for the stepper.

 required
model_state ModelStateSpecification | None

Specification describing the semantic ordering of the state entries.

 None
**kwargs Any

Additional keyword arguments for the engine.

 {}

Returns:

Type Description
Float64NDArray

The evolved time x state array.
Source code in src/flepimop2/engine/abc/__init__.py 
 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
def run(
    self,
    system: SystemABC,
    eval_times: Float64NDArray,
    initial_state: dict[IdentifierString, ParameterValue],
    params: Mapping[IdentifierString, ParameterValue],
    model_state: ModelStateSpecification | None = None,
    **kwargs: Any,
) -> Float64NDArray:
    """
    Run the engine with the provided system and parameters.

    Args:
        system: The dynamic system to be evolved.
        eval_times: Array of time points for evaluation.
        initial_state: Structured initial-state entries sampled from
            parameters.
        params: Additional parameters for the stepper.
        model_state: Specification describing the semantic ordering of the
            state entries.
        **kwargs: Additional keyword arguments for the engine.

    Returns:
        The evolved time x state array.
    """
    return self._runner(  # type: ignore[no-any-return]
        system.bind(),
        eval_times,
        initial_state,
        params,
        model_state=model_state,
        **kwargs,
    )

validate_system(system)

Validation hook for system properties.

Parameters:

Name Type Description Default
system SystemABC

The system to validate.

 required

Returns:

Type Description
list[ValidationIssue] | None

A list of validation issues, or None if not implemented.
Source code in src/flepimop2/engine/abc/__init__.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def validate_system(  # noqa: PLR6301
    self,
    system: SystemABC,  # noqa: ARG002
) -> list[ValidationIssue] | None:
    """
    Validation hook for system properties.

    Args:
        system: The system to validate.

    Returns:
        A list of validation issues, or `None` if not implemented.
    """
    return None

EngineProtocol

Bases: Protocol

Type-definition (Protocol) for engine runner functions.

__call__(stepper, times, initial_state, params, model_state=None, **kwargs)

Protocol for engine runner functions.

Source code in src/flepimop2/engine/abc/__init__.py 
41
42
43
44
45
46
47
48
49
50
51
def __call__(
    self,
    stepper: SystemProtocol,
    times: Float64NDArray,
    initial_state: dict[IdentifierString, ParameterValue],
    params: Mapping[IdentifierString, ParameterValue],
    model_state: ModelStateSpecification | None = None,
    **kwargs: Any,
) -> Float64NDArray:
    """Protocol for engine runner functions."""
    ...

ParameterABC

Bases: ModuleBase

Abstract base class for parameter modules.

Notes

Concrete parameter implementations should use request to decide how to materialize their output for a particular system. For example, a fixed scalar parameter may broadcast to a requested age axis, while a data-backed parameter may validate that its loaded data already matches the requested shape.

sample(*, axes=None, request=None) abstractmethod

Sample a value from the parameter.

Parameters:

Name Type Description Default
axes AxisCollection | None

Resolved runtime axes available for this simulation.

 None
request ParameterRequest | None

Optional system-declared request describing the expected shape and advisory type for the parameter.

 None

Returns:

Type Description
ParameterValue

A sampled parameter value with resolved shape metadata.
Source code in src/flepimop2/parameter/abc/__init__.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
@abstractmethod
def sample(
    self,
    *,
    axes: AxisCollection | None = None,
    request: ParameterRequest | None = None,
) -> ParameterValue:
    """
    Sample a value from the parameter.

    Args:
        axes: Resolved runtime axes available for this simulation.
        request: Optional system-declared request describing the expected shape
            and advisory type for the parameter.

    Returns:
        A sampled parameter value with resolved shape metadata.
    """
    raise NotImplementedError

ProcessABC

Bases: ModuleBase

Abstract base class for flepimop2 processing steps.

execute(*, dry_run=False)

Execute a processing step.

Parameters:

Name Type Description Default
dry_run bool

If True, the process will not actually execute but will simulate execution.

 False

Raises:

Type Description
Flepimop2ValidationError

If validation fails during a dry run.
Source code in src/flepimop2/process/abc/__init__.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def execute(self, *, dry_run: bool = False) -> None:
    """
    Execute a processing step.

    Args:
        dry_run: If True, the process will not actually execute but will simulate
            execution.

    Raises:
        Flepimop2ValidationError: If validation fails during a dry run.
    """
    if dry_run and (result := self._process_validate()) is not None:
        if result:
            raise Flepimop2ValidationError(result)
        return None
    return self._process(dry_run=dry_run)

SystemABC

Bases: ModuleBase

Abstract class for Dynamic Systems.

Attributes:

Name Type Description
module str

The fully-qualified module name for the system.
state_change StateChangeEnum

The type of state change.
options dict[str, Any] | None

Optional dictionary of additional options the system exposes for flepimop2 to take advantage of.

bind(params=None, **kwargs)

Create a particular SystemProtocol.

bind() translates a generic model specification into a particular realization. This can include statically defining parameters, but can also be called with no arguments to simply get the most flexible SystemProtocol available.

Parameters:

Name Type Description Default
params dict[IdentifierString, Any] | None

A dictionary of parameters to statically define for the system.

 None
**kwargs Any

Additional parameters to statically define for the system.

 {}

Returns:

Type Description
SystemProtocol

A stepper function for this system with static parameters defined.

Raises:

Type Description
TypeError

If params contains "time" or "state" keys, or parameters not in the System definition, or if the parameter values are incompatible with System definition.
Source code in src/flepimop2/system/abc/__init__.py 
 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
def bind(
    self,
    params: dict[IdentifierString, Any] | None = None,
    **kwargs: Any,
) -> SystemProtocol:
    """
    Create a particular SystemProtocol.

    `bind()` translates a generic model specification into a particular
    realization. This can include statically defining parameters, but can
    also be called with no arguments to simply get the most flexible
    SystemProtocol available.

    Args:
        params: A dictionary of parameters to statically define for the system.
        **kwargs: Additional parameters to statically define for the system.

    Returns:
        A stepper function for this system with static parameters defined.

    Raises:
        TypeError: If params contains "time" or "state" keys,
            or parameters not in the System definition,
            or if the parameter values are incompatible with System definition.
    """  # noqa: DOC502
    checked_pars = _consolidate_args(
        forbidden={"time", "state"}, params=params, **kwargs
    )
    return self._bind_impl(params=checked_pars)

model_state(axes)

Return metadata describing how parameters assemble the model state.

Parameters:

Name Type Description Default
axes AxisCollection

Resolved runtime axes for the active simulation.

 required

Returns:

Type Description
ModelStateSpecification | None

The runtime model-state specification, if defined.
Notes

Override this in systems whose evolving state is assembled from configured parameter entries. For an age-by-region SEIR system, this could return parameter_names=("S0", "E0", "I0", "R0") with axes=("region", "age"), allowing an engine to stack those entries into an array shaped (4, n_region, n_age) or to keep them in dictionary form if that is more natural for the engine.

Source code in src/flepimop2/system/abc/__init__.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def model_state(  # noqa: PLR6301
    self,
    axes: AxisCollection,  # noqa: ARG002
) -> ModelStateSpecification | None:
    """
    Return metadata describing how parameters assemble the model state.

    Args:
        axes: Resolved runtime axes for the active simulation.

    Returns:
        The runtime model-state specification, if defined.

    Notes:
        Override this in systems whose evolving state is assembled from configured
        parameter entries. For an age-by-region SEIR system, this could return
        `parameter_names=("S0", "E0", "I0", "R0")` with
        `axes=("region", "age")`, allowing an engine to stack those entries into
        an array shaped `(4, n_region, n_age)` or to keep them in dictionary form
        if that is more natural for the engine.
    """
    return None

requested_parameters(axes)

Infer parameter requests from the unbound stepper signature by default.

Parameters:

Name Type Description Default
axes AxisCollection

Resolved runtime axes for the active simulation.

 required

Returns:

Type Description
dict[IdentifierString, ParameterRequest]

A mapping of parameter names to runtime parameter requests.
Notes

Parameters with default values are treated as optional scalar inputs. Subclasses are encouraged to override this method when they need broadcasting behavior or named-axis shapes that cannot be recovered from plain Python annotations.

Source code in src/flepimop2/system/abc/__init__.py 
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
def requested_parameters(
    self,
    axes: AxisCollection,  # noqa: ARG002
) -> dict[IdentifierString, ParameterRequest]:
    """
    Infer parameter requests from the unbound stepper signature by default.

    Args:
        axes: Resolved runtime axes for the active simulation.

    Returns:
        A mapping of parameter names to runtime parameter requests.

    Notes:
        Parameters with default values are treated as optional scalar inputs.
        Subclasses are encouraged to override this method when they need
        broadcasting behavior or named-axis shapes that cannot be recovered
        from plain Python annotations.
    """
    requests: dict[IdentifierString, ParameterRequest] = {}
    signature = inspect.signature(self.bind())
    for name, parameter in signature.parameters.items():
        if name in {"time", "state"}:
            continue
        if parameter.kind in {
            inspect.Parameter.VAR_KEYWORD,
            inspect.Parameter.VAR_POSITIONAL,
        }:
            continue
        requests[name] = ParameterRequest(
            name=name,
            optional=parameter.default is not inspect.Parameter.empty,
        )
    return requests

step(time, state, **params)

Perform a single step of the system's dynamics.

Details

This method is only intended to be used for troubleshooting. Calling this method simply routes to bind() and then invokes the resulting SystemProtocol with the provided arguments.

Parameters:

Name Type Description Default
time float64

The current time.

 required
state Float64NDArray

The current state array.

 required
**params ParameterValue

Additional parameters for the stepper.

 {}

Returns:

Type Description
Float64NDArray

The next state array after one step.
Source code in src/flepimop2/system/abc/__init__.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def step(
    self, time: np.float64, state: Float64NDArray, **params: ParameterValue
) -> Float64NDArray:
    """
    Perform a single step of the system's dynamics.

    Details:
        This method is only intended to be used for troubleshooting. Calling
        this method simply routes to `bind()` and then invokes the resulting
        SystemProtocol with the provided arguments.

    Args:
        time: The current time.
        state: The current state array.
        **params: Additional parameters for the stepper.

    Returns:
        The next state array after one step.
    """
    return self.bind()(time, state, **params)

SystemProtocol

Bases: Protocol

Type-definition (Protocol) for system stepper functions.

__call__(time, state, **kwargs)

Protocol for system stepper functions.

Source code in src/flepimop2/typing.py 
285
286
287
288
289
def __call__(
    self, time: np.float64, state: Float64NDArray, **kwargs: Any
) -> Float64NDArray:
    """Protocol for system stepper functions."""
    ...