ode_environment

IntegrationError()

Bases: Exception

Error while integrating an ODE.

This exception is raised when an error occurs while integrating an ordinary differential equation.

Initialize the exception.

Source code in src/flowcean/ode/ode_environment.py

def __init__(self) -> None:
    """Initialize the exception."""
    super().__init__("failed to integrate ODE")

OdeState

Bases: ABC

State of a differential equation.

This class represents the state of a differential equation. It provides methods to convert the state to and from a numpy array for integration.

as_numpy() abstractmethod

Convert the state to a numpy array.

Returns:

Type	Description
NDArray[float64]	State as a numpy array.

Source code in src/flowcean/ode/ode_environment.py

@abstractmethod
def as_numpy(self) -> NDArray[np.float64]:
    """Convert the state to a numpy array.

    Returns:
        State as a numpy array.
    """

from_numpy(state) abstractmethod classmethod

Create a state from a numpy array.

Parameters:

Name	Type	Description	Default
state	NDArray[float64]	State as a numpy array.	required

Returns:

Type	Description
Self	State instance.

Source code in src/flowcean/ode/ode_environment.py

@classmethod
@abstractmethod
def from_numpy(cls, state: NDArray[np.float64]) -> Self:
    """Create a state from a numpy array.

    Args:
        state: State as a numpy array.

    Returns:
        State instance.
    """

OdeSystem(t, state)

Bases: ABC, Generic[X]

System governed by an ordinary differential equation.

This class represents a continuous system. The system is defined by a differential flow function \(f\) that governs the evolution of the state \(x\).

\[ \begin{aligned} \dot{x} &= f(t, x) \\ \end{aligned} \]

The system can be integrated to obtain the state at a future time.

Attributes:

Name	Type	Description
t	float	Current time.
state	X	Current state.

Initialize the system.

Parameters:

Name	Type	Description	Default
t	float	Initial time.	required
state	X	Initial state.	required

Source code in src/flowcean/ode/ode_environment.py

def __init__(self, t: float, state: X) -> None:
    """Initialize the system.

    Args:
        t: Initial time.
        state: Initial state.
    """
    self.t = t
    self.state = state

flow(t, state) abstractmethod

Ordinary differential equation.

Compute the derivative of the state \(x\) at time \(t\).

Parameters:

Name	Type	Description	Default
t	float	Time.	required
state	NDArray[float64]	State.	required

Returns:

Type	Description
NDArray[float64]	Derivative of the state.

Source code in src/flowcean/ode/ode_environment.py

@abstractmethod
def flow(
    self,
    t: float,
    state: NDArray[np.float64],
) -> NDArray[np.float64]:
    """Ordinary differential equation.

    Compute the derivative of the state $x$ at time $t$.

    Args:
        t: Time.
        state: State.

    Returns:
        Derivative of the state.
    """

step(dt)

Step the mode forward in time.

Step the mode forward in time by integrating the differential equation for a time step of dt.

Parameters:

Name	Type	Description	Default
dt	float	Time step.	required

Returns:

Type	Description
tuple[Sequence[float], Sequence[X]]	Tuple of times and states of the integration.

Source code in src/flowcean/ode/ode_environment.py

def step(self, dt: float) -> tuple[Sequence[float], Sequence[X]]:
    """Step the mode forward in time.

    Step the mode forward in time by integrating the differential equation
    for a time step of dt.

    Args:
        dt: Time step.

    Returns:
        Tuple of times and states of the integration.
    """
    y0 = self.state.as_numpy()
    solution = solve_ivp(
        self.flow,
        t_span=[self.t, self.t + dt],
        y0=y0,
    )
    if not solution.success:
        raise IntegrationError

    ts = cast(Sequence[float], solution.t[1:])
    states = [self.state.from_numpy(y) for y in solution.y.T[1:]]

    self.t = ts[-1]
    self.state = states[-1]

    return ts, states

OdeEnvironment(system, *, dt=1.0, map_to_dataframe)

Bases: IncrementalEnvironment, Generic[X]

Environment governed by an ordinary differential equation.

This environment integrates an OdeSystem to generate a sequence of states.

Initialize the environment.

Parameters:

Name	Type	Description	Default
system	OdeSystem[X]	ODE system.	required
dt	float	Time step.	1.0
map_to_dataframe	Callable[[Sequence[float], Sequence[X]], DataFrame]	Function to map states to a DataFrame.	required

Source code in src/flowcean/ode/ode_environment.py

def __init__(
    self,
    system: OdeSystem[X],
    *,
    dt: float = 1.0,
    map_to_dataframe: Callable[
        [Sequence[float], Sequence[X]],
        pl.DataFrame,
    ],
) -> None:
    """Initialize the environment.

    Args:
        system: ODE system.
        dt: Time step.
        map_to_dataframe: Function to map states to a DataFrame.
    """
    super().__init__()
    self.system = system
    self.dt = dt
    self.map_to_dataframe = map_to_dataframe
    self.ts = [self.system.t]
    self.states = [self.system.state]