Skip to content

GreedyObjective

sgptools.methods.GreedyObjective

Bases: Method

Informative sensor placement / path optimization using a greedy selection based on a generic objective function.

The method iteratively adds sensing locations from a discrete candidate set to maximize a user-specified objective (e.g., mutual information), using apricot.CustomSelection as the selection engine. Only single-robot scenarios are supported.

Refer to the following papers for more details
  • Krause et al., 2008. Near-Optimal Sensor Placements in Gaussian Processes: Theory, Efficient Algorithms and Empirical Studies.
  • Ma et al., 2018. Data-driven learning and planning for environmental sampling.

Attributes:

Name Type Description
objective Objective

Objective object to maximize over the chosen locations.

transform Transform | None

Optional transform applied to selected locations.

Source code in sgptools/methods.py
class GreedyObjective(Method):
    """Informative sensor placement / path optimization using a greedy selection
    based on a generic objective function.

    The method iteratively adds sensing locations from a discrete candidate
    set to maximize a user-specified objective (e.g., mutual information),
    using `apricot.CustomSelection` as the selection engine. Only single-robot
    scenarios are supported.

    Refer to the following papers for more details:
        - Krause et al., 2008. *Near-Optimal Sensor Placements in Gaussian
        Processes: Theory, Efficient Algorithms and Empirical Studies.*
        - Ma et al., 2018. *Data-driven learning and planning for environmental
        sampling.*

    Attributes:
        objective (Objective):
            Objective object to maximize over the chosen locations.
        transform (Transform | None):
            Optional transform applied to selected locations.
    """

    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,
                 objective: Union[str, Objective] = 'SLogMI',
                 **kwargs: Any):
        """Initialize a greedy objective-based method.

        Args:
            num_sensing (int):
                Number of sensing locations to select.
            X_objective (np.ndarray):
                Array of shape `(n, d)` used to define the objective (e.g. GP
                training inputs).
            kernel (gpflow.kernels.Kernel):
                GPflow kernel used inside the objective.
            noise_variance (float):
                Observation noise variance used inside the objective.
            transform (Transform | None):
                Optional transform applied to selected locations before evaluating
                the objective and constraints.
            num_robots (int):
                Number of robots / agents. `GreedyObjective` currently supports
                only `num_robots = 1` and will assert otherwise.
            X_candidates (np.ndarray | None):
                Discrete candidate locations with shape `(c, d)`. If `None`,
                defaults to `X_objective`.
            num_dim (int | None):
                Dimensionality of the sensing locations. If `None`, defaults to
                `X_objective.shape[-1]`.
            objective (str | Objective):
                Objective specification (string key or `Objective` instance) used
                by `get_objective` when a string is given.
            **kwargs (Any):
                Additional keyword arguments forwarded to the objective constructor
                when `objective` is a string.
        """
        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 GreedyObjective: {self.num_robots}"
            assert self.num_robots == num_robots_transform, error

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

        self.transform = transform

        if isinstance(objective, str):
            self.objective = get_objective(objective)(X_objective, kernel,
                                                      noise_variance, **kwargs)
        else:
            self.objective = objective

    def update(self, kernel: gpflow.kernels.Kernel,
               noise_variance: float) -> None:
        """Update the kernel and noise variance used by the objective.

        Args:
            kernel (gpflow.kernels.Kernel):
                New GPflow kernel instance.
            noise_variance (float):
                New observation noise variance.
        """
        self.objective.update(kernel, noise_variance)

    def get_hyperparameters(self) -> Tuple[gpflow.kernels.Kernel, float]:
        """Return the current kernel and noise variance used by the objective.

        Returns:
            Tuple[gpflow.kernels.Kernel, float]:
                A deep copy of the kernel and the current noise variance.
        """
        return deepcopy(self.objective.kernel), \
               self.objective.noise_variance

    def optimize(self,
                 optimizer: str = 'naive',
                 verbose: bool = False,
                 **kwargs: Any) -> np.ndarray:
        """Run greedy selection over the candidate set.

        Args:
            optimizer (str):
                Greedy strategy identifier passed to `apricot.CustomSelection`
                (e.g., `'naive'`, `'lazy'`).
            verbose (bool):
                If `True`, print progress information from apricot.
            **kwargs (Any):
                Additional keyword arguments forwarded to `CustomSelection`
                (currently unused in this wrapper but accepted for flexibility).

        Returns:
            np.ndarray:
                Array of shape `(num_robots, num_sensing, num_dim)` containing the
                selected sensing locations.
        """
        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)
        if self.transform is not None:
            sol_locations = self.transform.expand(
                sol_locations, expand_sensor_model=False)
            if not isinstance(sol_locations, np.ndarray):
                sol_locations = sol_locations.numpy()
        sol_locations = sol_locations.reshape(self.num_robots, -1,
                                              self.num_dim)
        return sol_locations

    def _objective(self, X_indices: np.ndarray) -> float:
        """Objective callback used by `apricot.CustomSelection`.

        The input is an array of candidate indices. The method:
        1. Maps indices to candidate locations.
        2. Optionally applies the transform (and constraints).
        3. Evaluates the underlying objective.
        4. Adds the transform constraint penalty.
        5. Returns the resulting scalar as a Python float.

        Args:
            X_indices (np.ndarray):
                Array of shape `(n, 1)` containing indices into `self.X_objective`
                / `self.X_candidates`.

        Returns:
            float:
                Objective value to be maximized by apricot's greedy selection
                routine.
        """
        # Map solution location indices to locations
        X_indices_flat = np.array(X_indices).reshape(-1).astype(int)
        X_locations = self.X_objective[X_indices_flat].reshape(
            -1, self.num_dim)

        constraint_penality: float = 0.0
        if self.transform is not None:
            X_expanded = self.transform.expand(X_locations)
            constraint_penality = self.transform.constraints(X_locations)
            reward = self.objective(X_expanded)  # maximize
        else:
            reward = self.objective(X_locations)  # maximize

        reward += constraint_penality
        return reward.numpy()

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

