Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, path):
self.path = os.path.abspath(path)
self.train_path = os.path.join(self.path, 'data-train.csv')
self.test_path = os.path.join(self.path, 'data-test.csv')
self.valid_path = os.path.join(self.path, 'data-validation.csv')
self.saver = Saver(self)
self.loader = Loader(self)
self.do_save = True
self.is_flat = True
self.n_labels = 0
self.train = None
self.test = None
self.validation = None
self.X_train = None
self.Y_train = None
self.X_test = None
self.Y_test = None
self.X_val = None
self.Y_val = None
self.X = None
self.Y = None
def save(self):
train_path = self.dataset.train_path
test_path = self.dataset.test_path
valid_path = self.dataset.valid_path
flat = self.dataset.is_flat
if not self.dataset.is_flat:
train_path = train_path.replace('.csv', '.pkl')
test_path = test_path.replace('.csv', '.pkl')
valid_path = valid_path.replace('.csv', '.pkl')
self.threads = [ \
threading.Thread(target=Saver._worker, args=( self.dataset.train, train_path, flat, )),
threading.Thread(target=Saver._worker, args=( self.dataset.test, test_path, flat, )),
threading.Thread(target=Saver._worker, args=( self.dataset.validation, valid_path, flat, ))
]
for t in self.threads:
t.start()