diff --git a/src/matcher/pattern_decoder.py b/src/matcher/pattern_decoder.py index afef7f0..fae6639 100644 --- a/src/matcher/pattern_decoder.py +++ b/src/matcher/pattern_decoder.py @@ -20,6 +20,7 @@ from src.matcher.protocol_database import ( ProtocolDatabase, get_protocol_database ) +from src.matcher.timing_analyzer import get_timing_analyzer @dataclass @@ -82,6 +83,7 @@ class PatternDecoder: def __init__(self, protocol_db: Optional[ProtocolDatabase] = None): self.protocol_db = protocol_db or get_protocol_database() + self.timing_analyzer = get_timing_analyzer() def decode(self, metadata: SignalMetadata) -> List[DeviceMatch]: """ @@ -118,7 +120,7 @@ class PatternDecoder: def _identify_pulse_widths(self, pulses: List[int]) -> Tuple[int, int, int, int]: """ - Identify SHORT/LONG pulse and gap durations using K-means clustering + Identify SHORT/LONG pulse and gap durations using robust timing analyzer Returns: (short_pulse, long_pulse, short_gap, long_gap) in microseconds @@ -126,17 +128,15 @@ class PatternDecoder: if not pulses: return (0, 0, 0, 0) - # Separate positive (HIGH pulses) and negative (LOW gaps) - high_pulses = [abs(p) for p in pulses if p > 0] - low_pulses = [abs(p) for p in pulses if p < 0] + # Use robust timing analyzer for improved extraction + timing = self.timing_analyzer.extract_timing(pulses) - # Cluster high pulses into SHORT/LONG - short_pulse, long_pulse = self._cluster_durations(high_pulses) - - # Cluster low pulses into SHORT/LONG - short_gap, long_gap = self._cluster_durations(low_pulses) - - return (short_pulse, long_pulse, short_gap, long_gap) + return ( + timing.short_pulse_us, + timing.long_pulse_us, + timing.short_gap_us, + timing.long_gap_us + ) def _cluster_durations(self, durations: List[int]) -> Tuple[int, int]: """ diff --git a/src/matcher/timing_analyzer.py b/src/matcher/timing_analyzer.py new file mode 100644 index 0000000..052b669 --- /dev/null +++ b/src/matcher/timing_analyzer.py @@ -0,0 +1,550 @@ +#!/usr/bin/env python3 +""" +Robust Timing Analyzer for RF Device Identification + +Extracts timing characteristics from raw RF pulse data and scores against +known protocol signatures. Handles noisy/imperfect captures with statistical +methods and tolerance windows. + +Key features: + - Multi-method clustering (K-means, histogram, percentile) + - Outlier removal for noise tolerance + - Timing ratio calculation (short/long pulse ratios) + - Similarity scoring with configurable tolerance + - Support for 2-4 level modulation schemes +""" + +import numpy as np +from dataclasses import dataclass +from typing import List, Tuple, Optional, Dict +from collections import defaultdict + + +@dataclass +class TimingCharacteristics: + """ + Extracted timing characteristics from raw signal + + Attributes: + short_pulse_us: Short HIGH pulse duration (microseconds) + long_pulse_us: Long HIGH pulse duration (microseconds) + short_gap_us: Short LOW gap duration (microseconds) + long_gap_us: Long LOW gap duration (microseconds) + pulse_ratio: Ratio of long_pulse / short_pulse + gap_ratio: Ratio of long_gap / short_gap + duty_cycle: Fraction of time signal is HIGH + pulse_count: Total number of pulses + confidence: Confidence in timing extraction (0.0-1.0) + method: Extraction method used ('kmeans', 'histogram', 'percentile') + """ + short_pulse_us: int + long_pulse_us: int + short_gap_us: int + long_gap_us: int + pulse_ratio: float + gap_ratio: float + duty_cycle: float + pulse_count: int + confidence: float + method: str + + +@dataclass +class TimingScore: + """ + Similarity score between extracted timing and protocol signature + + Attributes: + protocol_name: Name of the protocol + overall_score: Overall similarity score (0.0-1.0) + pulse_score: Similarity of pulse timings (0.0-1.0) + gap_score: Similarity of gap timings (0.0-1.0) + ratio_score: Similarity of timing ratios (0.0-1.0) + frequency_score: Frequency match score (0.0-1.0) + details: Additional scoring details + """ + protocol_name: str + overall_score: float + pulse_score: float + gap_score: float + ratio_score: float + frequency_score: float + details: Dict + + +class RobustTimingAnalyzer: + """ + Robust timing analyzer with noise tolerance and multi-method extraction + + Workflow: + 1. Clean pulse data (remove outliers, normalize) + 2. Extract timing characteristics (try multiple methods) + 3. Calculate timing ratios and statistics + 4. Score against protocol database + 5. Return ranked matches + """ + + def __init__(self, outlier_threshold: float = 1.5): + """ + Initialize analyzer + + Args: + outlier_threshold: IQR multiplier for outlier detection (default: 1.5) + """ + self.outlier_threshold = outlier_threshold + + def extract_timing(self, pulses: List[int]) -> TimingCharacteristics: + """ + Extract timing characteristics from pulse data + + Uses multi-method approach and returns best result: + 1. K-means clustering (preferred for clean signals) + 2. Histogram peak detection (good for multi-modal) + 3. Percentile-based (fallback for noisy signals) + + Args: + pulses: List of pulse durations (positive=HIGH, negative=LOW) + + Returns: + TimingCharacteristics with confidence score + """ + if not pulses or len(pulses) < 4: + return self._empty_timing() + + # Step 1: Separate HIGH and LOW pulses BEFORE outlier removal + # (outlier removal can corrupt alternating pattern) + high_pulses_raw = [abs(p) for p in pulses if p > 0] + low_pulses_raw = [abs(p) for p in pulses if p < 0] + + if not high_pulses_raw: + return self._empty_timing() + + # Step 2: Remove outliers from HIGH and LOW separately + high_pulses = self._remove_outliers_from_values(high_pulses_raw) + low_pulses = self._remove_outliers_from_values(low_pulses_raw) if low_pulses_raw else [] + + if not high_pulses: + return self._empty_timing() + + # Step 3: Try multiple extraction methods + methods = [ + ('kmeans', self._extract_kmeans), + ('histogram', self._extract_histogram), + ('percentile', self._extract_percentile) + ] + + best_timing = None + best_confidence = 0.0 + + for method_name, method_func in methods: + try: + timing = method_func(high_pulses, low_pulses) + if timing and timing.confidence > best_confidence: + best_timing = timing + best_confidence = timing.confidence + best_timing.method = method_name + except Exception: + continue + + if best_timing: + return best_timing + else: + return self._empty_timing() + + def score_against_protocol( + self, + timing: TimingCharacteristics, + protocol_short_us: int, + protocol_long_us: int, + protocol_frequency: int, + signal_frequency: int, + timing_tolerance: float = 0.25 + ) -> TimingScore: + """ + Score timing characteristics against a protocol signature + + Args: + timing: Extracted timing characteristics + protocol_short_us: Protocol's short pulse width + protocol_long_us: Protocol's long pulse width + protocol_frequency: Protocol's expected frequency + signal_frequency: Actual signal frequency + timing_tolerance: Tolerance for timing match (default: ±25%) + + Returns: + TimingScore with component scores and overall score + """ + # Pulse timing similarity + short_error = abs(timing.short_pulse_us - protocol_short_us) / protocol_short_us + long_error = abs(timing.long_pulse_us - protocol_long_us) / protocol_long_us + + pulse_score = self._error_to_score( + max(short_error, long_error), + tolerance=timing_tolerance + ) + + # Gap timing similarity (if available) + gap_score = 1.0 # Default: don't penalize gaps + + # Ratio similarity + protocol_ratio = protocol_long_us / protocol_short_us if protocol_short_us > 0 else 0 + ratio_error = abs(timing.pulse_ratio - protocol_ratio) / max(protocol_ratio, 0.01) + ratio_score = self._error_to_score(ratio_error, tolerance=0.3) + + # Frequency similarity + freq_error = abs(signal_frequency - protocol_frequency) / protocol_frequency + frequency_score = self._error_to_score(freq_error, tolerance=0.01) # ±1% + + # Overall score (weighted average) + overall_score = ( + pulse_score * 0.50 + # Pulse timing most important + ratio_score * 0.25 + # Ratio consistency + frequency_score * 0.20 + # Frequency match + gap_score * 0.05 # Gaps less important + ) * timing.confidence # Scale by extraction confidence + + return TimingScore( + protocol_name="", # Will be set by caller + overall_score=overall_score, + pulse_score=pulse_score, + gap_score=gap_score, + ratio_score=ratio_score, + frequency_score=frequency_score, + details={ + 'short_error_pct': f"{short_error:.1%}", + 'long_error_pct': f"{long_error:.1%}", + 'ratio_error_pct': f"{ratio_error:.1%}", + 'freq_error_pct': f"{freq_error:.1%}", + 'extraction_method': timing.method + } + ) + + # === Internal Methods === + + def _remove_outliers_from_values(self, values: List[int]) -> List[int]: + """ + Remove statistical outliers from a list of values using IQR method + + Args: + values: List of positive values (already abs()) + + Returns: + Cleaned values + """ + if len(values) < 10: + return values # Too few samples for outlier detection + + q1 = np.percentile(values, 25) + q3 = np.percentile(values, 75) + iqr = q3 - q1 + + # Use more lenient threshold (2.5 instead of 1.5) + lower_bound = q1 - 2.5 * iqr + upper_bound = q3 + 2.5 * iqr + + cleaned = [v for v in values if lower_bound <= v <= upper_bound] + + # Only apply if doesn't remove too much + if len(cleaned) >= len(values) * 0.7: + return cleaned + else: + return values + + def _extract_kmeans( + self, + high_pulses: List[int], + low_pulses: List[int] + ) -> Optional[TimingCharacteristics]: + """Extract timing using K-means clustering (k=2)""" + try: + from sklearn.cluster import KMeans + from sklearn.metrics import silhouette_score + except ImportError: + return None + + if len(high_pulses) < 10: + return None + + X = np.array(high_pulses).reshape(-1, 1) + kmeans = KMeans(n_clusters=2, random_state=42, n_init=10) + labels = kmeans.fit_predict(X) + + # Silhouette score: quality of clustering + confidence = silhouette_score(X, labels) + + # Get cluster centers + centers = sorted(kmeans.cluster_centers_.flatten()) + short_pulse = int(centers[0]) + long_pulse = int(centers[1]) + + # Extract gap timings + short_gap, long_gap = self._extract_gaps_percentile(low_pulses) + + # Calculate metrics + pulse_ratio = long_pulse / short_pulse if short_pulse > 0 else 0 + gap_ratio = long_gap / short_gap if short_gap > 0 else 0 + duty_cycle = self._calculate_duty_cycle(high_pulses, low_pulses) + + return TimingCharacteristics( + short_pulse_us=short_pulse, + long_pulse_us=long_pulse, + short_gap_us=short_gap, + long_gap_us=long_gap, + pulse_ratio=pulse_ratio, + gap_ratio=gap_ratio, + duty_cycle=duty_cycle, + pulse_count=len(high_pulses), + confidence=max(0.0, min(1.0, confidence)), # Clamp to [0, 1] + method='kmeans' + ) + + def _extract_histogram( + self, + high_pulses: List[int], + low_pulses: List[int] + ) -> Optional[TimingCharacteristics]: + """Extract timing using histogram peak detection""" + if len(high_pulses) < 10: + return None + + # Create histogram + hist, bin_edges = np.histogram(high_pulses, bins=min(50, len(high_pulses) // 5)) + + # Find peaks + try: + from scipy.signal import find_peaks + peaks, properties = find_peaks(hist, height=len(high_pulses) * 0.05) + except ImportError: + # Fallback: simple peak detection + peaks = [] + for i in range(1, len(hist) - 1): + if hist[i] > hist[i-1] and hist[i] > hist[i+1]: + peaks.append(i) + + if len(peaks) < 2: + return None + + # Get pulse widths at peak locations + peak_widths = [int(bin_edges[p]) for p in peaks] + peak_widths_sorted = sorted(peak_widths) + + short_pulse = peak_widths_sorted[0] + long_pulse = peak_widths_sorted[1] + + # Extract gaps + short_gap, long_gap = self._extract_gaps_percentile(low_pulses) + + # Confidence: ratio of peak heights to baseline + peak_heights = [hist[p] for p in peaks[:2]] + baseline = np.median(hist) + snr = np.mean(peak_heights) / (baseline + 1) + confidence = min(1.0, snr / 10) + + # Calculate metrics + pulse_ratio = long_pulse / short_pulse if short_pulse > 0 else 0 + gap_ratio = long_gap / short_gap if short_gap > 0 else 0 + duty_cycle = self._calculate_duty_cycle(high_pulses, low_pulses) + + return TimingCharacteristics( + short_pulse_us=short_pulse, + long_pulse_us=long_pulse, + short_gap_us=short_gap, + long_gap_us=long_gap, + pulse_ratio=pulse_ratio, + gap_ratio=gap_ratio, + duty_cycle=duty_cycle, + pulse_count=len(high_pulses), + confidence=confidence, + method='histogram' + ) + + def _extract_percentile( + self, + high_pulses: List[int], + low_pulses: List[int] + ) -> TimingCharacteristics: + """Extract timing using percentile-based approach (fallback)""" + # Use percentiles that are robust to outliers (15th/85th instead of min/max) + short_pulse = int(np.percentile(high_pulses, 15)) + long_pulse = int(np.percentile(high_pulses, 85)) + + # Extract gaps + short_gap, long_gap = self._extract_gaps_percentile(low_pulses) + + # Lower confidence for percentile method + confidence = 0.6 + + # Calculate metrics + pulse_ratio = long_pulse / short_pulse if short_pulse > 0 else 0 + gap_ratio = long_gap / short_gap if short_gap > 0 else 0 + duty_cycle = self._calculate_duty_cycle(high_pulses, low_pulses) + + return TimingCharacteristics( + short_pulse_us=short_pulse, + long_pulse_us=long_pulse, + short_gap_us=short_gap, + long_gap_us=long_gap, + pulse_ratio=pulse_ratio, + gap_ratio=gap_ratio, + duty_cycle=duty_cycle, + pulse_count=len(high_pulses), + confidence=confidence, + method='percentile' + ) + + def _extract_gaps_percentile(self, low_pulses: List[int]) -> Tuple[int, int]: + """Extract gap timings using percentile method""" + if not low_pulses: + return (0, 0) + + short_gap = int(np.percentile(low_pulses, 25)) + long_gap = int(np.percentile(low_pulses, 75)) + + return (short_gap, long_gap) + + def _calculate_duty_cycle( + self, + high_pulses: List[int], + low_pulses: List[int] + ) -> float: + """Calculate duty cycle (fraction of time signal is HIGH)""" + total_high = sum(high_pulses) + total_low = sum(low_pulses) + total_time = total_high + total_low + + if total_time == 0: + return 0.0 + + return total_high / total_time + + def _error_to_score(self, error: float, tolerance: float) -> float: + """ + Convert error percentage to similarity score + + Uses smooth falloff within tolerance window: + - error = 0 → score = 1.0 + - error = tolerance → score = 0.5 + - error = 2*tolerance → score = 0.0 + + Args: + error: Error as fraction (0.2 = 20% error) + tolerance: Tolerance threshold + + Returns: + Score between 0.0 and 1.0 + """ + if error <= tolerance: + # Linear falloff within tolerance + return 1.0 - (error / tolerance) * 0.5 + elif error <= 2 * tolerance: + # Continue falloff to 0 + normalized = (error - tolerance) / tolerance + return 0.5 * (1.0 - normalized) + else: + return 0.0 + + def _empty_timing(self) -> TimingCharacteristics: + """Return empty timing characteristics""" + return TimingCharacteristics( + short_pulse_us=0, + long_pulse_us=0, + short_gap_us=0, + long_gap_us=0, + pulse_ratio=0.0, + gap_ratio=0.0, + duty_cycle=0.0, + pulse_count=0, + confidence=0.0, + method='none' + ) + + +# === Strategy Integration === + +class TimingMatchStrategy: + """ + Matching strategy that uses timing analyzer + + Integrates with engine.py's MatchStrategy interface + """ + + def __init__(self, protocol_db): + """ + Initialize strategy + + Args: + protocol_db: Protocol database instance + """ + self.protocol_db = protocol_db + self.analyzer = RobustTimingAnalyzer() + + def match(self, metadata, db) -> List: + """ + Match signal metadata using timing analysis + + Args: + metadata: SignalMetadata object + db: Database connection (unused, uses protocol_db) + + Returns: + List of MatchResult objects + """ + from ..matcher.engine import MatchResult + + if not metadata.has_raw_data: + return [] + + # Extract timing from signal + timing = self.analyzer.extract_timing(metadata.raw_data) + + if timing.confidence < 0.3: # Too noisy + return [] + + # Score against all protocols + matches = [] + for protocol in self.protocol_db.get_all(): + score = self.analyzer.score_against_protocol( + timing=timing, + protocol_short_us=protocol.short_pulse_us, + protocol_long_us=protocol.long_pulse_us, + protocol_frequency=protocol.frequency, + signal_frequency=metadata.frequency, + timing_tolerance=protocol.timing_tolerance + ) + + # Only keep reasonable matches + if score.overall_score >= protocol.min_confidence: + matches.append(MatchResult( + device_id=0, # Will be filled by database lookup + device_name=protocol.name, + manufacturer=protocol.manufacturer or "Unknown", + confidence=score.overall_score, + match_method='timing', + match_details={ + 'pulse_score': score.pulse_score, + 'ratio_score': score.ratio_score, + 'frequency_score': score.frequency_score, + 'timing_method': timing.method, + 'short_pulse_us': timing.short_pulse_us, + 'long_pulse_us': timing.long_pulse_us, + 'pulse_ratio': f"{timing.pulse_ratio:.2f}", + **score.details + } + )) + + # Sort by confidence + matches.sort(key=lambda m: m.confidence, reverse=True) + + return matches[:20] # Return top 20 + + +# Singleton instance +_analyzer: Optional[RobustTimingAnalyzer] = None + + +def get_timing_analyzer() -> RobustTimingAnalyzer: + """Get singleton timing analyzer instance""" + global _analyzer + if _analyzer is None: + _analyzer = RobustTimingAnalyzer() + return _analyzer diff --git a/tests/unit/test_timing_analyzer.py b/tests/unit/test_timing_analyzer.py new file mode 100644 index 0000000..3995f02 --- /dev/null +++ b/tests/unit/test_timing_analyzer.py @@ -0,0 +1,405 @@ +#!/usr/bin/env python3 +""" +Unit Tests for Robust Timing Analyzer + +Tests timing extraction, scoring, and noise tolerance capabilities. +""" + +import pytest +import numpy as np +from src.matcher.timing_analyzer import ( + RobustTimingAnalyzer, + TimingCharacteristics, + TimingScore +) + + +class TestTimingExtraction: + """Test timing characteristic extraction""" + + def test_clean_signal_extraction(self): + """Test 1: Extract timing from clean PWM signal""" + analyzer = RobustTimingAnalyzer() + + # Simulated PWM signal: SHORT=500us, LONG=1000us + # Pattern: 0101010101 (10 bits) + pulses = [ + 500, -500, # 0 + 1000, -500, # 1 + 500, -500, # 0 + 1000, -500, # 1 + 500, -500, # 0 + 1000, -500, # 1 + 500, -500, # 0 + 1000, -500, # 1 + 500, -500, # 0 + 1000, -500, # 1 + ] + + timing = analyzer.extract_timing(pulses) + + # Verify extracted timings + assert 450 <= timing.short_pulse_us <= 550, f"Expected short ~500, got {timing.short_pulse_us}" + assert 950 <= timing.long_pulse_us <= 1050, f"Expected long ~1000, got {timing.long_pulse_us}" + assert timing.pulse_ratio >= 1.8 and timing.pulse_ratio <= 2.2, f"Expected ratio ~2.0, got {timing.pulse_ratio}" + assert timing.confidence > 0.5, f"Expected confidence >0.5, got {timing.confidence}" + assert timing.pulse_count == 10 + + def test_noisy_signal_extraction(self): + """Test 2: Extract timing from noisy signal with outliers""" + analyzer = RobustTimingAnalyzer() + + # Signal with noise and outliers + pulses = [ + 500, -500, + 1000, -500, + 500, -500, + 5000, # OUTLIER: glitch + -500, + 1000, -500, + 500, -500, + 1000, -500, + 50, # OUTLIER: noise spike + -500, + 1000, -500, + ] + + timing = analyzer.extract_timing(pulses) + + # Should still extract reasonable timings despite outliers + assert 400 <= timing.short_pulse_us <= 600, f"Short pulse should be ~500, got {timing.short_pulse_us}" + assert 900 <= timing.long_pulse_us <= 1100, f"Long pulse should be ~1000, got {timing.long_pulse_us}" + assert timing.confidence > 0.3, "Should have some confidence despite noise" + + def test_multi_level_signal(self): + """Test 3: Handle signal with multiple pulse widths""" + analyzer = RobustTimingAnalyzer() + + # Tri-level signal: 250us, 500us, 1000us + pulses = [ + 250, -250, + 500, -250, + 1000, -250, + 250, -250, + 500, -250, + 1000, -250, + 250, -250, + 500, -250, + 1000, -250, + ] + + timing = analyzer.extract_timing(pulses) + + # Should extract shortest and longest + assert timing.short_pulse_us < 400, "Should identify shortest pulse" + assert timing.long_pulse_us > 800, "Should identify longest pulse" + assert timing.pulse_count > 0 + + def test_empty_signal(self): + """Test 4: Handle empty/invalid input gracefully""" + analyzer = RobustTimingAnalyzer() + + # Empty signal + timing = analyzer.extract_timing([]) + assert timing.confidence == 0.0 + assert timing.short_pulse_us == 0 + + # Too short signal + timing = analyzer.extract_timing([500, -500]) + assert timing.confidence == 0.0 + + def test_duty_cycle_calculation(self): + """Test 5: Verify duty cycle calculation""" + analyzer = RobustTimingAnalyzer() + + # 50% duty cycle: equal HIGH and LOW time + pulses = [1000, -1000, 1000, -1000, 1000, -1000] + timing = analyzer.extract_timing(pulses) + assert 0.45 <= timing.duty_cycle <= 0.55, f"Expected duty cycle ~0.5, got {timing.duty_cycle}" + + # 25% duty cycle: SHORT HIGH, long LOW + pulses = [500, -1500, 500, -1500, 500, -1500] + timing = analyzer.extract_timing(pulses) + assert 0.20 <= timing.duty_cycle <= 0.30, f"Expected duty cycle ~0.25, got {timing.duty_cycle}" + + +class TestProtocolScoring: + """Test scoring against protocol signatures""" + + def test_perfect_match_scoring(self): + """Test 6: Score perfect timing match""" + analyzer = RobustTimingAnalyzer() + + # Perfect match timing + timing = TimingCharacteristics( + short_pulse_us=500, + long_pulse_us=1000, + short_gap_us=500, + long_gap_us=500, + pulse_ratio=2.0, + gap_ratio=1.0, + duty_cycle=0.5, + pulse_count=20, + confidence=1.0, + method='kmeans' + ) + + # Score against matching protocol + score = analyzer.score_against_protocol( + timing=timing, + protocol_short_us=500, + protocol_long_us=1000, + protocol_frequency=433920000, + signal_frequency=433920000, + timing_tolerance=0.25 + ) + + # Should have very high scores + assert score.overall_score > 0.9, f"Expected score >0.9, got {score.overall_score}" + assert score.pulse_score > 0.95, f"Expected pulse score >0.95, got {score.pulse_score}" + assert score.frequency_score > 0.95, f"Expected frequency score >0.95, got {score.frequency_score}" + + def test_partial_match_scoring(self): + """Test 7: Score partial timing match within tolerance""" + analyzer = RobustTimingAnalyzer() + + # 10% timing error (within 25% tolerance) + timing = TimingCharacteristics( + short_pulse_us=550, # 10% higher than 500 + long_pulse_us=1100, # 10% higher than 1000 + short_gap_us=500, + long_gap_us=500, + pulse_ratio=2.0, + gap_ratio=1.0, + duty_cycle=0.5, + pulse_count=20, + confidence=0.9, + method='kmeans' + ) + + score = analyzer.score_against_protocol( + timing=timing, + protocol_short_us=500, + protocol_long_us=1000, + protocol_frequency=433920000, + signal_frequency=433920000, + timing_tolerance=0.25 # ±25% + ) + + # Should still score well (within tolerance) + assert score.overall_score > 0.7, f"Expected score >0.7 for 10% error, got {score.overall_score}" + assert score.pulse_score > 0.6, f"Expected pulse score >0.6, got {score.pulse_score}" + + def test_mismatch_scoring(self): + """Test 8: Score timing mismatch outside tolerance""" + analyzer = RobustTimingAnalyzer() + + # Large timing error (>50%) + timing = TimingCharacteristics( + short_pulse_us=250, # 50% lower than 500 + long_pulse_us=2000, # 100% higher than 1000 + short_gap_us=500, + long_gap_us=500, + pulse_ratio=8.0, # Wrong ratio + gap_ratio=1.0, + duty_cycle=0.5, + pulse_count=20, + confidence=0.8, + method='kmeans' + ) + + score = analyzer.score_against_protocol( + timing=timing, + protocol_short_us=500, + protocol_long_us=1000, + protocol_frequency=433920000, + signal_frequency=433920000, + timing_tolerance=0.25 + ) + + # Should have low score (outside tolerance) + assert score.overall_score < 0.3, f"Expected score <0.3 for large error, got {score.overall_score}" + assert score.pulse_score < 0.2, f"Expected pulse score <0.2, got {score.pulse_score}" + + def test_frequency_mismatch_penalty(self): + """Test 9: Frequency mismatch reduces score""" + analyzer = RobustTimingAnalyzer() + + # Perfect timing, wrong frequency + timing = TimingCharacteristics( + short_pulse_us=500, + long_pulse_us=1000, + short_gap_us=500, + long_gap_us=500, + pulse_ratio=2.0, + gap_ratio=1.0, + duty_cycle=0.5, + pulse_count=20, + confidence=1.0, + method='kmeans' + ) + + score = analyzer.score_against_protocol( + timing=timing, + protocol_short_us=500, + protocol_long_us=1000, + protocol_frequency=433920000, + signal_frequency=315000000, # Wrong frequency! + timing_tolerance=0.25 + ) + + # Frequency mismatch should reduce overall score + assert score.frequency_score < 0.2, f"Expected low frequency score, got {score.frequency_score}" + assert score.overall_score < 0.9, "Overall score should be reduced by frequency mismatch" + + def test_confidence_scaling(self): + """Test 10: Low extraction confidence scales down score""" + analyzer = RobustTimingAnalyzer() + + # Perfect timing but low confidence (noisy extraction) + timing = TimingCharacteristics( + short_pulse_us=500, + long_pulse_us=1000, + short_gap_us=500, + long_gap_us=500, + pulse_ratio=2.0, + gap_ratio=1.0, + duty_cycle=0.5, + pulse_count=20, + confidence=0.4, # Low confidence + method='percentile' + ) + + score = analyzer.score_against_protocol( + timing=timing, + protocol_short_us=500, + protocol_long_us=1000, + protocol_frequency=433920000, + signal_frequency=433920000, + timing_tolerance=0.25 + ) + + # Low confidence should scale down overall score + assert score.overall_score < 0.5, f"Expected score <0.5 with low confidence, got {score.overall_score}" + + +class TestOutlierRemoval: + """Test outlier detection and removal""" + + def test_iqr_outlier_removal(self): + """Test 11: IQR method removes outliers correctly""" + analyzer = RobustTimingAnalyzer() + + # Normal pulses + outliers + values = [500] * 10 + [1000] * 10 + [10000] # Large outlier + + cleaned = analyzer._remove_outliers_from_values(values) + + # Large outlier should be removed (outside 2.5*IQR) + assert 10000 not in cleaned, "Large outlier should be removed" + # Should keep most valid data + assert len(cleaned) == 20, f"Should have 20 pulses after cleaning, got {len(cleaned)}" + + def test_small_sample_no_removal(self): + """Test 12: Don't remove outliers from small samples""" + analyzer = RobustTimingAnalyzer() + + # Too few samples for outlier detection + values = [500, 1000, 5000] + + cleaned = analyzer._remove_outliers_from_values(values) + + # Should return all samples + assert len(cleaned) == 3, "Small samples should not be filtered" + + +class TestMultiMethodExtraction: + """Test multi-method extraction approach""" + + def test_kmeans_preferred_for_clean(self): + """Test 13: K-means chosen for clean bimodal signals""" + try: + from sklearn.cluster import KMeans + has_sklearn = True + except ImportError: + has_sklearn = False + + if not has_sklearn: + pytest.skip("sklearn not available") + + analyzer = RobustTimingAnalyzer() + + # Clean bimodal signal (perfect for K-means) + pulses = [500] * 10 + [1000] * 10 + + timing = analyzer.extract_timing(pulses) + + # K-means should be selected (highest confidence) + assert timing.method == 'kmeans', f"Expected kmeans, got {timing.method}" + assert timing.confidence > 0.7, f"Expected high confidence, got {timing.confidence}" + + def test_fallback_to_percentile(self): + """Test 14: Fallback to percentile for difficult signals""" + analyzer = RobustTimingAnalyzer() + + # Challenging signal: gradual variation + pulses = list(range(400, 600, 10)) + list(range(900, 1100, 10)) + + timing = analyzer.extract_timing(pulses) + + # Should extract reasonable timings using some method + assert timing.short_pulse_us > 0, "Should extract short pulse" + assert timing.long_pulse_us > timing.short_pulse_us, "Long > short" + assert timing.confidence > 0.0, "Should have some confidence" + + +# Integration test +def test_real_world_lacrosse_signal(): + """Test 15: Process real-world LaCrosse TX141 signal pattern""" + analyzer = RobustTimingAnalyzer() + + # Simulated LaCrosse TX141-BV2 timing (500us/1000us PWM) + # 40-bit transmission + lacrosse_pulses = [] + # Preamble: 4x "10" + for _ in range(4): + lacrosse_pulses.extend([1000, -500, 500, -500]) + # Data: random pattern + data_bits = "10110100" * 4 # 32 bits + for bit in data_bits: + if bit == '1': + lacrosse_pulses.extend([1000, -500]) + else: + lacrosse_pulses.extend([500, -500]) + + # Add some noise + np.random.seed(42) + noisy_pulses = [ + int(p * np.random.normal(1.0, 0.05)) # ±5% jitter + for p in lacrosse_pulses + ] + + timing = analyzer.extract_timing(noisy_pulses) + + # Should extract LaCrosse-like timing + assert 450 <= timing.short_pulse_us <= 550, f"Expected short ~500, got {timing.short_pulse_us}" + assert 950 <= timing.long_pulse_us <= 1050, f"Expected long ~1000, got {timing.long_pulse_us}" + assert timing.pulse_ratio >= 1.7 and timing.pulse_ratio <= 2.3, "Ratio should be ~2.0" + assert timing.confidence > 0.5, "Should have good confidence" + + # Score against LaCrosse protocol + score = analyzer.score_against_protocol( + timing=timing, + protocol_short_us=500, + protocol_long_us=1000, + protocol_frequency=433920000, + signal_frequency=433920000, + timing_tolerance=0.25 + ) + + # Should match well + assert score.overall_score > 0.7, f"Expected good match, got {score.overall_score}" + + +if __name__ == '__main__': + pytest.main([__file__, '-v'])