GreedySGP

sgptools.methods.GreedySGP

Bases: Method

Implements informative sensor placement/path optimization using a greedy approach combined with a Sparse Gaussian Process (SGP) ELBO objective.

This method iteratively selects inducing points to maximize the SGP's ELBO. It currently supports only single-robot scenarios.

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]

Attributes:

Name	Type	Description
sgpr	AugmentedSGPR	The Augmented Sparse Gaussian Process Regression model.

Source code in sgptools/methods.py

class GreedySGP(Method):
    """
    Implements informative sensor placement/path optimization using a greedy approach combined with a Sparse Gaussian Process (SGP) ELBO objective.

    This method iteratively selects inducing points to maximize the SGP's ELBO.
    It currently supports only single-robot scenarios.

    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](https://www.itskalvik.com/publication/sgp-sp/)]

    Attributes:
        sgpr (AugmentedSGPR): The Augmented Sparse Gaussian Process Regression model.
    """

    def __init__(self,
                 num_sensing: int,
                 X_objective: np.ndarray,
                 kernel: gpflow.kernels.Kernel,
                 noise_variance: float,
                 transform: Optional[Transform] = None,
                 num_robots: int = 1,
                 X_candidates: Optional[np.ndarray] = None,
                 num_dim: Optional[int] = None,
                 **kwargs: Any):
        """
        Initializes the GreedySGP optimizer.

        Args:
            num_sensing (int): Number of sensing locations (inducing points) to select.
            X_objective (np.ndarray): (n, d); Data points used to train the SGP model.
            kernel (gpflow.kernels.Kernel): GPflow kernel function.
            noise_variance (float): Data noise variance.
            transform (Optional[Transform]): Transform object to apply to inducing points. Defaults to None.
            num_robots (int): Number of robots/agents. Defaults to 1.
            X_candidates (Optional[np.ndarray]): (c, d); Discrete set of candidate locations for sensor placement.
                                                 If None, X_objective is used as candidates.
            num_dim (Optional[int]): Dimensionality of the sensing locations. Defaults to dimensonality of X_objective.
            **kwargs: Additional keyword arguments.
        """
        super().__init__(num_sensing, X_objective, kernel, noise_variance,
                         transform, num_robots, X_candidates, num_dim)
        self.X_objective = X_objective
        if X_candidates is None:
            self.X_candidates = X_objective  # Default candidates to objective points

        if transform is not None:
            try:
                num_robots_transform = transform.num_robots
            except AttributeError:
                num_robots_transform = 1  # Assume single robot if num_robots not defined in transform
            error = f"num_robots is not equal in transform: {num_robots_transform} and GreedySGP: {self.num_robots}"
            assert self.num_robots == num_robots_transform, error

        error = f"num_robots={self.num_robots}; GreedySGP only supports num_robots=1"
        assert self.num_robots == 1, error

        # Fit the SGP
        dtype = X_objective.dtype
        train_set: Tuple[tf.Tensor, tf.Tensor] = (tf.constant(X_objective,
                                                              dtype=dtype),
                                                  tf.zeros(
                                                      (len(X_objective), 1),
                                                      dtype=dtype))

        X_init = get_inducing_pts(X_objective, num_sensing)
        self.sgpr = AugmentedSGPR(train_set,
                                  noise_variance=noise_variance,
                                  kernel=kernel,
                                  inducing_variable=X_init,
                                  transform=transform)

    def update(self, kernel: gpflow.kernels.Kernel,
               noise_variance: float) -> None:
        """
        Updates the kernel and noise variance parameters of the SGP model.

        Args:
            kernel (gpflow.kernels.Kernel): Updated GPflow kernel function.
            noise_variance (float): Updated data noise variance.
        """
        self.sgpr.update(kernel, noise_variance)

    def get_hyperparameters(self) -> Tuple[gpflow.kernels.Kernel, float]:
        """
        Retrieves the current kernel and noise variance hyperparameters from the SGP model.

        Returns:
            Tuple[gpflow.kernels.Kernel, float]: A tuple containing a deep copy of the kernel and the noise variance.
        """
        return deepcopy(self.sgpr.kernel), \
               self.sgpr.likelihood.variance.numpy()

    def optimize(self,
                 optimizer: str = 'naive',
                 verbose: bool = False,
                 **kwargs: Any) -> np.ndarray:
        """
        Optimizes sensor placement using a greedy SGP approach.

        Args:
            optimizer (str): The greedy optimizer strategy (e.g., 'naive', 'lazy'). Defaults to 'naive'.
            verbose (bool): Verbosity, if True additional details will by reported. Defaults to False.
            **kwargs: Additional keyword arguments.

        Returns:
            np.ndarray: (num_robots, num_sensing, num_dim); Optimized sensing locations.

        Usage:
            ```python
            # Assuming X_train, candidates, kernel_opt, noise_variance_opt are defined
            greedy_sgp_method = GreedySGP(
                num_sensing=5,
                X_objective=X_train,
                kernel=kernel_opt,
                noise_variance=noise_variance_opt,
                X_candidates=candidates
            )
            optimized_solution = greedy_sgp_method.optimize(optimizer='naive')
            ```
        """
        model = CustomSelection(self.num_sensing,
                                self._objective,
                                optimizer=optimizer,
                                verbose=verbose)

        # apricot's CustomSelection expects indices, so pass a dummy array of indices
        sol_indices = model.fit_transform(
            np.arange(len(self.X_candidates)).reshape(-1, 1))
        sol_indices = np.array(sol_indices).reshape(-1).astype(int)
        sol_locations = self.X_candidates[sol_indices]

        sol_locations = np.array(sol_locations).reshape(-1, self.num_dim)
        try:
            sol_expanded = self.transform.expand(sol_locations,
                                                 expand_sensor_model=False)
        except AttributeError:
            sol_expanded = sol_locations
        if not isinstance(sol_expanded, np.ndarray):
            sol_np = sol_expanded.numpy()
        else:
            sol_np = sol_expanded

        sol_np = sol_np.reshape(self.num_robots, -1, self.num_dim)
        return sol_np

    def _objective(self, X_indices: np.ndarray) -> float:
        """
        Internal objective function for the greedy SGP selection.

        This function maps the input indices to actual locations and updates
        the SGP model's inducing points to calculate the ELBO. The ELBO is
        then used as the objective for greedy maximization.

        Args:
            X_indices (np.ndarray): (n, 1); Array of indices corresponding to candidate locations.

        Returns:
            float: The ELBO of the SGP model for the given inducing points.
        """
        # Map solution location indices to locations
        # Since SGP requires num_sensing points,
        # pad the current greedy solution with the
        # first location in the solution (or zeros if no points selected yet)
        X_indices_flat = np.array(X_indices).reshape(-1).astype(int)
        num_pad = self.num_sensing - len(X_indices_flat)

        # Ensure that if X_indices_flat is empty, we still create a valid padding array
        if len(X_indices_flat) == 0 and num_pad > 0:
            X_pad = np.zeros(num_pad, dtype=int)
        elif len(X_indices_flat) > 0 and num_pad > 0:
            X_pad = np.full(num_pad, X_indices_flat[0], dtype=int)
        else:  # num_pad is 0 or negative
            X_pad = np.array([], dtype=int)

        X_combined_indices = np.concatenate([X_indices_flat, X_pad])
        X_locations = self.X_objective[X_combined_indices].reshape(
            -1, self.num_dim)

        # Update the SGP inducing points
        self.sgpr.inducing_variable.Z.assign(X_locations)
        return self.sgpr.elbo().numpy()

    @property
    def transform(self) -> Transform:
        """
        Gets the transform object associated with the SGP model.

        Returns:
            Transform: The transform object.
        """
        return self.sgpr.transform

