{ "cells": [ { "cell_type": "code", "execution_count": null, "id": "0915fab8", "metadata": {}, "outputs": [], "source": [ "import glob\n", "import os\n", "\n", "\n", "import numpy as np\n", "import pandas as pd\n", "\n", "\n", "# ============================================================\n", "# CONFIGURATION\n", "# ============================================================\n", "\n", "\n", "BASE_DIR = os.getcwd()\n", "\n", "\n", "INPUT_BASE = os.path.join(BASE_DIR, \"Reorganized-Data\")\n", "\n", "\n", "FFT_OUTPUT_BASE = os.path.join(\n", " BASE_DIR,\n", " \"CNN-Arrays\",\n", " \"FFT\",\n", ")\n", "\n", "\n", "# Lowest sampling rate\n", "FS = 12800\n", "FS_FOLDER = \"12,8kHz\"\n", "\n", "\n", "# Higher-load points only\n", "LOADS = [\"323Nm_2000RPM\"]\n", "\n", "\n", "# Folder names created by your reorganization script\n", "STATES = [\"Healthy\", \"IR\", \"OR\", \"Corrosion\"]\n", "\n", "\n", "# FFT / 1D CNN settings\n", "N_FFT_BINS = FS // 2\n", "\n", "\n", "# ============================================================\n", "# HELPER FUNCTIONS\n", "# ============================================================\n", "\n", "\n", "def load_vibration_csv(path):\n", " \"\"\"\n", " Load vibration data from a CSV file.\n", "\n", "\n", " Tries common vibration-column names first. If none are found,\n", " it uses the first numeric column.\n", " \"\"\"\n", " df = pd.read_csv(path)\n", "\n", "\n", " for col in [\n", " \"Vibration\",\n", " \"vibration\",\n", " \"Acceleration\",\n", " \"acceleration\",\n", " \"Signal\",\n", " \"signal\",\n", " ]:\n", " if col in df.columns:\n", " return df[col].to_numpy(dtype=float)\n", "\n", "\n", " numeric_cols = df.select_dtypes(include=[np.number]).columns\n", "\n", "\n", " if len(numeric_cols) == 0:\n", " raise ValueError(f\"No numeric columns found in: {path}\")\n", "\n", "\n", " return df[numeric_cols[0]].to_numpy(dtype=float)\n", "\n", "\n", "\n", "def compute_fft_array(signal, fs, n_bins=N_FFT_BINS):\n", " \"\"\"\n", " Return a fixed-length float32 log-magnitude FFT array.\n", "\n", "\n", " Adjacent frequency bins are averaged.\n", " No per-recording min-max normalization is applied.\n", " \"\"\"\n", " signal = np.asarray(signal, dtype=np.float64)\n", "\n", "\n", " if signal.ndim != 1:\n", " raise ValueError(f\"Expected a 1D signal, got {signal.shape}\")\n", "\n", "\n", " if not np.all(np.isfinite(signal)):\n", " raise ValueError(\"Signal contains NaN or infinite values\")\n", "\n", "\n", " if len(signal) < 4:\n", " raise ValueError(\"Signal is too short\")\n", "\n", "\n", " if fs <= 0:\n", " raise ValueError(\"Sampling rate must be positive\")\n", "\n", "\n", " signal = signal - signal.mean()\n", "\n", "\n", " window = np.hanning(len(signal))\n", " magnitude = np.abs(np.fft.rfft(signal * window)) / window.sum()\n", "\n", "\n", " # One-sided amplitude scaling\n", " if len(signal) % 2 == 0:\n", " magnitude[1:-1] *= 2.0\n", " else:\n", " magnitude[1:] *= 2.0\n", "\n", "\n", " # Exclude DC before pooling\n", " magnitude = magnitude[1:]\n", "\n", "\n", " if magnitude.size < n_bins:\n", " raise ValueError(\n", " f\"Only {magnitude.size} non-DC FFT bins are available; \"\n", " f\"need at least {n_bins}. Use a longer signal or fewer bins.\"\n", " )\n", "\n", "\n", " # Partition all available bins into consecutive frequency groups\n", " edges = np.linspace(\n", " 0,\n", " magnitude.size,\n", " n_bins + 1,\n", " dtype=np.int64,\n", " )\n", "\n", "\n", " pooled = np.array(\n", " [\n", " magnitude[start:stop].mean()\n", " for start, stop in zip(edges[:-1], edges[1:])\n", " ],\n", " dtype=np.float64,\n", " )\n", "\n", "\n", " features_db = 20.0 * np.log10(np.maximum(pooled, 1e-12))\n", "\n", "\n", " return features_db.astype(np.float32)\n", "\n", "\n", "\n", "# ============================================================\n", "# MAIN PROCESSING\n", "# ============================================================\n", "\n", "\n", "def main():\n", " os.makedirs(FFT_OUTPUT_BASE, exist_ok=True)\n", "\n", "\n", " total_csv_files = 0\n", " total_fft_arrays = 0\n", "\n", "\n", " for load in LOADS:\n", " for state in STATES:\n", " input_folder = os.path.join(\n", " INPUT_BASE,\n", " FS_FOLDER,\n", " load,\n", " state,\n", " )\n", "\n", "\n", " fft_folder = os.path.join(\n", " FFT_OUTPUT_BASE,\n", " FS_FOLDER,\n", " load,\n", " state,\n", " )\n", "\n", "\n", " if not os.path.isdir(input_folder):\n", " print(f\"Skipping missing input folder: {input_folder}\")\n", " continue\n", "\n", "\n", " os.makedirs(fft_folder, exist_ok=True)\n", "\n", "\n", " csv_files = sorted(\n", " glob.glob(os.path.join(input_folder, \"*.csv\"))\n", " )\n", "\n", "\n", " if not csv_files:\n", " print(f\"No CSV files found in: {input_folder}\")\n", " continue\n", "\n", "\n", " print(\"=\" * 70)\n", " print(f\"Processing: {input_folder}\")\n", " print(f\"Files found: {len(csv_files)}\")\n", "\n", "\n", " total_csv_files += len(csv_files)\n", "\n", "\n", " for csv_path in csv_files:\n", " base_name = os.path.splitext(\n", " os.path.basename(csv_path)\n", " )[0]\n", "\n", "\n", " fft_path = os.path.join(\n", " fft_folder,\n", " f\"{base_name}_FFT.npy\",\n", " )\n", "\n", "\n", " try:\n", " signal = load_vibration_csv(csv_path)\n", "\n", "\n", " fft_array = compute_fft_array(\n", " signal,\n", " FS,\n", " N_FFT_BINS,\n", " )\n", "\n", "\n", " np.save(fft_path, fft_array)\n", "\n", "\n", " total_fft_arrays += 1\n", "\n", "\n", " print(\n", " f\" Saved FFT array: \"\n", " f\"{os.path.basename(fft_path)}\"\n", " )\n", "\n", "\n", " except Exception as e:\n", " print(\n", " f\" ERROR processing {csv_path}: {e}\"\n", " )\n", "\n", "\n", " print(\"=\" * 70)\n", " print(\"Processing complete!\")\n", " print(f\"Total CSV files processed: {total_csv_files}\")\n", " print(f\"Total FFT arrays: {total_fft_arrays}\")\n", " print(f\"FFT array output: {FFT_OUTPUT_BASE}\")\n", "\n", "\n", "\n", "if __name__ == \"__main__\":\n", " main()" ] }, { "cell_type": "code", "execution_count": null, "id": "75dfcb65", "metadata": {}, "outputs": [], "source": [ "import json\n", "from pathlib import Path\n", "\n", "\n", "import matplotlib\n", "matplotlib.use(\"Agg\")\n", "\n", "\n", "import matplotlib.pyplot as plt\n", "import numpy as np\n", "import pandas as pd\n", "import tensorflow as tf\n", "\n", "\n", "from sklearn.metrics import (\n", " accuracy_score,\n", " classification_report,\n", " ConfusionMatrixDisplay,\n", " confusion_matrix,\n", " precision_recall_fscore_support,\n", ")\n", "from sklearn.model_selection import train_test_split\n", "\n", "\n", "# ============================================================\n", "# CONFIGURATION\n", "# ============================================================\n", "\n", "\n", "BASE_DIR = Path.cwd()\n", "\n", "\n", "DATA_DIR = (\n", " BASE_DIR / \"CNN-Arrays\" / \"FFT\" / \"12,8kHz\"\n", ")\n", "\n", "\n", "OUTPUT_DIR = (\n", " BASE_DIR / \"CNN-Results\" / \"FFT_1D_CNN_70_15_15\"\n", ")\n", "\n", "\n", "LOADS = [\"323Nm_2000RPM\"]\n", "\n", "\n", "CLASS_NAMES = [\"Healthy\", \"IR\", \"OR\", \"Corrosion\"]\n", "\n", "\n", "CLASS_FOLDERS = {\n", " \"Healthy\": [\"Healthy\"],\n", " \"IR\": [\"IR\"],\n", " \"OR\": [\"OR\"],\n", " \"Corrosion\": [\"Corrosion\", \"C\"],\n", "}\n", "\n", "\n", "INPUT_LENGTH = 6400\n", "INPUT_SHAPE = (INPUT_LENGTH, 1)\n", "\n", "\n", "TRAIN_FRACTION = 0.70\n", "VALIDATION_FRACTION = 0.15\n", "TEST_FRACTION = 0.15\n", "\n", "\n", "BATCH_SIZE = 32\n", "EPOCHS = 50\n", "LEARNING_RATE = 1e-3\n", "SEED = 42\n", "\n", "\n", "LABEL_IDS = np.arange(len(CLASS_NAMES))\n", "AUTOTUNE = tf.data.AUTOTUNE\n", "\n", "\n", "# Architecture source:\n", "# https://keras.io/examples/timeseries/timeseries_classification_from_scratch/\n", "\n", "\n", "\n", "# ============================================================\n", "# DATA DISCOVERY\n", "# ============================================================\n", "\n", "\n", "def collect_fft_files():\n", " paths = []\n", " labels = []\n", " operating_points = []\n", "\n", "\n", " for load in LOADS:\n", " for class_id, class_name in enumerate(CLASS_NAMES):\n", " count = 0\n", "\n", "\n", " for folder_name in CLASS_FOLDERS[class_name]:\n", " folder = DATA_DIR / load / folder_name\n", "\n", "\n", " if not folder.is_dir():\n", " continue\n", "\n", "\n", " files = sorted(folder.glob(\"*.npy\"))\n", "\n", "\n", " for path in files:\n", " paths.append(str(path.resolve()))\n", " labels.append(class_id)\n", " operating_points.append(load)\n", "\n", "\n", " count += len(files)\n", "\n", "\n", " print(\n", " f\"{load:20s} | \"\n", " f\"{class_name:10s} | \"\n", " f\"{count:4d} FFT arrays\"\n", " )\n", "\n", "\n", " if not paths:\n", " raise FileNotFoundError(\n", " f\"No FFT arrays found under:\\n{DATA_DIR}\\n\"\n", " \"Expected: ///*.npy\"\n", " )\n", "\n", "\n", " paths = np.asarray(paths, dtype=str)\n", " labels = np.asarray(labels, dtype=np.int32)\n", " operating_points = np.asarray(operating_points, dtype=str)\n", "\n", "\n", " counts = np.bincount(\n", " labels,\n", " minlength=len(CLASS_NAMES),\n", " )\n", "\n", "\n", " print(\"\\nTotal samples by class:\")\n", "\n", "\n", " for name, count in zip(CLASS_NAMES, counts):\n", " print(f\" {name:10s}: {count}\")\n", "\n", "\n", " return paths, labels, operating_points\n", "\n", "\n", "\n", "# ============================================================\n", "# TENSORFLOW DATA PIPELINE\n", "# ============================================================\n", "\n", "\n", "def load_fft_array(path, label):\n", " def read_array(path_bytes):\n", " file_path = path_bytes.decode(\"utf-8\")\n", "\n", "\n", " array = np.load(\n", " file_path,\n", " allow_pickle=False,\n", " ).astype(np.float32)\n", "\n", "\n", " if array.shape != (INPUT_LENGTH,):\n", " raise ValueError(\n", " f\"Expected shape ({INPUT_LENGTH},), \"\n", " f\"found {array.shape} in:\\n{file_path}\"\n", " )\n", "\n", "\n", " if not np.all(np.isfinite(array)):\n", " raise ValueError(\n", " f\"FFT array contains NaN or infinite values:\\n\"\n", " f\"{file_path}\"\n", " )\n", "\n", "\n", " return array\n", "\n", "\n", " array = tf.numpy_function(\n", " read_array,\n", " [path],\n", " tf.float32,\n", " )\n", "\n", "\n", " array = tf.ensure_shape(array, (INPUT_LENGTH,))\n", " array = tf.expand_dims(array, axis=-1)\n", " array = tf.ensure_shape(array, INPUT_SHAPE)\n", "\n", "\n", " return array, tf.cast(label, tf.int32)\n", "\n", "\n", "\n", "def make_dataset(paths, labels, training=False, seed=SEED):\n", " dataset = tf.data.Dataset.from_tensor_slices(\n", " (paths, labels)\n", " )\n", "\n", "\n", " dataset = dataset.map(\n", " load_fft_array,\n", " num_parallel_calls=AUTOTUNE,\n", " )\n", "\n", "\n", " if training:\n", " dataset = dataset.shuffle(\n", " buffer_size=len(paths),\n", " seed=seed,\n", " reshuffle_each_iteration=True,\n", " )\n", "\n", "\n", " return dataset.batch(BATCH_SIZE).prefetch(AUTOTUNE)\n", "\n", "\n", "\n", "# ============================================================\n", "# MODEL\n", "# ============================================================\n", "\n", "\n", "def build_model():\n", " \"\"\"\n", " FCN architecture from the Keras example:\n", " \"Timeseries classification from scratch\".\n", "\n", "\n", " Input: 6400-value FFT array with one channel.\n", " Output: probabilities for four bearing conditions.\n", " \"\"\"\n", " inputs = tf.keras.Input(shape=INPUT_SHAPE)\n", "\n", "\n", " x = tf.keras.layers.Conv1D(\n", " filters=64,\n", " kernel_size=3,\n", " padding=\"same\",\n", " )(inputs)\n", " x = tf.keras.layers.BatchNormalization()(x)\n", " x = tf.keras.layers.ReLU()(x)\n", "\n", "\n", " x = tf.keras.layers.Conv1D(\n", " filters=64,\n", " kernel_size=3,\n", " padding=\"same\",\n", " )(x)\n", " x = tf.keras.layers.BatchNormalization()(x)\n", " x = tf.keras.layers.ReLU()(x)\n", "\n", "\n", " x = tf.keras.layers.Conv1D(\n", " filters=64,\n", " kernel_size=3,\n", " padding=\"same\",\n", " )(x)\n", " x = tf.keras.layers.BatchNormalization()(x)\n", " x = tf.keras.layers.ReLU()(x)\n", "\n", "\n", " x = tf.keras.layers.GlobalAveragePooling1D()(x)\n", "\n", "\n", " outputs = tf.keras.layers.Dense(\n", " len(CLASS_NAMES),\n", " activation=\"softmax\",\n", " )(x)\n", "\n", "\n", " model = tf.keras.Model(\n", " inputs=inputs,\n", " outputs=outputs,\n", " name=\"keras_example_fft_fcn\",\n", " )\n", "\n", "\n", " model.compile(\n", " optimizer=tf.keras.optimizers.Adam(\n", " learning_rate=LEARNING_RATE\n", " ),\n", " loss=\"sparse_categorical_crossentropy\",\n", " metrics=[\"accuracy\"],\n", " )\n", "\n", "\n", " return model\n", "\n", "\n", "\n", "# ============================================================\n", "# PLOTS AND METRICS\n", "# ============================================================\n", "\n", "\n", "def save_history_plot(history, path):\n", " fig, axes = plt.subplots(\n", " 1, 2,\n", " figsize=(11, 4),\n", " )\n", "\n", "\n", " epochs = np.arange(\n", " 1,\n", " len(history.history[\"loss\"]) + 1,\n", " )\n", "\n", "\n", " for ax, metric in zip(axes, [\"accuracy\", \"loss\"]):\n", " ax.plot(\n", " epochs,\n", " history.history[metric],\n", " label=\"Training\",\n", " )\n", " ax.plot(\n", " epochs,\n", " history.history[f\"val_{metric}\"],\n", " label=\"Validation\",\n", " )\n", "\n", "\n", " ax.set_xlabel(\"Epoch\")\n", " ax.set_ylabel(metric.capitalize())\n", " ax.legend()\n", " ax.grid(alpha=0.3)\n", "\n", "\n", " fig.tight_layout()\n", " fig.savefig(path, dpi=160)\n", " plt.close(fig)\n", "\n", "\n", "\n", "def save_confusion_plot(\n", " y_true,\n", " y_pred,\n", " path,\n", " title,\n", "):\n", " cm = confusion_matrix(\n", " y_true,\n", " y_pred,\n", " labels=LABEL_IDS,\n", " )\n", "\n", "\n", " fig, ax = plt.subplots(figsize=(7, 6))\n", "\n", "\n", " ConfusionMatrixDisplay(\n", " confusion_matrix=cm,\n", " display_labels=CLASS_NAMES,\n", " ).plot(\n", " ax=ax,\n", " cmap=\"Blues\",\n", " values_format=\"d\",\n", " colorbar=False,\n", " )\n", "\n", "\n", " ax.set_title(title)\n", " fig.tight_layout()\n", " fig.savefig(path, dpi=160)\n", " plt.close(fig)\n", "\n", "\n", "\n", "def calculate_metrics(y_true, y_pred):\n", " metrics = {\n", " \"accuracy\": float(\n", " accuracy_score(y_true, y_pred)\n", " )\n", " }\n", "\n", "\n", " for average in [\"macro\", \"weighted\"]:\n", " precision, recall, f1, _ = (\n", " precision_recall_fscore_support(\n", " y_true,\n", " y_pred,\n", " labels=LABEL_IDS,\n", " average=average,\n", " zero_division=0,\n", " )\n", " )\n", "\n", "\n", " metrics[f\"precision_{average}\"] = float(precision)\n", " metrics[f\"recall_{average}\"] = float(recall)\n", " metrics[f\"f1_{average}\"] = float(f1)\n", "\n", "\n", " return metrics\n", "\n", "\n", "\n", "# ============================================================\n", "# MAIN\n", "# ============================================================\n", "\n", "\n", "def main():\n", " OUTPUT_DIR.mkdir(\n", " parents=True,\n", " exist_ok=True,\n", " )\n", "\n", "\n", " paths, labels, loads = collect_fft_files()\n", "\n", "\n", " config = {\n", " \"model_source\": (\n", " \"https://keras.io/examples/timeseries/\"\n", " \"timeseries_classification_from_scratch/\"\n", " ),\n", " \"class_names\": CLASS_NAMES,\n", " \"input_length\": INPUT_LENGTH,\n", " \"loads\": LOADS,\n", " \"train_fraction\": TRAIN_FRACTION,\n", " \"validation_fraction\": VALIDATION_FRACTION,\n", " \"test_fraction\": TEST_FRACTION,\n", " \"batch_size\": BATCH_SIZE,\n", " \"epochs\": EPOCHS,\n", " \"learning_rate\": LEARNING_RATE,\n", " \"seed\": SEED,\n", " \"normalization\": (\n", " \"No additional normalization; \"\n", " \"arrays are used as stored\"\n", " ),\n", " }\n", "\n", "\n", " with open(OUTPUT_DIR / \"config.json\", \"w\") as f:\n", " json.dump(config, f, indent=2)\n", "\n", "\n", " # ========================================================\n", " # SINGLE STRATIFIED 70 / 15 / 15 SPLIT\n", " # ========================================================\n", "\n", "\n", " all_indices = np.arange(len(paths))\n", "\n", "\n", " # First split:\n", " # 70% training, 30% temporary validation + test\n", " train_idx, temp_idx = train_test_split(\n", " all_indices,\n", " test_size=1.0 - TRAIN_FRACTION,\n", " stratify=labels,\n", " random_state=SEED,\n", " )\n", "\n", "\n", " # Second split:\n", " # Split the temporary 30% evenly:\n", " # 15% validation and 15% test\n", " val_idx, test_idx = train_test_split(\n", " temp_idx,\n", " test_size=0.50,\n", " stratify=labels[temp_idx],\n", " random_state=SEED,\n", " )\n", "\n", "\n", " print(\"\\n\" + \"=\" * 70)\n", " print(\"SINGLE STRATIFIED SPLIT\")\n", " print(\"=\" * 70)\n", " print(\n", " f\"Train: {len(train_idx)} \"\n", " f\"({len(train_idx) / len(paths):.1%})\"\n", " )\n", " print(\n", " f\"Validation: {len(val_idx)} \"\n", " f\"({len(val_idx) / len(paths):.1%})\"\n", " )\n", " print(\n", " f\"Test: {len(test_idx)} \"\n", " f\"({len(test_idx) / len(paths):.1%})\"\n", " )\n", "\n", "\n", " # Save the exact split for reproducibility.\n", " split_rows = []\n", "\n", "\n", " for split_name, indices in [\n", " (\"train\", train_idx),\n", " (\"validation\", val_idx),\n", " (\"test\", test_idx),\n", " ]:\n", " for idx in indices:\n", " split_rows.append(\n", " {\n", " \"path\": paths[idx],\n", " \"label\": int(labels[idx]),\n", " \"class\": CLASS_NAMES[labels[idx]],\n", " \"load\": loads[idx],\n", " \"split\": split_name,\n", " }\n", " )\n", "\n", "\n", " pd.DataFrame(split_rows).to_csv(\n", " OUTPUT_DIR / \"split_manifest.csv\",\n", " index=False,\n", " )\n", "\n", "\n", " tf.keras.backend.clear_session()\n", " tf.keras.utils.set_random_seed(SEED)\n", "\n", "\n", " train_ds = make_dataset(\n", " paths[train_idx],\n", " labels[train_idx],\n", " training=True,\n", " seed=SEED,\n", " )\n", "\n", "\n", " val_ds = make_dataset(\n", " paths[val_idx],\n", " labels[val_idx],\n", " )\n", "\n", "\n", " test_ds = make_dataset(\n", " paths[test_idx],\n", " labels[test_idx],\n", " )\n", "\n", "\n", " model = build_model()\n", " model.summary()\n", "\n", "\n", " best_model_path = OUTPUT_DIR / \"best_model.keras\"\n", "\n", "\n", " callbacks = [\n", " tf.keras.callbacks.ModelCheckpoint(\n", " str(best_model_path),\n", " monitor=\"val_loss\",\n", " mode=\"min\",\n", " save_best_only=True,\n", " verbose=0,\n", " ),\n", " tf.keras.callbacks.EarlyStopping(\n", " monitor=\"val_loss\",\n", " mode=\"min\",\n", " patience=10,\n", " restore_best_weights=True,\n", " verbose=1,\n", " ),\n", " tf.keras.callbacks.ReduceLROnPlateau(\n", " monitor=\"val_loss\",\n", " mode=\"min\",\n", " factor=0.5,\n", " patience=5,\n", " min_lr=1e-6,\n", " verbose=1,\n", " ),\n", " ]\n", "\n", "\n", " history = model.fit(\n", " train_ds,\n", " validation_data=val_ds,\n", " epochs=EPOCHS,\n", " callbacks=callbacks,\n", " verbose=2,\n", " )\n", "\n", "\n", " pd.DataFrame(history.history).to_csv(\n", " OUTPUT_DIR / \"history.csv\",\n", " index=False,\n", " )\n", "\n", "\n", " save_history_plot(\n", " history,\n", " OUTPUT_DIR / \"training_curves.png\",\n", " )\n", "\n", "\n", " # ========================================================\n", " # TEST EVALUATION\n", " # ========================================================\n", "\n", "\n", " best_model = tf.keras.models.load_model(\n", " str(best_model_path)\n", " )\n", "\n", "\n", " evaluation = best_model.evaluate(\n", " test_ds,\n", " return_dict=True,\n", " verbose=0,\n", " )\n", "\n", "\n", " probabilities = best_model.predict(\n", " test_ds,\n", " verbose=0,\n", " )\n", "\n", "\n", " predictions = probabilities.argmax(axis=1)\n", " y_true = labels[test_idx]\n", "\n", "\n", " metrics = calculate_metrics(\n", " y_true,\n", " predictions,\n", " )\n", "\n", "\n", " metrics.update(\n", " {\n", " \"test_loss\": float(evaluation[\"loss\"]),\n", " \"n_train\": len(train_idx),\n", " \"n_validation\": len(val_idx),\n", " \"n_test\": len(test_idx),\n", " \"epochs_run\": len(history.history[\"loss\"]),\n", " \"best_epoch\": (\n", " int(\n", " np.argmin(\n", " history.history[\"val_loss\"]\n", " )\n", " ) + 1\n", " ),\n", " }\n", " )\n", "\n", "\n", " results_df = pd.DataFrame([metrics])\n", " results_df.to_csv(\n", " OUTPUT_DIR / \"metrics.csv\",\n", " index=False,\n", " )\n", "\n", "\n", " report = classification_report(\n", " y_true,\n", " predictions,\n", " labels=LABEL_IDS,\n", " target_names=CLASS_NAMES,\n", " digits=4,\n", " zero_division=0,\n", " )\n", "\n", "\n", " print(\"\\n\" + \"=\" * 70)\n", " print(\"TEST-SET RESULTS\")\n", " print(\"=\" * 70)\n", " print(f\"Test accuracy: {metrics['accuracy']:.4f}\")\n", " print(f\"Test loss: {metrics['test_loss']:.4f}\")\n", " print(report)\n", "\n", "\n", " (\n", " OUTPUT_DIR / \"classification_report.txt\"\n", " ).write_text(\n", " report,\n", " encoding=\"utf-8\",\n", " )\n", "\n", "\n", " save_confusion_plot(\n", " y_true,\n", " predictions,\n", " OUTPUT_DIR / \"confusion_matrix.png\",\n", " \"FFT — 1D FCN: held-out test predictions\",\n", " )\n", "\n", "\n", " # ========================================================\n", " # SAVE TEST PREDICTIONS\n", " # ========================================================\n", "\n", "\n", " predictions_df = pd.DataFrame(\n", " {\n", " \"path\": paths[test_idx],\n", " \"load\": loads[test_idx],\n", " \"true_label\": y_true,\n", " \"predicted_label\": predictions,\n", " \"true_class\": [\n", " CLASS_NAMES[i] for i in y_true\n", " ],\n", " \"predicted_class\": [\n", " CLASS_NAMES[i] for i in predictions\n", " ],\n", " }\n", " )\n", "\n", "\n", " for class_id, name in enumerate(CLASS_NAMES):\n", " predictions_df[f\"probability_{name}\"] = (\n", " probabilities[:, class_id]\n", " )\n", "\n", "\n", " predictions_df.to_csv(\n", " OUTPUT_DIR / \"test_predictions.csv\",\n", " index=False,\n", " )\n", "\n", "\n", " # ========================================================\n", " # SAVE SUMMARY\n", " # ========================================================\n", "\n", "\n", " score_columns = [\n", " \"accuracy\",\n", " \"precision_macro\",\n", " \"recall_macro\",\n", " \"f1_macro\",\n", " \"precision_weighted\",\n", " \"recall_weighted\",\n", " \"f1_weighted\",\n", " \"test_loss\",\n", " ]\n", "\n", "\n", " summary = {\n", " metric: float(metrics[metric])\n", " for metric in score_columns\n", " }\n", "\n", "\n", " with open(OUTPUT_DIR / \"summary.json\", \"w\") as f:\n", " json.dump(summary, f, indent=2)\n", "\n", "\n", " print(f\"\\nResults saved to: {OUTPUT_DIR}\")\n", "\n", "\n", "\n", "if __name__ == \"__main__\":\n", " main()" ] } ], "metadata": { "kernelspec": { "display_name": "tf (3.12.3)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.3" } }, "nbformat": 4, "nbformat_minor": 5 }