Our upcoming research path aims to deploy the neural network in real-time, closed-loop clinical environments. One priority is moving beyond pre-rendered static spectrogram images to a dynamic, real-time sliding Short-Time Fourier Transform (STFT) pipeline. This setup will feed continuous scalp EEG streams directly into the model to enable sub-second detection latency. We are also evaluating deeper architecture structures, such as 1D-CNN temporal networks and vision transformers (ViT) that utilize multi-head self-attention to process global correlations across long time scales, which can help capture the subtle changes that precede a seizure.
Furthermore, we plan to validate these models against more diverse, multi-center datasets to test their generalization capability across different patient demographics and varying EEG sensor setups. Finally, to help clinicians interpret model findings, we are integrating explainable AI techniques like Grad-CAM (Gradient-weighted Class Activation Mapping). Grad-CAM overlays a visual heat map on the spectrogram, highlighting the specific frequency bands and time frames that drove the model's prediction and helping clinicians verify its reasoning [INDEX].
import os
import numpy as np
from PIL import Image
import tensorflow as tf
from tensorflow.keras.preprocessing.image import ImageDataGenerator
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv2D, MaxPooling2D, Flatten, Dense, Dropout
from tensorflow.keras import Input
from tensorflow.keras.callbacks import ModelCheckpoint, EarlyStopping
from sklearn.metrics import confusion_matrix
import matplotlib.pyplot as plt
import seaborn as sns
import time
# Set random seeds for reproducibility
np.random.seed(42)
tf.random.set_seed(42)
# Parameters
img_size = (280, 274) # Image size from the study
batch_size = 64 # Better results
epochs = 100 # Number of epochs for training
initial_lr = 0.001 # Better learning rate
augmentation = False # Data augmentation flag
shuffling = True # Data shuffling flag
# Define base directory
base_dir = os.path.expanduser('./RhythmScan/static/data/dataset')
train_dir = os.path.join(base_dir, 'train')
val_dir = os.path.join(base_dir, 'val')
test_dir = os.path.join(base_dir, 'test')
# Function to load images from a folder
def load_images_from_folder(folder, label):
images = []
labels = []
if not os.path.exists(folder):
print(f"Directory {folder} does not exist.")
return np.array(images), np.array(labels)
for filename in os.listdir(folder):
img_path = os.path.join(folder, filename)
try:
img = Image.open(img_path).convert('RGB')
img = img.resize(img_size)
img_array = np.array(img)
images.append(img_array)
labels.append(label)
except Exception as e:
print(f"Error loading image {img_path}: {e}")
return np.array(images), np.array(labels)
# Load images
print("Loading and processing Train set images:")
seizure_train_images, seizure_train_labels = load_images_from_folder(os.path.join(train_dir, 'seizure'), 1)
non_seizure_train_images, non_seizure_train_labels = load_images_from_folder(os.path.join(train_dir, 'non seizure'), 0)
print("Loading and processing Validation set images:")
seizure_val_images, seizure_val_labels = load_images_from_folder(os.path.join(val_dir, 'seizure'), 1)
non_seizure_val_images, non_seizure_val_labels = load_images_from_folder(os.path.join(val_dir, 'non seizure'), 0)
print("Loading and processing Test set images:")
seizure_test_images, seizure_test_labels = load_images_from_folder(os.path.join(test_dir, 'seizure'), 1)
non_seizure_test_images, non_seizure_test_labels = load_images_from_folder(os.path.join(test_dir, 'non seizure'), 0)
# Combine datasets
X_train = np.concatenate((seizure_train_images, non_seizure_train_images), axis=0)
y_train = np.concatenate((seizure_train_labels, non_seizure_train_labels), axis=0)
X_val = np.concatenate((seizure_val_images, non_seizure_val_images), axis=0)
y_val = np.concatenate((seizure_val_labels, non_seizure_val_labels), axis=0)
X_test = np.concatenate((seizure_test_images, non_seizure_test_images), axis=0)
y_test = np.concatenate((seizure_test_labels, non_seizure_test_labels), axis=0)
# Confirm that both 0 and 1 labels are present
print("Unique labels in y_train:", np.unique(y_train))
print("Unique labels in y_val:", np.unique(y_val))
print("Unique labels in y_test:", np.unique(y_test))
# Confirm dataset shapes
print(f"X_train shape: {X_train.shape}")
print(f"y_train shape: {y_train.shape}")
print(f"X_val shape: {X_val.shape}")
print(f"y_val shape: {y_val.shape}")
print(f"X_test shape: {X_test.shape}")
print(f"y_test shape: {y_test.shape}")
# Normalize and reshape data
X_train = X_train / 255.0
X_val = X_val / 255.0
X_test = X_test / 255.0
# One-hot encode labels
y_train = to_categorical(y_train, 2)
y_val = to_categorical(y_val, 2)
y_test = to_categorical(y_test, 2)
# Define the model
model = Sequential([
Input(shape=(img_size[0], img_size[1], 3)),
Conv2D(16, (3, 3), activation='relu'),
MaxPooling2D((2, 2)),
Conv2D(32, (3, 3), activation='relu'),
MaxPooling2D((2, 2)),
Flatten(),
Dense(64, activation='relu'),
Dense(2, activation='softmax')
])
# Compile the model
optimizer = tf.keras.optimizers.Adam(learning_rate=initial_lr)
optimizer_name = 'Adam'
model.compile(optimizer=optimizer, loss='binary_crossentropy', metrics=['accuracy'])
# Data augmentation (only if augmentation is set to True)
if augmentation:
datagen = ImageDataGenerator(
rotation_range=20,
width_shift_range=0.2,
height_shift_range=0.2,
horizontal_flip=True
)
else:
datagen = ImageDataGenerator()
# Prepare data generators
train_generator = datagen.flow(X_train, y_train, batch_size=batch_size, shuffle=shuffling)
val_generator = datagen.flow(X_val, y_val, batch_size=batch_size, shuffle=shuffling)
# Callbacks for saving the best model and early stopping
model_checkpoint = ModelCheckpoint(os.path.expanduser('./RhythmScan/best_model.keras'), monitor='val_loss', save_best_only=True)
early_stopping = EarlyStopping(monitor='val_loss', patience=10, restore_best_weights=True)
# Track training time
start_time = time.time()
# Train the model
history = model.fit(
train_generator,
validation_data=val_generator,
epochs=epochs,
steps_per_epoch=len(X_train) // batch_size,
validation_steps=len(X_val) // batch_size,
callbacks=[model_checkpoint, early_stopping]
)
# Calculate training time per epoch
training_time_per_epoch = (time.time() - start_time) / epochs
# Save the final model
model.save(os.path.expanduser('./RhythmScan/seizure_detection_final_model.keras'))
# Evaluate the model
val_loss, val_acc = model.evaluate(X_test, y_test)
print(f'Test accuracy: {val_acc * 100:.2f}%')
# Model size
model_size = os.path.getsize(os.path.expanduser('./RhythmScan/seizure_detection_final_model.keras')) / (1024 * 1024) # in MB
# Calculate averages for accuracy and loss
avg_train_acc = np.mean(history.history['accuracy'])
avg_val_acc = np.mean(history.history['val_accuracy'])
avg_train_loss = np.mean(history.history['loss'])
avg_val_loss = np.mean(history.history['val_loss'])
# Output results
print(f'Average Training Accuracy: {avg_train_acc * 100:.2f}%')
print(f'Average Validation Accuracy: {avg_val_acc * 100:.2f}%')
print(f'Average Training Loss: {avg_train_loss:.4f}')
print(f'Average Validation Loss: {avg_val_loss:.4f}')
print(f'Training time per epoch: {training_time_per_epoch:.2f} seconds')
print(f'Model size: {model_size:.2f} MB')
# Plot accuracy and loss over epochs
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.plot(history.history['accuracy'], label='Train')
plt.plot(history.history['val_accuracy'], label='Validation')
plt.title('Model accuracy')
plt.ylabel('Accuracy')
plt.xlabel('Epoch')
plt.legend(loc='upper left')
plt.subplot(1, 2, 2)
plt.plot(history.history['loss'], label='Train')
plt.plot(history.history['val_loss'], label='Validation')
plt.title('Model loss')
plt.ylabel('Loss')
plt.xlabel('Epoch')
plt.legend(loc='upper left')
# Save the figure
timestamp = time.strftime("%Y%m%d-%H%M%S")
save_dir = os.path.expanduser(f'./RhythmScan/static/results_{timestamp}')
os.makedirs(save_dir, exist_ok=True)
plt.savefig(os.path.join(save_dir, f'accuracy_loss_{timestamp}.png'))
plt.show()
# Predict the classes for the test set
y_pred = model.predict(X_test)
y_pred_classes = np.argmax(y_pred, axis=1)
y_true_classes = np.argmax(y_test, axis=1)
# Find misclassified indices
misclassified_indices = np.where(y_pred_classes != y_true_classes)[0]
# Plot the confusion matrix
conf_matrix = confusion_matrix(y_true_classes, y_pred_classes)
plt.figure(figsize=(10, 7))
sns.heatmap(conf_matrix, annot=True, fmt="d", cmap="Blues", xticklabels=['Non-Seizure', 'Seizure'], yticklabels=['Non-Seizure', 'Seizure'])
plt.title('Confusion Matrix')
plt.xlabel('Predicted Label')
plt.ylabel('True Label')
# Save the confusion matrix figure
plt.savefig(os.path.join(save_dir, f'confusion_matrix_{timestamp}.png'))
plt.show()
# Save the terminal output to a log file
log_file_path = os.path.join(save_dir, f'training_log_{timestamp}.txt')
output_text = f"""Test accuracy: {val_acc * 100:.2f}%
Average Training Accuracy: {avg_train_acc * 100:.2f}%
Average Validation Accuracy: {avg_val_acc * 100:.2f}%
Average Training Loss: {avg_train_loss:.4f}
Average Validation Loss: {avg_val_loss:.4f}
Training time per epoch: {training_time_per_epoch:.2f} seconds
Model size: {model_size:.2f} MB
# Model Settings:
Batch Size: {batch_size}
Epochs: {epochs}
Initial Learning Rate: {initial_lr}
Augmentation: {augmentation}
Shuffling: {shuffling}
Optimizer: {optimizer_name}
Random Seed: 42
Early Stopping Patience: {early_stopping.patience}
"""
# Save log text to the file
with open(log_file_path, 'w') as log_file:
log_file.write(output_text)
# Plot the first 10 misclassified images
num_images_to_display = 10
plt.figure(figsize=(15, 15))
for i, index in enumerate(misclassified_indices[:num_images_to_display]):
plt.subplot(5, 2, i + 1)
plt.imshow(X_test[index])
plt.title(f"True label: {y_true_classes[index]}, Predicted: {y_pred_classes[index]}")
plt.axis('off')
# Save misclassified images plot
plt.tight_layout()
plt.savefig(os.path.join(save_dir, f'misclassified_images_{timestamp}.png'))
plt.show()