""" Class for testing the performance of Generative Adversarial Networks in generating synthetic samples for datasets with a minority class. """ import numpy as np from sklearn.decomposition import PCA from sklearn.preprocessing import StandardScaler import matplotlib.pyplot as plt from library.dataset import DataSet, TrainTestData from library.testers import lr,knn, gb, TestResult class Exercise: """ Exercising a test for a minority class extension class. """ def __init__(self, testFunctions=None, shuffleFunction=None, numOfSlices=5, numOfShuffles=5): """ Creates a instance of this class. *testFunctions* is a dictionary /(String : Function)/ of functions for testing a generated dataset. The functions have the signature: /(TrainTestData, TrainTestData) -> TestResult/ *shuffleFunction* is either None or a function /numpy.array -> numpy.array/ that shuffles a given array. *numOfSlices* is an integer > 0. The dataset given for the run function will be divided in such many slices. *numOfShuffles* is an integer > 0. It gives the number of exercised tests. The GAN will be trained and tested (numOfShuffles * numOfSlices) times. """ self.numOfSlices = int(numOfSlices) self.numOfShuffles = int(numOfShuffles) self.shuffleFunction = shuffleFunction self.debug = print self.testFunctions = testFunctions if self.testFunctions is None: self.testFunctions = { "LR": lr, "GB": gb, "KNN": knn } self.results = { name: [] for name in self.testFunctions } # Check if the given values are in valid range. if self.numOfSlices < 0: raise AttributeError(f"Expected numOfSlices to be > 0 but got {self.numOfSlices}") if self.numOfShuffles < 0: raise AttributeError(f"Expected numOfShuffles to be > 0 but got {self.numOfShuffles}") def run(self, gan, dataset): """ Exercise all tests for a given GAN. *gan* is a implemention of library.interfaces.GanBaseClass. It defines the GAN to test. *dataset* is a library.dataset.DataSet that contains the majority class (dataset.data0) and the minority class (dataset.data1) of data for training and testing. """ # Check if the given values are in valid range. if len(dataset.data1) > len(dataset.data0): raise AttributeError( "Expected class 1 to be the minority class but class 1 is bigger than class 0.") # Reset results array. self.results = { name: [] for name in self.testFunctions } # If a shuffle function is given then shuffle the data before the # exercise starts. if self.shuffleFunction is not None: self.debug("-> Shuffling data") for _n in range(3): dataset.shuffleWith(self.shuffleFunction) # Repeat numOfShuffles times self.debug("### Start exercise for synthetic point generator") for shuffleStep in range(self.numOfShuffles): stepTitle = f"Step {shuffleStep + 1}/{self.numOfShuffles}" self.debug(f"\n====== {stepTitle} =======") # If a shuffle function is given then shuffle the data before the next # exercise starts. if self.shuffleFunction is not None: self.debug("-> Shuffling data") dataset.shuffleWith(self.shuffleFunction) # Split the (shuffled) data into numOfSlices slices. # dataSlices is a list of TrainTestData instances. # # If numOfSlices=3 then the data will be splited in D1, D2, D3. # dataSlices will contain: # [(train=D2+D3, test=D1), (train=D1+D3, test=D2), (train=D1+D2, test=D3)] self.debug("-> Spliting data to slices") dataSlices = TrainTestData.splitDataToSlices(dataset, self.numOfSlices) # Do a exercise for every slice. for (sliceNr, sliceData) in enumerate(dataSlices): sliceTitle = f"Slice {sliceNr + 1}/{self.numOfSlices}" self.debug(f"\n------ {stepTitle}: {sliceTitle} -------") self._exerciseWithDataSlice(gan, sliceData) self.debug("### Exercise is done.") for (n, name) in enumerate(self.results): stats = None for (m, result) in enumerate(self.results[name]): stats = result.addMinMaxAvg(stats) (mi, mx, avg) = TestResult.finishMinMaxAvg(stats) self.debug("") self.debug(f"-----[ {avg.title} ]-----") self.debug("maximum:") self.debug(str(mx)) self.debug("") self.debug("average:") self.debug(str(avg)) self.debug("") self.debug("minimum:") self.debug(str(mi)) def _exerciseWithDataSlice(self, gan, dataSlice): """ Runs one test for the given gan and dataSlice. *gan* is a implemention of library.interfaces.GanBaseClass. It defines the GAN to test. *dataSlice* is a library.dataset.TrainTestData instance that contains one data slice with training and testing data. """ # Start over with a new GAN instance. self.debug("-> Reset the GAN") gan.reset() # Train the gan so it can produce synthetic samples. self.debug("-> Train generator for synthetic samples") gan.train(dataSlice.train) # Count how many syhthetic samples are needed. numOfNeededSamples = dataSlice.train.size0 - dataSlice.train.size1 # Add synthetic samples (generated by the GAN) to the minority class. if numOfNeededSamples > 0: self.debug(f"-> create {numOfNeededSamples} synthetic samples") newSamples = gan.generateData(numOfNeededSamples) # Print out an overview of the new dataset. plotCloud(dataSlice.train.data0, dataSlice.train.data1, newSamples) dataSlice.train = DataSet( data0=dataSlice.train.data0, data1=np.concatenate((dataSlice.train.data1, newSamples)) ) # Test this dataset with every given test-function. # The results are printed out and stored to the results dictionary. for testerName in self.testFunctions: self.debug(f"-> test with '{testerName}'") testResult = (self.testFunctions[testerName])(dataSlice) self.debug(str(testResult)) self.results[testerName].append(testResult) def saveResultsTo(self, fileName): avgResults = {} with open(fileName, "w") as f: for (n, name) in enumerate(self.results): if n > 0: f.write("---\n") f.write(name + "\n") isFirst = True stats = None for (m, result) in enumerate(self.results[name]): if isFirst: isFirst = False f.write("Nr.;" + result.csvHeading() + "\n") stats = result.addMinMaxAvg(stats) f.write(f"{m + 1};" + result.toCSV() + "\n") (mi, mx, avg) = TestResult.finishMinMaxAvg(stats) f.write(f"max;" + mx.toCSV() + "\n") f.write(f"avg;" + avg.toCSV() + "\n") f.write(f"min;" + mi.toCSV() + "\n") avgResults[name] = avg return avgResults def plotCloud(data0, data1, dataNew=None, outputFile=None, title=""): """ Does a PCA analysis of the given data and plot the both important axis. """ # Normalizes the data. if dataNew is None: data_t = StandardScaler().fit_transform(np.concatenate([data0, data1])) else: data_t = StandardScaler().fit_transform(np.concatenate([data0, data1, dataNew])) # Run the PCA analysis. pca = PCA(n_components=2) pc = pca.fit_transform(data_t) fig, ax = plt.subplots(sharex=True, sharey=True) fig.set_dpi(600) fig.set_figwidth(10) fig.set_figheight(10) fig.set_facecolor("white") ax.set_title(title) def doSubplot(m, n, c): pca0 = [x[0] for x in pc[m : m + n]] pca1 = [x[1] for x in pc[m : m + n]] s = ax.scatter(pca0, pca1, c=c) m = 0 n = len(data0) doSubplot(m, n, "gray") m += n n = len(data1) doSubplot(m, n, "red") if dataNew is not None: m += n n = len(dataNew) doSubplot(m, n, "blue") ax.legend(title="", loc='upper left', labels=['majority', 'minority', 'synthetic minority']) ax.set_xlabel("PCA0") ax.set_ylabel("PCA1") plt.show() if outputFile is not None: fig.savefig(outputFile)