AugmentedGPR

sgptools.core.augmented_gpr.AugmentedGPR

Bases: GPR

GPR model from the GPFlow library augmented to use a transform object's expand and aggregate functions on the data points where necessary.

Refer to the following papers for more details

• Efficient Sensor Placement from Regression with Sparse Gaussian Processes in Continuous and Discrete Spaces [Jakkala and Akella, 2023]
• Multi-Robot Informative Path Planning from Regression with Sparse Gaussian Processes [Jakkala and Akella, 2024]

Parameters:

Name	Type	Description	Default
data	tuple	(X, y) ndarrays with inputs (n, d) and labels (n, 1)	required
kernel	Kernel	gpflow kernel function	required
noise_variance	float	data variance	required
transform	Transform	Transform object	None

Source code in sgptools/core/augmented_gpr.py

class AugmentedGPR(GPR):
    """GPR model from the GPFlow library augmented to use a transform object's
    expand and aggregate functions on the data points where necessary.  

    Refer to the following papers for more details:
        - Efficient Sensor Placement from Regression with Sparse Gaussian Processes in Continuous and Discrete Spaces [Jakkala and Akella, 2023]
        - Multi-Robot Informative Path Planning from Regression with Sparse Gaussian Processes [Jakkala and Akella, 2024]

    Args:
        data (tuple): (X, y) ndarrays with inputs (n, d) and labels (n, 1)
        kernel (gpflow.kernels.Kernel): gpflow kernel function
        noise_variance (float): data variance
        transform (Transform): Transform object
    """

    def __init__(self, *args, transform=None, **kwargs):
        super().__init__(*args, **kwargs)
        if transform is None:
            self.transform = Transform()
        else:
            self.transform = transform

    def predict_f(
        self,
        Xnew: InputData,
        full_cov: bool = True,
        full_output_cov: bool = False,
        aggregate_train: bool = False,
    ) -> MeanAndVariance:
        assert_params_false(self.predict_f, full_output_cov=full_output_cov)
        if self.transform is not None:
            Xnew = self.transform.expand(Xnew)

        X, Y = self.data
        err = Y - self.mean_function(X)

        kmm = self.kernel(X)
        knn = self.kernel(Xnew, full_cov=full_cov)
        kmn = self.kernel(X, Xnew)
        kmm_plus_s = add_likelihood_noise_cov(kmm, self.likelihood, X)

        if self.transform is not None:
            kmn = self.transform.aggregate(tf.transpose(kmn))
            kmn = tf.transpose(kmn)
            knn = self.transform.aggregate(knn)

        if aggregate_train:
            kmm_plus_s = self.transform.aggregate(kmm_plus_s)
            err = self.transform.aggregate(err)
            # reduce kmn only if it was not reduced before
            # which can when train and test data are the same size
            if kmn.shape[0] != kmn.shape[1]:
                kmn = self.transform.aggregate(kmn)

        conditional = gpflow.conditionals.base_conditional
        f_mean_zero, f_var = conditional(
            kmn, kmm_plus_s, knn, err, full_cov=full_cov,
            white=False)  # [N, P], [N, P] or [P, N, N]
        f_mean = f_mean_zero + self.mean_function(Xnew)
        return f_mean, f_var