def __init__(self, config_file_loc):
""" SPECIFIC TO READING FROM .INI CONFIGURATION FILE
self.config_parser = configparser.ConfigParser()
self.config_parser.read(config_file_loc)
self.training_params = self.config_parser["TRAINING"]
# Set training parameters from the configuration file
self.noEpochs = int(self.training_params['noEpochs'])
self.batch_size = int(self.training_params["batch_size"])
self.nNeurons = int(self.training_params["nNeurons"]) # Number of neurons in LSTM layers
self.loss = self.training_params["loss"]
self.h5_file_loc = self.training_params["h5file_loc"] # Location to save the trained model
# READING FROM A YAML CONFIG FILE
self.activation_model = None
self.layer_outputs = None
with open('config.yaml', 'r') as file:
config = yaml.safe_load(file)
self.noEpochs = config['TRAINING']['noEpochs']
self.batch_size = config['TRAINING']["batch_size"]
self.nNeurons = config['TRAINING']["nNeurons"] # Number of neurons in LSTM layers
self.loss = config['TRAINING']["loss"]
self.h5_file_loc = config['TRAINING']["h5file_loc"] # Location to save the trained model
self.model = None # Placeholder for the LSTM model
def load_data(features_csv: str, labels_csv: str, key_column: str, win_size: int) -> tuple[ndarray[Any, Any], Any]:
Loads features and labels from separate CSV files, merges them on a common key, reshapes the features
into sequences, and one-hot encodes the labels. The function also saves the label encoding to a YAML file.
features_csv (str): Path to the CSV file containing the feature data.
labels_csv (str): Path to the CSV file containing the label data.
key_column (str): The column name used as the key for merging the features and labels.
win_size (int): The window size for reshaping the data into sequences of time steps.
Tuple[pd.DataFrame, pd.DataFrame]:
- X_reshaped (pd.DataFrame): A 3D array of reshaped feature data with dimensions
[number_of_sequences, win_size, number_of_features].
- y_one_hot (pd.DataFrame): A 2D array of one-hot encoded labels with dimensions
[number_of_sequences, number_of_classes].
ValueError: If there is an issue with the merging or reshaping of data.
1. Loads the features and labels from the provided CSV files.
2. Merges the two datasets based on the specified key column.
3. Reshapes the features into sequences of `win_size` time steps.
4. Extracts labels, encodes them using a label encoder, and one-hot encodes the labels.
5. Saves the mapping of original label names to their encoded form in a `model_labels.yaml` file.
# Load features and labels
features_df = pd.read_csv(features_csv)
labels_df = pd.read_csv(labels_csv)
# Merge features and labels on the key column
combined_df = pd.merge(features_df, labels_df, on=key_column)
# Adjust 'Labels' to your actual label column name
columns=[combined_df.columns[0],
combined_df.columns[len(combined_df.columns) - 2],
# Reshape X to have sequences of [win_size] timesteps: [number_of_sequences, win_size, number_of_features]
number_of_features = X.shape[1]
number_of_sequences = X.shape[0] // win_size
print("Number of features: ", number_of_features)
print("number of sequences: ", number_of_sequences)
# X_reshaped = X.reshape((number_of_sequences, 90, number_of_features))
X_reshaped = X.reshape((number_of_sequences, win_size, number_of_features))
print("Shape of X:", X_reshaped.shape) # Debugging line to check the shape of X
# Extract labels, taking one label for every win_size timesteps
y = combined_df['Labels'].values[::win_size] # Adjust 'Labels' to your label column name
print("Shape of Y:", y.shape)
label_encoder = LabelEncoder()
y_encoded = label_encoder.fit_transform(y)
y_one_hot = to_categorical(y_encoded)
# To get the list of original labels in the order used during encoding:
original_labels_order = label_encoder.classes_
# print("Original labels in their encoded order:", original_labels_order)
# Create the labels data structure for YAML
'labels': {f'{i}': label for i, label in enumerate(original_labels_order)}
with open('model_labels.yaml', 'w') as file:
yaml.dump(labels_dict, file, sort_keys=False, default_flow_style=False)
return X_reshaped, y_one_hot
def create_model(self, input_shape, num_classes):
Creates and compiles the LSTM model.
input_shape (tuple): Shape of the input data (time steps, features).
num_classes (int): Number of classes in the dataset.
# A Sequential model in Keras is a linear stack of layers
self.model = Sequential()
# The requirements to use the cuDNN implementation are:
# ------------------------------------------------------
# https://www.tensorflow.org/api_docs/python/tf/keras/layers/LSTM
# recurrent_activation == sigmoid
# An LSTM layer is added to the model as the first layer
kernel_regularizer=l2(0.01),
recurrent_regularizer=l2(0.05),
recurrent_dropout=0.2, # needs to be 0 for cuDNN
# self.model.add(BatchNormalization())
self.model.add(TimeDistributed(BatchNormalization()))
# Another LSTM layer is added to the model
self.model.add(LSTM(self.nNeurons,
kernel_regularizer=l2(0.01),
recurrent_regularizer=l2(0.05),
recurrent_dropout=0.0, # needs to be 0 for cuDNN
self.model.add(BatchNormalization())
self.model.add(Dense(num_classes, activation='softmax'))
# Compile the model with Nadam optimizer and a learning rate scheduler
optimizer = Nadam(learning_rate=0.0006)
self.model.compile(optimizer=optimizer, loss='categorical_crossentropy', metrics=['accuracy'])
# prints a summary representation of the model, showing the layout of the layers,
# the shape of the output from each layer, and the number of parameters (weights and biases) in each layer
def train_model(self, X, y):
X (numpy array): Feature data.
y (numpy array): One-hot encoded labels.
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.1, random_state=42)
history = self.model.fit(X_train, y_train,
validation_data=(X_val, y_val),
batch_size=self.batch_size,
callbacks=[early_stopping, lr_scheduler])
# plot_training_history(history)
self.model.save(self.h5_file_loc)
print(f"Model saved at {self.h5_file_loc}")
def run(self, features_csv: str, labels_csv: str, key_column: str, win_size: str):
X, y = self.load_data(features_csv, labels_csv, key_column, win_size)
print("X shape: ", X.shape)
print("y shape: ", y.shape)
input_shape = (X.shape[1], X.shape[2]) # LSTM expects input as [samples, time steps, features]
print(f"number of classes {num_classes}")
self.create_model(input_shape, num_classes)