giglez/scripts/test_rtl433_with_known_devices.py

#!/usr/bin/env python3
"""
Test RTL_433 Decoder with Known Devices

Uses the test dataset created by create_test_dataset_with_known_devices.py
to validate RTL_433 decoding against known device captures.
"""

import os
import sys
from pathlib import Path

# Add project to path
sys.path.insert(0, str(Path(__file__).parent.parent))

from src.parser.sub_parser import parse_sub_file
from src.matcher.rtl433_decoder import get_decoder


def test_known_devices(test_dir="data/test_known_devices"):
    """Test RTL_433 decoder against known device captures"""

    test_path = Path(test_dir)

    if not test_path.exists():
        print(f"❌ Test directory not found: {test_dir}")
        print(f"   Run: python3 scripts/create_test_dataset_with_known_devices.py")
        return False

    # Get RTL_433 decoder
    decoder = get_decoder()

    # Check RTL_433 availability
    if not decoder._check_rtl433_available():
        print("❌ RTL_433 is not available on this system")
        print("   Install: sudo apt-get install rtl-433")
        return False

    print("=" * 70)
    print("Testing RTL_433 Decoder with Known Devices")
    print("=" * 70)
    print()
    print(f"RTL_433 Version: {decoder.get_version()}")
    print(f"Test Directory: {test_path}")
    print()

    # Get all .sub files
    sub_files = sorted(test_path.glob("*.sub"))

    if not sub_files:
        print(f"❌ No .sub files found in {test_dir}")
        return False

    print(f"Found {len(sub_files)} test files")
    print()

    # Test each file
    results = []

    for sub_file in sub_files:
        print("─" * 70)
        print(f"Testing: {sub_file.name}")
        print("─" * 70)

        try:
            # Parse .sub file
            metadata = parse_sub_file(str(sub_file))

            print(f"Protocol:  {metadata.protocol}")
            print(f"Frequency: {metadata.frequency / 1_000_000:.3f} MHz")
            print(f"Format:    {metadata.file_format}")

            if metadata.has_raw_data:
                print(f"RAW Data:  {metadata.pulse_count} pulses")
            else:
                print(f"KEY Data:  {metadata.key if hasattr(metadata, 'key') else 'N/A'}")

            print()

            # Try RTL_433 decoding
            print("🔍 Running RTL_433 decode...")
            devices = decoder.decode(metadata, enable_all_protocols=True)

            if devices:
                print(f"✅ Successfully decoded {len(devices)} device(s):")
                print()

                for i, device in enumerate(devices, 1):
                    print(f"  Device #{i}:")
                    print(f"    Model:        {device.model}")
                    print(f"    Manufacturer: {device.manufacturer or 'Unknown'}")
                    print(f"    Protocol ID:  {device.protocol_id}")
                    print(f"    Device ID:    {device.device_id or 'N/A'}")
                    print(f"    Confidence:   {device.confidence:.2%}")

                    if device.raw_data:
                        print(f"    Data Fields:  {len(device.raw_data)} fields")
                        # Show first 3 fields
                        for key, value in list(device.raw_data.items())[:3]:
                            print(f"      - {key}: {value}")

                    print()

                results.append({
                    "filename": sub_file.name,
                    "expected_protocol": metadata.protocol,
                    "decoded": True,
                    "device_count": len(devices),
                    "devices": devices
                })
            else:
                print("⚠️  No devices decoded")
                print("   This might be expected for KEY format captures")
                print("   RTL_433 works best with RAW pulse data")
                print()

                results.append({
                    "filename": sub_file.name,
                    "expected_protocol": metadata.protocol,
                    "decoded": False,
                    "device_count": 0,
                    "devices": []
                })

        except Exception as e:
            print(f"❌ Error processing {sub_file.name}: {e}")
            print()

            results.append({
                "filename": sub_file.name,
                "decoded": False,
                "error": str(e)
            })

    # Summary
    print("=" * 70)
    print("SUMMARY")
    print("=" * 70)
    print()

    total_files = len(results)
    successful_decodes = sum(1 for r in results if r.get("decoded", False))
    total_devices = sum(r.get("device_count", 0) for r in results)

    print(f"Total Files:       {total_files}")
    print(f"Successful Decodes: {successful_decodes} ({successful_decodes/total_files*100:.0f}%)")
    print(f"Total Devices:     {total_devices}")
    print()

    # Detailed results
    print("Results by File:")
    print()

    for result in results:
        filename = result["filename"]
        decoded = result.get("decoded", False)
        device_count = result.get("device_count", 0)

        if decoded:
            status = f"✅ {device_count} device(s)"
        elif "error" in result:
            status = f"❌ Error: {result['error']}"
        else:
            status = "⚠️  No decode"

        print(f"  {filename:<50} {status}")

    print()

    # Expected Results Analysis
    print("=" * 70)
    print("EXPECTED RESULTS ANALYSIS")
    print("=" * 70)
    print()
    print("NOTE: RTL_433 decoding success depends on:")
    print("  1. Signal format (RAW pulse data vs KEY format)")
    print("  2. Protocol support (RTL_433 has 244 protocols)")
    print("  3. Signal quality and completeness")
    print()
    print("Test Captures:")
    print("  - GateTX:    KEY format - decode unlikely (proprietary)")
    print("  - Honeywell: RAW format - decode LIKELY (supported)")
    print("  - Magellan:  KEY format - decode unlikely (proprietary)")
    print("  - MegaCode:  KEY format - decode unlikely (proprietary)")
    print("  - Holtek:    RAW format - decode POSSIBLE (generic)")
    print()
    print("Expected Success Rate: 20-40% (1-2 out of 5 files)")
    print("  → RAW captures more likely to decode")
    print("  → KEY format captures contain pre-decoded data")
    print("  → RTL_433 excels at decoding RAW pulse streams")
    print()

    # Integration Test
    print("=" * 70)
    print("INTEGRATION STATUS")
    print("=" * 70)
    print()

    if successful_decodes > 0:
        print("✅ RTL_433 integration is WORKING")
        print(f"   Successfully decoded {successful_decodes} file(s)")
        print()
        print("Next Steps:")
        print("  1. Test with real RAW captures from actual devices")
        print("  2. Monitor decode success rate in production")
        print("  3. Consider Phase 7 (caching) if performance needed")
    else:
        print("⚠️  RTL_433 integration is functional but no decodes")
        print()
        print("This is expected because:")
        print("  - Test captures are mostly KEY format (pre-decoded)")
        print("  - RTL_433 needs RAW pulse data for decoding")
        print("  - Test captures use Flipper-specific protocols")
        print()
        print("Recommendations:")
        print("  1. Test with weather sensor captures (well-supported)")
        print("  2. Capture RAW signals from known RTL_433 devices")
        print("  3. Check RTL_433 protocol list for supported devices")

    print()
    print("=" * 70)

    return successful_decodes > 0


if __name__ == '__main__':
    success = test_known_devices()
    sys.exit(0 if success else 1)