Initialize a greedy objective-based method.

Parameters:

Name Type Description Default
num_sensing int

Number of sensing locations to select.

required
X_objective ndarray

Array of shape (n, d) used to define the objective (e.g. GP training inputs).

required
kernel Kernel

GPflow kernel used inside the objective.

required
noise_variance float

Observation noise variance used inside the objective.

required
transform Transform | None

Optional transform applied to selected locations before evaluating the objective and constraints.

None
num_robots int

Number of robots / agents. GreedyObjective currently supports only num_robots = 1 and will assert otherwise.

1
X_candidates ndarray | None

Discrete candidate locations with shape (c, d). If None, defaults to X_objective.

None
num_dim int | None

Dimensionality of the sensing locations. If None, defaults to X_objective.shape[-1].

None
objective str | Objective

Objective specification (string key or Objective instance) used by get_objective when a string is given.

'SLogMI'
**kwargs Any

Additional keyword arguments forwarded to the objective constructor when objective is a string.

{}
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,
             objective: Union[str, Objective] = 'SLogMI',
             **kwargs: Any):
    """Initialize a greedy objective-based method.

    Args:
        num_sensing (int):
            Number of sensing locations to select.
        X_objective (np.ndarray):
            Array of shape `(n, d)` used to define the objective (e.g. GP
            training inputs).
        kernel (gpflow.kernels.Kernel):
            GPflow kernel used inside the objective.
        noise_variance (float):
            Observation noise variance used inside the objective.
        transform (Transform | None):
            Optional transform applied to selected locations before evaluating
            the objective and constraints.
        num_robots (int):
            Number of robots / agents. `GreedyObjective` currently supports
            only `num_robots = 1` and will assert otherwise.
        X_candidates (np.ndarray | None):
            Discrete candidate locations with shape `(c, d)`. If `None`,
            defaults to `X_objective`.
        num_dim (int | None):
            Dimensionality of the sensing locations. If `None`, defaults to
            `X_objective.shape[-1]`.
        objective (str | Objective):
            Objective specification (string key or `Objective` instance) used
            by `get_objective` when a string is given.
        **kwargs (Any):
            Additional keyword arguments forwarded to the objective constructor
            when `objective` is a string.
    """
    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 GreedyObjective: {self.num_robots}"
        assert self.num_robots == num_robots_transform, error

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

    self.transform = transform

    if isinstance(objective, str):
        self.objective = get_objective(objective)(X_objective, kernel,
                                                  noise_variance, **kwargs)
    else:
        self.objective = objective

get_hyperparameters()

Return the current kernel and noise variance used by the objective.

Returns:

Type Description
Tuple[Kernel, float]

Tuple[gpflow.kernels.Kernel, float]: A deep copy of the kernel and the current noise variance.

Source code in sgptools/methods.py
def get_hyperparameters(self) -> Tuple[gpflow.kernels.Kernel, float]:
    """Return the current kernel and noise variance used by the objective.

    Returns:
        Tuple[gpflow.kernels.Kernel, float]:
            A deep copy of the kernel and the current noise variance.
    """
    return deepcopy(self.objective.kernel), \
           self.objective.noise_variance

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

Run greedy selection over the candidate set.

Parameters:

Name Type Description Default
optimizer str

Greedy strategy identifier passed to apricot.CustomSelection (e.g., 'naive', 'lazy').

'naive'
verbose bool

If True, print progress information from apricot.

False
**kwargs Any

Additional keyword arguments forwarded to CustomSelection (currently unused in this wrapper but accepted for flexibility).

{}

Returns:

Type Description
ndarray

np.ndarray: Array of shape (num_robots, num_sensing, num_dim) containing the selected sensing locations.

Source code in sgptools/methods.py
def optimize(self,
             optimizer: str = 'naive',
             verbose: bool = False,
             **kwargs: Any) -> np.ndarray:
    """Run greedy selection over the candidate set.

    Args:
        optimizer (str):
            Greedy strategy identifier passed to `apricot.CustomSelection`
            (e.g., `'naive'`, `'lazy'`).
        verbose (bool):
            If `True`, print progress information from apricot.
        **kwargs (Any):
            Additional keyword arguments forwarded to `CustomSelection`
            (currently unused in this wrapper but accepted for flexibility).

    Returns:
        np.ndarray:
            Array of shape `(num_robots, num_sensing, num_dim)` containing the
            selected sensing locations.
    """
    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)
    if self.transform is not None:
        sol_locations = self.transform.expand(
            sol_locations, expand_sensor_model=False)
        if not isinstance(sol_locations, np.ndarray):
            sol_locations = sol_locations.numpy()
    sol_locations = sol_locations.reshape(self.num_robots, -1,
                                          self.num_dim)
    return sol_locations

update(kernel, noise_variance)

Update the kernel and noise variance used by the objective.

Parameters:

Name Type Description Default
kernel Kernel

New GPflow kernel instance.

required
noise_variance float

New observation noise variance.

required
Source code in sgptools/methods.py
def update(self, kernel: gpflow.kernels.Kernel,
           noise_variance: float) -> None:
    """Update the kernel and noise variance used by the objective.

    Args:
        kernel (gpflow.kernels.Kernel):
            New GPflow kernel instance.
        noise_variance (float):
            New observation noise variance.
    """
    self.objective.update(kernel, noise_variance)