""" RTL_433 Device Matcher Loads and indexes the RTL_433 protocol database for high-accuracy device matching. Provides fuzzy name matching, timing-based matching, and category lookups. Author: GigLez Team Date: January 2026 """ import json import logging from pathlib import Path from typing import List, Dict, Optional, Tuple from dataclasses import dataclass from difflib import SequenceMatcher logger = logging.getLogger(__name__) @dataclass class RTL433Match: """Match result from RTL_433 database""" device_id: str device_name: str category: str manufacturer: Optional[str] modulation: Optional[str] confidence: float match_method: str timing_data: Optional[Dict] = None class RTL433Matcher: """ High-performance matcher using RTL_433 protocol database Features: - Fast device ID and name lookups - Fuzzy name matching for partial matches - Modulation-based filtering - Timing signature matching - Category-based fallbacks """ def __init__(self, database_path: str = "data/rtl_433_protocols.json"): """ Initialize RTL_433 matcher with protocol database Args: database_path: Path to rtl_433_protocols.json """ self.database_path = Path(database_path) self.devices: List[Dict] = [] self.device_by_id: Dict[str, Dict] = {} self.device_by_name: Dict[str, Dict] = {} self.devices_by_category: Dict[str, List[Dict]] = {} self.devices_by_modulation: Dict[str, List[Dict]] = {} self._load_database() self._build_indexes() logger.info(f"RTL_433 matcher initialized with {len(self.devices)} devices") def _load_database(self): """Load RTL_433 protocol database from JSON""" try: if not self.database_path.exists(): logger.warning(f"RTL_433 database not found: {self.database_path}") return with open(self.database_path, 'r') as f: data = json.load(f) self.devices = data.get('devices', []) logger.info(f"Loaded {len(self.devices)} RTL_433 protocols") except Exception as e: logger.error(f"Failed to load RTL_433 database: {e}") self.devices = [] def _build_indexes(self): """Build search indexes for fast lookups""" for device in self.devices: # Index by device ID device_id = device.get('device_id', '').lower() if device_id: self.device_by_id[device_id] = device # Index by device name (lowercase for case-insensitive) device_name = device.get('name', '').lower() if device_name: self.device_by_name[device_name] = device # Index by category category = device.get('category', 'other') if category not in self.devices_by_category: self.devices_by_category[category] = [] self.devices_by_category[category].append(device) # Index by modulation modulation = device.get('modulation') if modulation: if modulation not in self.devices_by_modulation: self.devices_by_modulation[modulation] = [] self.devices_by_modulation[modulation].append(device) logger.info(f"Built indexes: {len(self.device_by_id)} IDs, " f"{len(self.devices_by_category)} categories, " f"{len(self.devices_by_modulation)} modulations") def match_by_protocol_name(self, protocol: str) -> List[RTL433Match]: """ Match by protocol/device name Tries: 1. Exact device ID match 2. Exact device name match 3. Fuzzy name match (similarity > 0.6) Args: protocol: Protocol name from .sub file (e.g., "acurite_rain_896", "Oregon") Returns: List of matches sorted by confidence """ matches = [] protocol_lower = protocol.lower() # 1. Exact device ID match if protocol_lower in self.device_by_id: device = self.device_by_id[protocol_lower] matches.append(RTL433Match( device_id=device['device_id'], device_name=device['name'], category=device.get('category', 'other'), manufacturer=device.get('manufacturer'), modulation=device.get('modulation'), confidence=0.95, match_method='rtl433_exact_id' )) return matches # 2. Exact device name match if protocol_lower in self.device_by_name: device = self.device_by_name[protocol_lower] matches.append(RTL433Match( device_id=device['device_id'], device_name=device['name'], category=device.get('category', 'other'), manufacturer=device.get('manufacturer'), modulation=device.get('modulation'), confidence=0.90, match_method='rtl433_exact_name' )) return matches # 3. Fuzzy matching - check all devices for device in self.devices: device_id = device.get('device_id', '').lower() device_name = device.get('name', '').lower() # Calculate similarity scores id_similarity = SequenceMatcher(None, protocol_lower, device_id).ratio() name_similarity = SequenceMatcher(None, protocol_lower, device_name).ratio() # Also check if protocol is substring substring_match = protocol_lower in device_id or protocol_lower in device_name # Take best similarity similarity = max(id_similarity, name_similarity) if substring_match: similarity = max(similarity, 0.7) if similarity > 0.6: matches.append(RTL433Match( device_id=device['device_id'], device_name=device['name'], category=device.get('category', 'other'), manufacturer=device.get('manufacturer'), modulation=device.get('modulation'), confidence=0.70 + (similarity * 0.15), # 0.70-0.85 range match_method='rtl433_fuzzy' )) # Sort by confidence matches.sort(key=lambda x: x.confidence, reverse=True) return matches[:5] # Top 5 matches def match_by_modulation(self, modulation: str, category: Optional[str] = None) -> List[RTL433Match]: """ Match devices by modulation type Args: modulation: Modulation type (OOK, FSK, etc.) category: Optional category filter Returns: List of matches """ matches = [] devices = self.devices_by_modulation.get(modulation, []) # Filter by category if provided if category: devices = [d for d in devices if d.get('category') == category] # Return top matches for device in devices[:10]: matches.append(RTL433Match( device_id=device['device_id'], device_name=device['name'], category=device.get('category', 'other'), manufacturer=device.get('manufacturer'), modulation=device.get('modulation'), confidence=0.55, # Lower confidence for modulation-only match match_method='rtl433_modulation' )) return matches def match_by_timing(self, short_pulse: int, long_pulse: int, gap_limit: Optional[int] = None, tolerance: float = 0.15) -> List[RTL433Match]: """ Match devices by timing signature Args: short_pulse: Short pulse width in microseconds long_pulse: Long pulse width in microseconds gap_limit: Gap limit in microseconds (optional) tolerance: Matching tolerance (default 15%) Returns: List of matches sorted by timing similarity """ matches = [] for device in self.devices: device_short = device.get('short_width') device_long = device.get('long_width') device_gap = device.get('gap_limit') if not (device_short and device_long): continue # Calculate timing similarity short_diff = abs(short_pulse - device_short) / device_short long_diff = abs(long_pulse - device_long) / device_long # Both must be within tolerance if short_diff <= tolerance and long_diff <= tolerance: # Calculate confidence based on closeness similarity = 1.0 - ((short_diff + long_diff) / 2) # Bonus for gap match gap_bonus = 0.0 if gap_limit and device_gap: gap_diff = abs(gap_limit - device_gap) / device_gap if gap_diff <= tolerance: gap_bonus = 0.05 confidence = 0.70 + (similarity * 0.20) + gap_bonus matches.append(RTL433Match( device_id=device['device_id'], device_name=device['name'], category=device.get('category', 'other'), manufacturer=device.get('manufacturer'), modulation=device.get('modulation'), confidence=min(confidence, 0.95), match_method='rtl433_timing', timing_data={ 'short_pulse': device_short, 'long_pulse': device_long, 'gap_limit': device_gap, 'similarity': similarity } )) # Sort by confidence matches.sort(key=lambda x: x.confidence, reverse=True) return matches[:5] def get_devices_by_category(self, category: str) -> List[Dict]: """Get all devices in a category""" return self.devices_by_category.get(category, []) def get_statistics(self) -> Dict: """Get database statistics""" return { 'total_devices': len(self.devices), 'categories': {cat: len(devs) for cat, devs in self.devices_by_category.items()}, 'modulations': {mod: len(devs) for mod, devs in self.devices_by_modulation.items()}, 'manufacturers': len(set(d.get('manufacturer') for d in self.devices if d.get('manufacturer'))) } # Global singleton instance _rtl433_matcher: Optional[RTL433Matcher] = None def get_rtl433_matcher() -> RTL433Matcher: """Get or create RTL_433 matcher singleton""" global _rtl433_matcher if _rtl433_matcher is None: _rtl433_matcher = RTL433Matcher() return _rtl433_matcher if __name__ == '__main__': # Test the matcher logging.basicConfig(level=logging.INFO) matcher = RTL433Matcher() print("\n" + "="*60) print("RTL_433 MATCHER TEST") print("="*60) # Test 1: Exact match print("\n1. Exact Protocol Match: 'acurite_rain_896'") matches = matcher.match_by_protocol_name('acurite_rain_896') for match in matches: print(f" {match.device_name} ({match.confidence:.2f}) - {match.match_method}") # Test 2: Fuzzy match print("\n2. Fuzzy Match: 'Oregon'") matches = matcher.match_by_protocol_name('Oregon') for match in matches: print(f" {match.device_name} ({match.confidence:.2f}) - {match.match_method}") # Test 3: Modulation match print("\n3. Modulation Match: 'OOK'") matches = matcher.match_by_modulation('OOK', category='weather')[:3] for match in matches: print(f" {match.device_name} ({match.confidence:.2f})") # Test 4: Timing match print("\n4. Timing Match: short=1000µs, long=2000µs") matches = matcher.match_by_timing(1000, 2000) for match in matches: print(f" {match.device_name} ({match.confidence:.2f})") # Statistics print("\n" + "="*60) stats = matcher.get_statistics() print(f"Total devices: {stats['total_devices']}") print(f"Categories: {len(stats['categories'])}") print(f"Modulations: {len(stats['modulations'])}") print("="*60)