rpi-tulivision/test_custom_decoder.py

#!/usr/bin/env python3
"""
Test script for the custom IR protocol decoder
Tests the decoder against captured signal data
"""

import json
import sys
import os
from custom_ir_protocol import CustomIRProtocol

def test_decoder_with_captured_data():
    """Test the custom decoder with captured signal data"""
    
    # Load captured signals
    try:
        with open("ir_analysis_20250927_190536.json", 'r') as f:
            signals = json.load(f)
    except FileNotFoundError:
        print("Error: ir_analysis_20250927_190536.json not found!")
        return False
    
    # Create custom protocol decoder
    protocol = CustomIRProtocol("TEST_CUSTOM")
    
    print("Testing Custom IR Protocol Decoder")
    print("=" * 50)
    print(f"Loaded {len(signals)} captured signals")
    print()
    
    # Test signals with 71 pulses (most common)
    successful_decodes = 0
    failed_decodes = 0
    decoded_commands = {}
    
    for i, signal_data in enumerate(signals):
        pulse_count = signal_data['pulse_count']
        pulses = signal_data['pulses']
        
        # Convert to the format expected by the decoder
        # (is_pulse, duration_in_seconds)
        formatted_pulses = []
        for j, duration_us in enumerate(pulses):
            is_pulse = (j % 2 == 0)  # Alternating pulse/space
            duration_seconds = duration_us / 1000000.0
            formatted_pulses.append((is_pulse, duration_seconds))
        
        # Try to decode
        command = protocol.decode(formatted_pulses)
        
        if command:
            successful_decodes += 1
            if command not in decoded_commands:
                decoded_commands[command] = 0
            decoded_commands[command] += 1
            
            print(f"Signal {i+1:2d} ({pulse_count:2d} pulses): {command}")
        else:
            failed_decodes += 1
            if pulse_count == 71:  # Only show failed 71-pulse signals
                print(f"Signal {i+1:2d} ({pulse_count:2d} pulses): FAILED TO DECODE")
    
    print()
    print("=" * 50)
    print("DECODING RESULTS")
    print("=" * 50)
    print(f"Successful decodes: {successful_decodes}")
    print(f"Failed decodes: {failed_decodes}")
    print(f"Success rate: {successful_decodes/(successful_decodes+failed_decodes)*100:.1f}%")
    print()
    
    if decoded_commands:
        print("Decoded commands:")
        for command, count in sorted(decoded_commands.items()):
            print(f"  {command}: {count} occurrences")
        print()
    
    # Analyze timing patterns for failed decodes
    print("Analyzing timing patterns...")
    analyze_timing_patterns(signals, protocol)
    
    return successful_decodes > 0

def analyze_timing_patterns(signals, protocol):
    """Analyze timing patterns to help debug the decoder"""
    
    print("\nTiming analysis for 71-pulse signals:")
    print("-" * 40)
    
    for signal_data in signals:
        if signal_data['pulse_count'] == 71:
            pulses = signal_data['pulses']
            
            # Check header
            if len(pulses) >= 2:
                header_pulse = pulses[0]
                header_space = pulses[1]
                
                print(f"Header: {header_pulse:.0f}μs pulse, {header_space:.0f}μs space")
                
                # Check if header matches expected timing
                pulse_match = protocol._is_timing_match(header_pulse, protocol.HEADER_PULSE)
                space_match = protocol._is_timing_match(header_space, protocol.HEADER_SPACE)
                
                print(f"  Header pulse match: {pulse_match}")
                print(f"  Header space match: {space_match}")
                
                # Analyze first few data bits
                if len(pulses) >= 6:
                    print("  First data bits:")
                    for i in range(2, min(8, len(pulses)), 2):
                        if i + 1 < len(pulses):
                            pulse_time = pulses[i]
                            space_time = pulses[i + 1]
                            
                            pulse_match = protocol._is_timing_match(pulse_time, protocol.BIT_PULSE)
                            space_0_match = protocol._is_timing_match(space_time, protocol.BIT_0_SPACE)
                            space_1_match = protocol._is_timing_match(space_time, protocol.BIT_1_SPACE)
                            
                            bit_value = "?"
                            if space_0_match:
                                bit_value = "0"
                            elif space_1_match:
                                bit_value = "1"
                            
                            print(f"    Bit {(i-2)//2}: {pulse_time:.0f}μs pulse, {space_time:.0f}μs space -> {bit_value}")
                
                break  # Only analyze first 71-pulse signal

def create_mapping_file():
    """Create a mapping file for the decoded commands"""
    
    # Example mapping based on common IR remote patterns
    mapping = {
        "CUSTOM_0000_0001": {
            "command": "power_toggle",
            "description": "Power on/off",
            "repeatable": True
        },
        "CUSTOM_0000_0002": {
            "command": "channel_1",
            "description": "Channel 1",
            "repeatable": False
        },
        "CUSTOM_0000_0003": {
            "command": "channel_2",
            "description": "Channel 2", 
            "repeatable": False
        },
        "CUSTOM_0000_0004": {
            "command": "channel_3",
            "description": "Channel 3",
            "repeatable": False
        },
        "CUSTOM_0000_0005": {
            "command": "volume_up",
            "description": "Volume up",
            "repeatable": True
        },
        "CUSTOM_0000_0006": {
            "command": "volume_down",
            "description": "Volume down",
            "repeatable": True
        },
        "REPEAT": {
            "command": "repeat_last",
            "description": "Repeat last command",
            "repeatable": False
        }
    }
    
    with open("custom_ir_mapping.json", "w") as f:
        json.dump(mapping, f, indent=2)
    
    print("Created custom_ir_mapping.json with example mappings")

if __name__ == "__main__":
    success = test_decoder_with_captured_data()
    
    if success:
        print("\nDecoder test completed successfully!")
        print("You can now integrate the custom protocol into your IR system.")
        create_mapping_file()
    else:
        print("\nDecoder test failed. Check the timing constants and protocol structure.")
        print("You may need to adjust the timing values in custom_ir_protocol.py")
    
    sys.exit(0 if success else 1)