System

abc

Abstract class for Dynamic Systems.

Flepimop2ValidationError(issues)

Bases: Flepimop2Error

Exception raised for validation errors from flepimop2.

This exception's main purpose is to signal that validation has failed within the flepimop2 library and provides detailed information about the validation issues encountered, typically provided by an external provider package. The main benefit of this exception is to encapsulate multiple validation issues into a single error object, making it easier to handle and report validation failures as well as providing a consistent interface for validation errors across different parts of the flepimop2 library.

Attributes:

Name	Type	Description
issues		A list of ValidationIssue instances representing the validation errors.

Examples:

>>> from pprint import pprint
>>> from flepimop2.exceptions import (
...     Flepimop2ValidationError,
...     ValidationIssue,
... )
>>> issues = [
...     ValidationIssue(
...         msg="Model requires undefined parameter 'gamma'.",
...         kind="missing_parameter",
...         ctx={"parameter": "gamma", "transition": "gamma * (S / N)"},
...     ),
...     ValidationIssue(
...         msg="Compartment 'E' is unreachable.",
...         kind="unreachable_compartment",
...         ctx={"compartment": "E"},
...     ),
... ]
>>> exception = Flepimop2ValidationError(issues)
>>> pprint(exception.issues)
[ValidationIssue(msg="Model requires undefined parameter 'gamma'.",
                 kind='missing_parameter',
                 ctx={'parameter': 'gamma', 'transition': 'gamma * (S / N)'}),
 ValidationIssue(msg="Compartment 'E' is unreachable.",
                 kind='unreachable_compartment',
                 ctx={'compartment': 'E'})]
>>> print(exception)
2 validation issues encountered:
- [missing_parameter] Model requires undefined parameter 'gamma'. (parameter=gamma, transition=gamma * (S / N))
- [unreachable_compartment] Compartment 'E' is unreachable. (compartment=E)
>>> raise Flepimop2ValidationError(issues)
Traceback (most recent call last):
    ...
flepimop2.exceptions._flepimop2_validation_error.Flepimop2ValidationError: 2 validation issues encountered:
- [missing_parameter] Model requires undefined parameter 'gamma'. (parameter=gamma, transition=gamma * (S / N))
- [unreachable_compartment] Compartment 'E' is unreachable. (compartment=E)

Initialize the Flepimop2ValidationError.

Parameters:

Name	Type	Description	Default
issues	list[ValidationIssue]	A list of ValidationIssue instances representing the validation errors.	required

Source code in src/flepimop2/exceptions/_flepimop2_validation_error.py

def __init__(self, issues: list[ValidationIssue]) -> None:
    """
    Initialize the Flepimop2ValidationError.

    Args:
        issues: A list of `ValidationIssue` instances representing the validation
            errors.
    """
    self.issues = issues
    message = self._format_issues(issues)
    super().__init__(message)

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

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

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

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

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

def __call__(
    self, time: np.float64, state: Float64NDArray, **kwargs: Any
) -> Float64NDArray:
    """Protocol for system stepper functions."""
    ...

ValidationIssue(msg, kind, ctx=None) dataclass

Represents a single validation issue encountered during data validation.

Attributes:

Name	Type	Description
msg	str	A human readable description of the validation issue.
kind	str	The type/category of the validation issue.
ctx	dict[str, Any] | None	Optional context providing additional information about the issue.

Examples:

>>> from pprint import pprint
>>> from flepimop2.exceptions import ValidationIssue
>>> issue = ValidationIssue(
...     msg="Invalid wrapper data format.",
...     kind="invalid_format",
...     ctx={"expected_format": "JSON", "line": 42},
... )
>>> pprint(issue)
ValidationIssue(msg='Invalid wrapper data format.',
                kind='invalid_format',
                ctx={'expected_format': 'JSON', 'line': 42})

build(config)

Build a SystemABC from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict[IdentifierString, Any] | ModuleBase | str	Configuration dictionary or a ModuleBase instance. The configuration must define a module.	required

Returns:

Type	Description
SystemABC	The constructed system instance.

Source code in src/flepimop2/system/abc/__init__.py

