pykoop.KoopmanPipeline
- class KoopmanPipeline(lifting_functions=None, regressor=None)
Bases:
_BaseComposition
,KoopmanLiftingFn
Meta-estimator for chaining lifting functions with an estimator.
- Parameters:
lifting_functions (List[Tuple[str, KoopmanLiftingFn]] | None) –
regressor (KoopmanRegressor | None) –
- liting_functions_
Fit lifting functions (and their names).
- Type:
List[Tuple[str, KoopmanLiftingFn]]
- regressor_
Fit regressor.
- Type:
- feature_names_in_
Array of input feature name strings.
- Type:
np.ndarray
Examples
Apply a basic Koopman pipeline to mass-spring-damper data
>>> kp = pykoop.KoopmanPipeline( ... lifting_functions=[('pl', pykoop.PolynomialLiftingFn(order=2))], ... regressor=pykoop.Edmd(), ... ) >>> kp.fit(X_msd, n_inputs=1, episode_feature=True) KoopmanPipeline(lifting_functions=[('pl', PolynomialLiftingFn(order=2))], regressor=Edmd()) >>> kp.get_feature_names_in().tolist() ['ep', 'x0', 'x1', 'u0'] >>> kp.get_feature_names_out().tolist() ['ep', 'x0', 'x1', 'x0^2', 'x0*x1', 'x1^2', 'u0', 'x0*u0', 'x1*u0', 'u0^2']
Apply more sophisticated Koopman pipeline to mass-spring-damper data
>>> kp = KoopmanPipeline( ... lifting_functions=[ ... ('ma', pykoop.SkLearnLiftingFn( ... sklearn.preprocessing.MaxAbsScaler())), ... ('pl', pykoop.PolynomialLiftingFn(order=2)), ... ('ss', pykoop.SkLearnLiftingFn( ... sklearn.preprocessing.StandardScaler())), ... ], ... regressor=pykoop.Edmd(), ... ) >>> kp.fit(X_msd, n_inputs=1, episode_feature=True) KoopmanPipeline(lifting_functions=[('ma', SkLearnLiftingFn(transformer=MaxAbsScaler())), ('pl', PolynomialLiftingFn(order=2)), ('ss', SkLearnLiftingFn(transformer=StandardScaler()))], regressor=Edmd()) >>> Xt_msd = kp.transform(X_msd[:2, :])
Apply bilinear Koopman pipeline to mass-spring-damper data
>>> kp = KoopmanPipeline( ... lifting_functions=[ ... ('ma', pykoop.SkLearnLiftingFn( ... sklearn.preprocessing.MaxAbsScaler())), ... ('sp', pykoop.SplitPipeline( ... lifting_functions_state=[ ... ('pl', pykoop.PolynomialLiftingFn(order=2)), ... ], ... lifting_functions_input=None, ... )), ... ('bi', pykoop.BilinearInputLiftingFn()), ... ('ss', pykoop.SkLearnLiftingFn( ... sklearn.preprocessing.StandardScaler())), ... ], ... regressor=pykoop.Edmd(), ... ) >>> kp.fit(X_msd, n_inputs=1, episode_feature=True) KoopmanPipeline(lifting_functions=[('ma', SkLearnLiftingFn(transformer=MaxAbsScaler())), ('sp', SplitPipeline(lifting_functions_state=[('pl', PolynomialLiftingFn(order=2))])), ('bi', BilinearInputLiftingFn()), ('ss', SkLearnLiftingFn(transformer=StandardScaler()))], regressor=Edmd())
- __init__(lifting_functions=None, regressor=None)
Instantiate for
KoopmanPipeline
.- Parameters:
lifting_functions (Optional[List[Tuple[str, KoopmanLiftingFn]]]) – List of names and lifting function objects.
regressor (Optional[KoopmanRegressor]) – Koopman regressor.
- Return type:
None
Methods
__init__
([lifting_functions, regressor])Instantiate for
KoopmanPipeline
.fit
(X[, y, n_inputs, episode_feature])Fit the Koopman pipeline.
fit_transform
(X[, y])Fit to data, then transform it.
fit_transformers
(X[, y, n_inputs, ...])Fit only the lifting functions in the pipeline.
frequency_response
(t_step[, f_min, f_max, ...])Compute frequency response of Koopman system.
get_feature_names_in
([format, episode_feature])Automatically generate input feature names.
get_feature_names_out
([input_features, ...])Get output feature names.
Get metadata routing of this object.
get_params
([deep])Get parameters for this estimator.
Invert transformed data.
lift
(X[, episode_feature])Lift state and input.
lift_input
(X[, episode_feature])Lift input only.
lift_state
(X[, episode_feature])Lift state only.
make_scorer
([n_steps, discount_factor, ...])Make a Koopman pipeline scorer.
n_samples_in
([n_samples_out])Calculate number of input samples required for given output length.
plot_bode
(t_step[, f_min, f_max, n_points, ...])Plot frequency response of Koopman system.
plot_eigenvalues
([unit_circle, figure_kw, ...])Plot eigenvalues of Koopman
A
matrix.plot_koopman_matrix
([subplots_kw, plot_kw])Plot heatmap of Koopman matrices.
plot_lifted_trajectory
(X[, episode_feature, ...])Plot lifted data matrix.
plot_predicted_trajectory
(X0_or_X[, U, ...])Plot predicted trajectory.
plot_svd
([subplots_kw, plot_kw])Plot singular values of Koopman matrices.
predict
(X)Perform a single-step prediction for each state in each episode.
Perform a multi-step prediction for the first state of each episode.
predict_trajectory
(X0_or_X[, U, ...])Predict state trajectory given input for each episode.
retract
(X[, episode_feature])Retract lifted state and input.
retract_input
(X[, episode_feature])Retract lifted input only.
retract_state
(X[, episode_feature])Retract lifted state only.
score
(X[, y])Calculate prediction score.
set_fit_request
(*[, episode_feature, n_inputs])Request metadata passed to the
fit
method.set_output
(*[, transform])Set output container.
set_params
(**kwargs)Set the parameters of this estimator.
transform
(X)Transform data.
- fit(X, y=None, n_inputs=0, episode_feature=False)
Fit the Koopman pipeline.
- Parameters:
- Returns:
Instance of itself.
- Return type:
- Raises:
ValueError – If constructor or fit parameters are incorrect.
- fit_transform(X, y=None, **fit_params)
Fit to data, then transform it.
Fits transformer to X and y with optional parameters fit_params and returns a transformed version of X.
- Parameters:
X (array-like of shape (n_samples, n_features)) – Input samples.
y (array-like of shape (n_samples,) or (n_samples, n_outputs), default=None) – Target values (None for unsupervised transformations).
**fit_params (dict) – Additional fit parameters.
- Returns:
X_new – Transformed array.
- Return type:
ndarray array of shape (n_samples, n_features_new)
- fit_transformers(X, y=None, n_inputs=0, episode_feature=False)
Fit only the lifting functions in the pipeline.
- Parameters:
- Returns:
Instance of itself.
- Return type:
- Raises:
ValueError – If constructor or fit parameters are incorrect.
- frequency_response(t_step, f_min=0, f_max=None, n_points=1000, decibels=True)
Compute frequency response of Koopman system.
- Parameters:
- Returns:
Frequency (Hz) and frequency response (gain or dB).
- Return type:
Tuple[np.ndarray, np.ndarray]
- Raises:
ValueError – If
f_min
is less than zero orf_max
is greater than the Nyquist frequency.
- get_feature_names_in(format=None, episode_feature=None)
Automatically generate input feature names.
- Parameters:
- Returns:
Automatically generated input feaure names.
- Return type:
np.ndarray
- get_feature_names_out(input_features=None, symbols_only=False, format=None, episode_feature=None)
Get output feature names.
- Parameters:
input_features (Optional[np.ndarray]) – Array of string input feature names. If provided, they are checked against
feature_names_in_
. IfNone
, ignored.symbols_only (bool) – If true, only return symbols (
theta_0
,upsilon_0
, etc.). Otherwise, returns the full equations (default).format (Optional[str]) – Feature name formatting method. Possible values are
'plaintext'
(default ifNone
) or'latex'
.episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
Output feature names.
- Return type:
np.ndarray
- get_metadata_routing()
Get metadata routing of this object.
Please check User Guide on how the routing mechanism works.
- Returns:
routing – A
MetadataRequest
encapsulating routing information.- Return type:
MetadataRequest
- get_params(deep=True)
Get parameters for this estimator.
- inverse_transform(X)
Invert transformed data.
- Parameters:
X (np.ndarray) – Transformed data matrix.
- Returns:
Inverted transformed data matrix.
- Return type:
np.ndarray
- lift(X, episode_feature=None)
Lift state and input.
Potentially more convenient alternative to calling
transform()
.- Parameters:
X (np.ndarray) – State and input.
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
Lifted state and input.
- Return type:
np.ndarray
- lift_input(X, episode_feature=None)
Lift input only.
More convenient alternative to calling
transform()
, then stripping the unwanted lifted states.- Parameters:
X (np.ndarray) – State and input.
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
Lifted input.
- Return type:
np.ndarray
- lift_state(X, episode_feature=None)
Lift state only.
More convenient alternative to padding the state with dummy inputs, calling
transform()
, then stripping the unwanted lifted inputs.- Parameters:
X (np.ndarray) – State.
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
Lifted state.
- Return type:
np.ndarray
- static make_scorer(n_steps=None, discount_factor=1, regression_metric='neg_mean_squared_error', regression_metric_kw=None, error_score=nan, multistep=True, relift_state=True)
Make a Koopman pipeline scorer.
A
scikit-learn
scorer accepts the parameters(estimator, X, y)
and returns a float representing the prediction quality ofestimator
onX
with reference toy
. Uses existingscikit-learn
regression metrics [1]. Higher numbers are better. Metrics corresponding to losses are negated.Technically, the scorer will predict the entire episode, regardless of how
n_steps
is set. It will then assign a zero weight to all errors beyondn_steps
.- Parameters:
n_steps (Optional[int]) – Number of steps ahead to predict. If
None
or longer than the episode, will score the entire episode.discount_factor (float) – Discount factor used to weight the error timeseries. Should be positive, with magnitude 1 or slightly less. The error at each timestep is weighted by
discount_factor**k
, wherek
is the timestep.regression_metric (Union[str, Callable]) –
Regression metric to use. One of
'explained_variance'
,'neg_mean_absolute_error'
,'neg_mean_squared_error'
,'neg_mean_squared_log_error'
,'neg_median_absolute_error'
,'r2'
, or'neg_mean_absolute_percentage_error'
,
which are existing
scikit-learn
regression metrics [1]. Can also directly specify a function with the same keyword arguments as thescikit-learn
ones. That is, at leasty_true
,y_pred
,sample_weight
, andmultioutput
.regression_metric_kw (Optional[Dict[str, Any]]) – Keyword arguments for
regression_method
. Ifsample_weight
keyword argument is specified,discount_factor
is ignored.error_score (Union[str, float]) – Value to assign to the score if
X_predicted
has diverged or if an error has occured in estimator fitting. If set to'raise'
, aValueError
is raised. If a numerical value is given, asklearn.exceptions.FitFailedWarning
warning is raised and the specified score is returned. The error score defines the worst possible score. If a score is finite but lower than the error score, the error score will be returned instead.multistep (bool) – If true, predict using
predict_trajectory()
. Otherwise, predict usingpredict()
(one-step-ahead prediction). Multistep prediction is highly recommended unless debugging. If one-step-ahead prediciton is used, n_steps and discount_factor are ignored.relift_state (bool) – If true, retract and re-lift state between prediction steps (default). Otherwise, only retract the state after all predictions are made. Correspond to the local and global error definitions of [MAM22]. Ignored if
multistep
is false.
- Returns:
Scorer compatible with
scikit-learn
.- Return type:
Callable[[KoopmanPipeline, np.ndarray, Optional[np.ndarray]], float]
- Raises:
ValueError – If
discount_factor
is negative or greater than one or iferror_score='raise'
and an error occurs in scoring.
References
- n_samples_in(n_samples_out=1)
Calculate number of input samples required for given output length.
- plot_bode(t_step, f_min=0, f_max=None, n_points=1000, decibels=True, subplots_kw=None, plot_kw=None)
Plot frequency response of Koopman system.
- Parameters:
t_step (float) – Sampling timestep.
f_min (float) – Minimum frequency to plot.
f_max (Optional[float]) – Maximum frequency to plot. If
None
, uses the Nyquist frequency.n_points (int) – Number of frequecy points to plot.
decibels (bool) – Plot gain in dB (default is true).
subplots_kw (Optional[Dict[str, Any]]) – Keyword arguments for
plt.subplots()
.plot_kw (Optional[Dict[str, Any]]) – Keyword arguments for Matplotlib
plt.Axes.plot()
.
- Returns:
Matplotlib
plt.Figure
andplt.Axes
objects.- Return type:
Tuple[plt.Figure, plt.Axes]
- Raises:
ValueError – If
f_min
is less than zero orf_max
is greater than the Nyquist frequency.
- plot_eigenvalues(unit_circle=True, figure_kw=None, subplot_kw=None, plot_kw=None)
Plot eigenvalues of Koopman
A
matrix.- Parameters:
- Returns:
Matplotlib
plt.Figure
andplt.Axes
objects.- Return type:
Tuple[plt.Figure, plt.Axes]
- plot_koopman_matrix(subplots_kw=None, plot_kw=None)
Plot heatmap of Koopman matrices.
- plot_lifted_trajectory(X, episode_feature=None, episode_style=None, subplots_kw=None, plot_kw=None)
Plot lifted data matrix.
- Parameters:
X (np.ndarray) – Data matrix.
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.episode_style (Optional[str]) – If
'columns'
, each episode is a column (default). If'overlay'
, states from each episode are plotted overtop of each other in different colors.subplots_kw (Optional[Dict[str, Any]]) – Keyword arguments for
plt.subplots()
.plot_kw (Optional[Dict[str, Any]]) – Keyword arguments for Matplotlib
plt.Axes.plot()
.
- Returns:
Matplotlib
plt.Figure
object and two-dimensional array ofplt.Axes
objects.- Return type:
Tuple[plt.Figure, np.ndarray]
- plot_predicted_trajectory(X0_or_X, U=None, relift_state=True, plot_lifted=False, plot_input=False, plot_error=False, episode_feature=None, plot_ground_truth=True, episode_style=None, subplots_kw=None, plot_kw=None)
Plot predicted trajectory.
- Parameters:
X0_or_X (np.ndarray) – Initial state if
U
is specified. IfU
isNone
, then treated as the ground truth trajectory from which the initial state and full input are extracted.U (Optional[np.ndarray]) – Input. Length of prediction is governed by length of input. If
None
, input is taken from last features ofX0_or_X
.relift_state (bool) – If true, retract and re-lift state between prediction steps (default). Otherwise, only retract the state after all predictions are made. Correspond to the local and global error definitions of [MAM22].
plot_lifted (bool) – If true, plot the lifted state. If false, plot the original state (default).
plot_input (bool) – If true, plot the input as well as the state. If false, plot only the original state (default).
plot_error (bool) – If true, plot the prediction error instead of the state. If false, plot the predicted state and ground truth (default).
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.plot_ground_truth (bool) – Plot contents of
X0_or_X
as ground truth ifU
isNone
. Ignored ifU
is notNone
.episode_style (Optional[str]) – If
'columns'
, each episode is a column (default). If'overlay'
, states from each episode are plotted overtop of each other in different colors.subplots_kw (Optional[Dict[str, Any]]) – Keyword arguments for
plt.subplots()
.plot_kw (Optional[Dict[str, Any]]) – Keyword arguments for Matplotlib
plt.Axes.plot()
.
- Returns:
Matplotlib
plt.Figure
object and two-dimensional array ofplt.Axes
objects.- Return type:
Tuple[plt.Figure, np.ndarray]
- plot_svd(subplots_kw=None, plot_kw=None)
Plot singular values of Koopman matrices.
- predict(X)
Perform a single-step prediction for each state in each episode.
Lifts the state, preforms a single-step prediction in the lifted space, then retracts to the original state space.
- Parameters:
X (np.ndarray) – Data matrix.
- Returns:
Predicted data matrix.
- Return type:
np.ndarray
- predict_multistep(X)
Perform a multi-step prediction for the first state of each episode.
This function takes the first
min_samples_
states of the input, along with all of its inputs, and predicts the nextX.shape[0]
states of the system. This action is performed on a per-episode basis. The state features ofX
(other than the firstmin_samples_
features) are not used at all.If prediction fails numerically, missing predictions are filled with
np.nan
.- Parameters:
X (np.ndarray) – Data matrix.
- Returns:
Predicted data matrix.
- Return type:
np.ndarray
- Raises:
ValueError – If an episode is shorter than
min_samples_
.
Warning
Deprecated in favour of
pykoop.KoopmanPipeline.predict_trajectory()
.
- predict_trajectory(X0_or_X, U=None, relift_state=True, return_lifted=False, return_input=False, episode_feature=None)
Predict state trajectory given input for each episode.
- Parameters:
X0_or_X (np.ndarray) – Initial state if
U
is specified. IfU
isNone
, then treated as the initial state and full input in one matrix, where the remaining states are ignored.U (Optional[np.ndarray]) – Input. Length of prediction is governed by length of input. If
None
, input is taken from last features ofX0_or_X
.relift_state (bool) – If true, retract and re-lift state between prediction steps (default). Otherwise, only retract the state after all predictions are made. Correspond to the local and global error definitions of [MAM22].
return_lifted (bool) – If true, return the lifted state. If false, return the original state (default).
return_input (bool) – If true, return the input as well as the state. If false, return only the original state (default).
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
Predicted state. If
return_input
, input is appended to the array. Ifreturn_lifted
, the predicted state (and input) are returned in the lifted space.- Return type:
np.ndarray
- Raises:
ValueError – If an episode is shorter than
min_samples_
.
Examples
Predict trajectory with one argument
>>> kp = pykoop.KoopmanPipeline( ... lifting_functions=[ ... ('pl', pykoop.PolynomialLiftingFn(order=2)), ... ], ... regressor=pykoop.Edmd(), ... ) >>> kp.fit(X_msd, n_inputs=1, episode_feature=True) KoopmanPipeline(lifting_functions=[('pl', PolynomialLiftingFn(order=2))], regressor=Edmd()) >>> X_pred = kp.predict_trajectory(X_msd)
Predict trajectory with two arguments >>> x0 = pykoop.extract_initial_conditions( … X_msd, … min_samples=kp.min_samples_, … n_inputs=1, … episode_feature=True, … ) >>> u = pykoop.extract_input( … X_msd, … n_inputs=1, … episode_feature=True, … ) >>> X_pred = kp.predict_trajectory(x0, u)
- retract(X, episode_feature=None)
Retract lifted state and input.
Potentially more convenient alternative to calling
inverse_transform()
.- Parameters:
X (np.ndarray) – Lifted state and input.
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
State and input.
- Return type:
np.ndarray
- retract_input(X, episode_feature=None)
Retract lifted input only.
More convenient alternative to padding the lifted state with dummy lifted states, calling
inverse_transform()
, then stripping the unwanted states.- Parameters:
X (np.ndarray) – Lifted input.
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
Input.
- Return type:
np.ndarray
- retract_state(X, episode_feature=None)
Retract lifted state only.
More convenient alternative to padding the lifted state with dummy lifted inputs, calling
inverse_transform()
.- Parameters:
X (np.ndarray) – Lifted state.
episode_feature (Optional[bool]) – True if first feature indicates which episode a timestep is from. If
None
,self.episode_feature_
is used.
- Returns:
State.
- Return type:
np.ndarray
- score(X, y=None)
Calculate prediction score.
For more flexible scoring, see
make_scorer()
.- Parameters:
X (np.ndarray) – Data matrix.
y (Optional[np.ndarray]) – Ignored.
- Returns:
Mean squared error prediction score.
- Return type:
- set_fit_request(*, episode_feature='$UNCHANGED$', n_inputs='$UNCHANGED$')
Request metadata passed to the
fit
method.Note that this method is only relevant if
enable_metadata_routing=True
(seesklearn.set_config()
). Please see User Guide on how the routing mechanism works.The options for each parameter are:
True
: metadata is requested, and passed tofit
if provided. The request is ignored if metadata is not provided.False
: metadata is not requested and the meta-estimator will not pass it tofit
.None
: metadata is not requested, and the meta-estimator will raise an error if the user provides it.str
: metadata should be passed to the meta-estimator with this given alias instead of the original name.
The default (
sklearn.utils.metadata_routing.UNCHANGED
) retains the existing request. This allows you to change the request for some parameters and not others.New in version 1.3.
Note
This method is only relevant if this estimator is used as a sub-estimator of a meta-estimator, e.g. used inside a
Pipeline
. Otherwise it has no effect.- Parameters:
episode_feature (str, True, False, or None, default=sklearn.utils.metadata_routing.UNCHANGED) – Metadata routing for
episode_feature
parameter infit
.n_inputs (str, True, False, or None, default=sklearn.utils.metadata_routing.UNCHANGED) – Metadata routing for
n_inputs
parameter infit
.self (KoopmanPipeline) –
- Returns:
self – The updated object.
- Return type:
- set_output(*, transform=None)
Set output container.
See Introducing the set_output API for an example on how to use the API.
- Parameters:
transform ({"default", "pandas"}, default=None) –
Configure output of transform and fit_transform.
”default”: Default output format of a transformer
”pandas”: DataFrame output
None: Transform configuration is unchanged
- Returns:
self – Estimator instance.
- Return type:
estimator instance
- set_params(**kwargs)
Set the parameters of this estimator.
The method works on simple estimators as well as on nested objects (such as
Pipeline
). The latter have parameters of the form<component>__<parameter>
so that it’s possible to update each component of a nested object.- Parameters:
**params (dict) – Estimator parameters.
- Returns:
self – Estimator instance.
- Return type:
estimator instance
- transform(X)
Transform data.
- Parameters:
X (np.ndarray) – Data matrix.
- Returns:
Transformed data matrix.
- Return type:
np.ndarray