pykoop.KoopmanPipeline
- class KoopmanPipeline(lifting_functions=None, regressor=None)
Bases:
_BaseComposition,KoopmanLiftingFnMeta-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
Amatrix.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.
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
fitmethod.set_output(*[, transform])Set output container.
set_params(**kwargs)Set the parameters of this estimator.
transform(X)Transform data.
- 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-learnscorer accepts the parameters(estimator, X, y)and returns a float representing the prediction quality ofestimatoronXwith reference toy. Uses existingscikit-learnregression metrics [1]. Higher numbers are better. Metrics corresponding to losses are negated.Technically, the scorer will predict the entire episode, regardless of how
n_stepsis 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
Noneor 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, wherekis 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-learnregression metrics [1]. Can also directly specify a function with the same keyword arguments as thescikit-learnones. That is, at leasty_true,y_pred,sample_weight, andmultioutput.regression_metric_kw (Optional[Dict[str, Any]]) – Keyword arguments for
regression_method. Ifsample_weightkeyword argument is specified,discount_factoris ignored.error_score (Union[str, float]) – Value to assign to the score if
X_predictedhas diverged or if an error has occured in estimator fitting. If set to'raise', aValueErroris raised. If a numerical value is given, asklearn.exceptions.FitFailedWarningwarning 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
multistepis false.
- Returns:
Scorer compatible with
scikit-learn.- Return type:
Callable[[KoopmanPipeline, np.ndarray, Optional[np.ndarray]], float]
- Raises:
ValueError – If
discount_factoris negative or greater than one or iferror_score='raise'and an error occurs in scoring.
References
- 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_minis less than zero orf_maxis 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
MetadataRequestencapsulating 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
- 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.Figureandplt.Axesobjects.- Return type:
Tuple[plt.Figure, plt.Axes]
- Raises:
ValueError – If
f_minis less than zero orf_maxis greater than the Nyquist frequency.
- plot_eigenvalues(unit_circle=True, figure_kw=None, subplot_kw=None, plot_kw=None)
Plot eigenvalues of Koopman
Amatrix.- Parameters:
- Returns:
Matplotlib
plt.Figureandplt.Axesobjects.- 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.Figureobject and two-dimensional array ofplt.Axesobjects.- 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
Uis specified. IfUisNone, 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_Xas ground truth ifUisNone. Ignored ifUis 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.Figureobject and two-dimensional array ofplt.Axesobjects.- 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_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
Uis specified. IfUisNone, 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
fitmethod.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 tofitif 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.Added 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_featureparameter infit.n_inputs (str, True, False, or None, default=sklearn.utils.metadata_routing.UNCHANGED) – Metadata routing for
n_inputsparameter 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", "polars"}, default=None) –
Configure output of transform and fit_transform.
”default”: Default output format of a transformer
”pandas”: DataFrame output
”polars”: Polars output
None: Transform configuration is unchanged
Added in version 1.4: “polars” option was added.
- 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