feat: RTL_433 protocol database integration (Phase 1)
Parse and import 286 device protocols from RTL_433: - Created RTL_433 C source parser (scripts/parse_rtl433_devices.py) - Extracted device names, modulation, timing parameters, categories - Generated structured JSON database (data/rtl_433_protocols.json) Protocol breakdown: - 116 weather stations/sensors (Acurite, Oregon, LaCrosse, Fine Offset) - 39 generic sensors (temperature, humidity, leak detectors) - 25 TPMS tire pressure sensors (BMW, Ford, Toyota, etc.) - 23 security devices (alarms, motion, smoke) - 18 home automation (remotes, doorbells, garage doors) - 52 other (energy meters, lighting, automotive) Modulation analysis: - 152 OOK (On-Off Keying) devices - 105 FSK (Frequency Shift Keying) devices - Timing parameters extracted (pulse widths, gaps, resets) Next steps: - Integrate RTL_433 DB into matching engine - Add timing analysis for RAW captures - Target: 75-80% accuracy (from current 60-70%)
This commit is contained in:
File diff suppressed because it is too large
Load Diff
Executable
+257
@@ -0,0 +1,257 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
RTL_433 Device Protocol Parser
|
||||
|
||||
Parses RTL_433 C source code to extract device protocol definitions
|
||||
and convert them to a structured JSON format for GigLez matching engine.
|
||||
|
||||
Usage:
|
||||
python3 scripts/parse_rtl433_devices.py /path/to/rtl_433/
|
||||
|
||||
Output:
|
||||
data/rtl_433_protocols.json
|
||||
"""
|
||||
|
||||
import re
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional
|
||||
from dataclasses import dataclass, asdict
|
||||
|
||||
@dataclass
|
||||
class RTL433Device:
|
||||
"""Represents an RTL_433 device protocol definition"""
|
||||
device_id: str # e.g., "acurite_rain_896"
|
||||
name: str # Human-readable name
|
||||
modulation: Optional[str] = None # OOK_PULSE_PPM, FSK_PULSE_PWM, etc.
|
||||
short_width: Optional[int] = None # Microseconds
|
||||
long_width: Optional[int] = None
|
||||
gap_limit: Optional[int] = None
|
||||
reset_limit: Optional[int] = None
|
||||
sync_width: Optional[int] = None
|
||||
tolerance: Optional[int] = None
|
||||
frequency: Optional[int] = None # Hz
|
||||
category: Optional[str] = None # Inferred from device type
|
||||
manufacturer: Optional[str] = None # Extracted from name
|
||||
source_file: Optional[str] = None
|
||||
|
||||
class RTL433Parser:
|
||||
"""Parses RTL_433 device definitions from C source"""
|
||||
|
||||
# Modulation types mapping
|
||||
MODULATION_TYPES = {
|
||||
'OOK_PULSE_PPM': 'OOK',
|
||||
'OOK_PULSE_PWM': 'OOK',
|
||||
'OOK_PULSE_Manchester_Zerobit': 'OOK',
|
||||
'OOK_PULSE_PCM': 'OOK',
|
||||
'OOK_PULSE_DMC': 'OOK',
|
||||
'OOK_PULSE_PIWM_RAW': 'OOK',
|
||||
'OOK_PULSE_PIWM_DC': 'OOK',
|
||||
'FSK_PULSE_PCM': 'FSK',
|
||||
'FSK_PULSE_PWM': 'FSK',
|
||||
'FSK_PULSE_Manchester_Zerobit': 'FSK',
|
||||
}
|
||||
|
||||
# Device category patterns (from device name)
|
||||
CATEGORY_PATTERNS = {
|
||||
'weather': ['weather', 'rain', 'wind', 'humidity', 'temperature', 'barometer', 'therm'],
|
||||
'tpms': ['tpms', 'tire', 'pressure'],
|
||||
'automotive': ['car', 'vehicle', 'key', 'fob', 'honda', 'toyota', 'ford', 'bmw', 'audi'],
|
||||
'security': ['alarm', 'security', 'motion', 'door', 'window', 'smoke', 'pir'],
|
||||
'home_automation': ['remote', 'switch', 'socket', 'plug', 'doorbell', 'garage', 'gate'],
|
||||
'sensors': ['sensor', 'detector', 'meter', 'monitor'],
|
||||
'energy': ['power', 'energy', 'electric', 'gas', 'water', 'current', 'voltage'],
|
||||
'lighting': ['light', 'lamp', 'dimmer', 'bulb'],
|
||||
}
|
||||
|
||||
def __init__(self, rtl433_path: str):
|
||||
self.rtl433_path = Path(rtl433_path)
|
||||
self.devices_dir = self.rtl433_path / 'src' / 'devices'
|
||||
self.devices: List[RTL433Device] = []
|
||||
|
||||
def parse_all(self) -> List[RTL433Device]:
|
||||
"""Parse all device definitions from RTL_433"""
|
||||
print(f"Parsing RTL_433 devices from: {self.devices_dir}")
|
||||
|
||||
if not self.devices_dir.exists():
|
||||
raise FileNotFoundError(f"Devices directory not found: {self.devices_dir}")
|
||||
|
||||
# Get list of device files
|
||||
device_files = sorted(self.devices_dir.glob('*.c'))
|
||||
print(f"Found {len(device_files)} device files")
|
||||
|
||||
for device_file in device_files:
|
||||
devices_in_file = self.parse_file(device_file)
|
||||
self.devices.extend(devices_in_file)
|
||||
if devices_in_file:
|
||||
print(f" {device_file.name}: {len(devices_in_file)} devices")
|
||||
|
||||
print(f"\nTotal devices parsed: {len(self.devices)}")
|
||||
return self.devices
|
||||
|
||||
def parse_file(self, file_path: Path) -> List[RTL433Device]:
|
||||
"""Parse device definitions from a single C file"""
|
||||
devices = []
|
||||
|
||||
try:
|
||||
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||||
content = f.read()
|
||||
|
||||
# Find all r_device struct definitions
|
||||
# Pattern: r_device const device_name = { ... };
|
||||
device_pattern = r'r_device\s+const\s+(\w+)\s*=\s*\{([^}]+)\}'
|
||||
|
||||
matches = re.finditer(device_pattern, content, re.MULTILINE | re.DOTALL)
|
||||
|
||||
for match in matches:
|
||||
device_id = match.group(1)
|
||||
struct_body = match.group(2)
|
||||
|
||||
device = self.parse_device_struct(device_id, struct_body, file_path.name)
|
||||
if device:
|
||||
devices.append(device)
|
||||
|
||||
except Exception as e:
|
||||
print(f" Error parsing {file_path.name}: {e}")
|
||||
|
||||
return devices
|
||||
|
||||
def parse_device_struct(self, device_id: str, struct_body: str, source_file: str) -> Optional[RTL433Device]:
|
||||
"""Parse individual r_device struct"""
|
||||
|
||||
# Extract .name field
|
||||
name_match = re.search(r'\.name\s*=\s*"([^"]+)"', struct_body)
|
||||
if not name_match:
|
||||
return None
|
||||
|
||||
name = name_match.group(1)
|
||||
|
||||
# Extract other fields
|
||||
device = RTL433Device(
|
||||
device_id=device_id,
|
||||
name=name,
|
||||
source_file=source_file
|
||||
)
|
||||
|
||||
# Modulation
|
||||
mod_match = re.search(r'\.modulation\s*=\s*(\w+)', struct_body)
|
||||
if mod_match:
|
||||
modulation_raw = mod_match.group(1)
|
||||
device.modulation = self.MODULATION_TYPES.get(modulation_raw, modulation_raw)
|
||||
|
||||
# Timing parameters
|
||||
for field_name in ['short_width', 'long_width', 'gap_limit', 'reset_limit', 'sync_width', 'tolerance']:
|
||||
field_match = re.search(rf'\.{field_name}\s*=\s*(\d+)', struct_body)
|
||||
if field_match:
|
||||
setattr(device, field_name, int(field_match.group(1)))
|
||||
|
||||
# Infer category from device name
|
||||
device.category = self.infer_category(name.lower())
|
||||
|
||||
# Extract manufacturer from name (first word usually)
|
||||
device.manufacturer = self.extract_manufacturer(name)
|
||||
|
||||
return device
|
||||
|
||||
def infer_category(self, name_lower: str) -> str:
|
||||
"""Infer device category from name"""
|
||||
for category, keywords in self.CATEGORY_PATTERNS.items():
|
||||
for keyword in keywords:
|
||||
if keyword in name_lower:
|
||||
return category
|
||||
return 'other'
|
||||
|
||||
def extract_manufacturer(self, name: str) -> Optional[str]:
|
||||
"""Extract manufacturer from device name"""
|
||||
# First word is usually manufacturer
|
||||
words = name.split()
|
||||
if words:
|
||||
manufacturer = words[0]
|
||||
# Filter out generic words
|
||||
generic = ['wireless', 'generic', 'remote', 'sensor', 'temperature', 'humidity']
|
||||
if manufacturer.lower() not in generic:
|
||||
return manufacturer
|
||||
return None
|
||||
|
||||
def to_json(self, output_path: str):
|
||||
"""Export devices to JSON"""
|
||||
devices_dict = [asdict(d) for d in self.devices]
|
||||
|
||||
with open(output_path, 'w') as f:
|
||||
json.dump({
|
||||
'source': 'RTL_433',
|
||||
'total_devices': len(self.devices),
|
||||
'devices': devices_dict
|
||||
}, f, indent=2)
|
||||
|
||||
print(f"\nExported to: {output_path}")
|
||||
|
||||
def print_summary(self):
|
||||
"""Print summary statistics"""
|
||||
print("\n" + "="*60)
|
||||
print("RTL_433 DEVICE SUMMARY")
|
||||
print("="*60)
|
||||
|
||||
# By category
|
||||
categories = {}
|
||||
for device in self.devices:
|
||||
cat = device.category or 'unknown'
|
||||
categories[cat] = categories.get(cat, 0) + 1
|
||||
|
||||
print("\nDevices by Category:")
|
||||
for cat, count in sorted(categories.items(), key=lambda x: x[1], reverse=True):
|
||||
print(f" {cat:20s}: {count:3d}")
|
||||
|
||||
# By modulation
|
||||
modulations = {}
|
||||
for device in self.devices:
|
||||
mod = device.modulation or 'unknown'
|
||||
modulations[mod] = modulations.get(mod, 0) + 1
|
||||
|
||||
print("\nDevices by Modulation:")
|
||||
for mod, count in sorted(modulations.items(), key=lambda x: x[1], reverse=True):
|
||||
print(f" {mod:20s}: {count:3d}")
|
||||
|
||||
# Top manufacturers
|
||||
manufacturers = {}
|
||||
for device in self.devices:
|
||||
mfg = device.manufacturer or 'unknown'
|
||||
manufacturers[mfg] = manufacturers.get(mfg, 0) + 1
|
||||
|
||||
print("\nTop 10 Manufacturers:")
|
||||
for mfg, count in sorted(manufacturers.items(), key=lambda x: x[1], reverse=True)[:10]:
|
||||
print(f" {mfg:20s}: {count:3d}")
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: python3 parse_rtl433_devices.py /path/to/rtl_433/")
|
||||
print("\nExample:")
|
||||
print(" python3 scripts/parse_rtl433_devices.py /tmp/rtl_433/")
|
||||
sys.exit(1)
|
||||
|
||||
rtl433_path = sys.argv[1]
|
||||
|
||||
parser = RTL433Parser(rtl433_path)
|
||||
devices = parser.parse_all()
|
||||
|
||||
# Create data directory if needed
|
||||
data_dir = Path('data')
|
||||
data_dir.mkdir(exist_ok=True)
|
||||
|
||||
# Export to JSON
|
||||
output_file = data_dir / 'rtl_433_protocols.json'
|
||||
parser.to_json(str(output_file))
|
||||
|
||||
# Print summary
|
||||
parser.print_summary()
|
||||
|
||||
print("\n" + "="*60)
|
||||
print(f"SUCCESS: Parsed {len(devices)} RTL_433 device protocols")
|
||||
print("="*60)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user