Best Python code snippet using pyresttest_python
MGP.py
Source:MGP.py  
1import torch2import gpytorch3import numpy as np4import matplotlib.pyplot as plt5from matplotlib.gridspec import GridSpec6import h5py7import os8import pdb9from MGP_subclasses import *10from data_processing import *11class Block_MGP():12    def __init__(self, kernel, learning_rate, n_training_iter, block_indices):13        self.kernel = kernel14        self.learning_rate = learning_rate15        self.n_training_iter = n_training_iter16        self.block_indices = block_indices17        self.number_of_block = len(block_indices)18        self.total_nb_tasks = len([item for sublist in self.block_indices for item in sublist])19        self.model = []20        self.likelihood = []21        self.loss_list = []22    def build_and_train_single_model(self, x_train, y_train, block_number=0, smart_end = False):23        '''24        :param x_train: array size nb_timesteps *1, represents time25        :param y_train: array size nb_timesteps * nb_tasks26        :param block_number: the number of the block, starts from 027        :return: modifies the attributes model and likelihood according to the training data28        '''29        nb_tasks = y_train.shape[-1]30        if nb_tasks == 1:31            self.likelihood.append(gpytorch.likelihoods.GaussianLikelihood())32            y_train = y_train[:,0]33            self.model.append(Single_task_GP_model(x_train, y_train, self.likelihood[block_number], self.kernel))34        if nb_tasks>1: #if no model has been ever trained, create a model35            self.likelihood.append(gpytorch.likelihoods.MultitaskGaussianLikelihood(num_tasks=nb_tasks))36            self.model.append(Multitask_GP_Model(x_train, y_train, self.likelihood[block_number], nb_tasks, self.kernel))37        self.model[block_number].train()38        self.likelihood[block_number].train()39        optimizer = torch.optim.Adam([{'params': self.model[block_number].parameters()}, ], lr=self.learning_rate)40        mll = gpytorch.mlls.ExactMarginalLogLikelihood(self.likelihood[block_number], self.model[block_number])41        loss_list_cur = []42        plot_frequency = self.n_training_iter // 1043        if smart_end:44            loss_hist = 045        for i in range(self.n_training_iter):46            optimizer.zero_grad()47            output = self.model[block_number](x_train)48            loss = -mll(output, y_train)49            if i>120 and smart_end:50                min_loss_variation = np.min(np.array(loss_list_cur[1:30])-np.array(loss_list_cur[0:29]))51                if loss - loss_hist > - min_loss_variation :52                    break53                else:54                    loss.backward()55                    optimizer.step()56                    if i % plot_frequency == 0:57                        print('Iter %d/%d - Loss: %.3f' % (i + 1, self.n_training_iter, loss.item()))58                    loss_list_cur.append(loss.item())59            else:60                loss.backward()61                optimizer.step()62                if i % plot_frequency == 0:63                    print('Iter %d/%d - Loss: %.3f' % (i + 1, self.n_training_iter, loss.item()))64                loss_list_cur.append(loss.item())65            loss_hist = loss.item()66        self.loss_list.append(loss_list_cur)67    def build_and_train_block_models(self, x_train, y_train, smart_end = False):68        '''69        :param x_train: array size nb_timesteps *1, represents time70        :param y_train: array size nb_timesteps * nb_tasks71        :return: train the multiple MGP, one for each block72        '''73        for i in range(self.number_of_block):74            print('### BLOCK %d ###'%i)75            self.build_and_train_single_model(x_train, y_train[:,self.block_indices[i]], i, smart_end)76    def test_block_model(self, x_test):77        '''78        :param x_test: array size nb_timesteps_test * 1, represents time79        :return:  test_mean_list : the mean of the posterior MGPs80                  test_covar_matrix_list : the psoetrior covariance matrices81                  test_std :the standard deviation of the MGPs82        BE CAREFUL : the outputs are list, each block has then its own mean /coavriances arrays83        '''84        test_mean_list = []85        test_covar_matrix_list = []86        test_std = []87        for i in range(self.number_of_block):88            self.model[i].eval()89            self.likelihood[i].eval()90            with torch.no_grad(), gpytorch.settings.fast_pred_var():91                test_observed_pred = self.likelihood[i](self.model[i](x_test))92                test_mean= test_observed_pred.mean.detach().numpy()93                test_covar_matrix = self.model[i].return_covar_matrix(x_test).detach().numpy()94            test_mean_list.append(test_mean)95            test_covar_matrix_list.append(test_covar_matrix)96            test_lower, test_upper = test_observed_pred.confidence_region() #95% confidence interval97            test_lower, test_upper = test_lower.detach().numpy(), test_upper.detach().numpy()98            test_std.append((test_upper - test_lower) / 2*1.96) # 95% confidence interval to std99        return test_mean_list, test_covar_matrix_list, test_std100    def plot_model(self, x_train, y_train, x_test, train_filter):101        '''102        :param x_train: array size nb_timesteps * 1, represents time103        :param y_train: array size nb_timesteps * nb_tasks104        :param x_test: array size nb_timesteps_test * 1, represents time105        :param train_filter : indices of the selected points for the training106        :return: a plot of the losses, the covariance matrices and the regression for each block107        '''108        test_mean_list, test_covar_matrix_list, test_std_deviation = self.test_block_model(x_test)109        fig = plt.figure(figsize=(18.5,9))110        gs = GridSpec(2, max(self.total_nb_tasks, 2*self.number_of_block))111        iter = 0112        for j in range(self.number_of_block):113            if len(self.block_indices[j])==1: #Single GP114                ax = fig.add_subplot(gs[0, iter])115                ax.plot(x_test.detach().numpy(), test_mean_list[j])116                ax.fill_between(x_test, test_mean_list[j] + test_std_deviation[j],117                                test_mean_list[j] - test_std_deviation[j], alpha=0.3)118                ax.set_title('Block %d Level %d' % (j, self.block_indices[j][0]))119                ax.plot(x_train.detach().numpy(), y_train.detach().numpy()[:, self.block_indices[j]],color='tomato')120                ax.plot(x_train.detach().numpy()[train_filter],y_train.detach().numpy()[train_filter, self.block_indices[j][0]], 'k*', color='red')121                iter = iter + 1122                ax.axvline(x_train.shape[0]/x_test.shape[0], color='green')123            else: #MGP124                for i in range(len(self.block_indices[j])):125                    ax = fig.add_subplot(gs[0, iter])126                    ax.plot(x_test.detach().numpy(), test_mean_list[j][:,i])127                    ax.fill_between(x_test, test_mean_list[j][:,i] + test_std_deviation[j][:,i], test_mean_list[j][:,i] - test_std_deviation[j][:,i], alpha=0.3)128                    ax.set_title('Block %d Level %d'%(j,self.block_indices[j][i]))129                    ax.plot(x_train.detach().numpy(), y_train.detach().numpy()[:, self.block_indices[j][i]], color='tomato')130                    ax.plot(x_train.detach().numpy()[train_filter], y_train.detach().numpy()[train_filter, self.block_indices[j][i]], 'k*', color='red')131                    ax.axvline(x_train.shape[0]/x_test.shape[0], color='green')132                    iter=iter+1133        for j in range(self.number_of_block):134            nb_tasks = len(self.block_indices[j])135            if nb_tasks ==1: #single GP136                ax1 = fig.add_subplot(gs[1, 2*j])137                ax1.imshow(test_covar_matrix_list[j])138                ax1.set_title('Block %d Covar Matrix' % j)139            if nb_tasks > 1 :  # multi GP140                ax1 = fig.add_subplot(gs[1, 2*j])141                matrix = change_representation_covariance_matrix(test_covar_matrix_list[j], nb_tasks)142                ax1.imshow(matrix)143                ax1.set_title('Block %d Covar Matrix' % j)144            ax2 = fig.add_subplot(gs[1, 2*j+1])145            ax2.plot(self.loss_list[j])146            ax2.set_title('Block %d Loss' % j)147        plt.show()148def train_Block_MGP_multiple_individuals(x_train, y_train, x_test, y_test, block_indices,149                                            kernel, learning_rate, n_iter,150                                            train_sample_subset = np.array([]), main_dir='unknown_dir', exec_type='unknown_exec', train_sampling_type = 'unknown_sampling',151                                            activate_plot=False, smart_end = False):152    '''153    :param x_train: array size nb_timesteps_test * 1, represents time154    :param y_train: array size nb_individuals * nb_timesteps_test * number_tasks155    :param block_indices: list of lists of indices (ex: [[0,1],[2,3],[4]]156    :param x_test: array size nb_timesteps_test * 1, represents time157    :param y_test: array size nb_individuals * nb_timesteps_test * number_tasks158    :param save_h5: boolean, to save the test values in a h5 file or not159    :param activate_plot: to plot for each individual the resulted regressions, losses...160    :return: train Block MGP for multiple individuals161    :return: predicted values (of same size as y_test) at x_test162    BE CAREFUL : x_train and x_test must be the same for all the individuals...163    '''164    flat_block_indices = [item for sublist in block_indices for item in sublist]165    y_predicted = np.nan*np.ones(y_test.numpy().shape)166    a = []167    for i in range(len(block_indices)):168        a.append([])169        for j in range(len(block_indices[i])):170            a[i].append(flat_block_indices.index(block_indices[i][j]))171    block_indices = a172    if len(x_train.shape)>1:173        raise ValueError('Wrong dimensions for the input X_train, x_train should be a 1D Vector')174    if len(x_test.shape)>1:175        raise ValueError('Wrong dimensions for the input X_test, x_test should be a 1D Vector')176    if x_train.shape[0] != y_train.shape[1]:177        raise ValueError('Number of time steps is different for x_train and y_train')178    flat_indices = [item for sublist in block_indices for item in sublist]179    nb_individuals, _, nb_tasks = y_train.shape180    if max(flat_indices) > nb_tasks:181        raise ValueError('One of the block indices is higher than the number of tasks in Y_train')182    list_means = []183    list_covariance_matrix = []184    for i in range(nb_individuals):185        # Training subset?186        if len(train_sample_subset.shape)==0:187            this_train_sample_subset = np.arange(x_train.shape[0])188        elif len(train_sample_subset.shape)==1:189            this_train_sample_subset=train_sample_subset190        elif len(train_sample_subset.shape)==2:191            this_train_sample_subset=train_sample_subset[i]192        else:193            raise ValueError('Error with train_sample_subset.shape={}'.format(train_sample_subset.shape))194        # Just use subset for training195        x_train_cur = x_train[this_train_sample_subset]196        y_train_cur = y_train[i, this_train_sample_subset]197        print('###########      INDIVIDUAL %d    ###########'%i)198        # Define and train199        mgp = Block_MGP(kernel, learning_rate, n_iter, block_indices)200        mgp.build_and_train_block_models(x_train_cur, y_train_cur, smart_end)201        # Plot if desired202        if activate_plot:203            mgp.plot_model(x_train, y_train[i], x_test, train_filter = this_train_sample_subset)204        205        # Predict for this individual206        test_mean_list, test_covar_matrix_list, _ = mgp.test_block_model(x_test)207        list_means.append(test_mean_list)208        list_covariance_matrix.append(test_covar_matrix_list)209        # Keep predicted mean210        for k in range(len(block_indices)):211            y_predicted[i,:,block_indices[k]]=test_mean_list[k].T212    # Save dataset213    h5_dataset_path='{}/{}/trained_models/MGP{}blocks_{}.h5'.format(main_dir, exec_type, len(block_indices), train_sampling_type)214    h5_dataset = h5py.File(h5_dataset_path, 'w')215    # Per block216    for i in range(len(block_indices)):217        cur_mean = np.array([list_means[j][i] for j in range(y_train.shape[0])])218        cur_covariance = np.array([list_covariance_matrix[j][i] for j in range(y_train.shape[0])])219        h5_dataset.create_dataset('mean_block_%d'%i, data=cur_mean)220        h5_dataset.create_dataset('covar_block_%d'%i, data=cur_covariance)221    # All predictions222    h5_dataset.create_dataset('y_predicted', data=y_predicted)223    h5_dataset.close()224    return h5_dataset_path, y_predicted225# Making sure the main program is not executed when the module is imported226if __name__ == '__main__':...Learning_Curve.py
Source:Learning_Curve.py  
1from Cross_Validation import Cross_Validation as crv2from sklearn.utils import shuffle3import matplotlib.pyplot as plt4import numpy as np5import time6class Learning_Curve:7    """8    This class is contains utility methods to compute and plot learning curves9    """10    """ Takes the following parameters as an input:11        learner: a SKLearn Classifier (SKLearn Classifier)12        k: number of iteration of Cross-Validation (float)13        examples: the Bag of Words (sparse matrix of integers)14        labels: labels for each sample (list of integers)15        sizes: sizes to be tested (list of floats between 0 and 1)16        Returns:17        trains_sizes: the tested sizes (list of integers)18        train_scores: the scores achieved on the train set (matrix of floats between 0 and 100)19        test_scores: the scores achieved on the test set (matrix of floats between 0 and 100)20    """21    @staticmethod22    def Learning_Curve(learner, k, examples, labels, sizes, type="shuffle"):23        train_scores = []24        test_scores = []25        train_sizes = []26        examples, labels = shuffle(examples, labels, random_state=int(time.time()))27        for s in sizes:28            size = s*examples.shape[0]29            data_slice, labels_slice = shuffle(examples[:size], labels[:size], random_state=int(time.time()))30            if type == "shuffle":31                score = crv.Shuffle_Cross_Validation(learner, k, data_slice, labels_slice, 0.1)32            elif type == "k-fold":33                score = crv.K_Fold_Cross_Validation(learner, k, data_slice, labels_slice)34            train_scores.append(score[0])35            test_scores.append(score[1])36            train_sizes.append(size)37        return train_sizes, train_scores, test_scores38    """ Takes the following parameters as an input:39        trains_sizes: the tested sizes (list of integers)40        avg_train_scores: the scores achieved on the train set for each size(list of floats between 0 and 100)41        avg_test_scores: the scores achieved on the test set for each size(list of floats between 0 and 100)42        test_std_deviation: the standard deviation on the test set for each size (list of floats)43        train_std_deviation: the standard deviation on the train set for each size (list of floats)44        name: the choosen name (string)45        Returns:46        Nothing.47        Plots the learning curve with the given data48    """49    @staticmethod50    def plot_curve(train_size, avg_test_scores, avg_train_scores, test_std_deviation, train_std_deviation, name):51        print name52        for i in zip(train_size, avg_test_scores):53            print i54        for j in zip(train_size, avg_train_scores):55            print j56        plt.figure()57        plt.plot(train_size, avg_test_scores, 'o-', label=name + " Test", color='blue',)58        plt.plot(train_size, avg_train_scores, 'o-', label=name + " Train", color='green')59        plt.fill_between(train_size, [x[0]+x[1] for x in zip(avg_test_scores, test_std_deviation)],60                 [x[0]-x[1] for x in zip(avg_test_scores, test_std_deviation)], color='blue', alpha=0.2)61        plt.fill_between(train_size, [x[0]+x[1] for x in zip(avg_train_scores, train_std_deviation)],62                 [x[0]-x[1] for x in zip(avg_train_scores, train_std_deviation)], color='green', alpha=0.2)63        plt.legend(bbox_to_anchor=(0., 1.02, 1., .102), loc=1, ncol=1, mode="expand", borderaxespad=0.)64        plt.axis([0, 20000, 0.25, 1.0])65        plt.xticks(range(0, 20000, 1000))66        plt.yticks(np.arange(0.30, 1.0, 0.1))67        for k in range(0, len(train_size)):68            plt.text(train_size[k], avg_test_scores[k], str(round(avg_test_scores[k], 4)))...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
