#!/usr/bin/env python3 """ Protocol Database for Pattern-Based Decoder Contains timing signatures and patterns for known RF protocols. Extracted from Flipper Zero firmware and RTL_433 protocol definitions. """ from dataclasses import dataclass from typing import Optional, List, Dict, Tuple from enum import Enum class Modulation(Enum): """Signal modulation types""" OOK = "OOK" # On-Off Keying FSK = "FSK" # Frequency Shift Keying ASK = "ASK" # Amplitude Shift Keying class Encoding(Enum): """Pulse encoding schemes""" PWM = "PWM" # Pulse Width Modulation (SHORT=0, LONG=1) PPM = "PPM" # Pulse Position Modulation MANCHESTER = "Manchester" DIFFERENTIAL_MANCHESTER = "Differential Manchester" @dataclass class ProtocolSignature: """Signature for a known RF protocol""" # Identification name: str category: str # Weather, Remote, Door, Sensor, etc. manufacturer: Optional[str] = None model: Optional[str] = None # Frequency frequency: int = 433920000 # Default 433.92 MHz frequency_tolerance: int = 100000 # ±100 kHz # Modulation modulation: Modulation = Modulation.OOK encoding: Encoding = Encoding.PWM # Timing (microseconds) short_pulse_us: int = 500 long_pulse_us: int = 1000 timing_tolerance: float = 0.2 # ±20% # Patterns preamble_pattern: Optional[str] = None # Binary pattern (e.g., "101010") sync_pattern: Optional[str] = None # Data min_bits: int = 24 max_bits: int = 64 typical_pulse_count: int = 100 # Confidence thresholds min_confidence: float = 0.6 def matches_timing(self, short_us: int, long_us: int) -> bool: """Check if observed timing matches this protocol""" short_min = self.short_pulse_us * (1 - self.timing_tolerance) short_max = self.short_pulse_us * (1 + self.timing_tolerance) long_min = self.long_pulse_us * (1 - self.timing_tolerance) long_max = self.long_pulse_us * (1 + self.timing_tolerance) return (short_min <= short_us <= short_max and long_min <= long_us <= long_max) def matches_frequency(self, freq: int) -> bool: """Check if frequency matches this protocol""" return abs(freq - self.frequency) <= self.frequency_tolerance # Known Protocol Signatures # Extracted from Flipper Zero firmware and RTL_433 protocol definitions WEATHER_SENSORS = [ ProtocolSignature( name="Oregon Scientific v2.1", category="Weather Sensor", manufacturer="Oregon Scientific", short_pulse_us=488, long_pulse_us=976, encoding=Encoding.MANCHESTER, preamble_pattern="1010" * 8, # 32-bit preamble sync_pattern="1000", min_bits=64, max_bits=128, typical_pulse_count=200, ), ProtocolSignature( name="Oregon Scientific v3.0", category="Weather Sensor", manufacturer="Oregon Scientific", short_pulse_us=500, long_pulse_us=1000, encoding=Encoding.MANCHESTER, preamble_pattern="1010" * 12, sync_pattern="1000", min_bits=64, max_bits=128, typical_pulse_count=250, ), ProtocolSignature( name="Acurite Tower Sensor", category="Weather Sensor", manufacturer="Acurite", short_pulse_us=220, long_pulse_us=440, encoding=Encoding.PWM, preamble_pattern=None, sync_pattern="10", min_bits=56, max_bits=64, typical_pulse_count=130, ), ProtocolSignature( name="Acurite 5n1 Weather Station", category="Weather Sensor", manufacturer="Acurite", short_pulse_us=220, long_pulse_us=440, encoding=Encoding.PWM, min_bits=64, max_bits=80, typical_pulse_count=160, ), ProtocolSignature( name="LaCrosse TX141TH-Bv2", category="Weather Sensor", manufacturer="LaCrosse", short_pulse_us=500, long_pulse_us=1000, encoding=Encoding.PWM, preamble_pattern="10" * 4, min_bits=40, max_bits=48, typical_pulse_count=100, ), ProtocolSignature( name="Nexus Temperature/Humidity", category="Weather Sensor", manufacturer="Nexus", short_pulse_us=500, long_pulse_us=1000, encoding=Encoding.PWM, preamble_pattern="1" * 8, min_bits=36, max_bits=40, typical_pulse_count=90, ), ProtocolSignature( name="Ambient Weather F007TH", category="Weather Sensor", manufacturer="Ambient Weather", short_pulse_us=500, long_pulse_us=1000, encoding=Encoding.PWM, min_bits=64, max_bits=72, typical_pulse_count=150, ), ] GARAGE_DOOR_OPENERS = [ ProtocolSignature( name="Princeton", category="Garage Door Opener", manufacturer=None, short_pulse_us=400, long_pulse_us=1200, encoding=Encoding.PWM, preamble_pattern="1" * 4, sync_pattern="10", min_bits=24, max_bits=32, typical_pulse_count=60, ), ProtocolSignature( name="Chamberlain/LiftMaster 315MHz", category="Garage Door Opener", manufacturer="Chamberlain", short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, min_bits=32, max_bits=40, typical_pulse_count=80, frequency=315000000, ), ProtocolSignature( name="LiftMaster 433MHz", category="Garage Door Opener", manufacturer="LiftMaster", # Fixed-code LiftMaster (pre-Security+) at 433 MHz # Uses ~350µs SHORT, ~1050µs LONG (3:1 ratio like Princeton but 40-bit) short_pulse_us=350, long_pulse_us=1050, encoding=Encoding.PWM, min_bits=32, max_bits=40, typical_pulse_count=90, frequency=433920000, ), ProtocolSignature( name="LiftMaster Security+ 2.0", category="Garage Door Opener", manufacturer="LiftMaster", # Security+ 2.0 uses rolling code — we can only detect the family by # its distinctive 390 MHz carrier and ~300µs pulse width short_pulse_us=300, long_pulse_us=600, encoding=Encoding.PWM, min_bits=40, max_bits=66, typical_pulse_count=100, frequency=390000000, frequency_tolerance=2000000, ), ProtocolSignature( name="Linear MegaCode", category="Garage Door Opener", manufacturer="Linear", short_pulse_us=250, long_pulse_us=500, encoding=Encoding.PWM, min_bits=32, max_bits=32, typical_pulse_count=70, frequency=318000000, ), ProtocolSignature( name="Marantec D302 / D304", category="Garage Door Opener", manufacturer="Marantec", short_pulse_us=1000, long_pulse_us=2000, encoding=Encoding.PWM, min_bits=12, max_bits=16, typical_pulse_count=35, frequency=433920000, ), ProtocolSignature( name="FAAC XT2 / XT4", category="Garage Door Opener", manufacturer="FAAC", short_pulse_us=500, long_pulse_us=1500, encoding=Encoding.PWM, min_bits=40, max_bits=64, typical_pulse_count=110, frequency=433920000, ), ProtocolSignature( name="Came TOP432 / BRC802", category="Garage Door Opener", manufacturer="Came", short_pulse_us=500, long_pulse_us=1000, encoding=Encoding.PWM, min_bits=12, max_bits=24, typical_pulse_count=55, frequency=433920000, ), ProtocolSignature( name="BFT Mitto Rolling Code", category="Garage Door Opener", manufacturer="BFT", short_pulse_us=500, long_pulse_us=1500, encoding=Encoding.PWM, min_bits=52, max_bits=64, typical_pulse_count=125, frequency=433920000, ), ProtocolSignature( name="CAME-АТЛАС Gate Remote", category="Garage Door Opener", manufacturer="Came", short_pulse_us=320, long_pulse_us=960, encoding=Encoding.PWM, min_bits=24, max_bits=24, typical_pulse_count=55, frequency=433920000, ), ] DOORBELLS = [ ProtocolSignature( name="Honeywell Doorbell", category="Doorbell", manufacturer="Honeywell", short_pulse_us=175, long_pulse_us=340, encoding=Encoding.PWM, min_bits=48, max_bits=48, typical_pulse_count=100, ), ProtocolSignature( name="Byron Doorbell", category="Doorbell", manufacturer="Byron", # Byron WE-series: Princeton-like OOK, 300-400us SHORT, 3:1 ratio short_pulse_us=350, long_pulse_us=1050, encoding=Encoding.PWM, preamble_pattern="1" * 4, sync_pattern="10000", # Long sync gap min_bits=24, max_bits=32, typical_pulse_count=60, frequency=433920000, ), ProtocolSignature( name="GE Doorbell", category="Doorbell", manufacturer="GE", # GE wireless doorbell 433 MHz, ~250us SHORT, 2:1 ratio short_pulse_us=250, long_pulse_us=500, encoding=Encoding.PWM, min_bits=24, max_bits=48, typical_pulse_count=80, frequency=433920000, ), ProtocolSignature( name="Heidemann Doorbell", category="Doorbell", manufacturer="Heidemann", short_pulse_us=400, long_pulse_us=1200, encoding=Encoding.PWM, min_bits=12, max_bits=24, typical_pulse_count=40, frequency=433920000, ), ProtocolSignature( name="Elro DB286A Doorbell", category="Doorbell", manufacturer="Elro", short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, preamble_pattern="1" * 4, min_bits=24, max_bits=24, typical_pulse_count=50, frequency=433920000, ), ProtocolSignature( name="Blyss Doorbell", category="Doorbell", manufacturer="Blyss", short_pulse_us=500, long_pulse_us=1500, encoding=Encoding.PWM, min_bits=32, max_bits=40, typical_pulse_count=80, frequency=433920000, ), ] TIRE_PRESSURE = [ ProtocolSignature( name="Toyota TPMS", category="TPMS", manufacturer="Toyota", short_pulse_us=50, long_pulse_us=100, encoding=Encoding.MANCHESTER, min_bits=64, max_bits=80, typical_pulse_count=160, frequency=315000000, ), ProtocolSignature( name="Schrader TPMS", category="TPMS", manufacturer="Schrader", short_pulse_us=50, long_pulse_us=100, encoding=Encoding.MANCHESTER, min_bits=64, max_bits=80, typical_pulse_count=160, frequency=433920000, ), ] SECURITY_SENSORS = [ ProtocolSignature( name="Magellan", category="Security Sensor", manufacturer="Paradox", short_pulse_us=250, long_pulse_us=500, encoding=Encoding.PWM, min_bits=32, max_bits=48, typical_pulse_count=80, frequency=433920000, ), ] REMOTE_CONTROLS = [ ProtocolSignature( name="PT2262", category="Remote Control", manufacturer=None, short_pulse_us=350, long_pulse_us=1050, encoding=Encoding.PWM, preamble_pattern="1" * 4, min_bits=24, max_bits=24, typical_pulse_count=50, ), ProtocolSignature( name="PT2260", category="Remote Control", manufacturer=None, short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, min_bits=24, max_bits=24, typical_pulse_count=50, ), ProtocolSignature( name="EV1527", category="Remote Control", manufacturer=None, short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, min_bits=24, max_bits=24, typical_pulse_count=50, ), ProtocolSignature( name="Generic Remote SC226x EV1527", category="Remote Control", manufacturer=None, short_pulse_us=320, long_pulse_us=960, encoding=Encoding.PWM, min_bits=24, max_bits=24, typical_pulse_count=52, frequency=433920000, ), ProtocolSignature( name="HCS301 Rolling Code", category="Remote Control", manufacturer="Microchip", short_pulse_us=400, long_pulse_us=800, encoding=Encoding.PWM, min_bits=66, max_bits=66, typical_pulse_count=140, ), ProtocolSignature( name="RGB LED Remote Controller", category="Remote Control", manufacturer=None, # Most 433 MHz RGB LED remotes use EV1527-like encoding short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, min_bits=24, max_bits=24, typical_pulse_count=52, frequency=433920000, ), ProtocolSignature( name="Nice Flor-S Rolling Code", category="Remote Control", manufacturer="Nice", short_pulse_us=500, long_pulse_us=1000, encoding=Encoding.PWM, min_bits=52, max_bits=56, typical_pulse_count=115, frequency=433920000, ), ProtocolSignature( name="FAAC SLH Rolling Code", category="Remote Control", manufacturer="FAAC", short_pulse_us=500, long_pulse_us=1500, encoding=Encoding.PWM, min_bits=40, max_bits=64, typical_pulse_count=110, frequency=433920000, ), ProtocolSignature( name="Somfy RTS", category="Remote Control", manufacturer="Somfy", # Somfy RTS uses Manchester at 433.42 MHz short_pulse_us=604, long_pulse_us=1208, encoding=Encoding.MANCHESTER, min_bits=56, max_bits=56, typical_pulse_count=120, frequency=433420000, frequency_tolerance=50000, ), ProtocolSignature( name="Holtek HT12X Remote", category="Remote Control", manufacturer="Holtek", short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, preamble_pattern="1" * 36, # Long burst sync min_bits=12, max_bits=12, typical_pulse_count=52, frequency=433920000, ), ] FAN_CONTROLLERS = [ ProtocolSignature( name="Hampton Bay Ceiling Fan Remote", category="Fan Controller", manufacturer="Hampton Bay", # Most common: 303 MHz or 433 MHz, OOK, 12-bit dip-switch code short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, min_bits=12, max_bits=24, typical_pulse_count=45, frequency=303900000, frequency_tolerance=200000, ), ProtocolSignature( name="Hampton Bay Ceiling Fan Remote 433MHz", category="Fan Controller", manufacturer="Hampton Bay", short_pulse_us=320, long_pulse_us=960, encoding=Encoding.PWM, min_bits=12, max_bits=24, typical_pulse_count=45, frequency=433920000, ), ProtocolSignature( name="Hunter Fan Remote", category="Fan Controller", manufacturer="Hunter", short_pulse_us=250, long_pulse_us=750, encoding=Encoding.PWM, min_bits=16, max_bits=32, typical_pulse_count=55, frequency=433920000, ), ProtocolSignature( name="Harbor Breeze Ceiling Fan Remote", category="Fan Controller", manufacturer="Harbor Breeze", short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, min_bits=12, max_bits=24, typical_pulse_count=50, frequency=433920000, ), ProtocolSignature( name="Westinghouse Fan Remote", category="Fan Controller", manufacturer="Westinghouse", short_pulse_us=330, long_pulse_us=990, encoding=Encoding.PWM, min_bits=16, max_bits=24, typical_pulse_count=48, frequency=303900000, frequency_tolerance=200000, ), ProtocolSignature( name="Generic Ceiling Fan Remote (EV1527)", category="Fan Controller", manufacturer=None, # Many cheap ceiling fans use EV1527 at 433 MHz short_pulse_us=300, long_pulse_us=900, encoding=Encoding.PWM, min_bits=24, max_bits=24, typical_pulse_count=52, frequency=433920000, ), ] # Import RTL_433 protocols try: from src.matcher.rtl433_protocols_imported import RTL433_PROTOCOLS except ImportError: print("Warning: RTL_433 protocols not imported yet. Run scripts/import_rtl433_protocols.py") RTL433_PROTOCOLS = [] # Compile all protocols into single list ALL_PROTOCOLS = ( WEATHER_SENSORS + GARAGE_DOOR_OPENERS + DOORBELLS + TIRE_PRESSURE + SECURITY_SENSORS + REMOTE_CONTROLS + FAN_CONTROLLERS + RTL433_PROTOCOLS # Imported from RTL_433 database ) class ProtocolDatabase: """Database of known RF protocol signatures""" def __init__(self): self.protocols = ALL_PROTOCOLS self._by_category: Dict[str, List[ProtocolSignature]] = {} self._by_frequency: Dict[int, List[ProtocolSignature]] = {} self._index_protocols() def _index_protocols(self): """Build indexes for fast lookup""" for proto in self.protocols: # Index by category if proto.category not in self._by_category: self._by_category[proto.category] = [] self._by_category[proto.category].append(proto) # Index by frequency (rounded to MHz) freq_mhz = round(proto.frequency / 1_000_000) if freq_mhz not in self._by_frequency: self._by_frequency[freq_mhz] = [] self._by_frequency[freq_mhz].append(proto) def find_by_timing( self, short_us: int, long_us: int, frequency: Optional[int] = None ) -> List[ProtocolSignature]: """Find protocols matching timing characteristics""" matches = [] candidates = self.protocols if frequency: freq_mhz = round(frequency / 1_000_000) candidates = self._by_frequency.get(freq_mhz, self.protocols) for proto in candidates: if proto.matches_timing(short_us, long_us): if not frequency or proto.matches_frequency(frequency): matches.append(proto) return matches def find_by_category(self, category: str) -> List[ProtocolSignature]: """Find all protocols in a category""" return self._by_category.get(category, []) def find_by_frequency(self, frequency: int) -> List[ProtocolSignature]: """Find protocols near a frequency""" matches = [] for proto in self.protocols: if proto.matches_frequency(frequency): matches.append(proto) return matches def get_all(self) -> List[ProtocolSignature]: """Get all protocols""" return self.protocols def get_statistics(self) -> Dict: """Get database statistics""" return { "total_protocols": len(self.protocols), "categories": list(self._by_category.keys()), "by_category": { cat: len(protos) for cat, protos in self._by_category.items() }, "frequency_bands": list(set( round(p.frequency / 1_000_000) for p in self.protocols )), } # Global instance _database: Optional[ProtocolDatabase] = None def get_protocol_database() -> ProtocolDatabase: """Get singleton protocol database instance""" global _database if _database is None: _database = ProtocolDatabase() return _database if __name__ == '__main__': # Test protocol database db = get_protocol_database() print("=== Protocol Database Statistics ===") stats = db.get_statistics() print(f"Total protocols: {stats['total_protocols']}") print(f"Categories: {', '.join(stats['categories'])}") print() print("Protocols by category:") for cat, count in stats['by_category'].items(): print(f" {cat}: {count}") print() print(f"Frequency bands: {', '.join(str(f) + ' MHz' for f in sorted(stats['frequency_bands']))}") print() # Test timing match print("=== Testing Timing Match ===") print("Looking for protocols with SHORT=500us, LONG=1000us @ 433.92 MHz") matches = db.find_by_timing(500, 1000, 433920000) print(f"Found {len(matches)} matches:") for proto in matches: print(f" - {proto.name} ({proto.manufacturer or 'Unknown'}) - {proto.category}")