feat: added gear-mesh physics + GearProvider (cross-domain: SEU gearbox), engine left unchanged.

Author: abhiprd2000

HANDOFF_TASFIN.md

@@ -0,0 +1,53 @@
+# SEU / Gearbox cross-domain validation - handoff
+
+Goal: run the full CNSD pipeline on SEU gears (a non-bearing domain) to
+demonstrate the provider interface is genuinely domain-agnostic. This is the
+universality result.
+
+## What's included (built + tested)
+
+- `cnsd/physics/gear.py` - gear-mesh physics: GMF (= teeth x shaft-rate),
+  shaft-rate sidebands, envelope-spectrum prominence. GMF math verified
+  (20 teeth @ 1800 rpm -> 600 Hz).
+- `cnsd/physics/providers/gear.py` - GearProvider, implements the same
+  PhysicsProvider interface as BearingProvider. Maps the SEU 5-class taxonomy
+  (Health/Chipped/Miss/Root/Surface) onto two evidence channels: localized tooth
+  faults -> GMF sidebands; surface wear -> GMF harmonics.
+- `cnsd/physics/providers/__init__.py` - registers domain 'gear'.
+- `cnsd/builder.py` - config path: `domain.type: gear` builds the GearProvider
+  from `physics.parameters.n_teeth_input` (+ optional n_teeth_output).
+- `cnsd/physics/configs.py` - PhysicsConfig.bearing is now Optional (gear configs
+  have no bearing geometry).
+- `validate_seu.py` - the run script. Only `load_seu()` is dataset-specific.
+- 3 new tests (gear math, interface conformance, registry) - 10/10 pass.
+
+## Your job
+
+1. Wire `load_seu()` in validate_seu.py - return X (n,1024), y (n,), cond (n,).
+   SEU gearset: tab-separated, 8 channels, header ends at the 'Data' line; pick
+   ONE channel and commit to it before seeing results (no channel cherry-picking).
+2. Set `N_TEETH_INPUT` to the real SEU driving-gear tooth count (I used 20 as a
+   placeholder - CONFIRM from the rig spec; the GMF is wrong if this is wrong).
+3. Confirm `SEU_FS` (sampling rate) and the condition->rpm map.
+4. Run `python validate_seu.py`, send me the output.
+
+## Honest caveats (expect these, they are not bugs)
+
+- The prominence threshold (3.0, same as bearings) will likely need tuning on
+  real SEU signals - same story as the CWRU sweep. Expect a high INCONCLUSIVE
+  rate at first; sweep tau on a held-out calibration split (NOT the test set, NOT
+  the training set where the CNN memorizes - the lesson from #11).
+- The Chipped/Miss/Root families all share the sideband channel, so the physics
+  cannot distinguish *between* them (they look alike in the envelope spectrum -
+  this is physically true, localized tooth faults all produce shaft-rate
+  sidebands). The provider confirms "localized tooth fault" vs "surface wear" vs
+  "health"; finer separation is the CNN's job. Report this honestly - it's a real
+  property of gear physics, not a limitation to hide.
+- N_TEETH_INPUT is the single most important parameter. If the GMF is off, every
+  verdict is meaningless. Triple-check it.
+
+## How it plugs in (no engine change)
+
+The engine never sees gear-specific code - it calls the PhysicsProvider
+interface. Same five layers, same diagnose() path, only the provider differs.
+That's the universality claim, demonstrated.

cnsd/builder.py

@@ -84,6 +84,12 @@ def build_provider(config):
         cond_to_rpm = _coerce_int_keys(params.get('motor_load_rpm', {0: 1797}))
         return provider_cls(bearing=geometry, cond_to_rpm=cond_to_rpm, fs=fs)
 
+    if domain_type == 'gear':
+        cond_to_rpm = _coerce_int_keys(params.get('motor_load_rpm', {0: 1800}))
+        return provider_cls(n_teeth_input=params['n_teeth_input'],
+                            n_teeth_output=params.get('n_teeth_output'),
+                            cond_to_rpm=cond_to_rpm, fs=fs)
+
     # other registered domains: pass parameters through + fs
     try:
         return provider_cls(fs=fs, **params)

cnsd/physics/configs.py

@@ -1,13 +1,13 @@
 """Physics configs: bearing geometry + condition->rpm + sampling rate per dataset."""
 from dataclasses import dataclass
-from typing import Dict
+from typing import Dict, Optional
 
 
 @dataclass
 class PhysicsConfig:
-    bearing: Dict[str, float]
     cond_to_rpm: Dict
     fs: int
+    bearing: Optional[Dict[str, float]] = None
     name: str = 'custom'
 
 

cnsd/physics/gear.py

@@ -0,0 +1,88 @@
+"""Gear-mesh physics for gearbox fault verification.
+
+A gear fault does not produce ball-pass frequencies. The dominant signature is
+the gear-mesh frequency (GMF) and its harmonics, with sidebands spaced at the
+shaft rotation rate. Different fault types modulate this differently:
+
+  - a localized tooth fault (chip / root crack / missing tooth) produces strong
+    sidebands around the GMF at the shaft rate of the faulty gear (the defect is
+    excited once per revolution),
+  - distributed surface wear raises the GMF harmonics without strong shaft-rate
+    sidebands.
+
+This module computes GMF + sidebands and measures their prominence in the
+envelope spectrum, mirroring the bearing module's role for rolling elements.
+"""
+import numpy as np
+from scipy.signal import hilbert
+
+DEFAULT_FS = 5120
+
+
+def gear_mesh_frequencies(rpm, n_teeth_input, n_teeth_output=None):
+    """Gear-mesh frequency and shaft rates (Hz) from teeth counts + input speed.
+
+    GMF = input_shaft_rate * n_teeth_input (the mesh rate is shared by both gears;
+    teeth_in * f_in == teeth_out * f_out). Returns the GMF, the input shaft rate,
+    and (if output teeth given) the output shaft rate - the sideband spacings.
+    """
+    f_in = rpm / 60.0
+    gmf = f_in * n_teeth_input
+    out = {'GMF': gmf, 'shaft_input': f_in}
+    if n_teeth_output:
+        out['shaft_output'] = gmf / n_teeth_output
+    return out
+
+
+def envelope_spectrum(signal, fs=DEFAULT_FS):
+    """Amplitude-envelope spectrum (Hilbert -> |.| -> remove DC -> rFFT)."""
+    x = np.asarray(signal, float).flatten()
+    env = np.abs(hilbert(x))
+    env = env - env.mean()
+    mag = np.abs(np.fft.rfft(env))
+    freqs = np.fft.rfftfreq(len(env), d=1.0 / fs)
+    return freqs, mag
+
+
+def _prominence(freqs, mag, f0, rel_tol=0.04):
+    if f0 <= 0:
+        return 0.0
+    tol = f0 * rel_tol
+    band = (freqs >= f0 - tol) & (freqs <= f0 + tol)
+    if not band.any():
+        return 0.0
+    peak = mag[band].max()
+    nb = (freqs >= f0 - 6 * tol) & (freqs <= f0 + 6 * tol) & ~band
+    baseline = np.median(mag[nb]) if nb.any() else np.median(mag)
+    return float(peak / (baseline + 1e-12))
+
+
+def sideband_strength(freqs, mag, gmf, shaft_rate, n_sidebands=2):
+    """Mean prominence of the shaft-rate sidebands around the GMF - the marker of
+    a localized tooth fault."""
+    vals = []
+    for k in range(1, n_sidebands + 1):
+        for f0 in (gmf - k * shaft_rate, gmf + k * shaft_rate):
+            vals.append(_prominence(freqs, mag, f0))
+    return float(np.mean(vals)) if vals else 0.0
+
+
+def gear_fault_evidence(signal, rpm, n_teeth_input, n_teeth_output=None,
+                        fs=DEFAULT_FS):
+    """Physical evidence for gear-fault families in one window.
+
+    Returns prominence of:
+      'mesh'     : GMF + harmonics (raised by distributed wear)
+      'sideband' : shaft-rate sidebands around GMF (raised by localized faults)
+    plus the characteristic frequencies in Hz.
+    """
+    freqs, mag = envelope_spectrum(signal, fs)
+    gf = gear_mesh_frequencies(rpm, n_teeth_input, n_teeth_output)
+    gmf, shaft = gf['GMF'], gf['shaft_input']
+    mesh = np.mean([_prominence(freqs, mag, gmf * h) for h in (1, 2)])
+    side = sideband_strength(freqs, mag, gmf, shaft)
+    return {
+        'mesh_strength': float(mesh),
+        'sideband_strength': float(side),
+        'freqs_hz': {k: float(v) for k, v in gf.items()},
+    }

cnsd/physics/providers/__init__.py

@@ -7,11 +7,13 @@
 from cnsd.physics.providers.base import PhysicsProvider
 from cnsd.physics.providers.bearing import BearingProvider
 from cnsd.physics.providers.spectral import SpectralProvider
+from cnsd.physics.providers.gear import GearProvider
 
 # domain.type (from config) -> provider builder. 'spectral' is the universal
 # zero-knowledge fallback used when a domain has no dedicated provider.
 _REGISTRY = {
     'bearing': BearingProvider,
+    'gear': GearProvider,
     'spectral': SpectralProvider,
     None: SpectralProvider,
 }

cnsd/physics/providers/gear.py

@@ -0,0 +1,77 @@
+"""Gear physics provider: gearbox fault verification via gear-mesh signatures.
+
+Implements the PhysicsProvider interface for gearboxes. Verifies a predicted gear
+fault against the gear-mesh frequency (GMF) and its shaft-rate sidebands:
+localized tooth faults (chip, root crack, missing tooth) raise the sidebands;
+distributed surface wear raises the mesh harmonics. The SEU gearbox taxonomy
+(Health, Chipped, Miss, Root, Surface) maps onto these two evidence channels.
+"""
+from cnsd.physics.providers.base import PhysicsProvider
+from cnsd.physics.gear import gear_fault_evidence
+
+# SEU gearbox fault families -> which evidence channel signifies them.
+# 'localized' faults show shaft-rate sidebands; 'distributed' wear shows mesh.
+_FAMILY_CHANNEL = {
+    'Chipped Tooth': 'sideband',
+    'Missing Tooth': 'sideband',
+    'Root Crack': 'sideband',
+    'Surface Wear': 'mesh',
+}
+
+_ROOT_CAUSE = {
+    'Chipped Tooth': ('gear tooth', 'GMF sidebands',
+                      'a chipped tooth striking once per revolution, producing '
+                      'shaft-rate sidebands around the gear-mesh frequency'),
+    'Missing Tooth': ('gear tooth', 'GMF sidebands',
+                      'a missing tooth producing a strong once-per-revolution '
+                      'impact and shaft-rate sidebands'),
+    'Root Crack': ('gear tooth root', 'GMF sidebands',
+                   'a root crack reducing tooth stiffness, modulating the mesh at '
+                   'the shaft rate'),
+    'Surface Wear': ('gear flank', 'GMF harmonics',
+                     'distributed surface wear raising the gear-mesh harmonics'),
+    'Health': (None, None, 'no gear fault; mesh spectrum is clean'),
+}
+
+
+class GearProvider(PhysicsProvider):
+    families = ['Chipped Tooth', 'Missing Tooth', 'Root Crack', 'Surface Wear']
+
+    def __init__(self, n_teeth_input, n_teeth_output=None, cond_to_rpm=None,
+                 fs=5120):
+        self.n_teeth_input = n_teeth_input
+        self.n_teeth_output = n_teeth_output
+        self.cond_to_rpm = cond_to_rpm or {0: 1800}
+        self.fs = fs
+
+    def evidence(self, signal, condition):
+        rpm = self.cond_to_rpm.get(int(condition),
+                                   next(iter(self.cond_to_rpm.values()), 1800))
+        ev = gear_fault_evidence(signal, rpm, self.n_teeth_input,
+                                 self.n_teeth_output, fs=self.fs)
+        # localized fault families share the sideband channel; surface wear the mesh
+        side, mesh = ev['sideband_strength'], ev['mesh_strength']
+        return {
+            'family_strength': {'Chipped Tooth': side, 'Missing Tooth': side,
+                                'Root Crack': side, 'Surface Wear': mesh},
+            'frequencies_hz': ev['freqs_hz'],
+            '_channels': {'sideband': side, 'mesh': mesh},
+        }
+
+    def root_cause(self, family, evidence):
+        comp, freq_name, mechanism = _ROOT_CAUSE.get(family, (None, None, ''))
+        channel = _FAMILY_CHANNEL.get(family)
+        strength = evidence.get('_channels', {}).get(channel, 0.0) if channel else 0.0
+        gmf = evidence.get('frequencies_hz', {}).get('GMF')
+        if family in ('Health', None):
+            statement = 'No gear fault detected; the mesh spectrum is clean.'
+        elif gmf is not None:
+            statement = (f'Defect on the {comp} ({family}), evidenced by '
+                         f'{freq_name} around GMF={gmf:.1f} Hz '
+                         f'(strength {strength:.1f}). {mechanism}.')
+        else:
+            statement = f'{family} on the {comp}.'
+        return {'component': comp, 'fault_type': family, 'mechanism': mechanism,
+                'evidence_frequency': freq_name,
+                'evidence_frequency_hz': gmf,
+                'evidence_strength': float(strength), 'statement': statement}

test_layers.py

@@ -0,0 +1,92 @@
+"""Smoke tests for the CNSD layers (no TensorFlow / DoWhy required)."""
+import numpy as np
+from cnsd.physics import characteristic_frequencies
+from cnsd.physics.providers import BearingProvider, SpectralProvider, get_provider
+from cnsd.symbolic import PhysicsRuleEngine
+from cnsd.causal import intervention_effect_of_condition
+from cnsd.consensus import fuse
+
+_BEARING = {'n_balls': 9, 'd_ball': 0.3126, 'd_pitch': 1.537, 'contact_angle': 0.0}
+_TAXONOMY = {0: ('Normal', 'None'), 5: ('Inner Race', 'Medium'), 8: ('Outer Race', 'Medium')}
+
+
+def _impulse(freq, fs=12000, n=1024, a=6.0):
+    p = int(fs / freq); s = np.zeros(n)
+    for i in range(0, n, p):
+        d = np.exp(-(np.arange(min(p, n - i))) / 25.0); s[i:i+len(d)] += a * d
+    return s + 0.3 * np.random.randn(n)
+
+
+def _bearing_engine():
+    prov = BearingProvider(bearing=_BEARING, cond_to_rpm={0: 1797}, fs=12000)
+    return PhysicsRuleEngine(provider=prov, taxonomy=_TAXONOMY)
+
+
+def test_physics_frequencies_match_published_cwru():
+    cf = characteristic_frequencies(1797, bearing=_BEARING)
+    assert abs(cf['BPFO'] - 107.4) < 1.0
+    assert abs(cf['BPFI'] - 162.2) < 1.0
+
+
+def test_symbolic_confirms_matching_fault():
+    r = _bearing_engine().diagnose(_impulse(characteristic_frequencies(1797, bearing=_BEARING)['BPFI']), 5, 0)
+    assert r['verdict'] == 'CONFIRMED'
+    assert r['predicted_family'] == 'Inner Race'
+
+
+def test_symbolic_conflicts_when_physics_disagrees():
+    # inner-race signal but the classifier said outer race (class 8)
+    r = _bearing_engine().diagnose(_impulse(characteristic_frequencies(1797, bearing=_BEARING)['BPFI']), 8, 0)
+    assert r['verdict'] == 'CONFLICT'
+    # root cause follows the physics, overriding the classifier
+    assert r['root_cause']['fault_type'] == 'Inner Race'
+
+
+def test_spectral_provider_runs_without_geometry():
+    # the universal fallback: no machine model, still flags a dominant fault freq
+    eng = PhysicsRuleEngine(provider=SpectralProvider(fs=12000))
+    r = eng.diagnose(_impulse(160.0), 1, 0)
+    assert r['verdict'] in ('CONFIRMED', 'INCONCLUSIVE')
+    assert 'root_cause' in r
+
+
+def test_get_provider_falls_back_to_spectral():
+    assert get_provider('unknown_machine') is SpectralProvider
+    assert get_provider('bearing') is BearingProvider
+
+
+def test_causal_doZ_is_rung2():
+    y = np.random.binomial(1, 0.6, 500); cond = np.random.choice([0, 1, 2, 3], 500)
+    assert intervention_effect_of_condition(y, cond, n_perm=50)['rung'] == 2
+
+
+def test_consensus_conflict_forces_review():
+    assert fuse('CONFLICT', 0.99) == 'MANUAL_REVIEW'
+    assert fuse('CONFIRMED', 0.99) == 'HIGH_CONFIDENCE'
+
+
+# ── gear domain (cross-domain universality) ──────────────────────────────────
+
+def test_gear_mesh_frequency_math():
+    from cnsd.physics.gear import gear_mesh_frequencies
+    gf = gear_mesh_frequencies(1800, 20)   # 20 teeth at 1800 rpm (30 Hz)
+    assert abs(gf['GMF'] - 600.0) < 0.1
+    assert abs(gf['shaft_input'] - 30.0) < 0.1
+
+
+def test_gear_provider_implements_interface():
+    from cnsd.physics.providers.gear import GearProvider
+    from cnsd.physics.providers.base import PhysicsProvider
+    p = GearProvider(n_teeth_input=20, cond_to_rpm={0: 1800}, fs=5120)
+    assert isinstance(p, PhysicsProvider)
+    sig = np.sin(2 * np.pi * 600 * np.arange(2048) / 5120) + 0.1 * np.random.randn(2048)
+    ev = p.evidence(sig, 0)
+    assert 'family_strength' in ev and 'frequencies_hz' in ev
+    assert p.dominant_family(ev) in p.families
+
+
+def test_gear_registered():
+    from cnsd.physics.providers import get_provider, available_domains
+    from cnsd.physics.providers.gear import GearProvider
+    assert get_provider('gear') is GearProvider
+    assert 'gear' in available_domains()