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

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

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

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

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._stepper,  # noqa: SLF001
        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

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/engine/abc/__init__.py

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

@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

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(*args, **kwargs)

Bases: ModuleABC

Abstract class for Dynamic Systems.

Attributes:

Name	Type	Description
module	str	The 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.

Initialize the SystemABC.

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

Parameters:

Name	Type	Description	Default
*args	Any	Positional arguments.	()
**kwargs	Any	Keyword arguments.	{}

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

def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ARG002
    """
    Initialize the SystemABC.

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

    Args:
        *args: Positional arguments.
        **kwargs: Keyword arguments.
    """
    self._stepper = _no_step_function

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

step(time, state, **params)

Perform a single step of the system's dynamics.

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.

    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._stepper(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/system/abc/__init__.py

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