giglez/scripts/test_real_weather_sensors.py

#!/usr/bin/env python3
"""
Test RTL_433 Decoder with Real Weather Sensor Captures

Tests the downloaded weather sensor files from FlipperZero-Subghz-DB
"""

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_real_weather_sensors(test_dir="data/test_rtl433_real"):
    """Test RTL_433 decoder against real weather sensor captures"""

    test_path = Path(test_dir)

    if not test_path.exists():
        print(f"❌ Test directory not found: {test_dir}")
        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")
        return False

    print("=" * 80)
    print("Testing RTL_433 with Real Weather Sensor Captures")
    print("=" * 80)
    print()
    print(f"Source: FlipperZero-Subghz-DB (13,716 files)")
    print(f"Selected: Weather station RAW captures")
    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"Testing {len(sub_files)} weather sensor captures")
    print()

    # Test each file
    results = []
    successful_decodes = 0
    total_devices_decoded = 0

    for i, sub_file in enumerate(sub_files, 1):
        print("─" * 80)
        print(f"[{i}/{len(sub_files)}] {sub_file.name}")
        print("─" * 80)

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

            print(f"Protocol:  {metadata.protocol or 'RAW'}")
            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"⚠️  No RAW data - skipping")
                results.append({
                    "filename": sub_file.name,
                    "decoded": False,
                    "reason": "No RAW data"
                })
                print()
                continue

            print()

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

            if devices:
                successful_decodes += 1
                total_devices_decoded += len(devices)

                print(f"✅ SUCCESS! Decoded {len(devices)} device(s):")
                print()

                for j, device in enumerate(devices, 1):
                    print(f"  Device #{j}:")
                    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 all data fields
                        for key, value in device.raw_data.items():
                            print(f"      {key}: {value}")

                    print()

                results.append({
                    "filename": sub_file.name,
                    "decoded": True,
                    "device_count": len(devices),
                    "devices": devices
                })
            else:
                print("❌ No devices decoded")
                print()

                results.append({
                    "filename": sub_file.name,
                    "decoded": False,
                    "reason": "RTL_433 found no matches"
                })

        except Exception as e:
            print(f"❌ Error: {e}")
            print()

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

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

    total_files = len(results)
    success_rate = (successful_decodes / total_files * 100) if total_files > 0 else 0

    print(f"Total Files:        {total_files}")
    print(f"Successful Decodes: {successful_decodes} ({success_rate:.1f}%)")
    print(f"Failed Decodes:     {total_files - successful_decodes}")
    print(f"Total Devices:      {total_devices_decoded}")
    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) decoded"
        elif "error" in result:
            status = f"❌ Error: {result['error'][:50]}"
        elif "reason" in result:
            status = f"⚠️  {result['reason']}"
        else:
            status = "❌ No decode"

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

    print()

    # Device breakdown
    if successful_decodes > 0:
        print("=" * 80)
        print("DECODED DEVICES")
        print("=" * 80)
        print()

        for result in results:
            if result.get("decoded") and result.get("devices"):
                print(f"{result['filename']}:")
                for device in result['devices']:
                    print(f"  → {device.manufacturer or 'Unknown'} {device.model}")
                print()

    # Analysis
    print("=" * 80)
    print("ANALYSIS")
    print("=" * 80)
    print()

    if success_rate >= 50:
        print("🎉 EXCELLENT! RTL_433 integration is working well!")
        print(f"   Achieved {success_rate:.1f}% decode rate with real weather sensors")
    elif success_rate >= 25:
        print("✅ GOOD! RTL_433 is successfully decoding devices")
        print(f"   {success_rate:.1f}% success rate is reasonable for mixed captures")
    elif success_rate > 0:
        print("⚠️  PARTIAL SUCCESS - Some devices decoded")
        print(f"   {success_rate:.1f}% success rate - check signal quality")
    else:
        print("❌ NO DECODES")
        print("   Possible causes:")
        print("   - Signal quality too low")
        print("   - Protocols not supported by RTL_433")
        print("   - Files may need preprocessing")

    print()

    # Comparison with previous test
    print("Comparison with Previous Test:")
    print("  Previous (Flipper unit tests): 0% (0/5 files)")
    print(f"  Current (Real weather sensors): {success_rate:.1f}% ({successful_decodes}/{total_files} files)")
    print()

    if successful_decodes > 0:
        improvement = "SIGNIFICANT IMPROVEMENT" if success_rate > 10 else "IMPROVEMENT"
        print(f"  Result: {improvement} ✅")
    else:
        print("  Result: No improvement (investigate further)")

    print()
    print("=" * 80)

    return successful_decodes > 0


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