TfData.py

#!/usr/bin/env python
 
 
 
 
import numpy as np
 
 
class TfDataBasf2():
    """
    handles data, necessary for the training
    """
 
    def __init__(self, train_x, train_y, valid_x, valid_y, batch_size, seed=None, epoch_random_shuffle=True):
        """
        declaration of class variables
        """
        
        self.train_xtrain_x = train_x.astype(np.float32)
        
        self.train_ytrain_y = train_y.astype(np.float32)
        
        self.valid_xvalid_x = valid_x.astype(np.float32)
        
        self.valid_yvalid_y = valid_y.astype(np.float32)
 
        
        self.batch_sizebatch_size = batch_size
        
        self.seedseed = seed
        
        self.epoch_random_shuffleepoch_random_shuffle = epoch_random_shuffle
 
        
        self.train_eventstrain_events = self.train_xtrain_x.shape[0]
        
        self.valid_eventsvalid_events = self.valid_xvalid_x.shape[0]
 
        
        self.feature_numberfeature_number = self.train_xtrain_x.shape[1]
 
        
        self.batchesbatches = self.train_xtrain_x.shape[0] // self.batch_sizebatch_size
 
        
        self.train_idxtrain_idx = np.zeros(self.train_xtrain_x.shape[0])
 
        
        self.batch_train_xbatch_train_x = np.zeros((self.feature_numberfeature_number, self.batch_sizebatch_size))
 
        
        self.batch_train_ybatch_train_y = np.zeros(self.batch_sizebatch_size)
 
        
        self.random_staterandom_state = np.random.RandomState(seed)
 
        
        self.sanitize_labelssanitize_labels()
 
    def sanitize_labels(self):
        """
        checks for a binary classification problem
        transforms the two class labels to {0,1}
        """
        # not binary
        assert len(np.unique(self.train_ytrain_y)) == 2
        # different classes
        assert np.array_equal(np.unique(self.train_ytrain_y), np.unique(self.valid_yvalid_y))
 
        # reject corner cases when classes would have special values
        if self.train_ytrain_y.min() > 0:
            self.train_ytrain_y[self.train_ytrain_y == self.train_ytrain_y.min()] = 0
            self.valid_yvalid_y[self.valid_yvalid_y == self.valid_yvalid_y.min()] = 0
 
        if self.train_ytrain_y.max() != 1:
            self.train_ytrain_y[self.train_ytrain_y == self.train_ytrain_y.max()] = 1
            self.valid_yvalid_y[self.valid_yvalid_y == self.valid_yvalid_y.max()] = 1
 
        # transform labels
        if self.train_ytrain_y.min() != 0:
            self.train_ytrain_y[self.train_ytrain_y == self.train_ytrain_y.min()] = 0
            self.valid_yvalid_y[self.valid_yvalid_y == self.valid_yvalid_y.min()] = 0
 
    def batch_iterator(self):
        """
        iterator to provide training batches
        """
        self.train_idxtrain_idx = np.arange(len(self.train_idxtrain_idx))
 
        if self.epoch_random_shuffleepoch_random_shuffle:
            self.random_staterandom_state.shuffle(self.train_idxtrain_idx)
 
        for i in range(self.batchesbatches):
            self.batch_train_xbatch_train_x = self.train_xtrain_x[self.train_idxtrain_idx[i * self.batch_sizebatch_size: (i + 1) * self.batch_sizebatch_size]]
            self.batch_train_ybatch_train_y = self.train_ytrain_y[self.train_idxtrain_idx[i * self.batch_sizebatch_size: (i + 1) * self.batch_sizebatch_size]]
 
            yield self.batch_train_xbatch_train_x, self.batch_train_ybatch_train_y
 
 
class TfDataBasf2Stub():
    """
    stub class just for initializing in basf2 begin_run
    """
 
    def __init__(self, batch_size, feature_number, event_number, train_fraction):
        """
        declare for initialization required batch parameters
        """
        
        self.batch_sizebatch_size = batch_size
 
        
        self.feature_numberfeature_number = feature_number
 
        
        self.batchesbatches = (event_number * train_fraction) // self.batch_sizebatch_size
 
        
        self.train_eventstrain_events = int(train_fraction * event_number)
 
        
        self.valid_eventsvalid_events = int((1 - train_fraction) * event_number)