From 60c07647aa36f9a33b9b64baa11c36e4611e2719 Mon Sep 17 00:00:00 2001 From: priestlypython Date: Wed, 14 Jan 2026 18:04:51 -0800 Subject: [PATCH] test: Add RTL_433 validation test script for known devices --- scripts/test_rtl433_with_known_devices.py | 228 ++++++++++++++++++++++ 1 file changed, 228 insertions(+) create mode 100755 scripts/test_rtl433_with_known_devices.py diff --git a/scripts/test_rtl433_with_known_devices.py b/scripts/test_rtl433_with_known_devices.py new file mode 100755 index 0000000..d1c4fc7 --- /dev/null +++ b/scripts/test_rtl433_with_known_devices.py @@ -0,0 +1,228 @@ +#!/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)