giglez/scripts/import_rtl433_protocols.py

#!/usr/bin/env python3
"""
Import RTL_433 Protocol Database

Expands GigLez protocol database from 18 → 200+ by importing RTL_433's open-source
protocol definitions.

Source: data/rtl_433_protocols.json (286 devices)
Output: src/matcher/rtl433_protocols_imported.py
"""

import json
from pathlib import Path
from typing import Dict, List
from collections import defaultdict


def load_rtl433_json() -> Dict:
    """Load RTL_433 protocol database JSON"""
    json_path = Path(__file__).parent.parent / "data" / "rtl_433_protocols.json"

    with open(json_path) as f:
        data = json.load(f)

    print(f"✓ Loaded {data['total_devices']} devices from RTL_433 database")
    return data


def estimate_frequency(device: Dict) -> int:
    """
    Estimate frequency for devices with null frequency field

    Based on:
      - Category (weather = 433MHz, automotive = 315MHz)
      - Manufacturer patterns
      - Default ISM bands
    """
    freq = device.get('frequency')
    if freq:
        return freq

    # Heuristics based on category and manufacturer
    category = (device.get('category') or '').lower()
    manufacturer = (device.get('manufacturer') or '').lower()
    name = (device.get('name') or '').lower()

    # Automotive TPMS → 315 MHz (North America) or 433 MHz (Europe)
    if category == 'automotive' or 'tpms' in name:
        return 315000000  # Default to NA frequency

    # Security systems → 433 MHz (most common)
    if category == 'security' or 'alarm' in name or 'security' in name:
        return 433920000

    # Doorbells → 433 MHz
    if 'doorbell' in name or 'bell' in name:
        return 433920000

    # Garage door openers → 315 MHz (NA) or 433 MHz
    if 'garage' in name or 'door' in name:
        # Chamberlain/LiftMaster = 315 MHz
        if 'chamberlain' in manufacturer or 'liftmaster' in manufacturer:
            return 315000000
        return 433920000

    # Weather sensors → 433 MHz (most common globally)
    if category == 'weather' or any(kw in name for kw in ['temperature', 'humidity', 'rain', 'wind', 'sensor']):
        # Some exceptions: Acurite (915 MHz for 5n1)
        if 'acurite' in manufacturer and '5n1' in name:
            return 915000000
        return 433920000

    # Default: 433.92 MHz (most popular ISM band)
    return 433920000


def estimate_encoding(device: Dict) -> str:
    """
    Estimate encoding scheme from modulation and name patterns

    Common mappings:
      - OOK → PWM (most common)
      - FSK → Manchester or custom
      - ASK → PWM
    """
    modulation = (device.get('modulation') or 'OOK').upper()
    name = (device.get('name') or '').lower()

    # Manchester encoding indicators
    if 'manchester' in name or 'oregon' in name:
        return 'Encoding.MANCHESTER'

    # Differential Manchester (rare)
    if 'diff' in name and 'manchester' in name:
        return 'Encoding.DIFFERENTIAL_MANCHESTER'

    # Default based on modulation
    if modulation in ['OOK', 'ASK']:
        return 'Encoding.PWM'
    elif modulation == 'FSK':
        return 'Encoding.MANCHESTER'  # FSK often uses Manchester

    return 'Encoding.PWM'  # Safe default


def estimate_bit_counts(device: Dict) -> tuple:
    """
    Estimate min/max bit counts from device characteristics

    Based on:
      - Typical payload sizes for device categories
      - Manufacturer patterns
    """
    category = (device.get('category') or '').lower()
    name = (device.get('name') or '').lower()

    # Weather sensors: typically 32-72 bits
    if category == 'weather':
        if 'oregon' in name:
            return (64, 128)  # Oregon Scientific uses longer messages
        return (32, 72)

    # Remotes/garage openers: typically 24-40 bits
    if 'remote' in name or 'garage' in name:
        return (24, 40)

    # TPMS: typically 64-80 bits
    if 'tpms' in name:
        return (64, 80)

    # Security sensors: 32-64 bits
    if category == 'security':
        return (32, 64)

    # Default
    return (24, 64)


def estimate_pulse_count(short_width: int, long_width: int, avg_bits: int) -> int:
    """
    Estimate typical pulse count for a transmission

    Formula: pulses ≈ 2 * bits (for PWM: each bit = HIGH + LOW pulse)
              + preamble overhead (~20-40 pulses)
    """
    base_pulses = avg_bits * 2  # Each bit = 2 transitions
    preamble_overhead = 30  # Typical preamble length

    return base_pulses + preamble_overhead


