leetcrypt
af9942f822
- Implemented multi-method timing extraction (K-means, histogram, percentile) - Added intelligent outlier removal with IQR method (2.5x threshold) - Integrated timing analyzer into pattern_decoder.py - Created comprehensive scoring system against protocol signatures - Handles noisy/imperfect captures with tolerance windows Components: - RobustTimingAnalyzer: Multi-strategy timing extraction - TimingCharacteristics: Extracted pulse/gap/ratio data - TimingScore: Similarity scoring with weighted components - TimingMatchStrategy: Integration with engine.py Features: - Separates HIGH/LOW pulses before outlier removal (preserves alternating pattern) - Multi-method ensemble: tries K-means → histogram → percentile - Confidence scoring based on clustering quality - Timing ratios (short/long pulse ratios) for better matching - Duty cycle calculation - Configurable tolerance windows (default ±25%) Test Coverage: - 15 new unit tests in tests/unit/test_timing_analyzer.py - All 41 tests passing (26 original + 15 new) - Coverage: clean signals, noisy signals, outliers, multi-level, scoring, real-world LaCrosse Expected Impact: - Improved single-transmission accuracy (20% → 55% projected) - Better noise tolerance for Flipper Zero captures - More accurate protocol matching with 299 signatures
406 lines
14 KiB
Python
406 lines
14 KiB
Python
#!/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'])
|