def build(config: dict[IdentifierString, Any] | ModuleBase | str) -> SystemABC:
    """
    Build a `SystemABC` from a configuration dictionary.

    Args:
        config: Configuration dictionary or a `ModuleBase` instance. The
            configuration must define a `module`.

    Returns:
        The constructed system instance.

    """
    return _build(config, "system", SystemABC)  # type: ignore[type-abstract]

wrap(stepper, state_change, options=None, *, requested_parameters=None, model_state=None)

Adapt a user-defined function into a SystemABC.

Parameters:

Name	Type	Description	Default
stepper	SystemProtocol	A user-defined function that implements the system's dynamics.	required
state_change	StateChangeEnum	The type of state change for the system.	required
options	dict[IdentifierString, Any] | None	Optional dictionary of additional information about the system.	None
requested_parameters	Callable[[AxisCollection], dict[IdentifierString, ParameterRequest]] | None	Optional callback declaring runtime parameter requests for the system.	None
model_state	Callable[[AxisCollection], ModelStateSpecification | None] | None	Optional callback declaring how parameters assemble the evolving model state.	None

Returns:

Type	Description
SystemABC	A SystemABC instance that wraps the provided stepper function.

Raises:

Type	Description
TypeError	If the provided stepper function does not conform to the expected signature or if offered an erroneous state_change.

Source code in src/flepimop2/system/abc/__init__.py

def wrap(
    stepper: SystemProtocol,
    state_change: StateChangeEnum,
    options: dict[IdentifierString, Any] | None = None,
    *,
    requested_parameters: (
        Callable[[AxisCollection], dict[IdentifierString, ParameterRequest]] | None
    ) = None,
    model_state: (
        Callable[[AxisCollection], ModelStateSpecification | None] | None
    ) = None,
) -> SystemABC:
    """
    Adapt a user-defined function into a `SystemABC`.

    Args:
        stepper: A user-defined function that implements the system's dynamics.
        state_change: The type of state change for the system.
        options: Optional dictionary of additional information about the system.
        requested_parameters: Optional callback declaring runtime parameter requests
            for the system.
        model_state: Optional callback declaring how parameters assemble the evolving
            model state.

    Returns:
        A `SystemABC` instance that wraps the provided stepper function.

    Raises:
        TypeError: If the provided stepper function does not conform to the expected
            signature or if offered an erroneous state_change.
    """
    if not isinstance(state_change, StateChangeEnum):
        msg = (
            "state_change must be an instance of StateChangeEnum; "
            f"got {type(state_change)}."
        )
        raise TypeError(msg)
    if state_change == StateChangeEnum.ERROR:
        msg = "state_change cannot be StateChangeEnum.ERROR for a valid system."
        raise TypeError(msg)

    return _AdapterSystem(
        state_change=state_change,
        stepper=stepper,
        options=options,
        requested_parameters=requested_parameters,
        model_state=model_state,
    )

wrapper

A SystemABC which wraps a user-defined script file.

WrapperSystem

Bases: SystemABC

A SystemABC that wraps a user-defined Python script file.

The script must define a stepper function compatible with SystemProtocol.

Attributes:

Name	Type	Description
script	Path	Path to the Python script containing the stepper function.
state_change	StateChangeEnum	State-change convention declared by the wrapped system.
options	dict[str, Any] | None	Opaque extra options exposed through ModuleBase.option.
requested_parameters	dict[IdentifierString, ParameterRequest]	Optional declarative runtime parameter metadata.
model_state	ModelStateSpecification | None	Optional declarative model-state metadata.

model_state(axes)

Return wrapper model-state metadata from an explicit callback.

Source code in src/flepimop2/system/wrapper/__init__.py

def model_state(self, axes: AxisCollection) -> ModelStateSpecification | None:
    """Return wrapper model-state metadata from an explicit callback."""
    if self._model_state_func is not None:
        return self._model_state_func(axes)  # type: ignore[no-any-return]
    return super().model_state(axes)

requested_parameters(axes)

Return wrapper parameter requests from an explicit callback when provided.

Source code in src/flepimop2/system/wrapper/__init__.py

def requested_parameters(
    self,
    axes: AxisCollection,
) -> dict[IdentifierString, ParameterRequest]:
    """Return wrapper parameter requests from an explicit callback when provided."""
    if self._requested_parameters_func is not None:
        return self._requested_parameters_func(axes)  # type: ignore[no-any-return]
    return super().requested_parameters(axes)