def convert_to_protocol_signature(device: Dict) -> Dict:
    """
    Convert RTL_433 device format to GigLez ProtocolSignature

    Input (RTL_433):
      {
        "device_id": "acurite_th",
        "name": "Acurite 609TXC Temperature and Humidity Sensor",
        "modulation": "OOK",
        "short_width": 1000,
        "long_width": 2000,
        "gap_limit": 3000,
        "reset_limit": 10000,
        "frequency": null,
        "category": "weather",
        "manufacturer": "Acurite"
      }

    Output (GigLez):
      ProtocolSignature(
        name="Acurite 609TXC Temperature and Humidity Sensor",
        category="Weather Sensor",
        manufacturer="Acurite",
        modulation=Modulation.OOK,
        encoding=Encoding.PWM,
        short_pulse_us=1000,
        long_pulse_us=2000,
        frequency=433920000,
        ...
      )
    """
    # Extract basic fields
    name = device.get('name', 'Unknown Device')
    manufacturer = device.get('manufacturer', 'Unknown')
    category = device.get('category', 'unknown').title()

    # Map modulation
    modulation_map = {
        'OOK': 'Modulation.OOK',
        'FSK': 'Modulation.FSK',
        'ASK': 'Modulation.ASK'
    }
    modulation = modulation_map.get(device.get('modulation', 'OOK'), 'Modulation.OOK')

    # Timing parameters
    short_width = device.get('short_width') or 500
    long_width = device.get('long_width') or 1000

    # Frequency estimation
    frequency = estimate_frequency(device)

    # Encoding estimation
    encoding = estimate_encoding(device)

    # Bit count estimation
    min_bits, max_bits = estimate_bit_counts(device)
    avg_bits = (min_bits + max_bits) // 2

    # Pulse count estimation
    typical_pulse_count = estimate_pulse_count(short_width, long_width, avg_bits)

    # Preamble/sync patterns (if detectable from name)
    preamble_pattern = None
    sync_pattern = None

    if 'oregon' in name.lower():
        preamble_pattern = '"1010" * 12'  # Oregon Scientific preamble
        sync_pattern = '"1000"'
    elif 'princeton' in name.lower() or 'pt2262' in name.lower():
        preamble_pattern = '"1" * 4'
        sync_pattern = '"10"'

    return {
        'name': name,
        'category': category,
        'manufacturer': manufacturer,
        'modulation': modulation,
        'encoding': encoding,
        'short_pulse_us': short_width,
        'long_pulse_us': long_width,
        'frequency': frequency,
        'frequency_tolerance': 100000,  # ±100 kHz
        'min_bits': min_bits,
        'max_bits': max_bits,
        'typical_pulse_count': typical_pulse_count,
        'preamble_pattern': preamble_pattern,
        'sync_pattern': sync_pattern,
        'timing_tolerance': 0.25,  # ±25% (more lenient than hand-curated)
        'min_confidence': 0.5,  # Lower threshold for imported protocols
        'source': f"RTL_433 (device_id: {device.get('device_id', 'unknown')})"
    }


def filter_valid_protocols(protocols: List[Dict]) -> List[Dict]:
    """
    Filter out invalid/incomplete protocol definitions

    Requirements:
      - Must have valid timing (short_width and long_width > 0)
      - Must have reasonable values (not extreme outliers)
      - Prefer unique entries (deduplicate by name)
    """
    valid = []
    seen_names = set()

    for proto in protocols:
        # Check timing validity
        if proto['short_pulse_us'] <= 0 or proto['long_pulse_us'] <= 0:
            continue

        # Check reasonable ranges (10μs to 10ms)
        if not (10 <= proto['short_pulse_us'] <= 10000):
            continue
        if not (10 <= proto['long_pulse_us'] <= 10000):
            continue

        # Check long > short (PWM assumption)
        if proto['long_pulse_us'] <= proto['short_pulse_us']:
            # Swap if reversed
            proto['short_pulse_us'], proto['long_pulse_us'] = \
                proto['long_pulse_us'], proto['short_pulse_us']

        # Deduplicate by name
        if proto['name'] in seen_names:
            continue

        seen_names.add(proto['name'])
        valid.append(proto)

    return valid


