#!/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)