style: fix ruff formatting errors in evaluate_baselines.py

Author: TasfinMahmud

evaluate_baselines.py

@@ -1,6 +1,6 @@
+import argparse
 import os
 import sys
-import argparse
 import traceback
 
 import numpy as np
@@ -13,7 +13,7 @@
     from cnsd import Dataset
     from cnsd.diagnosis.system import CNSD
     from cnsd.perception.cnn import _train_cnn
-    
+
     # Argparse
     parser = argparse.ArgumentParser()
     parser.add_argument('--dataset', type=str, required=True, choices=['cwru', 'pu', 'xjtusy'])
@@ -24,7 +24,10 @@
     if args.dataset == 'pu':
         from cnsd.physics import PhysicsConfig
         from validate_pu import load_pu_domain_split
-        (X_train_full, y_train_full, cond_train_full), (X_target, y_target, cond_target) = load_pu_domain_split()
+
+        (X_train_full, y_train_full, cond_train_full), (X_target, y_target, cond_target) = (
+            load_pu_domain_split()
+        )
         unique_rpm = set(cond_train_full).union(set(cond_target))
         rpm_map = {float(r): float(r) for r in unique_rpm}
         physics = PhysicsConfig(
@@ -35,42 +38,43 @@
         )
         taxonomy = {0: ('Normal', 'None'), 1: ('Outer Race', 'Medium'), 2: ('Inner Race', 'High')}
         fs = 64000
-        
+
     elif args.dataset == 'cwru':
-        from validate_run import load_cwru, CWRU, TAXONOMY
+        from validate_run import CWRU, TAXONOMY, load_cwru
+
         X, y, cond = load_cwru()
         X = np.asarray(X, np.float32)
         y = np.asarray(y)
         cond = np.asarray(cond)
-        
+
         train_mask = cond < 3
         target_mask = cond == 3
-        
+
         X_train_full = X[train_mask]
         y_train_full = y[train_mask]
         cond_train_full = cond[train_mask]
-        
+
         X_target = X[target_mask]
         y_target = y[target_mask]
         cond_target = cond[target_mask]
-        
+
         physics = CWRU
         taxonomy = TAXONOMY
         fs = 12000
-        
+
     elif args.dataset == 'xjtusy':
         from cnsd.datasets.xjtusy import load_xjtusy_domain_split
-        from cnsd.physics.configs import XJTUSY_PHYSICS
+
         train_ds, target_ds = load_xjtusy_domain_split(window_size=4096)
-        
+
         X_train_full = train_ds.X
         y_train_full = train_ds.y
         cond_train_full = train_ds.cond
-        
+
         X_target = target_ds.X
         y_target = target_ds.y
         cond_target = target_ds.cond
-        
+
         physics = target_ds.physics
         taxonomy = target_ds.taxonomy
         fs = target_ds.fs
@@ -112,8 +116,13 @@ def clone_for_mc(layer):
 
     # Test data processing
     test_ds = Dataset.from_arrays(
-        X_target, y_target, cond_target,
-        fs=fs, physics=physics, taxonomy=taxonomy, name=f'{args.dataset.upper()}_Test'
+        X_target,
+        y_target,
+        cond_target,
+        fs=fs,
+        physics=physics,
+        taxonomy=taxonomy,
+        name=f'{args.dataset.upper()}_Test',
     )
     sig_te = np.stack([test_ds.X[i].reshape(-1) for i in range(len(test_ds.X))]).astype(np.float32)
     yte = test_ds.y
@@ -134,14 +143,34 @@ def clone_for_mc(layer):
         train_idx = indices[:split_idx]
         calib_idx = indices[split_idx:]
 
-        X_tr, y_tr, cond_tr = X_train_full[train_idx], y_train_full[train_idx], cond_train_full[train_idx]
-        X_ca, y_ca, cond_ca = X_train_full[calib_idx], y_train_full[calib_idx], cond_train_full[calib_idx]
+        X_tr, y_tr, cond_tr = (
+            X_train_full[train_idx],
+            y_train_full[train_idx],
+            cond_train_full[train_idx],
+        )
+        X_ca, y_ca, cond_ca = (
+            X_train_full[calib_idx],
+            y_train_full[calib_idx],
+            cond_train_full[calib_idx],
+        )
 
         train_ds = Dataset.from_arrays(
-            X_tr, y_tr, cond_tr, fs=fs, physics=physics, taxonomy=taxonomy, name=f'{args.dataset.upper()}_Train'
+            X_tr,
+            y_tr,
+            cond_tr,
+            fs=fs,
+            physics=physics,
+            taxonomy=taxonomy,
+            name=f'{args.dataset.upper()}_Train',
         )
         calib_ds = Dataset.from_arrays(
-            X_ca, y_ca, cond_ca, fs=fs, physics=physics, taxonomy=taxonomy, name=f'{args.dataset.upper()}_Calib'
+            X_ca,
+            y_ca,
+            cond_ca,
+            fs=fs,
+            physics=physics,
+            taxonomy=taxonomy,
+            name=f'{args.dataset.upper()}_Calib',
         )
 
         # 2. Train Primary Model (Bypass SCM to prevent multiprocess deadlocks)
