Source code for mmgp.backends.gpy

# -*- coding: utf-8 -*-
#
# This file is subject to the terms and conditions defined in
# file 'LICENSE.txt', which is part of this source code package.
#
#
from mmgp.regressor import RegressorBase

import numpy as np
from typing import Self

import GPy



class Regressor(RegressorBase):
    """
    gpy regressor
    """

    def __init__(self, options: dict):
        """
        Args:
            algo (str): The regression algorithm to use.
            options (dict): A dictionary of options specific to the chosen algorithm. Allowed fields are "kernel", "optim", "num_restarts", "max_iters" and "anisotropic".

        """
        super(Regressor, self).__init__()


        self.algo = "GPy"



        self.options = options


        if self.options['show_warnings'] == False:
            import logging, warnings
            warnings.filterwarnings("ignore", category=RuntimeWarning)
            logger = logging.getLogger('GP')
            logger.propagate = False




    def fit(self, X: np.ndarray, y: np.ndarray) -> Self:
        """Train the regression model on the provided data.

        Args:
            X (np.ndarray): The input features for training.
            y (np.ndarray): The target values for training.
        """

        self.input_dim  = X.shape[1]
        self.output_dim = y.shape[1]

        kernel_class = getattr(GPy.kern, self.options["kernel"])
        kernel = kernel_class(
            input_dim=self.input_dim,
            ARD=self.options["anisotropic"])
        
        self.kmodel=[GPy.models.GPRegression(X, y[:,i].reshape(-1,1), kernel.copy()) for i in range(self.output_dim)]
        for i in range(self.output_dim):
            self.kmodel[i].optimize_restarts(
                optimizer=self.options["optim"],
                max_iters=self.options["max_iters"],
                num_restarts=self.options["num_restarts"],
                verbose=False)





    def predict(self, X: np.ndarray, return_var: bool) -> np.ndarray:
        """Make predictions using the trained regression model.

        Args:
            X (np.ndarray): The input features for making predictions. Array of size 1 x self.input_dim
            return_var (bool): True if prediction variance is computed

        Returns:
            np.ndarray: Predicted target values (and variances).  One array (two arrays) of size 1 x self.output_dim
        """
        
        mean=np.zeros((1,self.output_dim))
        var=np.zeros((1,self.output_dim))
        for i in range(self.output_dim):
            m,v = self.kmodel[i].predict(X)
            mean[0,i]=m[0,0]
            var[0,i]=v[0,0]
            
        if (return_var):
            return (mean,var)
        else:
            return mean





    def predict_Monte_Carlo_draw(
            self, X: np.ndarray) -> np.ndarray:
        """Generate Monte Carlo draws from the trained regression model.

        Args:
            X (np.ndarray): The input features for generating draws. Array of size 1 x self.input_dim

        Returns:
            np.ndarray: Monte Carlo draw from the posterior of the regression model. Array of size 1 x self.output_dim
        """
        
        mean,var = self.predict(X,return_var=True)
        K=np.zeros((self.output_dim, self.output_dim))
        np.fill_diagonal(K,np.sqrt(var))
        samples=np.random.normal(loc=0, scale=1, size=self.output_dim).reshape(-1,1)
        samples=np.dot(K,samples).T

        return samples