transform property

Gets the transform object associated with the SGP model.

Returns:

Name	Type	Description
Transform	Transform	The transform object.

__init__(num_sensing, X_objective, kernel, noise_variance, transform=None, num_robots=1, X_candidates=None, num_dim=None, **kwargs)

Initializes the GreedySGP optimizer.

Parameters:

Name	Type	Description	Default
num_sensing	int	Number of sensing locations (inducing points) to select.	required
X_objective	ndarray	(n, d); Data points used to train the SGP model.	required
kernel	Kernel	GPflow kernel function.	required
noise_variance	float	Data noise variance.	required
transform	Optional[Transform]	Transform object to apply to inducing points. Defaults to None.	None
num_robots	int	Number of robots/agents. Defaults to 1.	1
X_candidates	Optional[ndarray]	(c, d); Discrete set of candidate locations for sensor placement. If None, X_objective is used as candidates.	None
num_dim	Optional[int]	Dimensionality of the sensing locations. Defaults to dimensonality of X_objective.	None
**kwargs	Any	Additional keyword arguments.	{}

Source code in sgptools/methods.py

def __init__(self,
             num_sensing: int,
             X_objective: np.ndarray,
             kernel: gpflow.kernels.Kernel,
             noise_variance: float,
             transform: Optional[Transform] = None,
             num_robots: int = 1,
             X_candidates: Optional[np.ndarray] = None,
             num_dim: Optional[int] = None,
             **kwargs: Any):
    """
    Initializes the GreedySGP optimizer.

    Args:
        num_sensing (int): Number of sensing locations (inducing points) to select.
        X_objective (np.ndarray): (n, d); Data points used to train the SGP model.
        kernel (gpflow.kernels.Kernel): GPflow kernel function.
        noise_variance (float): Data noise variance.
        transform (Optional[Transform]): Transform object to apply to inducing points. Defaults to None.
        num_robots (int): Number of robots/agents. Defaults to 1.
        X_candidates (Optional[np.ndarray]): (c, d); Discrete set of candidate locations for sensor placement.
                                             If None, X_objective is used as candidates.
        num_dim (Optional[int]): Dimensionality of the sensing locations. Defaults to dimensonality of X_objective.
        **kwargs: Additional keyword arguments.
    """
    super().__init__(num_sensing, X_objective, kernel, noise_variance,
                     transform, num_robots, X_candidates, num_dim)
    self.X_objective = X_objective
    if X_candidates is None:
        self.X_candidates = X_objective  # Default candidates to objective points

    if transform is not None:
        try:
            num_robots_transform = transform.num_robots
        except AttributeError:
            num_robots_transform = 1  # Assume single robot if num_robots not defined in transform
        error = f"num_robots is not equal in transform: {num_robots_transform} and GreedySGP: {self.num_robots}"
        assert self.num_robots == num_robots_transform, error

    error = f"num_robots={self.num_robots}; GreedySGP only supports num_robots=1"
    assert self.num_robots == 1, error

    # Fit the SGP
    dtype = X_objective.dtype
    train_set: Tuple[tf.Tensor, tf.Tensor] = (tf.constant(X_objective,
                                                          dtype=dtype),
                                              tf.zeros(
                                                  (len(X_objective), 1),
                                                  dtype=dtype))

    X_init = get_inducing_pts(X_objective, num_sensing)
    self.sgpr = AugmentedSGPR(train_set,
                              noise_variance=noise_variance,
                              kernel=kernel,
                              inducing_variable=X_init,
                              transform=transform)