@@ -161,7 +190,9 @@ def clone_for_mc(layer):
             ens.append(m_cnn)
 
         # 4. Calibrate Tau
-        sig_ca = np.stack([calib_ds.X[i].reshape(-1) for i in range(len(calib_ds.X))]).astype(np.float32)
+        sig_ca = np.stack([calib_ds.X[i].reshape(-1) for i in range(len(calib_ds.X))]).astype(
+            np.float32
+        )
         Xin_ca = sig_ca[..., None]
         probs_ca = model.cnn.predict(Xin_ca, batch_size=128, verbose=0)
         pred_ca = probs_ca.argmax(1)
@@ -172,7 +203,10 @@ def clone_for_mc(layer):
         for tau in [1.0, 1.5, 2.0, 2.5, 3.0]:
             model.symbolic.tau = float(tau)
             verds = np.array(
-                [model.symbolic.diagnose(sig_ca[i], pred_ca[i], calib_ds.cond[i])['verdict'] for i in range(len(sig_ca))]
+                [
+                    model.symbolic.diagnose(sig_ca[i], pred_ca[i], calib_ds.cond[i])['verdict']
+                    for i in range(len(sig_ca))
+                ]
             )
             conf = verds == 'CONFIRMED'
             cnfl = verds == 'CONFLICT'
@@ -193,7 +227,10 @@ def clone_for_mc(layer):
 
         # Physics evaluation
         verds = np.array(
-            [model.symbolic.diagnose(sig_te[i], pred_te[i], cond_te[i])['verdict'] for i in range(len(sig_te))]
+            [
+                model.symbolic.diagnose(sig_te[i], pred_te[i], cond_te[i])['verdict']
+                for i in range(len(sig_te))
+            ]
         )
         conf = verds == 'CONFIRMED'
         cnfl = verds == 'CONFLICT'
@@ -235,7 +272,9 @@ def clone_for_mc(layer):
         ens_gap = get_matched_coverage_gap(ens_score, ens_correct, target_n)
         results['ens_gap'].append(ens_gap)
 
-        print(f'Matched Coverage GAPs -> Softmax:{soft_gap:+.3f} | MC-Drop:{mc_gap:+.3f} | Ens:{ens_gap:+.3f}')
+        print(
+            f'Matched Coverage GAPs -> Softmax:{soft_gap:+.3f} | MC-Drop:{mc_gap:+.3f} | Ens:{ens_gap:+.3f}'
+        )
 
         # 6. Noise Test
         rng = np.random.RandomState(seed)
@@ -248,13 +287,18 @@ def clone_for_mc(layer):
                 sig_n = sig_te + rng.randn(*sig_te.shape).astype(np.float32) * np.sqrt(npow)
 
             Xin_n = sig_n[..., None]
-            ep_class = np.stack([m.predict(Xin_n, batch_size=128, verbose=0).argmax(1) for m in ens])
+            ep_class = np.stack(
+                [m.predict(Xin_n, batch_size=128, verbose=0).argmax(1) for m in ens]
+            )
             v = stats.mode(ep_class, axis=0, keepdims=False).mode
             unan = (ep_class == v).all(0)
             ok = v == yte
 
             pv = np.array(
-                [model.symbolic.diagnose(sig_n[i], v[i], cond_te[i])['verdict'] for i in range(len(sig_n))]
+                [
+                    model.symbolic.diagnose(sig_n[i], v[i], cond_te[i])['verdict']
+                    for i in range(len(sig_n))
+                ]
             )
             pc = pv == 'CONFLICT'
             uw = unan & (~ok)
@@ -264,10 +308,16 @@ def clone_for_mc(layer):
             print(f'  Noise={s:>5} | catch_rate={catch:.3f}')
 
     print('\n' + '=' * 60 + f'\nFINAL AGGREGATED RESULTS ({len(seeds)} Seeds)\n' + '=' * 60)
-    print(f'Physics GAP: {np.nanmean(results["phys_gap"]):+.3f} ± {np.nanstd(results["phys_gap"]):.3f}')
-    print(f'Softmax GAP: {np.nanmean(results["soft_gap"]):+.3f} ± {np.nanstd(results["soft_gap"]):.3f}')
+    print(
+        f'Physics GAP: {np.nanmean(results["phys_gap"]):+.3f} ± {np.nanstd(results["phys_gap"]):.3f}'
+    )
+    print(
+        f'Softmax GAP: {np.nanmean(results["soft_gap"]):+.3f} ± {np.nanstd(results["soft_gap"]):.3f}'
+    )
     print(f'MC-Drop GAP: {np.nanmean(results["mc_gap"]):+.3f} ± {np.nanstd(results["mc_gap"]):.3f}')
-    print(f'Ensemble GAP: {np.nanmean(results["ens_gap"]):+.3f} ± {np.nanstd(results["ens_gap"]):.3f}')
+    print(
+        f'Ensemble GAP: {np.nanmean(results["ens_gap"]):+.3f} ± {np.nanstd(results["ens_gap"]):.3f}'
+    )
 
     t_stat, p_val = stats.ttest_rel(results['phys_gap'], results['ens_gap'])
     avg_ncov = np.mean(results['ncov'])