I have a big dataset (too big for my RAM) -> I though about using a Custom made Data Generator.
When I use the data generator, the values are pretty much stuck on these (note the Validation Recall and Precision, although the Training values aren't good at all):
2360/2360 [==============================] - 283s 119ms/step - loss: 1.8525 - accuracy: 0.9723 - Precision: 0.2250 - Recall: 0.2259 - val_loss: 0.7152 - val_accuracy: 0.7497 - val_Precision: 0.0000e+00 - val_Recall: 0.0000e+00
and when I run it without the generator (hence, loading all of the X, y pairs to the RAM at once) I get values like these (which are legit for first epoch):
2360/2360 [==============================] - 209s 129ms/step - loss: 2.9429 - accuracy: 0.9211 - Precision: 0.7750 - Recall: 0.7340 - val_loss: 0.2236 - val_accuracy: 0.9816 - val_Precision: 0.9592 - val_Recall: 0.9528
I can't understand why does it happen. I tested my generator offline (not on training) and it loads the data correctly.
the class:
import datetime
import time
import numpy as np
import mne
import pandas as pd
import pickle
import h5py
import threading
import queue
from tensorflow.python.keras.utils.all_utils import Sequence
from src.generate.MetaDict import MetaDict
from src.generate.window_and_labels import *
from src.generate.preprocess import *
from pathlib import Path
class DataGenerator(Sequence):
DATASET_FILE_NAME = 'dataset.h5'
META_DATASET_FILE_NAME = 'meta_dataset.pkl'
SUBJECT_NAME_LENGTH = 5
THREADS = 8
def __init__(self, batch_size=32, n_channels=23, shuffle=True, n_classes=1, balance_dataset=True,
**kwargs):
self._indexes = None
self._recording_file_open = None
self._meta_dict = MetaDict()
self._batch_size = batch_size
self._n_channels = n_channels
self._shuffle = shuffle
self._n_classes = n_classes
self._channels_last = kwargs.pop('channels_last', False)
self._allow_fft = kwargs.pop('allow_fft', True)
self._do_normalize_not_standardize = kwargs.pop('do_normalize_not_standardize', False)
# the ratio between the amount of the samples with label '0' and the samples with label '1'
# if save_ratio == 1 -> #1_samples = #0_samples
self._save_ratio = kwargs.pop('save_ratio', 1)
self._meta_dataset = kwargs.pop('meta_dataset', None)
self._win_size_seconds = kwargs.pop('win_size', None) # in seconds
self._overlap_size = kwargs.pop('overlap_size', None) # in seconds
current_dir = Path(__file__)
self._project_dir = [p for p in current_dir.parents if p.parts[-1] == 'SeizureDetectionResearch'][0]
if self._meta_dataset is None or not isinstance(self._meta_dataset, pd.DataFrame):
with open(os.path.join(self._project_dir, DataGenerator.META_DATASET_FILE_NAME), 'rb') as f:
dataset_params = pickle.load(f)
self._meta_dataset = dataset_params['meta_dataset']
self._win_size_seconds = dataset_params['win_size']
self._overlap_size = dataset_params['overlap_size']
self._dataset = h5py.File(os.path.join(self._project_dir, DataGenerator.DATASET_FILE_NAME), 'r')
self._loaded_recording = dict(recording_file=None, samples=None)
if balance_dataset:
self._remove_zero_seizure_files()
self._balance_dataset()
# unique returns the items based on their order!
self._recording_file_names_list = list(pd.unique(self._meta_dataset['recording_file']))
# dataset must be balanced.
self._samples_per_file_name = {f_name: sub_df
for f_name, sub_df in
self._meta_dataset.groupby('recording_file', as_index=False)}
# TODO: remove
self._is_run = True
self.batch = [None] * self._batch_size
self.labels = [None] * self._batch_size
self.lock = threading.Lock()
self.q = queue.Queue()
self.threads = [threading.Thread(target=self.worker) for _ in range(self.THREADS)]
for t in self.threads:
t.start()
# TODO: to here
self.on_epoch_end()
def __len__(self):
"""Denotes the number of batches per epoch"""
return len(self._meta_dataset) // self._batch_size
def stop(self):
with self.lock:
self._is_run = False
def worker(self):
while True:
if not self._is_run:
return
try:
samples_indices, placement = self.q.get(block=True, timeout=1)
except:
pass
else:
try:
samples, labels = self._data_generation(samples_indices)
except Exception as e:
print(e)
else:
with self.lock:
self.batch[placement:placement + int(self._batch_size / self.THREADS)] = samples
self.labels[placement:placement + int(self._batch_size / self.THREADS)] = labels
self.q.task_done()
def __getitem__(self, batch_index):
"""Generate one batch of data"""
# Generate indexes of the batch
indexes = self._indexes[batch_index * self._batch_size:(batch_index + 1) * self._batch_size]
for i in range(len(self.threads)):
samples_indices = indexes[
int(self._batch_size / self.THREADS) * i:int(self._batch_size / self.THREADS) * (i + 1)]
placement = int(self._batch_size / self.THREADS) * i
self.q.put((samples_indices, placement))
self.q.join()
samples = np.array(self.batch, dtype=np.float64)
labels = np.array(self.labels, dtype=np.float64)
# Generate data
# samples, labels = self._data_generation(indexes)
# samples = samples.reshape(-1, *samples.shape) # was removed since it was doing trouble.
return samples, labels
def __repr__(self):
return '\n'.join([f'Samples count = {len(self._meta_dataset)}',
f'Batches count = {len(self)}',
str(self._meta_dataset.label.value_counts())])
def _data_generation(self, indices):
"""Generates data containing batch_size samples"""
# X : (n_samples, *dim, n_channels)
# Initialization
# TODO: change here too
batch_shape = self.get_batch_shape()
samples = np.zeros((len(indices), *batch_shape[1:]))
labels = np.zeros(len(indices), dtype=int)
# samples = np.zeros(self.get_batch_shape())
# labels = np.zeros(self._batch_size, dtype=int)
# Generate data
for sample_order_idx, sample_abs_idx in enumerate(indices):
samples[sample_order_idx], labels[sample_order_idx] = self._load_sample_on_idx(sample_abs_idx)
# to_categorical makes kind of 'one_hot' label vector.
# 1, 2, 3, 4... -> [1,0,0,0], [0,1,0,0]...
if self._n_classes > 1:
return samples, tf.keras.utils.to_categorical(labels, num_classes=self._n_classes)
else:
return samples, labels
def _load_sample_on_idx(self, index: int) -> tuple:
row = self._meta_dataset.iloc[index]
rec_name = row['recording_file']
win_start_idx = row['win_start_idx']
label = row['label']
# with self.lock: # TODO: ALSO HERE
# if self._loaded_recording['recording_file'] != rec_name:
# print(rec_name, '->', self._loaded_recording['recording_file'])
# self._loaded_recording['recording_file'] = rec_name
# subject_folder = rec_name[:DataGenerator.SUBJECT_NAME_LENGTH]
# sample = self._dataset.get('/'.join([subject_folder, rec_name]))
# self._loaded_recording['samples'] = sample[()]
# TODO: Change Here
subject_folder = rec_name[:DataGenerator.SUBJECT_NAME_LENGTH]
samples = self._dataset.get('/'.join([subject_folder, rec_name]))
sample = samples[:, win_start_idx: win_start_idx + int(self._win_size_seconds * SAMPLING_FREQUENCY)]
# sample = \
# self._loaded_recording['samples'][:,
# win_start_idx: win_start_idx + int(self._win_size_seconds * SAMPLING_FREQUENCY)]
sample = self.sample_pipeline(sample, self._allow_fft, self._channels_last, self._do_normalize_not_standardize)
return sample, label
@staticmethod
def sample_pipeline(sample: np.ndarray, allow_fft: bool, channels_last: bool, normalize_vs_standardize: bool):
"""
the pipeline that each sample should go through when imported from database
:param normalize_vs_standardize: True = Normalize. False = Standardize
:param channels_last: are the channels last axis.
:param allow_fft: is FFT needed
:param sample: a sample.
:return: processed sample.
"""
# before transposing (if needed), do the fft channel wise.
if allow_fft:
sample = abs_fft(sample)
elif normalize_vs_standardize:
sample = normalize_data(sample)
else:
sample = standardize_data(sample)
if channels_last: # the original data is saved as (Channels, Samples)
sample = np.transpose(sample)
return sample
def on_epoch_end(self):
"""Updates indexes after each epoch"""
if self._shuffle:
self.create_random_index_list()
else:
self._indexes = np.arange(len(self._meta_dataset))
def create_random_index_list(self) -> None:
"""
creates random list of indices.
since the main dataset (dataframe) doesn't change, we roll random indices. main concern is that the indices order
won't require recording files opening frequently since it's a heavy operation. so for each recording,
we will roll the indices, randomize them and place them in the main indices
:return:
"""
self._indexes = []
indices = {}
last_idx = 0
# self._recording_files is sorted by occurrence appearance!
# randomizing the samples in the file
for file_name in self._recording_file_names_list:
temp_indices = np.arange(len(self._samples_per_file_name[file_name])) + last_idx
np.random.shuffle(temp_indices)
indices[file_name] = temp_indices
last_idx += len(temp_indices)
k = list(indices.keys())
np.random.shuffle(k)
# randomizing the file's occurrence.
for f_name in k:
self._indexes += indices[f_name].tolist()
def _remove_zero_seizure_files(self):
only_seizures = \
list(filter(lambda file_name: self._meta_dict.get_record_seizure_count(file_name) != 0,
list(pd.unique(self._meta_dataset['recording_file']))))
self._meta_dataset = self._meta_dataset.loc[self._meta_dataset.recording_file.isin(only_seizures)]
def _balance_dataset(self):
label_count = self._meta_dataset.label.value_counts()
diff = label_count[0] - label_count[1] * self._save_ratio
index_label_zero = self._meta_dataset[self._meta_dataset.label == 0.0].index[:diff]
self._meta_dataset.drop(index_label_zero, inplace=True)
def labels_to_list(self):
all_labels = []
for i in range(len(self)):
_, labels = self.__getitem__(i)
all_labels.extend(labels)
return all_labels
def get_batch_shape(self):
if self._channels_last:
return self._batch_size, int(self._win_size_seconds * SAMPLING_FREQUENCY), self._n_channels
else:
return self._batch_size, self._n_channels, int(self._win_size_seconds * SAMPLING_FREQUENCY)
def get_data_for_file(self, record_name) -> np.ndarray:
"""
returns all the data from a recording. has no use in training.
:param record_name: the name of the record.
:return: data.
"""
data = self._dataset[record_name[:DataGenerator.SUBJECT_NAME_LENGTH]][record_name][()]
return data
def get_all_data(self):
batch_shape = self.get_batch_shape()
X = np.zeros([len(self) * self._batch_size, batch_shape[1], batch_shape[2]])
y = np.zeros([len(self) * self._batch_size])
for batch_index in range(len(self)):
samples, labels = self.__getitem__(batch_index)
for sample_idx in range(self._batch_size):
X[batch_index * self._batch_size + sample_idx, :, :] = samples[sample_idx, :, :]
y[batch_index * self._batch_size + sample_idx] = labels[sample_idx]
return X, y
@property
def win_size(self):
return self._win_size_seconds
@property
def overlap_size(self):
return self._overlap_size
@property
def shuffle(self):
return self._shuffle
@shuffle.setter
def shuffle(self, new_val):
self._shuffle = new_val
#
# def fft_pipeline(samples) -> np.ndarray:
# """
# the function needs to take a window only! (23, #samples)
# :param samples: the window of samples
# :return: normalized FFT with 0 DC
# """
#
# # samples[:, 0] = 0
# # samples = samples[:, 1:len(samples[0, :])//2-1] # remove DC and mirror of fft
# # samples = normalize_data(samples, out_min_value=0)
# return samples