get_hyperparameters()

Retrieves the current kernel and noise variance hyperparameters from the SGP model.

Returns:

Type	Description
Tuple[Kernel, float]	Tuple[gpflow.kernels.Kernel, float]: A tuple containing a deep copy of the kernel and the noise variance.

Source code in sgptools/methods.py

def get_hyperparameters(self) -> Tuple[gpflow.kernels.Kernel, float]:
    """
    Retrieves the current kernel and noise variance hyperparameters from the SGP model.

    Returns:
        Tuple[gpflow.kernels.Kernel, float]: A tuple containing a deep copy of the kernel and the noise variance.
    """
    return deepcopy(self.sgpr.kernel), \
           self.sgpr.likelihood.variance.numpy()

optimize(optimizer='naive', verbose=False, **kwargs)

Optimizes sensor placement using a greedy SGP approach.

Parameters:

Name	Type	Description	Default
optimizer	str	The greedy optimizer strategy (e.g., 'naive', 'lazy'). Defaults to 'naive'.	'naive'
verbose	bool	Verbosity, if True additional details will by reported. Defaults to False.	False
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Type	Description
ndarray	np.ndarray: (num_robots, num_sensing, num_dim); Optimized sensing locations.

Usage

# Assuming X_train, candidates, kernel_opt, noise_variance_opt are defined
greedy_sgp_method = GreedySGP(
    num_sensing=5,
    X_objective=X_train,
    kernel=kernel_opt,
    noise_variance=noise_variance_opt,
    X_candidates=candidates
)
optimized_solution = greedy_sgp_method.optimize(optimizer='naive')

