Trilltechnician/giglez
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: robust timing analyzer for RF device identification - iteration 2/5

Browse Source 
- 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
This commit is contained in:
leetcrypt
2026-02-14 19:06:13 -08:00
parent 9f73595b20
commit af9942f822
3 changed files with 966 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_timing_analyzer.py
+405
View File
@@ -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'])