ode

DifferentialMode(t, state)

Bases: Generic[X, Input], OdeSystem[X]

Differential mode of a hybrid system.

This class represents a mode of a hybrid system by extending an OdeSystem with a transition function that determines the next mode.

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

transition(i) abstractmethod

Transition to the next mode.

Determine the next mode based on the current input. This method should return the current mode if no transition is needed.

Parameters:

Name	Type	Description	Default
i	Input	Input.	required

Returns:

Type	Description
DifferentialMode[X, Input]	Next mode.

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

@abstractmethod
def transition(
    self,
    i: Input,
) -> "DifferentialMode[X, Input]":
    """Transition to the next mode.

    Determine the next mode based on the current input. This method should
    return the current mode if no transition is needed.

    Args:
        i: Input.

    Returns:
        Next mode.
    """

HybridSystem(initial_mode, inputs, map_to_dataframe)

Bases: IncrementalEnvironment, Generic[X, Input]

Hybrid system environment.

This environment generates samples by simulating a hybrid system. The system is defined by a set of differential modes and a sequence of inputs that determine the transitions between the modes.

Initialize the hybrid system.

Parameters:

Name	Type	Description	Default
initial_mode	DifferentialMode[X, Input]	Initial mode of the system.	required
inputs	Iterator[tuple[float, Input]]	Timeseries of inputs (time, input).	required
map_to_dataframe	Callable[[Sequence[float], Sequence[Input], Sequence[X]], DataFrame]	Function to map times, inputs, and states to a DataFrame.	required

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

def __init__(
    self,
    initial_mode: DifferentialMode[X, Input],
    inputs: Iterator[tuple[float, Input]],
    map_to_dataframe: Callable[
        [Sequence[float], Sequence[Input], Sequence[X]],
        pl.DataFrame,
    ],
) -> None:
    """Initialize the hybrid system.

    Args:
        initial_mode: Initial mode of the system.
        inputs: Timeseries of inputs (time, input).
        map_to_dataframe: Function to map times, inputs, and states to a
            DataFrame.
    """
    super().__init__()
    self.mode = initial_mode
    self.inputs = inputs
    self.map_to_dataframe = map_to_dataframe
    self.last_t = 0.0
    self.data = pl.DataFrame()

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

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]

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