Source code in sgptools/methods.py

def optimize(self,
             optimizer: str = 'naive',
             verbose: bool = False,
             **kwargs: Any) -> np.ndarray:
    """
    Optimizes sensor placement using a greedy SGP approach.

    Args:
        optimizer (str): The greedy optimizer strategy (e.g., 'naive', 'lazy'). Defaults to 'naive'.
        verbose (bool): Verbosity, if True additional details will by reported. Defaults to False.
        **kwargs: Additional keyword arguments.

    Returns:
        np.ndarray: (num_robots, num_sensing, num_dim); Optimized sensing locations.

    Usage:
        ```python
        # Assuming X_train, candidates, kernel_opt, noise_variance_opt are defined
        greedy_sgp_method = GreedySGP(
            num_sensing=5,
            X_objective=X_train,
            kernel=kernel_opt,
            noise_variance=noise_variance_opt,
            X_candidates=candidates
        )
        optimized_solution = greedy_sgp_method.optimize(optimizer='naive')
        ```
    """
    model = CustomSelection(self.num_sensing,
                            self._objective,
                            optimizer=optimizer,
                            verbose=verbose)

    # apricot's CustomSelection expects indices, so pass a dummy array of indices
    sol_indices = model.fit_transform(
        np.arange(len(self.X_candidates)).reshape(-1, 1))
    sol_indices = np.array(sol_indices).reshape(-1).astype(int)
    sol_locations = self.X_candidates[sol_indices]

    sol_locations = np.array(sol_locations).reshape(-1, self.num_dim)
    try:
        sol_expanded = self.transform.expand(sol_locations,
                                             expand_sensor_model=False)
    except AttributeError:
        sol_expanded = sol_locations
    if not isinstance(sol_expanded, np.ndarray):
        sol_np = sol_expanded.numpy()
    else:
        sol_np = sol_expanded

    sol_np = sol_np.reshape(self.num_robots, -1, self.num_dim)
    return sol_np

update(kernel, noise_variance)

Updates the kernel and noise variance parameters of the SGP model.

Parameters:

Name	Type	Description	Default
kernel	Kernel	Updated GPflow kernel function.	required
noise_variance	float	Updated data noise variance.	required

Source code in sgptools/methods.py

def update(self, kernel: gpflow.kernels.Kernel,
           noise_variance: float) -> None:
    """
    Updates the kernel and noise variance parameters of the SGP model.

    Args:
        kernel (gpflow.kernels.Kernel): Updated GPflow kernel function.
        noise_variance (float): Updated data noise variance.
    """
    self.sgpr.update(kernel, noise_variance)