#!/usr/bin/env python3 """ Pattern-Based Decoder for RF Signals Decodes devices from single-transmission captures using timing patterns, statistical fingerprints, and protocol library matching. Designed specifically for short captures from Flipper Zero and LilyGo T-Embed that don't have enough repetitions for RTL_433. """ import numpy as np from dataclasses import dataclass from typing import List, Optional, Tuple, Dict from collections import Counter from src.parser.sub_parser import SignalMetadata from src.matcher.protocol_database import ( ProtocolSignature, ProtocolDatabase, 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 class PulseStatistics: """Statistical characteristics of pulse data""" mean_pulse_width: float std_pulse_width: float mean_gap_width: float std_gap_width: float pulse_gap_ratio: float duty_cycle: float pulse_count: int min_pulse: int max_pulse: int min_gap: int max_gap: int @dataclass class TimingPattern: """Detected timing pattern in signal""" short_pulse_us: int long_pulse_us: int short_gap_us: int long_gap_us: int bit_pattern: str # Binary string decoded from pulses confidence: float @dataclass class DeviceMatch: """Potential device match from pattern decoder""" protocol: ProtocolSignature confidence: float match_method: str # 'timing', 'fingerprint', 'protocol' details: Dict @property def name(self) -> str: return self.protocol.name @property def manufacturer(self) -> Optional[str]: return self.protocol.manufacturer @property def category(self) -> str: return self.protocol.category class PatternDecoder: """ Multi-strategy decoder for single-transmission RF captures Uses three complementary strategies: 1. Timing Pattern Analysis - Identify SHORT/LONG pulses and decode bits 2. Statistical Fingerprinting - Match signal characteristics 3. Protocol Library Matching - Compare against known protocol signatures """ 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]: """ Decode device from signal metadata using multi-strategy approach Args: metadata: Parsed .sub file metadata with RAW_Data Returns: List of DeviceMatch sorted by confidence (highest first) """ if not metadata.has_raw_data: return [] pulses = metadata.raw_data frequency = metadata.frequency matches = [] # Strategy 1: Timing Pattern Analysis timing_matches = self._decode_timing_patterns(pulses, frequency) matches.extend(timing_matches) # Strategy 2: Statistical Fingerprinting fingerprint = self._extract_fingerprint(pulses) fingerprint_matches = self._match_fingerprint(fingerprint, frequency) matches.extend(fingerprint_matches) # Strategy 3: Protocol Library Matching (combines timing + known protocols) # Already integrated into timing_matches above # Deduplicate and rank by confidence return self._rank_matches(matches) def _identify_pulse_widths(self, pulses: List[int]) -> Tuple[int, int, int, int]: """ Identify SHORT/LONG pulse and gap durations using robust timing analyzer Returns: (short_pulse, long_pulse, short_gap, long_gap) in microseconds """ if not pulses: return (0, 0, 0, 0) # Use robust timing analyzer for improved extraction timing = self.timing_analyzer.extract_timing(pulses) 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]: """ Cluster durations into two groups (SHORT and LONG) using K-means Falls back to percentile-based approach if K-means unavailable """ if not durations: return (0, 0) if len(durations) < 2: val = durations[0] return (val, val) # Try K-means clustering (sklearn) try: from sklearn.cluster import KMeans # Convert to 2D array for sklearn X = np.array(durations).reshape(-1, 1) # Cluster into 2 groups kmeans = KMeans(n_clusters=2, random_state=42, n_init=10) kmeans.fit(X) # Get cluster centers centers = sorted(kmeans.cluster_centers_.flatten()) return (int(centers[0]), int(centers[1])) except ImportError: # Fallback: Use percentile-based approach sorted_durations = sorted(durations) median_idx = len(sorted_durations) // 2 # Split at median lower_half = sorted_durations[:median_idx] upper_half = sorted_durations[median_idx:] short = int(np.mean(lower_half)) if lower_half else sorted_durations[0] long = int(np.mean(upper_half)) if upper_half else sorted_durations[-1] return (short, long) def _decode_timing_patterns( self, pulses: List[int], frequency: int ) -> List[DeviceMatch]: """ Decode signal using timing pattern analysis Steps: 1. Identify SHORT/LONG pulse widths 2. Decode binary pattern (SHORT=0, LONG=1) 3. Match against protocol database """ matches = [] # Identify pulse widths short_pulse, long_pulse, short_gap, long_gap = self._identify_pulse_widths(pulses) if short_pulse == 0 or long_pulse == 0: return [] # Decode to binary pattern (using HIGH pulses only) bit_pattern = self._decode_to_bits(pulses, short_pulse, long_pulse) # 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 ) # 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 (Tuned based on benchmarks) === # ORIGINAL: Timing(30%) + Frequency(25%) + BitCount(20%) + Preamble(15%) + Stats(10%) # TUNED: Timing(35%) + Preamble(25%) + BitCount(20%) + Frequency(15%) + Stats(5%) # Rationale: Preamble detection is highly discriminative, frequency less so (many protocols per band) # 1. Timing accuracy (35% - increased from 30%) timing_error = abs(proto.short_pulse_us - short_pulse) / proto.short_pulse_us timing_confidence = max(0, 1.0 - timing_error) # 2. Preamble match (25% - increased from 15%) preamble_match = self.preamble_detector.match_against_protocol( detected_preamble, proto.preamble_pattern, proto.sync_pattern ) preamble_confidence = preamble_match.similarity # 3. Bit count match (20% - unchanged) 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 # 4. Frequency match (15% - decreased from 25%) freq_match = self.frequency_fingerprinter.score_frequency_match( frequency, proto.frequency, proto.frequency_tolerance ) frequency_confidence = freq_match.score # 5. Statistical fingerprint (5% - decreased from 10%) stats_confidence = 0.8 # Default - could add pulse count matching # Overall confidence (weighted average) overall_confidence = ( timing_confidence * 0.35 + preamble_confidence * 0.25 + bit_confidence * 0.20 + frequency_confidence * 0.15 + stats_confidence * 0.05 ) # Confidence level classification if overall_confidence >= 0.8: confidence_level = 'high' elif overall_confidence >= 0.5: confidence_level = 'medium' else: confidence_level = 'low' if overall_confidence >= proto.min_confidence: matches.append(DeviceMatch( protocol=proto, confidence=overall_confidence, match_method='multi_factor', details={ 'short_pulse_us': short_pulse, 'long_pulse_us': long_pulse, 'bit_count': bit_count, 'bit_pattern': bit_pattern[:64], 'timing_score': f"{timing_confidence:.2%}", 'preamble_score': f"{preamble_confidence:.2%}", 'bit_count_score': f"{bit_confidence:.2%}", 'frequency_score': f"{frequency_confidence:.2%}", 'stats_score': f"{stats_confidence:.2%}", 'confidence_level': confidence_level, 'preamble_type': detected_preamble.type if detected_preamble else 'none', 'scoring_weights': 'T:35% P:25% B:20% F:15% S:5%', } )) return matches def _decode_to_bits( self, pulses: List[int], short_pulse: int, long_pulse: int ) -> str: """ Decode pulse train to binary string Uses PWM encoding: SHORT=0, LONG=1 """ bits = [] threshold = (short_pulse + long_pulse) / 2 for pulse in pulses: if pulse > 0: # Only decode HIGH pulses duration = abs(pulse) if duration < threshold: bits.append('0') else: bits.append('1') return ''.join(bits) def _extract_fingerprint(self, pulses: List[int]) -> PulseStatistics: """ Extract statistical fingerprint from pulse data Returns metrics that characterize the signal's timing properties """ if not pulses: return PulseStatistics( mean_pulse_width=0, std_pulse_width=0, mean_gap_width=0, std_gap_width=0, pulse_gap_ratio=0, duty_cycle=0, pulse_count=0, min_pulse=0, max_pulse=0, min_gap=0, max_gap=0 ) # Separate HIGH pulses and LOW gaps high_pulses = [p for p in pulses if p > 0] low_pulses = [abs(p) for p in pulses if p < 0] # Calculate statistics mean_pulse = np.mean(high_pulses) if high_pulses else 0 std_pulse = np.std(high_pulses) if high_pulses else 0 mean_gap = np.mean(low_pulses) if low_pulses else 0 std_gap = np.std(low_pulses) if low_pulses else 0 total_high = sum(high_pulses) total_low = sum(low_pulses) total_time = total_high + total_low pulse_gap_ratio = mean_pulse / mean_gap if mean_gap > 0 else 0 duty_cycle = total_high / total_time if total_time > 0 else 0 return PulseStatistics( mean_pulse_width=float(mean_pulse), std_pulse_width=float(std_pulse), mean_gap_width=float(mean_gap), std_gap_width=float(std_gap), pulse_gap_ratio=float(pulse_gap_ratio), duty_cycle=float(duty_cycle), pulse_count=len(pulses), min_pulse=min(high_pulses) if high_pulses else 0, max_pulse=max(high_pulses) if high_pulses else 0, min_gap=min(low_pulses) if low_pulses else 0, max_gap=max(low_pulses) if low_pulses else 0, ) def _match_fingerprint( self, fingerprint: PulseStatistics, frequency: int ) -> List[DeviceMatch]: """ Match statistical fingerprint against protocol database Uses signal characteristics to find similar protocols """ matches = [] # Get protocols near this frequency candidates = self.protocol_db.find_by_frequency(frequency) for proto in candidates: # Compare fingerprint characteristics # Use typical timing to estimate expected characteristics # Expected mean pulse ~ (short + long) / 2 expected_mean_pulse = (proto.short_pulse_us + proto.long_pulse_us) / 2 pulse_error = abs(expected_mean_pulse - fingerprint.mean_pulse_width) / expected_mean_pulse # Expected pulse count expected_count = proto.typical_pulse_count count_error = abs(expected_count - fingerprint.pulse_count) / expected_count # Similarity score (lower error = higher confidence) pulse_similarity = max(0, 1.0 - pulse_error) count_similarity = max(0, 1.0 - count_error) overall_confidence = (pulse_similarity * 0.6 + count_similarity * 0.4) # Lower threshold for fingerprint matches (less precise) if overall_confidence >= 0.4: matches.append(DeviceMatch( protocol=proto, confidence=overall_confidence * 0.8, # Scale down (less confident) match_method='fingerprint', details={ 'mean_pulse_us': f"{fingerprint.mean_pulse_width:.1f}", 'pulse_count': fingerprint.pulse_count, 'duty_cycle': f"{fingerprint.duty_cycle:.2%}", 'pulse_error': f"{pulse_error:.2%}", 'count_error': f"{count_error:.2%}", } )) return matches def _rank_matches(self, matches: List[DeviceMatch]) -> List[DeviceMatch]: """ Deduplicate and rank matches by confidence If same protocol matched by multiple methods, keep highest confidence """ # Group by protocol name by_protocol: Dict[str, List[DeviceMatch]] = {} for match in matches: name = match.protocol.name if name not in by_protocol: by_protocol[name] = [] by_protocol[name].append(match) # Keep best match per protocol best_matches = [] for name, protocol_matches in by_protocol.items(): best = max(protocol_matches, key=lambda m: m.confidence) best_matches.append(best) # Sort by confidence (highest first) return sorted(best_matches, key=lambda m: m.confidence, reverse=True) def get_version(self) -> str: """Get decoder version""" return "PatternDecoder v1.0" # Global instance _decoder: Optional[PatternDecoder] = None def get_pattern_decoder() -> PatternDecoder: """Get singleton pattern decoder instance""" global _decoder if _decoder is None: _decoder = PatternDecoder() return _decoder if __name__ == '__main__': # Test pattern decoder with sample data from src.parser.sub_parser import parse_sub_file from pathlib import Path print("=== Pattern Decoder Test ===") print() # Try to load a test file test_file = Path("data/test_rtl433_real/LaCrosse-TX141TH-BV2-raw.sub") if test_file.exists(): print(f"Testing with: {test_file.name}") print() metadata = parse_sub_file(str(test_file)) decoder = get_pattern_decoder() print(f"Signal Details:") print(f" Frequency: {metadata.frequency / 1_000_000:.3f} MHz") print(f" Pulse Count: {metadata.pulse_count}") print() print("Running pattern decoder...") matches = decoder.decode(metadata) print(f"Found {len(matches)} potential matches:") print() for i, match in enumerate(matches[:5], 1): # Show top 5 print(f"Match #{i}:") print(f" Device: {match.name}") print(f" Manufacturer: {match.manufacturer or 'Unknown'}") print(f" Category: {match.category}") print(f" Confidence: {match.confidence:.2%}") print(f" Method: {match.match_method}") print(f" Details: {match.details}") print() else: print(f"❌ Test file not found: {test_file}") print(" Run download script first to get test data")