fault_rules.py

"""Deterministic candidate fault ranking (the grounding layer).

This is what makes a 4B model reliable: instead of asking the model to BOTH
know acoustic fault signatures AND reason, we encode known signatures here as
transparent rules, produce a ranked shortlist with human-readable evidence,
and let the model only pick + explain the best-supported candidate.

Every diagnosis is therefore traceable to a measured feature value.
"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, List

from audio_analyzer import AudioFeatures


@dataclass
class Candidate:
    name: str
    urgency: str          # CRITICAL | HIGH | MEDIUM | LOW
    weight: float         # 0-1 strength of acoustic support
    evidence: str         # plain-language reason, formatted with feature values


@dataclass
class Rule:
    name: str
    urgency: str
    weight: float
    when: Callable[[AudioFeatures], bool]
    evidence: str         # an f-string-style template using {field} placeholders


def _fmt(template: str, f: AudioFeatures) -> str:
    try:
        return template.format(**f.to_dict())
    except Exception:
        return template


# --- Rule tables -----------------------------------------------------------
# Start broad; add appliances over time. Order within a list does not matter;
# candidates are sorted by weight after evaluation.

RULES: dict[str, List[Rule]] = {
    "Washing machine": [
        Rule("Worn drum bearing", "HIGH", 0.90,
             lambda f: f.has_regular_pattern and 50 < f.pattern_interval_ms < 400
                       and f.spectral_centroid_hz > 1500,
             "Regular click every {pattern_interval_ms:.0f} ms with a bright "
             "{spectral_centroid_hz:.0f} Hz spectrum is the classic spalled bearing-race signature."),
        Rule("Drive belt slip / wear", "MEDIUM", 0.70,
             lambda f: f.dominant_frequency_hz > 1800 and f.harmonic_ratio > 0.55,
             "A sustained {dominant_frequency_hz:.0f} Hz tone that is strongly harmonic "
             "(ratio {harmonic_ratio:.2f}) is typical of a slipping/worn drive belt."),
        Rule("Drum load imbalance", "LOW", 0.60,
             lambda f: f.rms_variance > 0.02 and not f.has_regular_pattern,
             "Large loudness swings (variance {rms_variance:.4f}) with no rhythm point to "
             "an unbalanced drum load rather than a mechanical fault."),
        Rule("Foreign object in drum/pump", "MEDIUM", 0.65,
             lambda f: f.onset_rate_per_sec > 4 and not f.has_regular_pattern
                       and f.zero_crossing_rate > 0.12,
             "Frequent irregular harsh knocks ({onset_rate_per_sec:.1f}/s) suggest a coin or "
             "clip loose in the drum or pump."),
    ],
    "Electric fan": [
        Rule("Blade imbalance", "MEDIUM", 0.80,
             lambda f: f.rms_variance > 0.008 and f.dominant_frequency_hz < 120,
             "Low-frequency hum at {dominant_frequency_hz:.0f} Hz modulating in amplitude "
             "(variance {rms_variance:.4f}) indicates an unbalanced or dusty blade."),
        Rule("Dry / failing motor bearing", "HIGH", 0.85,
             lambda f: f.spectral_centroid_hz > 2500 and f.zero_crossing_rate > 0.15,
             "A bright {spectral_centroid_hz:.0f} Hz, harsh (ZCR {zero_crossing_rate:.2f}) "
             "sound is a dry motor bearing approaching failure."),
        Rule("Blade striking housing", "HIGH", 0.75,
             lambda f: f.has_regular_pattern and f.pattern_interval_ms < 100,
             "Fast regular ticking every {pattern_interval_ms:.0f} ms means a blade is "
             "clipping the cage or housing on each rotation."),
    ],
    "Electric motor (generic)": [
        Rule("Bearing failure", "HIGH", 0.88,
             lambda f: f.has_regular_pattern and f.spectral_centroid_hz > 1800,
             "Periodic impacts every {pattern_interval_ms:.0f} ms with high spectral centroid "
             "{spectral_centroid_hz:.0f} Hz indicate bearing-race damage."),
        Rule("Electrical hum / loose lamination", "LOW", 0.60,
             lambda f: 90 < f.dominant_frequency_hz < 130 and f.harmonic_ratio > 0.6,
             "A steady {dominant_frequency_hz:.0f} Hz harmonic hum is mains-related "
             "(loose laminations); usually cosmetic."),
        Rule("Brush / commutator arcing", "MEDIUM", 0.70,
             lambda f: f.zero_crossing_rate > 0.2 and f.spectral_bandwidth_hz > 3000,
             "Very harsh broadband noise (ZCR {zero_crossing_rate:.2f}, bandwidth "
             "{spectral_bandwidth_hz:.0f} Hz) suggests worn brushes arcing on the commutator."),
        Rule("High-frequency squeal / bearing whine", "MEDIUM", 0.75,
             lambda f: f.dominant_frequency_hz > 1800 and f.harmonic_ratio > 0.5,
             "A sustained {dominant_frequency_hz:.0f} Hz tonal whine (harmonic ratio "
             "{harmonic_ratio:.2f}) points to a dry bearing or a glazed drive belt squealing."),
    ],
    "Tumble dryer": [
        Rule("Drum roller wear", "HIGH", 0.85,
             lambda f: f.has_regular_pattern and 80 < f.pattern_interval_ms < 300
                       and f.spectral_centroid_hz > 800,
             "Rhythmic thump every {pattern_interval_ms:.0f} ms with moderate spectral "
             "brightness ({spectral_centroid_hz:.0f} Hz) is classic drum-roller wear."),
        Rule("Belt slipping / glazing", "MEDIUM", 0.70,
             lambda f: f.dominant_frequency_hz > 1800 and f.harmonic_ratio > 0.50,
             "A sustained high {dominant_frequency_hz:.0f} Hz squeal with strong harmonic "
             "content (ratio {harmonic_ratio:.2f}) indicates a glazed or slipping belt."),
        Rule("Foreign object (coins / buttons)", "LOW", 0.60,
             lambda f: f.onset_rate_per_sec > 5 and not f.has_regular_pattern,
             "Frequent irregular rattling ({onset_rate_per_sec:.1f}/s) is typical of coins "
             "or buttons trapped between the drum and the tub."),
    ],
    "Refrigerator/Freezer": [
        Rule("Compressor bearing failure", "HIGH", 0.88,
             lambda f: f.has_regular_pattern and f.spectral_centroid_hz > 1500
                       and f.pattern_interval_ms < 150,
             "Fast regular clicking every {pattern_interval_ms:.0f} ms with a bright "
             "{spectral_centroid_hz:.0f} Hz signature points to compressor bearing wear."),
        Rule("Evaporator fan motor bearing", "MEDIUM", 0.72,
             lambda f: f.has_regular_pattern and f.dominant_frequency_hz > 400
                       and f.spectral_centroid_hz > 2000,
             "A steady {dominant_frequency_hz:.0f} Hz drone with high brightness "
             "({spectral_centroid_hz:.0f} Hz) suggests a failing evaporator fan bearing."),
        Rule("Condenser fan grinding", "MEDIUM", 0.65,
             lambda f: f.zero_crossing_rate > 0.15 and f.spectral_bandwidth_hz > 2500,
             "Broadband harsh noise (ZCR {zero_crossing_rate:.2f}, bandwidth "
             "{spectral_bandwidth_hz:.0f} Hz) is consistent with condenser fan rub or grind."),
    ],
    "Air conditioner": [
        Rule("Compressor failure", "CRITICAL", 0.92,
             lambda f: f.has_regular_pattern and f.spectral_centroid_hz > 1800
                       and f.rms_db > -30,
             "Loud ({rms_db:.0f} dB) rhythmic knocking every {pattern_interval_ms:.0f} ms "
             "with high spectral content ({spectral_centroid_hz:.0f} Hz) is compressor "
             "failure — shut down immediately."),
        Rule("Fan blade damage / debris", "MEDIUM", 0.70,
             lambda f: f.has_regular_pattern and f.pattern_interval_ms < 200
                       and f.spectral_centroid_hz < 1500,
             "Low-frequency thwack every {pattern_interval_ms:.0f} ms with a dull spectrum "
             "({spectral_centroid_hz:.0f} Hz) indicates a damaged or obstructed fan blade."),
        Rule("Refrigerant leak (hissing)", "MEDIUM", 0.65,
             lambda f: f.spectral_centroid_hz > 3000 and f.zero_crossing_rate > 0.12
                       and not f.has_regular_pattern,
             "A bright hiss ({spectral_centroid_hz:.0f} Hz) with no periodic structure is "
             "consistent with refrigerant escaping from a leak."),
    ],
    "Vacuum cleaner": [
        Rule("Brush roll bearing failure", "HIGH", 0.85,
             lambda f: f.has_regular_pattern and f.spectral_centroid_hz > 2000
                       and f.pattern_interval_ms < 80,
             "Very fast regular clicking every {pattern_interval_ms:.0f} ms with a bright "
             "{spectral_centroid_hz:.0f} Hz tone indicates brush-roll bearing failure."),
        Rule("Motor bearing whine", "MEDIUM", 0.75,
             lambda f: f.dominant_frequency_hz > 2000 and f.harmonic_ratio > 0.50,
             "A sustained high-pitched whine at {dominant_frequency_hz:.0f} Hz with strong "
             "harmonics (ratio {harmonic_ratio:.2f}) is a dry motor bearing."),
        Rule("Airway blockage", "MEDIUM", 0.68,
             lambda f: f.rms_db > -25 and f.spectral_centroid_hz > 2500
                       and not f.has_regular_pattern,
             "Unusually loud ({rms_db:.0f} dB) broadband rush ({spectral_centroid_hz:.0f} Hz) "
             "without periodic structure suggests a clogged airway or full dustbin."),
    ],
    "Dishwasher": [
        Rule("Wash pump bearing failure", "HIGH", 0.82,
             lambda f: f.has_regular_pattern and f.spectral_centroid_hz > 1500
                       and f.pattern_interval_ms < 200,
             "Rhythmic rattling every {pattern_interval_ms:.0f} ms with bright content "
             "({spectral_centroid_hz:.0f} Hz) indicates a worn wash-pump bearing."),
        Rule("Drain pump cavitating", "MEDIUM", 0.68,
             lambda f: f.onset_rate_per_sec > 4 and f.spectral_bandwidth_hz > 3000
                       and not f.has_regular_pattern,
             "Irregular gurgling ({onset_rate_per_sec:.1f}/s) with broad harsh spectrum "
             "({spectral_bandwidth_hz:.0f} Hz) suggests the drain pump is cavitating."),
        Rule("Spray arm imbalance", "LOW", 0.55,
             lambda f: f.has_regular_pattern and f.dominant_frequency_hz < 200
                       and f.rms_variance > 0.01,
             "Slow regular swish every {pattern_interval_ms:.0f} ms with loudness variation "
             "(variance {rms_variance:.4f}) points to a bent or blocked spray arm."),
    ],
    "Microwave": [
        Rule("Turntable motor failure", "MEDIUM", 0.70,
             lambda f: f.dominant_frequency_hz > 0 and f.dominant_frequency_hz < 100
                       and f.harmonic_ratio > 0.6 and not f.has_regular_pattern,
             "A steady low hum ({dominant_frequency_hz:.0f} Hz) with high harmonic purity "
             "(ratio {harmonic_ratio:.2f}) but no mechanical clicking means the turntable "
             "motor has seized while the magnetron runs."),
        Rule("Magnetron arcing / failure", "HIGH", 0.85,
             lambda f: f.zero_crossing_rate > 0.2 and f.spectral_centroid_hz > 2000
                       and f.rms_db > -25,
             "Loud ({rms_db:.0f} dB) harsh buzzing (ZCR {zero_crossing_rate:.2f}) with "
             "bright spectrum ({spectral_centroid_hz:.0f} Hz) indicates magnetron arcing."),
        Rule("Cooling fan bearing", "LOW", 0.60,
             lambda f: f.has_regular_pattern and f.pattern_interval_ms < 150
                       and f.spectral_centroid_hz > 1200,
             "Regular fast ticking every {pattern_interval_ms:.0f} ms at moderate brightness "
             "({spectral_centroid_hz:.0f} Hz) is the cooling fan bearing wearing out."),
    ],
    "Bicycle (chain/gears)": [
        Rule("Chain wear / dry links", "MEDIUM", 0.72,
             lambda f: f.has_regular_pattern and f.pattern_interval_ms < 200
                       and f.spectral_centroid_hz > 1500,
             "Fast rhythmic clicking every {pattern_interval_ms:.0f} ms with a bright "
             "{spectral_centroid_hz:.0f} Hz tone is classic chain-link wear."),
        Rule("Wheel bearing pitting", "HIGH", 0.85,
             lambda f: f.has_regular_pattern and 50 < f.pattern_interval_ms < 400
                       and f.spectral_centroid_hz > 2000,
             "Regular thump every {pattern_interval_ms:.0f} ms with high-frequency "
             "content ({spectral_centroid_hz:.0f} Hz) indicates a pitted wheel bearing."),
        Rule("Derailleur misalignment", "MEDIUM", 0.65,
             lambda f: f.onset_rate_per_sec > 3 and not f.has_regular_pattern
                       and f.spectral_centroid_hz > 1800,
             "Irregular metallic rattling ({onset_rate_per_sec:.1f}/s) with bright tones "
             "({spectral_centroid_hz:.0f} Hz) suggests the derailleur is out of alignment."),
    ],
    "Power drill": [
        Rule("Brush / commutator wear", "MEDIUM", 0.75,
             lambda f: f.zero_crossing_rate > 0.2 and f.spectral_bandwidth_hz > 3000,
             "Very harsh broadband noise (ZCR {zero_crossing_rate:.2f}, bandwidth "
             "{spectral_bandwidth_hz:.0f} Hz) is consistent with worn motor brushes."),
        Rule("Gear train grinding", "MEDIUM", 0.70,
             lambda f: f.spectral_centroid_hz > 1800 and f.harmonic_ratio < 0.4
                       and f.onset_rate_per_sec > 3,
             "Bright ({spectral_centroid_hz:.0f} Hz) but non-tonal (harmonic ratio "
             "{harmonic_ratio:.2f}) grinding with rapid onsets ({onset_rate_per_sec:.1f}/s) "
             "points to worn or chipped gears."),
        Rule("Bearing failure", "HIGH", 0.82,
             lambda f: f.has_regular_pattern and f.spectral_centroid_hz > 2000
                       and f.pattern_interval_ms < 100,
             "Fast regular ticking every {pattern_interval_ms:.0f} ms with high spectral "
             "brightness ({spectral_centroid_hz:.0f} Hz) is a spindle or armature bearing "
             "approaching failure."),
    ],
    "Car engine": [
        Rule("Rod knock / bearing", "CRITICAL", 0.90,
             lambda f: f.has_regular_pattern and 20 < f.pattern_interval_ms < 200
                       and f.spectral_centroid_hz > 1200,
             "A rhythmic deep knock every {pattern_interval_ms:.0f} ms that tracks RPM is a "
             "rod/main bearing knock — stop driving."),
        Rule("Belt squeal (serpentine/timing)", "MEDIUM", 0.72,
             lambda f: f.dominant_frequency_hz > 2000 and f.harmonic_ratio > 0.5,
             "A high {dominant_frequency_hz:.0f} Hz squeal is usually a glazed/loose "
             "serpentine belt or tensioner."),
        Rule("Exhaust leak / tappet", "MEDIUM", 0.65,
             lambda f: f.onset_rate_per_sec > 6 and f.zero_crossing_rate > 0.1,
             "Rapid ticking ({onset_rate_per_sec:.1f}/s) can be a noisy tappet or a small "
             "exhaust-manifold leak."),
    ],
}

# Appliances without a dedicated table fall back to the generic motor rules.
GENERIC_FALLBACK = "Electric motor (generic)"


def rank_candidates(features: AudioFeatures, appliance: str) -> List[Candidate]:
    """Return fault candidates whose rules fire, strongest first.

    If nothing fires, return a single low-confidence 'inconclusive' candidate so
    downstream code (and the model) always has something honest to say.
    """
    table = RULES.get(appliance) or RULES.get(GENERIC_FALLBACK, [])
    fired: List[Candidate] = []
    for rule in table:
        try:
            if rule.when(features):
                fired.append(Candidate(
                    name=rule.name, urgency=rule.urgency, weight=rule.weight,
                    evidence=_fmt(rule.evidence, features),
                ))
        except Exception:
            continue

    fired.sort(key=lambda c: c.weight, reverse=True)

    if not fired:
        fired.append(Candidate(
            name="Inconclusive",
            urgency="LOW",
            weight=0.0,
            evidence=(
                f"No known fault signature matched. Anomaly score is "
                f"{features.anomaly_score:.2f}; sound may be within normal range "
                f"or the recording is too quiet/short."
            ),
        ))
    return fired