giglez/tests/test_rtl433_integration.py

"""
RTL_433 Integration Test

Tests complete pipeline:
1. Parse .sub file
2. Convert to pulse data
3. Run RTL_433 decoder
4. Test matcher integration
"""

from pathlib import Path
import time
from loguru import logger

from src.parser.sub_parser import parse_sub_file
from src.parser.rtl433_converter import get_converter
from src.matcher.rtl433_decoder import get_decoder
from src.matcher.strategies import RTL433DecoderStrategy


def test_converter():
    """Test RTL_433 converter"""
    print("="*70)
    print("TEST 1: RTL_433 Converter")
    print("="*70)

    converter = get_converter()
    test_files = list(Path("signatures/t-embed-rf").glob("*.sub"))

    success_count = 0
    for test_file in test_files[:5]:  # Test first 5 files
        try:
            metadata = parse_sub_file(str(test_file))
            if metadata.has_raw_data:
                output_path = converter.convert_metadata(metadata)
                if output_path:
                    print(f"✓ {test_file.name}: {metadata.pulse_count} pulses → {output_path}")
                    success_count += 1
                else:
                    print(f"✗ {test_file.name}: Conversion failed")
            else:
                print(f"⚠ {test_file.name}: No RAW data")
        except Exception as e:
            print(f"✗ {test_file.name}: Error - {e}")

    print(f"\nConverter: {success_count}/{len(test_files[:5])} successful\n")
    return success_count > 0


def test_decoder():
    """Test RTL_433 decoder"""
    print("="*70)
    print("TEST 2: RTL_433 Decoder")
    print("="*70)

    decoder = get_decoder()

    # Check availability
    version = decoder.get_version()
    print(f"RTL_433 Version: {version}")

    # Get protocols
    protocols = decoder.get_supported_protocols()
    print(f"Supported Protocols: {len(protocols)}")
    print(f"Sample: {list(protocols.items())[:3]}\n")

    # Test decoding
    test_files = list(Path("signatures/t-embed-rf").glob("*.sub"))
    results = []

    for test_file in test_files:
        try:
            metadata = parse_sub_file(str(test_file))
            if not metadata.has_raw_data:
                continue

            start_time = time.time()
            devices = decoder.decode(metadata, enable_all_protocols=True)
            decode_time = time.time() - start_time

            result = {
                'file': test_file.name,
                'frequency': metadata.frequency / 1_000_000,
                'pulses': metadata.pulse_count,
                'devices': len(devices),
                'time': decode_time
            }

            if devices:
                result['decoded'] = [d.model for d in devices]
                print(f"✓ {test_file.name}: {len(devices)} device(s) in {decode_time:.2f}s")
                for dev in devices:
                    print(f"    - {dev.model} (protocol {dev.protocol_id})")
            else:
                print(f"✗ {test_file.name}: No decode ({decode_time:.2f}s)")

            results.append(result)

        except Exception as e:
            print(f"✗ {test_file.name}: Error - {e}")

    # Summary
    success = sum(1 for r in results if r['devices'] > 0)
    print(f"\nDecoder: {success}/{len(results)} files decoded successfully")

    if results:
        avg_time = sum(r['time'] for r in results) / len(results)
        print(f"Average decode time: {avg_time:.2f}s")

    print()
    return results


def test_matcher_strategy():
    """Test RTL433DecoderStrategy integration"""
    print("="*70)
    print("TEST 3: RTL433DecoderStrategy")
    print("="*70)

    # Mock database for testing
    class MockDB:
        def __init__(self):
            self.devices = []
            self.next_id = 1

        def execute(self, query, params=()):
            # Simple mock - always return empty for SELECT
            if query.strip().upper().startswith('SELECT'):
                return []

            # For INSERT, return new ID
            if query.strip().upper().startswith('INSERT'):
                result = [{
                    'id': self.next_id,
                    'manufacturer': params[0] if len(params) > 0 else None,
                    'model': params[1] if len(params) > 1 else 'Unknown'
                }]
                self.devices.append(result[0])
                self.next_id += 1
                return result

            return []

    strategy = RTL433DecoderStrategy()
    mock_db = MockDB()

    test_files = list(Path("signatures/t-embed-rf").glob("*.sub"))
    results = []

    for test_file in test_files[:10]:  # Test first 10
        try:
            metadata = parse_sub_file(str(test_file))
            if not metadata.has_raw_data:
                continue

            matches = strategy.match(metadata, mock_db)

            result = {
                'file': test_file.name,
                'matches': len(matches)
            }

            if matches:
                print(f"✓ {test_file.name}: {len(matches)} match(es)")
                for match in matches:
                    print(f"    - {match.device_name} (confidence: {match.confidence})")
                result['devices'] = [m.device_name for m in matches]
            else:
                print(f"✗ {test_file.name}: No matches")

            results.append(result)

        except Exception as e:
            print(f"✗ {test_file.name}: Error - {e}")

    # Summary
    success = sum(1 for r in results if r['matches'] > 0)
    print(f"\nStrategy: {success}/{len(results)} files matched")
    print(f"Created devices in mock DB: {len(mock_db.devices)}\n")

    return results


def test_comprehensive():
    """Run all tests"""
    print("\n" + "="*70)
    print("RTL_433 INTEGRATION TEST SUITE")
    print("="*70 + "\n")

    # Test 1: Converter
    converter_ok = test_converter()

    # Test 2: Decoder
    decoder_results = test_decoder()

    # Test 3: Matcher Strategy
    strategy_results = test_matcher_strategy()

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

    print(f"✓ Converter: {'PASS' if converter_ok else 'FAIL'}")

    decoder_success = sum(1 for r in decoder_results if r.get('devices', 0) > 0)
    decoder_rate = (decoder_success / len(decoder_results) * 100) if decoder_results else 0
    print(f"✓ Decoder: {decoder_success}/{len(decoder_results)} files ({decoder_rate:.1f}%)")

    strategy_success = sum(1 for r in strategy_results if r.get('matches', 0) > 0)
    strategy_rate = (strategy_success / len(strategy_results) * 100) if strategy_results else 0
    print(f"✓ Strategy: {strategy_success}/{len(strategy_results)} files ({strategy_rate:.1f}%)")

    # Check success criteria
    print("\nSuccess Criteria:")
    print(f"  - RTL_433 installed: {'✓ PASS' if converter_ok else '✗ FAIL'}")
    print(f"  - Converter working: {'✓ PASS' if converter_ok else '✗ FAIL'}")
    print(f"  - Decoder functional: {'✓ PASS' if len(decoder_results) > 0 else '✗ FAIL'}")
    print(f"  - Strategy integrated: {'✓ PASS' if len(strategy_results) > 0 else '✗ FAIL'}")

    # Note about decode rate
    print(f"\nNote: Low decode rate ({decoder_rate:.1f}%) is expected for")
    print("short/noisy test captures. Real devices will have higher success.")
    print("="*70)


if __name__ == '__main__':
    # Configure logging
    import sys
    logger.remove()
    logger.add(sys.stderr, level="WARNING")  # Reduce noise

    test_comprehensive()