Contiguous temporal withholding for hemispheric analysis in frontal EEG during cycling.

1. # -*- coding: utf-8 -*-
2. """
3. Created on Wed Feb 11 14:02:57 2026
5. @author: USER
6. """
8. # -*- coding: utf-8 -*-
9. """
10. BiLSTM baseline with contiguous temporal test-block evaluation.
12. Dataset overview
13. - The database contains 4 subjects.
14. - For each subject, multiple pedaling sessions were recorded.
15. - Each session was segmented into contraction epochs.
16. - For each contraction epoch, 18 electrode-wise segments were generated (one per electrode).
17. - The .mat file stores the concatenation of all sessions for that subject.
19. Therefore, the total number of samples in X is:
20. N = (pedaling_sessions) × (contraction_epochs_per_session) × 18
22. Per subject (from the dataset overview table):
23. - Subject 1: 2 × 100 × 18 = 3600
24. - Subject 2: 3 × 100 × 18 = 5400
25. - Subject 3: 3 × 62 × 18 = 3348
26. - Subject 4: 3 × 40 × 18 = 2160
28. Variables inside each .mat file
29. - X: array shaped (N, 1, T)
30. N = total electrode-wise segments across all sessions
31. T = number of time samples per segment (e.g., 164 for Subject 1 in the uploaded file)
32. Values are float64.
34. - Y_text: array shaped (N, 1)
35. Hemisphere labels as strings '0' and '1', one label per electrode-wise segment:
36. '0' = Left hemisphere electrode
37. '1' = Right hemisphere electrode
39. Important
40. - Each input sample to the BiLSTM corresponds to ONE electrode segment from ONE contraction epoch
41. (not a full 18-channel contraction).
43. Evaluation protocol (contiguous temporal ablation)
44. - Select a contiguous block of length floor(test_fraction * N) as TEST.
45. - Use the remaining samples as a pool; shuffle them; split into TRAIN and VALIDATION.
46. - Train the BiLSTM with early stopping and learning-rate reduction.
47. - Evaluate on the TEST block.
48. - Repeat across num_blocks start positions distributed across the time axis.
49. - Repeat the whole procedure repeat_index = 1..5 and save metrics to Excel each time.
50. """
52. import time
53. import scipy.io
54. import numpy as np
55. import tensorflow as tf
56. import pandas as pd
58. from sklearn.preprocessing import LabelEncoder
59. from sklearn.metrics import confusion_matrix, precision_recall_fscore_support
61. from tensorflow.keras.models import Sequential
62. from tensorflow.keras.layers import Bidirectional, LSTM, Dense, Dropout
63. from tensorflow.keras.utils import to_categorical
64. from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
65. from tensorflow.keras.optimizers import Adam
68. # ---------------- Baseline hyperparameters ----------------
70. for subject_number in range(1, 5):
72. validation_fraction = 0.1 # fraction of the remaining pool reserved for validation
73. epochs = 40
74. batch_size = 32
75. test_fraction = 0.20 # contiguous fraction of samples used as TEST
76. num_blocks = 20 # number of contiguous TEST blocks (start positions)
77. num_repeats = 5 # number of full repeats saved as separate Excel files
79. for repeat_index in range(1, num_repeats + 1):
81. # Reproducibility (same seed per repeat_index)
82. np.random.seed(42)
83. tf.random.set_seed(42)
85. print(f"\nSubject: {subject_number}")
86. print("Model: BiLSTM")
87. print("Protocol: contiguous TEST block; validation from shuffled remainder")
88. print("Labels: 0 = Left, 1 = Right")
90. # ---------------- Load data ----------------
91. # Keep filename convention consistent with your stored files.
92. # If your files are named 'XY_Voluntario_{k}_text.mat', change this line accordingly.
93. mat_filename = f"Subject_{subject_number}.mat"
94. mat_data = scipy.io.loadmat(mat_filename)
96. # Raw variables from file
97. signal_segments = mat_data["X"] # shape (N, 1, T)
98. hemisphere_labels_raw = mat_data["Y_text"].squeeze() # shape (N,) or (N,1) after squeeze
100. # Convert labels to a clean 1D string array, e.g., ['0','1',...]
101. hemisphere_labels_str = np.array([str(s[0]) for s in hemisphere_labels_raw])
103. # Encode labels (strings -> int -> one-hot)
104. label_encoder = LabelEncoder()
105. hemisphere_labels_int = label_encoder.fit_transform(hemisphere_labels_str)
106. hemisphere_labels_onehot = to_categorical(hemisphere_labels_int)
108. # Reshape signals for LSTM input: (N, T, 1)
109. # The middle dimension (1) is removed; each sample is an electrode-wise temporal segment.
110. signal_segments_flat = signal_segments.squeeze(axis=1) # (N, T)
111. signal_lstm_input = signal_segments_flat[..., np.newaxis] # (N, T, 1)
113. total_samples = signal_lstm_input.shape[0]
114. print("Total samples (electrode-wise segments):", total_samples)
116. # Length of contiguous TEST block
117. test_block_length = int(np.floor(test_fraction * total_samples))
118. if test_block_length < 1:
119. raise ValueError("Contiguous TEST block is too small. Increase N or test_fraction.")
121. # Start indices distributed across the time axis
122. test_start_indices = np.linspace(0, total_samples - test_block_length, num_blocks, dtype=int)
124. # ---------------- Callbacks (baseline) ----------------
125. callbacks = [
126. EarlyStopping(
127. monitor="val_loss",
128. patience=8,
129. restore_best_weights=True
130. ),
131. ReduceLROnPlateau(
132. monitor="val_loss",
133. factor=0.5,
134. patience=4,
135. verbose=1
136. )
137. ]
139. # Store metrics for each contiguous block run
140. all_results = []
142. # ---------------- Contiguous block evaluation ----------------
143. for block_index, test_start in enumerate(test_start_indices):
145. test_end = test_start + test_block_length
147. print(f"\n=== RUN {block_index + 1}/{num_blocks} ===")
148. print(f"TEST block indices: {test_start}–{test_end - 1}")
150. # Mask to exclude the TEST block from the training pool
151. keep_mask = np.ones(total_samples, dtype=bool)
152. keep_mask[test_start:test_end] = False
154. # Remaining pool (TRAIN + VALIDATION)
155. remaining_signals = signal_lstm_input[keep_mask]
156. remaining_labels = hemisphere_labels_onehot[keep_mask]
158. # TEST block (contiguous in time)
159. test_signals = signal_lstm_input[test_start:test_end]
160. test_labels = hemisphere_labels_onehot[test_start:test_end]
162. # Shuffle the remaining pool before splitting into TRAIN / VALIDATION
163. shuffled_indices = np.arange(remaining_signals.shape[0])
164. np.random.shuffle(shuffled_indices)
166. remaining_signals = remaining_signals[shuffled_indices]
167. remaining_labels = remaining_labels[shuffled_indices]
169. # Manual TRAIN / VALIDATION split
170. split_index = int((1.0 - validation_fraction) * remaining_signals.shape[0])
172. train_signals = remaining_signals[:split_index]
173. train_labels = remaining_labels[:split_index]
175. val_signals = remaining_signals[split_index:]
176. val_labels = remaining_labels[split_index:]
178. # ---------------- Model (rebuilt each run) ----------------
179. model = Sequential([
180. Bidirectional(
181. LSTM(
182. 32,
183. input_shape=(signal_lstm_input.shape[1], signal_lstm_input.shape[2])
184. )
185. ),
186. Dropout(0.5),
187. Dense(64, activation="relu"),
188. Dropout(0.4),
189. Dense(hemisphere_labels_onehot.shape[1], activation="softmax")
190. ])
192. model.compile(
193. optimizer=Adam(learning_rate=0.001, clipnorm=1.0),
194. loss="categorical_crossentropy",
195. metrics=["accuracy"]
196. )
198. start_time = time.time()
200. history = model.fit(
201. train_signals, train_labels,
202. epochs=epochs,
203. batch_size=batch_size,
204. validation_data=(val_signals, val_labels),
205. callbacks=callbacks,
206. verbose=0,
207. shuffle=True
208. )
210. elapsed_s = time.time() - start_time
212. # Final training/validation accuracy from the last epoch recorded
213. train_acc = history.history["accuracy"][-1]
214. val_acc = history.history["val_accuracy"][-1]
215. gap_train_val = train_acc - val_acc
217. # Evaluate on TEST block
218. test_loss, test_acc = model.evaluate(test_signals, test_labels, verbose=0)
220. # Predictions on TEST
221. test_prob = model.predict(test_signals, verbose=0)
222. test_pred = np.argmax(test_prob, axis=1)
223. test_true = np.argmax(test_labels, axis=1)
225. # Confusion matrix and per-class metrics
226. cm = confusion_matrix(test_true, test_pred)
228. # Using labels=[0,1] ensures consistent order: class 0 (Left), class 1 (Right)
229. precision, recall, f1, _ = precision_recall_fscore_support(
230. test_true,
231. test_pred,
232. average=None,
233. labels=[0, 1],
234. zero_division=0
235. )
237. print(f"Train acc (final): {train_acc:.4f}")
238. print(f"Val acc (final): {val_acc:.4f}")
239. print(f"Test acc: {test_acc:.4f}")
240. print(f"Recall (Left=0): {recall[0]:.4f}")
241. print(f"Recall (Right=1): {recall[1]:.4f}")
242. print(f"F1 (Left=0): {f1[0]:.4f}")
243. print(f"F1 (Right=1): {f1[1]:.4f}")
244. print(f"Time (s): {elapsed_s:.2f}")
246. # Store results for this contiguous block
247. all_results.append({
248. "run": block_index + 1,
249. "test_block_start": int(test_start),
250. "test_block_end": int(test_end - 1),
251. "n_train": int(train_signals.shape[0]),
252. "n_val": int(val_signals.shape[0]),
253. "n_test": int(test_signals.shape[0]),
254. "train_acc": float(train_acc),
255. "val_acc": float(val_acc),
256. "test_acc": float(test_acc),
257. "gap_train_val": float(gap_train_val),
258. "time_s": float(elapsed_s),
259. "epochs_used": int(len(history.history["accuracy"])),
261. # Confusion matrix entries for binary classification
262. "cm_00": int(cm[0, 0]),
263. "cm_01": int(cm[0, 1]),
264. "cm_10": int(cm[1, 0]),
265. "cm_11": int(cm[1, 1]),
267. # Class-wise metrics (0=Left, 1=Right)
268. "prec_0_left": float(precision[0]),
269. "rec_0_left": float(recall[0]),
270. "f1_0_left": float(f1[0]),
271. "prec_1_right": float(precision[1]),
272. "rec_1_right": float(recall[1]),
273. "f1_1_right": float(f1[1]),
274. })
276. # ---------------- End-of-repeat summary ----------------
277. print("\n===== SUMMARY OF CONTIGUOUS BLOCK RUNS =====")
278. for r in all_results:
279. print(
280. f"Run {r['run']:02d} | "
281. f"Train {r['train_acc']:.3f} | "
282. f"Val {r['val_acc']:.3f} | "
283. f"Test {r['test_acc']:.3f} | "
284. f"rec(L=0) {r['rec_0_left']:.3f} | rec(R=1) {r['rec_1_right']:.3f} | "
285. f"prec(L=0) {r['prec_0_left']:.3f} | prec(R=1) {r['prec_1_right']:.3f} | "
286. f"f1(L=0) {r['f1_0_left']:.3f} | f1(R=1) {r['f1_1_right']:.3f} | "
287. f"test_block {r['test_block_start']}-{r['test_block_end']}"
288. )
290. # ---------------- Save results to Excel ----------------
291. results_df = pd.DataFrame(all_results)
293. # Output name per subject and repeat index
294. excel_name = f"Subject{subject_number}_results_repeat{repeat_index}.xlsx"
295. results_df.to_excel(excel_name, index=False)
297. print(f"File saved: {excel_name}")

Test_block.py at commit a102e4b, no license · at the source