# Improved Pattern-Matching Algorithm Using FOSS Resources **Focus:** No ML/DL - Pure algorithmic improvements using open-source RF databases **Goal:** 3x accuracy improvement on unknown devices (5% → 40%) --- ## Core Strategy: Multi-Pass Heuristic Pipeline ``` ┌─────────────────────────────────────────────────────────────────┐ │ Stage 1: Signal Preprocessing & Feature Extraction │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ 1. Outlier Removal (IQR method) │ │ │ │ 2. Noise Floor Estimation │ │ │ │ 3. Pulse Normalization │ │ │ │ 4. Preamble Detection (autocorrelation) │ │ │ └───────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────┐ │ Stage 2: Protocol Database Filtering │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ • Frequency-based filtering (±100kHz) │ │ │ │ • Modulation type matching (OOK/FSK from preset) │ │ │ │ • Candidate pool: 286 → 20-30 protocols │ │ │ └───────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────┐ │ Stage 3: Timing Analysis (Multi-Method) │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ Method A: K-Means (k=2,3,4) + Silhouette scoring │ │ │ │ Method B: Histogram Peak Detection │ │ │ │ Method C: DBSCAN (density-based clustering) │ │ │ │ → Ensemble vote: Best method selected by confidence │ │ │ └───────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────┐ │ Stage 4: Pattern Matching & Scoring │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ Score = Σ weighted features: │ │ │ │ • Timing accuracy (30%) │ │ │ │ • Frequency match (25%) │ │ │ │ • Bit count match (20%) │ │ │ │ • Preamble pattern (15%) │ │ │ │ • Statistical fingerprint (10%) │ │ │ └───────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────────┐ │ Stage 5: Disambiguation & Ranking │ │ ┌───────────────────────────────────────────────────────────┐ │ │ │ • Geographic priors (common devices in region) │ │ │ │ • Category likelihood (residential vs industrial) │ │ │ │ • Time-of-day patterns (TPMS = daytime, security = 24/7) │ │ │ │ • Return top-3 matches with confidence intervals │ │ │ └───────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘ ``` --- ## Implementation Details ### Component 1: Robust Timing Extraction **File:** `src/matcher/timing_analyzer.py` (new) ```python import numpy as np from scipy import stats from sklearn.cluster import KMeans, DBSCAN from dataclasses import dataclass from typing import List, Tuple, Optional @dataclass class TimingSignature: """Extracted timing characteristics""" short_pulse_us: int long_pulse_us: int mid_pulse_us: Optional[int] # For tri-bit encoding short_gap_us: int long_gap_us: int num_levels: int # 2, 3, or 4 confidence: float # How clean the clustering is method: str # 'kmeans', 'histogram', 'dbscan' class RobustTimingAnalyzer: """Multi-method timing extraction with ensemble voting""" def __init__(self): self.methods = { 'kmeans': self._kmeans_method, 'histogram': self._histogram_method, 'dbscan': self._dbscan_method } def extract_timing(self, pulses: List[int]) -> TimingSignature: """ Try multiple methods, return best by confidence """ # Preprocess: remove outliers pulses_clean = self._remove_outliers(pulses) # Separate HIGH (positive) and LOW (negative) pulses high_pulses = [p for p in pulses_clean if p > 0] low_pulses = [abs(p) for p in pulses_clean if p < 0] # Try each method, score by quality results = [] for name, method_func in self.methods.items(): try: timing = method_func(high_pulses, low_pulses) timing.method = name results.append(timing) except Exception as e: continue # Return best result if results: return max(results, key=lambda t: t.confidence) else: # Fallback: simple mean return self._fallback_method(high_pulses, low_pulses) def _remove_outliers(self, pulses: List[int], method='iqr') -> List[int]: """ Remove statistical outliers (noise spikes, glitches) Methods: - IQR: Interquartile range (robust to outliers) - Z-score: Standard deviation based - MAD: Median Absolute Deviation """ if len(pulses) < 10: return pulses abs_pulses = [abs(p) for p in pulses] signs = [1 if p >= 0 else -1 for p in pulses] if method == 'iqr': q1, q3 = np.percentile(abs_pulses, [25, 75]) iqr = q3 - q1 lower = q1 - 1.5 * iqr upper = q3 + 1.5 * iqr cleaned = [ p for p in pulses if lower <= abs(p) <= upper ] return cleaned elif method == 'zscore': mean = np.mean(abs_pulses) std = np.std(abs_pulses) threshold = 3 # 3 standard deviations cleaned = [ p for p in pulses if abs(abs(p) - mean) <= threshold * std ] return cleaned return pulses def _kmeans_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature: """ K-means clustering for k=2,3,4 with silhouette scoring """ best_clustering = None best_score = -1 for k in [2, 3, 4]: if len(high_pulses) < k * 5: # Need enough samples continue X = np.array(high_pulses).reshape(-1, 1) kmeans = KMeans(n_clusters=k, random_state=42, n_init=10) labels = kmeans.fit_predict(X) # Silhouette score: how well-separated are clusters? from sklearn.metrics import silhouette_score score = silhouette_score(X, labels) if score > best_score: best_score = score centers = sorted(kmeans.cluster_centers_.flatten()) best_clustering = (k, centers, score) if not best_clustering: raise ValueError("K-means failed") k, centers, score = best_clustering # Extract timing based on number of levels if k == 2: short_pulse = int(centers[0]) long_pulse = int(centers[1]) mid_pulse = None elif k == 3: short_pulse = int(centers[0]) mid_pulse = int(centers[1]) long_pulse = int(centers[2]) else: # k == 4 # Rare: some protocols have 4 distinct widths short_pulse = int(centers[0]) mid_pulse = int(centers[1]) long_pulse = int(centers[2]) # Repeat for gaps if low_pulses: Y = np.array(low_pulses).reshape(-1, 1) kmeans_gaps = KMeans(n_clusters=min(2, len(set(low_pulses))), random_state=42, n_init=10) gap_centers = sorted(kmeans_gaps.fit_predict(Y)) short_gap = int(gap_centers[0]) if len(gap_centers) > 0 else 0 long_gap = int(gap_centers[1]) if len(gap_centers) > 1 else short_gap else: short_gap = 0 long_gap = 0 return TimingSignature( short_pulse_us=short_pulse, long_pulse_us=long_pulse, mid_pulse_us=mid_pulse, short_gap_us=short_gap, long_gap_us=long_gap, num_levels=k, confidence=score, # Silhouette score method='kmeans' ) def _histogram_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature: """ Histogram peak detection - find modes in distribution """ from scipy.signal import find_peaks # Create histogram hist, bin_edges = np.histogram(high_pulses, bins=50) # Find peaks (local maxima) peaks, properties = find_peaks(hist, height=len(high_pulses) * 0.05) # At least 5% of pulses if len(peaks) < 2: raise ValueError("Not enough peaks detected") # Get pulse widths at peak locations peak_widths = [bin_edges[p] for p in peaks] peak_widths_sorted = sorted(peak_widths) short_pulse = int(peak_widths_sorted[0]) long_pulse = int(peak_widths_sorted[1]) if len(peak_widths_sorted) > 1 else short_pulse * 2 mid_pulse = int(peak_widths_sorted[1]) if len(peak_widths_sorted) > 2 else None # Confidence: ratio of peak heights to noise peak_heights = [hist[p] for p in peaks] noise_level = np.median(hist) snr = np.mean(peak_heights) / (noise_level + 1) confidence = min(1.0, snr / 10) # Normalize to 0-1 # Gaps (simplified) short_gap = int(np.percentile(low_pulses, 25)) if low_pulses else 0 long_gap = int(np.percentile(low_pulses, 75)) if low_pulses else 0 return TimingSignature( short_pulse_us=short_pulse, long_pulse_us=long_pulse, mid_pulse_us=mid_pulse, short_gap_us=short_gap, long_gap_us=long_gap, num_levels=len(peak_widths_sorted), confidence=confidence, method='histogram' ) def _dbscan_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature: """ DBSCAN density-based clustering - handles noise naturally """ X = np.array(high_pulses).reshape(-1, 1) # DBSCAN parameters eps = np.std(high_pulses) * 0.3 # Cluster radius min_samples = max(3, len(high_pulses) // 20) # At least 5% of data dbscan = DBSCAN(eps=eps, min_samples=min_samples) labels = dbscan.fit_predict(X) # Extract cluster centers unique_labels = set(labels) if -1 in unique_labels: unique_labels.remove(-1) # Remove noise label if len(unique_labels) < 2: raise ValueError("DBSCAN found < 2 clusters") centers = [] for label in unique_labels: cluster_points = X[labels == label] center = np.mean(cluster_points) centers.append(center) centers_sorted = sorted(centers) short_pulse = int(centers_sorted[0]) long_pulse = int(centers_sorted[1]) if len(centers_sorted) > 1 else short_pulse * 2 mid_pulse = int(centers_sorted[1]) if len(centers_sorted) > 2 else None # Confidence: fraction of non-noise points noise_count = sum(labels == -1) confidence = 1.0 - (noise_count / len(labels)) # Gaps short_gap = int(np.percentile(low_pulses, 25)) if low_pulses else 0 long_gap = int(np.percentile(low_pulses, 75)) if low_pulses else 0 return TimingSignature( short_pulse_us=short_pulse, long_pulse_us=long_pulse, mid_pulse_us=mid_pulse, short_gap_us=short_gap, long_gap_us=long_gap, num_levels=len(centers_sorted), confidence=confidence, method='dbscan' ) def _fallback_method(self, high_pulses: List[int], low_pulses: List[int]) -> TimingSignature: """Simple percentile-based fallback""" short_pulse = int(np.percentile(high_pulses, 25)) long_pulse = int(np.percentile(high_pulses, 75)) short_gap = int(np.percentile(low_pulses, 25)) if low_pulses else 0 long_gap = int(np.percentile(low_pulses, 75)) if low_pulses else 0 return TimingSignature( short_pulse_us=short_pulse, long_pulse_us=long_pulse, mid_pulse_us=None, short_gap_us=short_gap, long_gap_us=long_gap, num_levels=2, confidence=0.5, method='fallback' ) ``` --- ### Component 2: Preamble Detection **File:** `src/matcher/preamble_detector.py` (new) ```python import numpy as np from typing import Optional, Dict class PreambleDetector: """Detect preamble/sync patterns at start of transmission""" def detect(self, pulses: List[int]) -> Optional[Dict]: """ Detect common preamble types: 1. Alternating pattern (101010...) - used by Oregon, Manchester 2. Long burst (11111...) - used by Princeton, PT2262 3. Sync word (specific pattern) - used by Acurite, LaCrosse """ if len(pulses) < 20: return None # Check first 50 pulses preamble_window = pulses[:50] # Method 1: Autocorrelation for periodic patterns periodic = self._detect_periodic(preamble_window) if periodic: return periodic # Method 2: Long burst detection burst = self._detect_long_burst(preamble_window) if burst: return burst # Method 3: Known sync patterns sync = self._detect_sync_pattern(preamble_window) if sync: return sync return None def _detect_periodic(self, pulses: List[int]) -> Optional[Dict]: """ Detect repeating patterns via autocorrelation Example: Oregon Scientific sends "1010101010..." (16 times) """ # Autocorrelation autocorr = np.correlate(pulses, pulses, mode='full') autocorr = autocorr[len(autocorr)//2:] # Keep positive lags # Find first peak after lag=1 peaks = [] for i in range(2, min(20, len(autocorr))): if autocorr[i] > autocorr[i-1] and autocorr[i] > autocorr[i+1]: peaks.append((i, autocorr[i])) if not peaks: return None # Strongest peak = period period, strength = max(peaks, key=lambda x: x[1]) # Verify: pattern repeats at least 4 times pattern = pulses[:period] repetitions = 0 for i in range(0, len(pulses) - period, period): chunk = pulses[i:i+period] if self._pattern_matches(pattern, chunk, tolerance=0.2): repetitions += 1 else: break if repetitions >= 4: return { 'type': 'periodic', 'period': period, 'repetitions': repetitions, 'pattern': pattern, 'length': period * repetitions } return None def _pattern_matches(self, pattern1: List[int], pattern2: List[int], tolerance: float) -> bool: """Check if two patterns match within tolerance""" if len(pattern1) != len(pattern2): return False for p1, p2 in zip(pattern1, pattern2): if abs(p1 - p2) > abs(p1) * tolerance: return False return True def _detect_long_burst(self, pulses: List[int]) -> Optional[Dict]: """ Detect long HIGH pulse at start Example: Princeton sends 4x long HIGH as preamble """ if pulses[0] <= 0: return None # First pulse must be HIGH mean_pulse = np.mean([abs(p) for p in pulses]) # First pulse significantly longer than average? if pulses[0] > mean_pulse * 2: return { 'type': 'long_burst', 'duration_us': pulses[0], 'length': 1 } return None def _detect_sync_pattern(self, pulses: List[int]) -> Optional[Dict]: """ Detect known sync patterns from protocol database Example: Acurite uses "10" (short HIGH, short LOW) as sync """ # Convert pulses to binary (simplified) mean_pulse = np.median([abs(p) for p in pulses if p > 0]) bits = [] for p in pulses[:20]: if p > 0: bits.append('1' if p > mean_pulse else '0') else: bits.append('0') bit_string = ''.join(bits) # Check against known sync patterns known_syncs = { '10': 'Acurite', '1000': 'Oregon Scientific', '1111': 'PT2262', '0110': 'LaCrosse' } for pattern, protocol in known_syncs.items(): if bit_string.startswith(pattern): return { 'type': 'sync_pattern', 'pattern': pattern, 'protocol_hint': protocol, 'length': len(pattern) } return None ``` --- ### Component 3: Protocol Database Loader (RTL_433 Import) **File:** `scripts/import_rtl433_database.py` (new) ```python #!/usr/bin/env python3 """ Import RTL_433 protocol database to expand GigLez signatures Reads: data/rtl_433_protocols.json (286 devices) Writes: Updated src/matcher/protocol_database.py """ import json from pathlib import Path def load_rtl433_protocols(): """Load RTL_433 JSON database""" json_path = Path("data/rtl_433_protocols.json") with open(json_path) as f: data = json.load(f) return data['devices'] def convert_to_protocol_signature(device): """ Convert RTL_433 format to ProtocolSignature RTL_433 format: { "name": "Acurite Tower Sensor", "modulation": "OOK", "short_width": 1000, "long_width": 2000, "gap_limit": 3500, "frequency": null, # Often missing "manufacturer": "Acurite", "category": "weather" } """ # Map modulation modulation_map = { 'OOK': 'Modulation.OOK', 'FSK': 'Modulation.FSK', 'ASK': 'Modulation.ASK' } # Estimate frequency from category if missing freq = device.get('frequency') if not freq: # Defaults by category if device['category'] == 'weather': freq = 433920000 # Most weather sensors elif device['category'] == 'automotive': freq = 315000000 # TPMS (North America) else: freq = 433920000 # Default ISM band # Calculate typical pulse count # Estimate: typical_pulses = 2 * (short + long) * 24 bits short = device.get('short_width', 500) long = device.get('long_width', 1000) typical_pulse_count = int(2.5 * (short + long) / 100) # Heuristic return { 'name': device['name'], 'manufacturer': device.get('manufacturer'), 'category': device.get('category', 'Unknown'), 'modulation': modulation_map.get(device.get('modulation'), 'Modulation.OOK'), 'short_pulse_us': short, 'long_pulse_us': long, 'frequency': freq, 'frequency_tolerance': 100000, # ±100kHz 'gap_limit': device.get('gap_limit', short * 4), 'reset_limit': device.get('reset_limit', long * 5), 'typical_pulse_count': typical_pulse_count, 'source': 'RTL_433' } def generate_python_code(protocols): """Generate Python code for protocol_database.py""" code = '''# Auto-generated from RTL_433 database # Total protocols: {} # Source: scripts/import_rtl433_database.py RTL433_PROTOCOLS = [ '''.format(len(protocols)) for proto in protocols: code += f''' ProtocolSignature( name="{proto['name']}", category="{proto['category']}", manufacturer="{proto.get('manufacturer', 'Unknown')}", modulation={proto['modulation']}, short_pulse_us={proto['short_pulse_us']}, long_pulse_us={proto['long_pulse_us']}, frequency={proto['frequency']}, frequency_tolerance={proto['frequency_tolerance']}, typical_pulse_count={proto['typical_pulse_count']}, ), ''' code += '''] # Combine with hand-curated protocols ALL_PROTOCOLS = ( WEATHER_SENSORS + GARAGE_DOOR_OPENERS + DOORBELLS + TIRE_PRESSURE + SECURITY_SENSORS + REMOTE_CONTROLS + RTL433_PROTOCOLS # Add imported protocols ) ''' return code if __name__ == '__main__': print("Loading RTL_433 database...") devices = load_rtl433_protocols() print(f"Found {len(devices)} devices") print("Converting to ProtocolSignature format...") protocols = [convert_to_protocol_signature(d) for d in devices] print("Generating Python code...") code = generate_python_code(protocols) output_file = "src/matcher/rtl433_protocols_generated.py" with open(output_file, 'w') as f: f.write(code) print(f"✅ Written to {output_file}") print(f" Total protocols: {len(protocols)}") print(f" Categories: {set(p['category'] for p in protocols)}") ``` --- ## Performance Targets | Feature | Current | Improved | Method | |---------|---------|----------|--------| | **Protocol Database** | 25 | 311 (25 + 286 RTL_433) | Auto-import script | | **Timing Accuracy** | 65% | 80% | Multi-method ensemble | | **Single-Tx Success** | 20% | 55% | Outlier removal + preamble | | **Processing Speed** | 150ms | 90ms | Frequency filtering | | **Unknown Accuracy** | 5% | 40% | Expanded DB + scoring | --- ## FOSS Resources Utilized 1. **RTL_433 Protocol Database:** 286 device signatures 2. **Flipper Zero Protocols:** 40+ protocol decoders 3. **Scikit-learn:** K-means, DBSCAN clustering 4. **SciPy:** Statistical analysis, peak detection 5. **NumPy:** Signal processing, autocorrelation 6. **FCC Database:** Device frequency validation --- ## Next Steps 1. **Implement Component 1:** Robust timing analyzer (`timing_analyzer.py`) 2. **Implement Component 2:** Preamble detector (`preamble_detector.py`) 3. **Run Component 3:** Import RTL_433 database (expand to 311 protocols) 4. **Integrate:** Update `pattern_decoder.py` to use new components 5. **Benchmark:** Test on labeled dataset, measure accuracy improvement --- **Implementation Time:** 1-2 weeks **Dependencies:** None (FOSS only) **Risk:** Low (incremental improvements to existing system)