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: ModuleABC

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.

__init_subclass__(**kwargs)

Ensure concrete subclasses define a valid state change type.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Additional keyword arguments passed to parent classes.	{}

Raises:

Type	Description
TypeError	If a concrete subclass does not define state_change.

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

def __init_subclass__(cls, **kwargs: Any) -> None:
    """
    Ensure concrete subclasses define a valid state change type.

    Args:
        **kwargs: Additional keyword arguments passed to parent classes.

    Raises:
        TypeError: If a concrete subclass does not define `state_change`.

    """
    super().__init_subclass__(**kwargs)
    if inspect.isabstract(cls):
        return
    annotations = inspect.get_annotations(cls)
    has_state_change = (
        "state_change" in cls.__dict__ or "state_change" in annotations
    )
    if not has_state_change:
        msg = (
            f"Concrete class '{cls.__name__}' must define 'state_change' as "
            "a class attribute or type annotation."
        )
        raise TypeError(msg)

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)

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	Any	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: Any
) -> 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] | ModuleModel	Configuration dictionary or a ModuleModel instance.	required

Returns:

Type	Description
SystemABC	The constructed system instance.

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

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

    Args:
        config: Configuration dictionary or a `ModuleModel` instance.

    Returns:
        The constructed system instance.

    """
    return _build(
        config,
        "system",
        "flepimop2.system.wrapper",
        SystemABC,
    )

wrap(stepper, state_change, options=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

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,
) -> 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.

    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,
    )

wrapper

A SystemABC which wraps a user-defined script file.

build(config)

Build a SystemABC from a configuration dictionary.

Parameters:

Name	Type	Description	Default
config	dict[IdentifierString, Any] | ModuleModel	Configuration dictionary or a ModuleModel instance.	required

Returns:

Type	Description
SystemABC	The constructed system instance.

Raises:

Type	Description
AttributeError	If the loaded module does not have a valid 'stepper' function.

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

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

    Args:
        config: Configuration dictionary or a `ModuleModel` instance.

    Returns:
        The constructed system instance.

    Raises:
        AttributeError: If the loaded module does not have a valid 'stepper' function.
    """
    config = _as_dict(config)
    config_options = config.get("options", {})

    if script := config.get("script", config_options.get("script")):
        mod = _load_module(script, "flepimop2.system.wrapped")
        if not _validate_function(mod, "stepper"):
            msg = f"Module at {script} does not have a valid 'stepper' function."
            raise AttributeError(msg)

        stepper = as_system_protocol(mod.stepper)
        state_change: StateChangeEnum = StateChangeEnum(
            config.get("state_change", StateChangeEnum.ERROR)
        )
        config_options["script"] = script
        return system_wrap(
            stepper,
            state_change,
            config_options,
        )

    msg = (
        "Configuration must have a 'script' key either at the top level "
        "or within 'options'."
    )
    raise AttributeError(msg)