def generate_python_code(protocols: List[Dict]) -> str:
    """
    Generate Python code for rtl433_protocols_imported.py

    Creates a list of ProtocolSignature objects that can be imported
    into protocol_database.py
    """
    # Group by category for organization
    by_category = defaultdict(list)
    for proto in protocols:
        by_category[proto['category']].append(proto)

    code = '''#!/usr/bin/env python3
"""
RTL_433 Imported Protocol Signatures

Auto-generated from RTL_433 open-source protocol database.
Source: data/rtl_433_protocols.json ({total} devices)

DO NOT EDIT MANUALLY - run scripts/import_rtl433_protocols.py to regenerate.

Generated: {date}
"""

from src.matcher.protocol_database import ProtocolSignature, Modulation, Encoding

# Imported RTL_433 Protocols ({count} total)
RTL433_PROTOCOLS = [
'''.format(
        total=len(protocols),
        count=len(protocols),
        date=__import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    )

    # Generate entries by category
    for category in sorted(by_category.keys()):
        protos = by_category[category]
        code += f'\n    # {category} ({len(protos)} devices)\n'

        for proto in protos:
            code += f'''    ProtocolSignature(
        name="{proto['name']}",
        category="{proto['category']}",
        manufacturer="{proto['manufacturer']}",
        modulation={proto['modulation']},
        encoding={proto['encoding']},
        short_pulse_us={proto['short_pulse_us']},
        long_pulse_us={proto['long_pulse_us']},
        frequency={proto['frequency']},
        frequency_tolerance={proto['frequency_tolerance']},
        min_bits={proto['min_bits']},
        max_bits={proto['max_bits']},
        typical_pulse_count={proto['typical_pulse_count']},
'''

            if proto['preamble_pattern']:
                code += f"        preamble_pattern={proto['preamble_pattern']},\n"
            if proto['sync_pattern']:
                code += f"        sync_pattern={proto['sync_pattern']},\n"

            code += f'''        timing_tolerance={proto['timing_tolerance']},
        min_confidence={proto['min_confidence']},
    ),
'''

    code += ''']

# Export for use in protocol_database.py
__all__ = ['RTL433_PROTOCOLS']
'''

    return code


def update_protocol_database():
    """
    Update protocol_database.py to include RTL433_PROTOCOLS
    """
    db_path = Path(__file__).parent.parent / "src" / "matcher" / "protocol_database.py"

    with open(db_path, 'r') as f:
        original_code = f.read()

    # Check if already updated
    if 'RTL433_PROTOCOLS' in original_code:
        print("⚠ protocol_database.py already includes RTL433_PROTOCOLS")
        return

    # Find the ALL_PROTOCOLS definition and update it
    old_all_protocols = '''# Compile all protocols into single list
ALL_PROTOCOLS = (
    WEATHER_SENSORS +
    GARAGE_DOOR_OPENERS +
    DOORBELLS +
    TIRE_PRESSURE +
    SECURITY_SENSORS +
    REMOTE_CONTROLS
)'''

    new_all_protocols = '''# 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 +
    RTL433_PROTOCOLS  # Imported from RTL_433 database
)'''

    updated_code = original_code.replace(old_all_protocols, new_all_protocols)

    with open(db_path, 'w') as f:
        f.write(updated_code)

    print(f"✓ Updated {db_path} to include RTL433_PROTOCOLS")


def main():
    """Main import workflow"""
    print("=== RTL_433 Protocol Database Import ===\n")

    # Step 1: Load RTL_433 JSON
    data = load_rtl433_json()
    devices = data['devices']

    # Step 2: Convert to GigLez format
    print(f"\n📝 Converting {len(devices)} devices to ProtocolSignature format...")
    protocols = [convert_to_protocol_signature(dev) for dev in devices]

    # Step 3: Filter valid entries
    print(f"🔍 Filtering valid protocols...")
    valid_protocols = filter_valid_protocols(protocols)
    print(f"✓ Kept {len(valid_protocols)} valid protocols (removed {len(protocols) - len(valid_protocols)} invalid)")

    # Step 4: Generate Python code
    print(f"\n📄 Generating Python code...")
    code = generate_python_code(valid_protocols)

    # Step 5: Write to file
    output_path = Path(__file__).parent.parent / "src" / "matcher" / "rtl433_protocols_imported.py"
    with open(output_path, 'w') as f:
        f.write(code)

    print(f"✓ Written to {output_path}")

    # Step 6: Update protocol_database.py
    print(f"\n🔧 Updating protocol_database.py...")
    update_protocol_database()

    # Step 7: Summary statistics
    print(f"\n=== Import Summary ===")
    print(f"Total RTL_433 devices: {len(devices)}")
    print(f"Valid protocols imported: {len(valid_protocols)}")

    # By category
    from collections import Counter
    categories = Counter(p['category'] for p in valid_protocols)
    print(f"\nProtocols by category:")
    for cat, count in categories.most_common():
        print(f"  {cat}: {count}")

    # By frequency
    frequencies = Counter(p['frequency'] for p in valid_protocols)
    print(f"\nProtocols by frequency:")
    for freq, count in sorted(frequencies.items()):
        freq_mhz = freq / 1_000_000
        print(f"  {freq_mhz:.2f} MHz: {count}")

    print(f"\n✅ Import complete!")
    print(f"   Original protocols: 18")
    print(f"   Imported protocols: {len(valid_protocols)}")
    print(f"   Total protocols: {18 + len(valid_protocols)}")

    # Verify import
    print(f"\n🧪 Verifying import...")
    try:
        from src.matcher.protocol_database import get_protocol_database
        db = get_protocol_database()
        stats = db.get_statistics()
        print(f"✓ Database loaded successfully")
        print(f"  Total protocols in database: {stats['total_protocols']}")
    except Exception as e:
        print(f"❌ Error loading database: {e}")


if __name__ == '__main__':
    main()