Towards real-time additive-free dopamine detection at 10<sup>-8</sup> mM with hardware accelerated platform integrated on camera.

1. import matplotlib.pyplot as plt
2. import numpy as np
3. from sklearn.decomposition import PCA
4. from sklearn.metrics import mean_squared_error, r2_score
5. from tensorflow import keras
6. from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau
7. from tensorflow.keras.layers import Dense
8. from tensorflow.keras.models import Sequential
11. class SpectraTrainer:
12. def __init__(self, loader, n_pca=8):
13. self.loader = loader
14. self.n_pca = n_pca
16. self._prepare_data()
17. self._build_model()
19. def _prepare_data(self):
20. self.X_train, self.Y_train = self.loader.get_train_data()
21. self.X_test, self.Y_test = self.loader.get_test_data()
22. all_X, all_Y, all_Y_reg = self.loader.get_all_data()
24. # PCA
25. self.pca = PCA(n_components=self.n_pca, svd_solver="full")
26. self.pca.fit(all_X)
27. self.pca_components = self.pca.components_
29. # Min-max normalize PC
30. self.pc_train = np.matmul(self.X_train, self.pca_components.T)
32. self.pc_min = np.min(self.pc_train, axis=0)
33. self.pc_max = np.max(self.pc_train, axis=0)
34. self.pc_train_normalized = (self.pc_train - self.pc_min) / (
35. self.pc_max - self.pc_min
36. )
38. # Transform test data
39. self.pc_test = np.matmul(self.X_test, self.pca_components.T)
40. self.pc_test_normalized = (self.pc_test - self.pc_min) / (
41. self.pc_max - self.pc_min
42. )
44. # print(f'PCA explained variance: {self.pca.explained_variance_ratio_}')
45. # print(f'Total variance explained: {np.sum(self.pca.explained_variance_ratio_):.4f}')
47. def _build_model(self):
48. self.model = Sequential(name="spectra_regression")
49. self.model.add(Dense(32, activation="relu", input_shape=(self.n_pca,)))
50. self.model.add(Dense(64, activation="relu"))
51. self.model.add(Dense(64, activation="relu"))
52. self.model.add(Dense(1, activation="sigmoid"))
54. optimizer = keras.optimizers.Adam(learning_rate=1e-3)
55. self.model.compile(optimizer=optimizer, loss="mse", metrics=["mae"])
57. def train(self, epochs=100, batch_size=8, verbose=1):
58. early_stopper = EarlyStopping(monitor="loss", patience=15)
59. reduce_lr = ReduceLROnPlateau(
60. monitor="loss", factor=0.2, verbose=0, patience=4, min_lr=1e-6
61. )
63. self.history = self.model.fit(
64. self.pc_train_normalized,
65. self.Y_train,
66. epochs=epochs,
67. verbose=verbose,
68. batch_size=batch_size,
69. callbacks=[reduce_lr, early_stopper],
70. )
72. return self.history
74. def evaluate(self):
75. Y_pred = self.model.predict(self.pc_test_normalized, verbose=0).flatten()
77. # Inverse transform to original concentration scale
78. Y_test_orig = np.array([self.loader.inverse_transform(y) for y in self.Y_test])
79. Y_pred_orig = np.array([self.loader.inverse_transform(y) for y in Y_pred])
81. # Calculate metrics: MSE on normalized Y, R2 on original scale
82. mse = mean_squared_error(self.Y_test, Y_pred)
83. r2 = r2_score(Y_test_orig, Y_pred_orig)
85. print(f"\nTest MSE (normalized): {mse:.6f}")
86. print(f"Test R2 (original scale): {r2:.6f}")
87. return Y_pred, Y_test_orig, Y_pred_orig, mse, r2
89. def evaluate_on_all_data(self):
90. all_X, all_Y, all_Y_reg = self.loader.get_all_data()
91. pc_all = np.matmul(all_X, self.pca_components.T)
92. pc_all_normalized = (pc_all - self.pc_min) / (self.pc_max - self.pc_min)
94. Y_pred = self.model.predict(pc_all_normalized, verbose=0).flatten()
96. # Inverse transform to original concentration scale
97. Y_orig = np.array([self.loader.inverse_transform(y) for y in all_Y_reg])
98. Y_pred_orig = np.array([self.loader.inverse_transform(y) for y in Y_pred])
100. # Calculate metrics: MSE on normalized Y, R2 on original scale
101. mse = mean_squared_error(all_Y_reg, Y_pred)
102. r2 = r2_score(Y_orig, Y_pred_orig)
104. print(f"\nAll Data MSE (normalized): {mse:.6f}")
105. print(f"All Data R2 (original scale): {r2:.6f}")
107. return Y_pred, Y_orig, Y_pred_orig, mse, r2
109. def visualize(self, mse, r2):
110. Y_pred = self.model.predict(self.pc_test_normalized, verbose=0).flatten()
111. Y_test_orig = np.array([self.loader.inverse_transform(y) for y in self.Y_test])
112. Y_pred_orig = np.array([self.loader.inverse_transform(y) for y in Y_pred])
114. # Get concentration list for x-axis
115. c_list = self.loader.selected_concentrations
116. x_ax = np.arange(np.min(c_list), np.max(c_list) + 1)
118. plt.figure(figsize=(10, 8))
119. plt.plot(x_ax, x_ax, "b", linewidth=2)
120. plt.scatter(
121. Y_test_orig, Y_pred_orig, marker="d", color="green", alpha=0.8, s=60
122. )
123. plt.legend(["Ideal fit", "Model Prediction"], fontsize=16)
125. # Create tick labels
126. ticks = []
127. for c in x_ax:
128. ticks.append("$10^{" + str(int(c)) + "}$")
130. plt.xticks(x_ax, ticks, fontsize=12)
131. plt.yticks(x_ax, ticks, fontsize=12)
132. plt.title("mse={:.4f}, r2={:.4f}".format(mse, r2))
134. return plt.gcf()
136. def save_model(self, save_path):
137. self.model.save(save_path)
138. # np.save(save_path.replace('.h5', '_pca.npy'), self.pca_components)
139. print(f"Model saved to {save_path}")
142. if __name__ == "__main__":
143. from dataloader import SpectraDataLoader
145. loader = SpectraDataLoader(
146. data_dir="./data/dopamine_pbs",
147. split_by_batch=False,
148. concentration_list=[0, -11, -10, -9, -8, -7, -6],
149. norm_type="anchor",
150. test_ratio=0.25,
151. )
153. trainer = SpectraTrainer(loader=loader, n_pca=8)
155. print("\nTraining model...")
156. trainer.train(epochs=100, batch_size=8, verbose=1)
158. print("\nEvaluating model...")
159. Y_pred, Y_test_orig, Y_pred_orig, mse, r2 = trainer.evaluate()
161. print("\nEvaluating on all data...")
162. _ = trainer.evaluate_on_all_data()
164. print("\nVisualizing results...")
165. fig = trainer.visualize(mse, r2)
166. plt.show()

trainer.py at commit 8e7e734, no license · at the source