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

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 
43
44
45
46
47
48
49
50
51
52
53
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 
33
34
35
36
37
38
39
40
41
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(*args, **kwargs)

Bases: ModuleABC

Abstract class for Engines to evolve Dynamic Systems.

Initialize the EngineABC.

The default initialization sets the runner to a no-op function. Concrete implementations should override this with a valid runner function.

Parameters:

Name Type Description Default
*args Any

Positional arguments.

 ()
**kwargs Any

Keyword arguments.

 {}
Source code in src/flepimop2/engine/abc/__init__.py 
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ARG002
    """
    Initialize the EngineABC.

    The default initialization sets the runner to a no-op function. Concrete
    implementations should override this with a valid runner function.

    Args:
        *args: Positional arguments.
        **kwargs: Keyword arguments.
    """
    self._runner = _no_run_func

run(system, eval_times, initial_state, params, **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 Float64NDArray

The initial state array.

 required
params dict[IdentifierString, Any]

Additional parameters for the stepper.

 required
**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 
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
def run(
    self,
    system: SystemABC,
    eval_times: Float64NDArray,
    initial_state: Float64NDArray,
    params: dict[IdentifierString, Any],
    **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: The initial state array.
        params: Additional parameters for the stepper.
        **kwargs: Additional keyword arguments for the engine.

    Returns:
        The evolved time x state array.
    """
    return self._runner(
        system.bind(),
        eval_times,
        initial_state,
        params,
        **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 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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, state, params, **kwargs)

Protocol for engine runner functions.

Source code in src/flepimop2/typing.py 
215
216
217
218
219
220
221
222
223
224
def __call__(
    self,
    stepper: SystemProtocol,
    times: Float64NDArray,
    state: Float64NDArray,
    params: dict[IdentifierString, Any],
    **kwargs: Any,
) -> Float64NDArray:
    """Protocol for engine runner functions."""
    ...

ParameterABC

Bases: ModuleABC

Abstract base class for parameters.

sample() abstractmethod

Sample a value from the parameter.

Returns:

Type Description
Float64NDArray

A sampled value from the parameter.
Source code in src/flepimop2/parameter/abc/__init__.py 
32
33
34
35
36
37
38
39
@abstractmethod
def sample(self) -> Float64NDArray:
    """Sample a value from the parameter.

    Returns:
        A sampled value from the parameter.
    """
    raise NotImplementedError

ProcessABC

Bases: ModuleABC

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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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: 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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
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 
 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 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 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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 
186
187
188
189
190
def __call__(
    self, time: np.float64, state: Float64NDArray, **kwargs: Any
) -> Float64NDArray:
    """Protocol for system stepper functions."""
    ...