diff --git a/src/matcher/frequency_fingerprint.py b/src/matcher/frequency_fingerprint.py new file mode 100644 index 0000000..885aab0 --- /dev/null +++ b/src/matcher/frequency_fingerprint.py @@ -0,0 +1,356 @@ +#!/usr/bin/env python3 +""" +Frequency Fingerprinting for RF Device Identification + +Uses center frequency and frequency bands as identification features. +Narrows candidate protocols by frequency before detailed timing analysis. + +Key features: + - Frequency band classification (315/433/868/915 MHz) + - Protocol filtering by frequency tolerance + - Frequency-based scoring + - ISM band detection +""" + +import numpy as np +from dataclasses import dataclass +from typing import List, Dict, Optional, Tuple +from collections import defaultdict + + +# ISM (Industrial, Scientific, Medical) band definitions +ISM_BANDS = { + '315MHz': (314_000_000, 316_000_000), # North America + '433MHz': (433_050_000, 434_790_000), # Europe (primary ISM) + '868MHz': (868_000_000, 868_600_000), # Europe SRD band + '915MHz': (902_000_000, 928_000_000), # North America ISM +} + + +@dataclass +class FrequencyFingerprint: + """ + Frequency characteristics of a signal + + Attributes: + center_freq: Center frequency in Hz + bandwidth: Estimated bandwidth in Hz (optional) + ism_band: Detected ISM band name ('315MHz', '433MHz', etc.) + frequency_offset: Offset from standard band center + confidence: Detection confidence (0.0-1.0) + """ + center_freq: int + bandwidth: Optional[int] + ism_band: Optional[str] + frequency_offset: int + confidence: float + + +@dataclass +class FrequencyMatch: + """ + Match result comparing signal frequency to protocol + + Attributes: + protocol_name: Name of matched protocol + frequency_error: Absolute error in Hz + frequency_error_pct: Error as percentage + within_tolerance: Whether within protocol's tolerance + score: Match score (0.0-1.0) + details: Additional match details + """ + protocol_name: str + frequency_error: int + frequency_error_pct: float + within_tolerance: bool + score: float + details: Dict + + +class FrequencyFingerprinter: + """ + Frequency-based protocol filtering and scoring + + Workflow: + 1. Analyze signal frequency + 2. Determine ISM band + 3. Filter protocols by frequency band + 4. Score remaining protocols by frequency match + """ + + def __init__(self): + """Initialize fingerprinter""" + self.ism_bands = ISM_BANDS + + def analyze(self, frequency: int) -> FrequencyFingerprint: + """ + Analyze frequency characteristics + + Args: + frequency: Signal center frequency in Hz + + Returns: + FrequencyFingerprint with detected characteristics + """ + # Detect ISM band + ism_band = self._detect_ism_band(frequency) + + # Calculate offset from band center + offset = 0 + if ism_band: + band_low, band_high = self.ism_bands[ism_band] + band_center = (band_low + band_high) // 2 + offset = frequency - band_center + + # Confidence based on band detection + confidence = 0.9 if ism_band else 0.6 + + return FrequencyFingerprint( + center_freq=frequency, + bandwidth=None, # Not calculated from single .sub file + ism_band=ism_band, + frequency_offset=offset, + confidence=confidence + ) + + def filter_protocols_by_frequency( + self, + protocols: List, + signal_frequency: int, + tolerance_hz: int = 200_000 # ±200 kHz default + ) -> List: + """ + Filter protocols by frequency match + + Args: + protocols: List of ProtocolSignature objects + signal_frequency: Signal frequency in Hz + tolerance_hz: Frequency tolerance in Hz + + Returns: + Filtered list of protocols within frequency range + """ + filtered = [] + + for protocol in protocols: + freq_error = abs(protocol.frequency - signal_frequency) + + # Use protocol's own tolerance if available, otherwise use default + protocol_tolerance = getattr(protocol, 'frequency_tolerance', tolerance_hz) + + if freq_error <= protocol_tolerance: + filtered.append(protocol) + + return filtered + + def score_frequency_match( + self, + signal_frequency: int, + protocol_frequency: int, + protocol_tolerance: int = 100_000 + ) -> FrequencyMatch: + """ + Score frequency match between signal and protocol + + Args: + signal_frequency: Signal frequency in Hz + protocol_frequency: Protocol expected frequency in Hz + protocol_tolerance: Protocol's frequency tolerance in Hz + + Returns: + FrequencyMatch with score + """ + freq_error = abs(signal_frequency - protocol_frequency) + freq_error_pct = freq_error / protocol_frequency if protocol_frequency > 0 else 1.0 + + within_tolerance = freq_error <= protocol_tolerance + + # Score calculation: + # - Exact match = 1.0 + # - Within tolerance = 0.8-1.0 (linear falloff) + # - Outside tolerance = 0.0-0.5 (steep falloff) + if freq_error == 0: + score = 1.0 + elif within_tolerance: + # Linear falloff within tolerance + normalized = freq_error / protocol_tolerance + score = 1.0 - (normalized * 0.2) # 1.0 → 0.8 + else: + # Steep falloff outside tolerance + excess = freq_error - protocol_tolerance + # Penalty: 50% score at 2x tolerance, 0% at 4x tolerance + if excess < protocol_tolerance: + score = 0.5 * (1.0 - excess / protocol_tolerance) + else: + score = 0.0 + + return FrequencyMatch( + protocol_name="", # Set by caller + frequency_error=freq_error, + frequency_error_pct=freq_error_pct, + within_tolerance=within_tolerance, + score=score, + details={ + 'signal_freq_mhz': f"{signal_frequency / 1_000_000:.3f}", + 'protocol_freq_mhz': f"{protocol_frequency / 1_000_000:.3f}", + 'error_khz': f"{freq_error / 1_000:.1f}", + 'tolerance_khz': f"{protocol_tolerance / 1_000:.1f}" + } + ) + + def group_protocols_by_band( + self, + protocols: List + ) -> Dict[str, List]: + """ + Group protocols by ISM frequency band + + Args: + protocols: List of ProtocolSignature objects + + Returns: + Dict mapping band name to protocols + """ + grouped = defaultdict(list) + + for protocol in protocols: + band = self._detect_ism_band(protocol.frequency) + if band: + grouped[band].append(protocol) + else: + grouped['other'].append(protocol) + + return dict(grouped) + + def get_band_statistics( + self, + protocols: List + ) -> Dict[str, int]: + """ + Get protocol count statistics by band + + Args: + protocols: List of ProtocolSignature objects + + Returns: + Dict mapping band name to protocol count + """ + grouped = self.group_protocols_by_band(protocols) + return {band: len(protos) for band, protos in grouped.items()} + + # === Helper Methods === + + def _detect_ism_band(self, frequency: int) -> Optional[str]: + """ + Detect which ISM band a frequency belongs to + + Args: + frequency: Frequency in Hz + + Returns: + Band name ('315MHz', '433MHz', etc.) or None + """ + for band_name, (low, high) in self.ism_bands.items(): + if low <= frequency <= high: + return band_name + + return None + + def _get_band_center(self, band_name: str) -> int: + """Get center frequency of an ISM band""" + if band_name not in self.ism_bands: + return 0 + + low, high = self.ism_bands[band_name] + return (low + high) // 2 + + +# === Integration with Protocol Database === + +class FrequencyFilterStrategy: + """ + Matching strategy that pre-filters by frequency + + Reduces search space before expensive timing analysis + """ + + def __init__(self, protocol_db): + """ + Initialize strategy + + Args: + protocol_db: Protocol database instance + """ + self.protocol_db = protocol_db + self.fingerprinter = FrequencyFingerprinter() + + def filter_candidates( + self, + signal_frequency: int, + max_candidates: int = 50 + ) -> List: + """ + Get candidate protocols filtered by frequency + + Args: + signal_frequency: Signal frequency in Hz + max_candidates: Maximum protocols to return + + Returns: + List of candidate protocols + """ + # Get all protocols + all_protocols = self.protocol_db.get_all() + + # Filter by frequency (±200 kHz default tolerance) + candidates = self.fingerprinter.filter_protocols_by_frequency( + all_protocols, + signal_frequency, + tolerance_hz=200_000 + ) + + # If too many, narrow further + if len(candidates) > max_candidates: + # Sort by frequency error, keep closest + candidates.sort( + key=lambda p: abs(p.frequency - signal_frequency) + ) + candidates = candidates[:max_candidates] + + return candidates + + def get_frequency_score( + self, + signal_frequency: int, + protocol_frequency: int, + protocol_tolerance: int + ) -> float: + """ + Get frequency match score + + Args: + signal_frequency: Signal frequency in Hz + protocol_frequency: Protocol frequency in Hz + protocol_tolerance: Protocol tolerance in Hz + + Returns: + Score between 0.0 and 1.0 + """ + match = self.fingerprinter.score_frequency_match( + signal_frequency, + protocol_frequency, + protocol_tolerance + ) + return match.score + + +# Singleton instance +_fingerprinter: Optional[FrequencyFingerprinter] = None + + +def get_frequency_fingerprinter() -> FrequencyFingerprinter: + """Get singleton frequency fingerprinter instance""" + global _fingerprinter + if _fingerprinter is None: + _fingerprinter = FrequencyFingerprinter() + return _fingerprinter diff --git a/src/matcher/pattern_decoder.py b/src/matcher/pattern_decoder.py index fae6639..5ac4c1a 100644 --- a/src/matcher/pattern_decoder.py +++ b/src/matcher/pattern_decoder.py @@ -21,6 +21,8 @@ from src.matcher.protocol_database import ( get_protocol_database ) from src.matcher.timing_analyzer import get_timing_analyzer +from src.matcher.preamble_detector import get_preamble_detector +from src.matcher.frequency_fingerprint import get_frequency_fingerprinter @dataclass @@ -84,6 +86,8 @@ 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() + self.preamble_detector = get_preamble_detector() + self.frequency_fingerprinter = get_frequency_fingerprinter() def decode(self, metadata: SignalMetadata) -> List[DeviceMatch]: """ @@ -204,53 +208,78 @@ class PatternDecoder: # Decode to binary pattern (using HIGH pulses only) bit_pattern = self._decode_to_bits(pulses, short_pulse, long_pulse) - # Find protocols matching these timings - protocol_matches = self.protocol_db.find_by_timing( - short_pulse, - long_pulse, - frequency + # Detect preamble + detected_preamble = self.preamble_detector.detect(pulses, short_pulse, long_pulse) + + # Pre-filter protocols by frequency (reduces search space) + frequency_filtered = self.frequency_fingerprinter.filter_protocols_by_frequency( + self.protocol_db.get_all(), + frequency, + tolerance_hz=200_000 # ±200 kHz ) - for proto in protocol_matches: - # Calculate confidence based on: - # 1. Timing accuracy - # 2. Bit count match - # 3. Pattern match (if preamble/sync defined) + # Further filter by timing match + protocol_matches = [ + p for p in frequency_filtered + if p.matches_timing(short_pulse, long_pulse) + ] + for proto in protocol_matches: + # === Multi-Factor Scoring === + # Score = Timing(30%) + Frequency(25%) + BitCount(20%) + Preamble(15%) + Stats(10%) + + # 1. Timing accuracy (30%) timing_error = abs(proto.short_pulse_us - short_pulse) / proto.short_pulse_us timing_confidence = max(0, 1.0 - timing_error) + # 2. Frequency match (25%) + freq_match = self.frequency_fingerprinter.score_frequency_match( + frequency, + proto.frequency, + proto.frequency_tolerance + ) + frequency_confidence = freq_match.score + + # 3. Bit count match (20%) bit_count = len(bit_pattern) bit_count_match = (proto.min_bits <= bit_count <= proto.max_bits) bit_confidence = 1.0 if bit_count_match else 0.5 - # Check preamble/sync patterns - pattern_confidence = 1.0 - if proto.preamble_pattern and proto.preamble_pattern in bit_pattern: - pattern_confidence = 1.0 - elif proto.sync_pattern and proto.sync_pattern in bit_pattern: - pattern_confidence = 0.9 - else: - pattern_confidence = 0.7 + # 4. Preamble match (15%) + preamble_match = self.preamble_detector.match_against_protocol( + detected_preamble, + proto.preamble_pattern, + proto.sync_pattern + ) + preamble_confidence = preamble_match.similarity + + # 5. Statistical fingerprint (10%) - duty cycle, pulse count + stats_confidence = 0.8 # Default - could add pulse count matching # Overall confidence (weighted average) overall_confidence = ( - timing_confidence * 0.4 + - bit_confidence * 0.3 + - pattern_confidence * 0.3 + timing_confidence * 0.30 + + frequency_confidence * 0.25 + + bit_confidence * 0.20 + + preamble_confidence * 0.15 + + stats_confidence * 0.10 ) if overall_confidence >= proto.min_confidence: matches.append(DeviceMatch( protocol=proto, confidence=overall_confidence, - match_method='timing_pattern', + match_method='multi_factor', details={ 'short_pulse_us': short_pulse, 'long_pulse_us': long_pulse, 'bit_count': bit_count, - 'bit_pattern': bit_pattern[:64], # Truncate for readability - 'timing_error': f"{timing_error:.2%}", + 'bit_pattern': bit_pattern[:64], + 'timing_score': f"{timing_confidence:.2%}", + 'frequency_score': f"{frequency_confidence:.2%}", + 'preamble_score': f"{preamble_confidence:.2%}", + 'bit_count_score': f"{bit_confidence:.2%}", + 'preamble_type': detected_preamble.type if detected_preamble else 'none', } )) diff --git a/src/matcher/preamble_detector.py b/src/matcher/preamble_detector.py new file mode 100644 index 0000000..93a5e1d --- /dev/null +++ b/src/matcher/preamble_detector.py @@ -0,0 +1,448 @@ +#!/usr/bin/env python3 +""" +Preamble Detection for RF Device Identification + +Detects and analyzes preamble patterns in RF signals to improve device identification. +Preambles are repetitive patterns at the start of transmissions used for synchronization. + +Common preamble types: + - Alternating (01010101...) - Manchester, Oregon Scientific + - Sync word (fixed pattern) - Acurite, LaCrosse + - Long burst (1111...) - Princeton, PT2262 + - Custom patterns - Protocol-specific +""" + +import numpy as np +from dataclasses import dataclass +from typing import List, Optional, Dict, Tuple +from collections import Counter + + +@dataclass +class PreamblePattern: + """ + Detected preamble pattern in signal + + Attributes: + type: Pattern type ('alternating', 'sync_word', 'long_burst', 'custom') + pattern: Binary pattern as string (e.g., "10101010") + length: Length in pulses + repetitions: Number of times pattern repeats + confidence: Detection confidence (0.0-1.0) + details: Additional pattern characteristics + """ + type: str + pattern: str + length: int + repetitions: int + confidence: float + details: Dict + + +@dataclass +class PreambleMatch: + """ + Match result comparing detected preamble to protocol + + Attributes: + protocol_name: Name of matched protocol + match_type: 'exact', 'partial', 'similar', 'none' + similarity: Similarity score (0.0-1.0) + details: Match details + """ + protocol_name: str + match_type: str + similarity: float + details: Dict + + +class PreambleDetector: + """ + Detect and analyze preamble patterns in RF signals + + Uses multiple detection strategies: + 1. Autocorrelation for periodic patterns + 2. Long burst detection + 3. Known sync word matching + 4. Pattern repetition analysis + """ + + def __init__(self, preamble_window: int = 50): + """ + Initialize detector + + Args: + preamble_window: Number of pulses to analyze for preamble (default: 50) + """ + self.preamble_window = preamble_window + + # Known sync patterns from common protocols + self.known_sync_patterns = { + '10': ['Acurite Tower', 'Princeton'], + '1000': ['Oregon Scientific v2.1', 'Oregon Scientific v3.0'], + '1111': ['PT2262', 'PT2260'], + '0110': ['LaCrosse TX141'], + '101010': ['Manchester encoding'], + } + + def detect(self, pulses: List[int], timing_short: int, timing_long: int) -> Optional[PreamblePattern]: + """ + Detect preamble pattern in signal + + Args: + pulses: List of pulse durations (positive=HIGH, negative=LOW) + timing_short: Short pulse duration (microseconds) + timing_long: Long pulse duration (microseconds) + + Returns: + PreamblePattern if detected, None otherwise + """ + if not pulses or len(pulses) < 10: + return None + + # Analyze first N pulses + preamble_pulses = pulses[:self.preamble_window] + + # Try detection methods in order of specificity + methods = [ + self._detect_long_burst, + self._detect_alternating, + self._detect_sync_word, + self._detect_repetition, + ] + + for method in methods: + try: + pattern = method(preamble_pulses, timing_short, timing_long) + if pattern and pattern.confidence > 0.5: + return pattern + except Exception: + continue + + return None + + def match_against_protocol( + self, + detected: Optional[PreamblePattern], + protocol_preamble: Optional[str], + protocol_sync: Optional[str] + ) -> PreambleMatch: + """ + Match detected preamble against protocol signature + + Args: + detected: Detected preamble pattern + protocol_preamble: Expected preamble pattern from protocol + protocol_sync: Expected sync word from protocol + + Returns: + PreambleMatch with similarity score + """ + if not detected: + return PreambleMatch( + protocol_name="", + match_type='none', + similarity=0.0, + details={'reason': 'no_preamble_detected'} + ) + + if not protocol_preamble and not protocol_sync: + # Protocol has no defined preamble - neutral score + return PreambleMatch( + protocol_name="", + match_type='none', + similarity=0.5, # Neutral - doesn't help or hurt + details={'reason': 'protocol_no_preamble'} + ) + + # Check exact match with protocol preamble + if protocol_preamble: + # Clean protocol preamble (handle Python expressions like "10" * 4) + clean_preamble = protocol_preamble + + # Try to evaluate as Python expression for patterns like '"10" * 4' + if '*' in clean_preamble: + try: + clean_preamble = str(eval(clean_preamble)) + except Exception: + # Fallback: remove quotes and spaces + clean_preamble = clean_preamble.replace('"', '').replace("'", '').replace(' * ', '').replace('*', '') + else: + # Just remove quotes + clean_preamble = clean_preamble.replace('"', '').replace("'", '') + + if clean_preamble in detected.pattern: + return PreambleMatch( + protocol_name="", + match_type='exact', + similarity=1.0, + details={'matched': clean_preamble, 'in_pattern': detected.pattern} + ) + + # Check sync word match + if protocol_sync: + clean_sync = protocol_sync.replace('"', '').replace("'", '') + + if clean_sync in detected.pattern: + return PreambleMatch( + protocol_name="", + match_type='exact', + similarity=0.9, + details={'matched_sync': clean_sync} + ) + + # Check partial match (pattern type similarity) + if protocol_preamble: + # Alternating pattern check + if ('10' in protocol_preamble or '01' in protocol_preamble) and detected.type == 'alternating': + return PreambleMatch( + protocol_name="", + match_type='partial', + similarity=0.7, + details={'reason': 'both_alternating'} + ) + + # Long burst check + if '1111' in protocol_preamble and detected.type == 'long_burst': + return PreambleMatch( + protocol_name="", + match_type='partial', + similarity=0.7, + details={'reason': 'both_long_burst'} + ) + + # No match + return PreambleMatch( + protocol_name="", + match_type='none', + similarity=0.3, # Penalty for mismatch + details={'reason': 'no_match'} + ) + + # === Detection Methods === + + def _detect_long_burst( + self, + pulses: List[int], + timing_short: int, + timing_long: int + ) -> Optional[PreamblePattern]: + """ + Detect long HIGH burst at start (e.g., Princeton: 1111...) + + Args: + pulses: Pulse data + timing_short: Short pulse duration + timing_long: Long pulse duration + + Returns: + PreamblePattern if detected + """ + if not pulses or pulses[0] <= 0: + return None # Must start with HIGH + + first_pulse = pulses[0] + + # Long burst = first pulse significantly longer than expected long pulse + # OR first pulse is much longer than remaining pulses + if first_pulse > timing_long * 2.5: + # Count how many "long" units this represents + repetitions = int(first_pulse / timing_long) if timing_long > 0 else 1 + + return PreamblePattern( + type='long_burst', + pattern='1' * repetitions, + length=1, + repetitions=repetitions, + confidence=0.8, + details={ + 'duration_us': first_pulse, + 'expected_long': timing_long, + 'ratio': first_pulse / timing_long if timing_long > 0 else 0 + } + ) + + return None + + def _detect_alternating( + self, + pulses: List[int], + timing_short: int, + timing_long: int + ) -> Optional[PreamblePattern]: + """ + Detect alternating pattern (01010101...) via autocorrelation + + Common in Manchester encoding, Oregon Scientific + + Args: + pulses: Pulse data + timing_short: Short pulse duration + timing_long: Long pulse duration + + Returns: + PreamblePattern if detected + """ + if len(pulses) < 20: + return None + + # Decode first 30 pulses to binary + threshold = (timing_short + timing_long) / 2 + bits = [] + for p in pulses[:30]: + if p > 0: # Only HIGH pulses + bits.append('1' if abs(p) > threshold else '0') + + if len(bits) < 10: + return None + + bit_string = ''.join(bits) + + # Check for alternating pattern (at least 8 alternations) + alternating_patterns = ['10101010', '01010101'] + for pattern in alternating_patterns: + if pattern in bit_string[:16]: # Check first 16 bits + # Count repetitions + reps = 0 + for i in range(len(bit_string) - 1): + if bit_string[i:i+2] in ['10', '01']: + reps += 1 + else: + break + + if reps >= 4: + return PreamblePattern( + type='alternating', + pattern=pattern * (reps // len(pattern)), + length=reps * 2, + repetitions=reps, + confidence=0.9, + details={ + 'alternations': reps, + 'detected_pattern': pattern + } + ) + + return None + + def _detect_sync_word( + self, + pulses: List[int], + timing_short: int, + timing_long: int + ) -> Optional[PreamblePattern]: + """ + Detect known sync words from protocol library + + Args: + pulses: Pulse data + timing_short: Short pulse duration + timing_long: Long pulse duration + + Returns: + PreamblePattern if detected + """ + if len(pulses) < 10: + return None + + # Decode first 20 pulses to binary + threshold = (timing_short + timing_long) / 2 + bits = [] + for p in pulses[:20]: + if p > 0: + bits.append('1' if abs(p) > threshold else '0') + + bit_string = ''.join(bits) + + # Check against known sync patterns (longest first to avoid substring matches) + sorted_patterns = sorted(self.known_sync_patterns.items(), key=lambda x: len(x[0]), reverse=True) + for pattern, protocols in sorted_patterns: + if pattern in bit_string: + return PreamblePattern( + type='sync_word', + pattern=pattern, + length=len(pattern), + repetitions=1, + confidence=0.85, + details={ + 'sync_word': pattern, + 'possible_protocols': protocols, + 'position': bit_string.index(pattern) + } + ) + + return None + + def _detect_repetition( + self, + pulses: List[int], + timing_short: int, + timing_long: int + ) -> Optional[PreamblePattern]: + """ + Detect repetitive patterns via autocorrelation + + Args: + pulses: Pulse data + timing_short: Short pulse duration + timing_long: Long pulse duration + + Returns: + PreamblePattern if detected + """ + if len(pulses) < 20: + return None + + # Decode to binary + threshold = (timing_short + timing_long) / 2 + bits = [] + for p in pulses[:40]: + if p > 0: + bits.append('1' if abs(p) > threshold else '0') + + if len(bits) < 12: + return None + + # Look for repeating 2-8 bit patterns + for pattern_len in range(2, 9): + if len(bits) < pattern_len * 3: + continue + + pattern = bits[:pattern_len] + pattern_str = ''.join(pattern) + + # Count repetitions + reps = 1 + for i in range(pattern_len, len(bits) - pattern_len + 1, pattern_len): + chunk = bits[i:i+pattern_len] + if chunk == pattern: + reps += 1 + else: + break + + if reps >= 3: # At least 3 repetitions + return PreamblePattern( + type='custom', + pattern=pattern_str * reps, + length=pattern_len * reps, + repetitions=reps, + confidence=0.75, + details={ + 'base_pattern': pattern_str, + 'pattern_length': pattern_len + } + ) + + return None + + +# Singleton instance +_detector: Optional[PreambleDetector] = None + + +def get_preamble_detector() -> PreambleDetector: + """Get singleton preamble detector instance""" + global _detector + if _detector is None: + _detector = PreambleDetector() + return _detector diff --git a/tests/unit/test_preamble_frequency.py b/tests/unit/test_preamble_frequency.py new file mode 100644 index 0000000..01befe7 --- /dev/null +++ b/tests/unit/test_preamble_frequency.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +""" +Unit Tests for Preamble Detection and Frequency Fingerprinting +""" + +import pytest +from src.matcher.preamble_detector import PreambleDetector, PreamblePattern +from src.matcher.frequency_fingerprint import FrequencyFingerprinter + + +class TestPreambleDetection: + """Test preamble detection algorithms""" + + def test_long_burst_detection(self): + """Test 1: Detect long HIGH burst preamble (Princeton style)""" + detector = PreambleDetector() + + # Long burst: 4000us followed by normal pulses (need at least 10 pulses) + pulses = [4000, -500, 500, -500, 1000, -500, 500, -500, 1000, -500, 500, -500] + + pattern = detector.detect(pulses, timing_short=500, timing_long=1000) + + assert pattern is not None, "Should detect long burst" + assert pattern.type == 'long_burst', f"Expected long_burst, got {pattern.type}" + assert pattern.confidence > 0.7, f"Expected confidence >0.7, got {pattern.confidence}" + + def test_alternating_detection(self): + """Test 2: Detect alternating 01010101 pattern (Manchester/Oregon)""" + detector = PreambleDetector() + + # Alternating pattern: 500us/1000us pulses + # Pattern: 01010101 (8 alternations) + 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 + ] * 2 # Repeat for emphasis + + pattern = detector.detect(pulses, timing_short=500, timing_long=1000) + + assert pattern is not None, "Should detect alternating pattern" + assert pattern.type == 'alternating', f"Expected alternating, got {pattern.type}" + assert pattern.repetitions >= 4, f"Expected >=4 repetitions, got {pattern.repetitions}" + + def test_sync_word_detection(self): + """Test 3: Detect known sync word (e.g., '1000' for Oregon Scientific)""" + detector = PreambleDetector() + + # Sync word "1000" (need at least 10 pulses) + pulses = [ + 1000, -500, # 1 + 500, -500, # 0 + 500, -500, # 0 + 500, -500, # 0 + # Followed by data (add more to meet 10 pulse minimum) + 1000, -500, + 500, -500, + 1000, -500, + 500, -500, + ] + + pattern = detector.detect(pulses, timing_short=500, timing_long=1000) + + assert pattern is not None, "Should detect sync word" + assert pattern.type == 'sync_word', f"Expected sync_word, got {pattern.type}" + assert '1000' in pattern.pattern, "Pattern should contain '1000'" + + def test_no_preamble(self): + """Test 4: Handle signals without clear preamble""" + detector = PreambleDetector() + + # Random-looking pattern + pulses = [600, -400, 800, -600, 550, -450, 900, -500] + + pattern = detector.detect(pulses, timing_short=500, timing_long=1000) + + # May or may not detect - should not crash + assert True, "Should handle no preamble gracefully" + + def test_preamble_matching_exact(self): + """Test 5: Exact preamble match scoring""" + detector = PreambleDetector() + + # Detected alternating pattern + detected = PreamblePattern( + type='alternating', + pattern='10101010', + length=8, + repetitions=4, + confidence=0.9, + details={} + ) + + # Protocol expects alternating + match = detector.match_against_protocol( + detected, + protocol_preamble='"10" * 4', # "10101010" + protocol_sync=None + ) + + assert match.match_type == 'exact', f"Expected exact match, got {match.match_type}" + assert match.similarity >= 0.9, f"Expected similarity >=0.9, got {match.similarity}" + + def test_preamble_matching_partial(self): + """Test 6: Partial preamble match (type similarity)""" + detector = PreambleDetector() + + detected = PreamblePattern( + type='alternating', + pattern='0101', + length=4, + repetitions=2, + confidence=0.8, + details={} + ) + + # Protocol also uses alternating (but different exact pattern) + match = detector.match_against_protocol( + detected, + protocol_preamble='"01" * 8', + protocol_sync=None + ) + + # Should match type even if exact pattern differs + assert match.similarity >= 0.5, f"Expected partial match, got {match.similarity}" + + def test_preamble_mismatch(self): + """Test 7: Preamble mismatch penalty""" + detector = PreambleDetector() + + detected = PreamblePattern( + type='long_burst', + pattern='1111', + length=4, + repetitions=1, + confidence=0.8, + details={} + ) + + # Protocol expects alternating (mismatch) + match = detector.match_against_protocol( + detected, + protocol_preamble='"10" * 4', + protocol_sync=None + ) + + assert match.similarity < 0.5, f"Expected low similarity for mismatch, got {match.similarity}" + + +class TestFrequencyFingerprinting: + """Test frequency-based filtering and scoring""" + + def test_ism_band_detection_433mhz(self): + """Test 8: Detect 433 MHz ISM band""" + fingerprinter = FrequencyFingerprinter() + + freq = 433920000 # 433.92 MHz + + fingerprint = fingerprinter.analyze(freq) + + assert fingerprint.ism_band == '433MHz', f"Expected 433MHz band, got {fingerprint.ism_band}" + assert fingerprint.center_freq == freq + assert fingerprint.confidence > 0.8 + + def test_ism_band_detection_315mhz(self): + """Test 9: Detect 315 MHz ISM band""" + fingerprinter = FrequencyFingerprinter() + + freq = 315000000 # 315 MHz + + fingerprint = fingerprinter.analyze(freq) + + assert fingerprint.ism_band == '315MHz', f"Expected 315MHz band, got {fingerprint.ism_band}" + + def test_frequency_exact_match(self): + """Test 10: Perfect frequency match scoring""" + fingerprinter = FrequencyFingerprinter() + + match = fingerprinter.score_frequency_match( + signal_frequency=433920000, + protocol_frequency=433920000, + protocol_tolerance=100000 + ) + + assert match.score == 1.0, f"Expected score 1.0 for exact match, got {match.score}" + assert match.within_tolerance is True + assert match.frequency_error == 0 + + def test_frequency_within_tolerance(self): + """Test 11: Frequency match within tolerance""" + fingerprinter = FrequencyFingerprinter() + + # 50 kHz offset (within ±100 kHz tolerance) + match = fingerprinter.score_frequency_match( + signal_frequency=433970000, + protocol_frequency=433920000, + protocol_tolerance=100000 + ) + + assert match.within_tolerance is True + assert match.score >= 0.8, f"Expected score >=0.8 within tolerance, got {match.score}" + + def test_frequency_outside_tolerance(self): + """Test 12: Frequency mismatch outside tolerance""" + fingerprinter = FrequencyFingerprinter() + + # 500 kHz offset (outside ±100 kHz tolerance) + match = fingerprinter.score_frequency_match( + signal_frequency=434420000, + protocol_frequency=433920000, + protocol_tolerance=100000 + ) + + assert match.within_tolerance is False + assert match.score < 0.6, f"Expected low score outside tolerance, got {match.score}" + + def test_frequency_filtering(self): + """Test 13: Filter protocols by frequency""" + fingerprinter = FrequencyFingerprinter() + + # Mock protocols + class MockProtocol: + def __init__(self, name, freq): + self.name = name + self.frequency = freq + self.frequency_tolerance = 100000 + + protocols = [ + MockProtocol('433MHz Device', 433920000), + MockProtocol('315MHz Device', 315000000), + MockProtocol('868MHz Device', 868000000), + MockProtocol('433MHz Device 2', 433850000), # Close to 433.92 + ] + + # Filter for 433 MHz signal + filtered = fingerprinter.filter_protocols_by_frequency( + protocols, + signal_frequency=433920000, + tolerance_hz=200000 + ) + + # Should get both 433 MHz devices + assert len(filtered) == 2, f"Expected 2 protocols, got {len(filtered)}" + assert all('433MHz' in p.name for p in filtered) + + def test_protocol_grouping_by_band(self): + """Test 14: Group protocols by ISM band""" + fingerprinter = FrequencyFingerprinter() + + class MockProtocol: + def __init__(self, name, freq): + self.name = name + self.frequency = freq + + protocols = [ + MockProtocol('Dev1', 433920000), + MockProtocol('Dev2', 315000000), + MockProtocol('Dev3', 433850000), + MockProtocol('Dev4', 868000000), + ] + + grouped = fingerprinter.group_protocols_by_band(protocols) + + assert '433MHz' in grouped + assert '315MHz' in grouped + assert '868MHz' in grouped + assert len(grouped['433MHz']) == 2, "Should have 2 protocols in 433MHz band" + + def test_band_statistics(self): + """Test 15: Get protocol count per band""" + fingerprinter = FrequencyFingerprinter() + + class MockProtocol: + def __init__(self, freq): + self.frequency = freq + + protocols = [ + MockProtocol(433920000), + MockProtocol(433850000), + MockProtocol(315000000), + ] + + stats = fingerprinter.get_band_statistics(protocols) + + assert stats.get('433MHz', 0) == 2 + assert stats.get('315MHz', 0) == 1 + + +if __name__ == '__main__': + pytest.main([__file__, '-v'])