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