Upload debug_protobuf_structure.py

debug_protobuf_structure.py

@@ -0,0 +1,320 @@
+#!/usr/bin/env python3
+"""
+Debug the exact protobuf structure needed by CoreML
+"""
+
+import struct
+import binascii
+from pathlib import Path
+
+def debug_protobuf_structure():
+    """Debug CoreML protobuf structure by analyzing working models"""
+    print("🔍 DEBUGGING COREML PROTOBUF STRUCTURE")
+    print("=" * 60)
+    
+    # Analyze working models to understand the exact structure
+    working_models = [
+        "/Users/kikow/Library/Application Support/FluidAudio/vad/silero_stft.mlmodel",
+        "/Users/kikow/Library/Application Support/FluidAudio/vad/silero_encoder.mlmodel", 
+        "/Users/kikow/Library/Application Support/FluidAudio/vad/silero_rnn_decoder.mlmodel"
+    ]
+    
+    for model_path in working_models:
+        if Path(model_path).exists():
+            print(f"\n🔬 Analyzing: {Path(model_path).name}")
+            analyze_model_structure(model_path)
+        else:
+            print(f"❌ Model not found: {model_path}")
+    
+    # Compare with our broken model
+    broken_model = "/Users/kikow/Library/Application Support/FluidAudio/vad/correct_classifier_conv1d.mlmodel"
+    if Path(broken_model).exists():
+        print(f"\n🔬 Analyzing BROKEN: {Path(broken_model).name}")
+        analyze_model_structure(broken_model)
+
+def analyze_model_structure(model_path):
+    """Analyze the protobuf structure of a CoreML model"""
+    with open(model_path, 'rb') as f:
+        data = f.read()
+    
+    print(f"📊 File size: {len(data)} bytes")
+    
+    # Analyze protobuf header
+    print("🔍 Protobuf Header Analysis:")
+    analyze_protobuf_header(data[:200])
+    
+    # Look for ModelDescription section
+    print("\n🔍 ModelDescription Analysis:")
+    desc_section = find_model_description(data)
+    if desc_section:
+        analyze_model_description(desc_section)
+    
+    # Look for input/output names
+    print("\n🔍 Input/Output Analysis:")
+    find_input_output_names(data)
+    
+    # Analyze neural network section
+    print("\n🔍 Neural Network Section:")
+    nn_section = find_neural_network_section(data)
+    if nn_section:
+        analyze_neural_network_section(nn_section[:500])  # First 500 bytes
+
+def analyze_protobuf_header(header_data):
+    """Analyze the protobuf header"""
+    print(f"First 50 bytes (hex): {header_data[:50].hex()}")
+    
+    # Decode protobuf fields
+    pos = 0
+    field_count = 0
+    
+    while pos < len(header_data) and field_count < 10:
+        try:
+            # Read field tag
+            field_tag = header_data[pos]
+            wire_type = field_tag & 0x07
+            field_number = field_tag >> 3
+            
+            print(f"  Field {field_number}, Wire Type {wire_type} at pos {pos}")
+            
+            if wire_type == 0:  # Varint
+                pos += 1
+                value, consumed = decode_varint(header_data[pos:])
+                print(f"    Varint value: {value}")
+                pos += consumed
+            elif wire_type == 2:  # Length-delimited
+                pos += 1
+                length, consumed = decode_varint(header_data[pos:])
+                pos += consumed
+                print(f"    Length-delimited, length: {length}")
+                if pos + length <= len(header_data):
+                    content = header_data[pos:pos+length]
+                    print(f"    Content preview: {content[:20]}")
+                pos += length
+            else:
+                pos += 1
+            
+            field_count += 1
+            
+        except:
+            break
+
+def decode_varint(data):
+    """Decode protobuf varint"""
+    value = 0
+    shift = 0
+    pos = 0
+    
+    while pos < len(data):
+        byte = data[pos]
+        value |= (byte & 0x7F) << shift
+        pos += 1
+        if (byte & 0x80) == 0:
+            break
+        shift += 7
+    
+    return value, pos
+
+def find_model_description(data):
+    """Find the ModelDescription section"""
+    # ModelDescription is typically field 2 in the top-level message
+    # Look for field tag 0x12 (field 2, wire type 2)
+    
+    for i in range(len(data) - 10):
+        if data[i] == 0x12:  # Field 2, length-delimited
+            try:
+                length, consumed = decode_varint(data[i+1:])
+                if length > 0 and length < 10000:  # Reasonable size
+                    start = i + 1 + consumed
+                    end = start + length
+                    if end <= len(data):
+                        print(f"Found potential ModelDescription at {i}, length: {length}")
+                        return data[start:end]
+            except:
+                continue
+    
+    return None
+
+def analyze_model_description(desc_data):
+    """Analyze ModelDescription protobuf data"""
+    print(f"ModelDescription size: {len(desc_data)} bytes")
+    print(f"First 100 bytes: {desc_data[:100].hex()}")
+    
+    # Look for string patterns (input/output names)
+    strings = extract_strings(desc_data)
+    print(f"Strings found: {strings[:10]}")  # First 10 strings
+
+def find_input_output_names(data):
+    """Find input and output names in the model"""
+    # Look for common input/output names
+    common_names = [
+        b'audio_input', b'rnn_features', b'encoder_features', b'stft_features',
+        b'vad_probability', b'output', b'var_', b'h_out', b'c_out'
+    ]
+    
+    for name in common_names:
+        positions = []
+        start = 0
+        while True:
+            pos = data.find(name, start)
+            if pos == -1:
+                break
+            positions.append(pos)
+            start = pos + 1
+        
+        if positions:
+            print(f"  '{name.decode()}' found at positions: {positions[:5]}")
+
+def find_neural_network_section(data):
+    """Find the neural network section"""
+    # Neural network is typically field 200 (0xC8, 0x01) or similar
+    # Look for field tags that might indicate neural network
+    
+    patterns = [
+        b'\x82\x0C',  # Field 200, length-delimited
+        b'\xA2\x06',  # Field 100, length-delimited
+        b'\x82\x06'   # Field 100, length-delimited
+    ]
+    
+    for pattern in patterns:
+        pos = data.find(pattern)
+        if pos != -1:
+            print(f"Found potential neural network section at {pos}")
+            try:
+                length, consumed = decode_varint(data[pos+2:])
+                start = pos + 2 + consumed
+                end = start + length
+                if end <= len(data):
+                    return data[start:end]
+            except:
+                continue
+    
+    return None
+
+def analyze_neural_network_section(nn_data):
+    """Analyze neural network section"""
+    print(f"Neural network size: {len(nn_data)} bytes")
+    print(f"First 100 bytes: {nn_data[:100].hex()}")
+
+def extract_strings(data):
+    """Extract readable strings from binary data"""
+    strings = []
+    current_string = b""
+    
+    for byte in data:
+        if 32 <= byte <= 126:  # Printable ASCII
+            current_string += bytes([byte])
+        else:
+            if len(current_string) >= 3:  # Minimum length
+                try:
+                    strings.append(current_string.decode('utf-8'))
+                except:
+                    pass
+            current_string = b""
+    
+    # Add final string if any
+    if len(current_string) >= 3:
+        try:
+            strings.append(current_string.decode('utf-8'))
+        except:
+            pass
+    
+    return strings
+
+def create_corrected_protobuf():
+    """Create a corrected protobuf based on analysis"""
+    print("\n🔧 Creating corrected protobuf based on analysis...")
+    
+    # Use the STFT model as the most reliable template
+    stft_path = "/Users/kikow/Library/Application Support/FluidAudio/vad/silero_stft.mlmodel"
+    output_path = "/Users/kikow/Library/Application Support/FluidAudio/vad/correct_classifier_conv1d.mlmodel"
+    
+    if not Path(stft_path).exists():
+        print("❌ STFT template not available")
+        return False
+    
+    with open(stft_path, 'rb') as f:
+        template_data = bytearray(f.read())
+    
+    print(f"📊 Template size: {len(template_data)} bytes")
+    
+    # Step 1: Replace input name
+    if b'audio_input' in template_data:
+        # Replace with rnn_features (same length to avoid shifting offsets)
+        template_data = template_data.replace(b'audio_input', b'rnn_feature')
+        print("✅ Replaced audio_input with rnn_feature")
+    
+    # Step 2: Find and replace output names
+    output_patterns = [b'var_71', b'output']
+    for pattern in output_patterns:
+        if pattern in template_data:
+            # Replace with vad_probability (pad/truncate as needed)
+            if len(pattern) >= len(b'vad_prob'):
+                replacement = b'vad_prob' + b'\x00' * (len(pattern) - len(b'vad_prob'))
+            else:
+                replacement = b'vad_prob'[:len(pattern)]
+            template_data = template_data.replace(pattern, replacement)
+            print(f"✅ Replaced {pattern} with {replacement}")
+    
+    # Step 3: Try to modify tensor shapes in the protobuf
+    # Look for shape patterns and try to modify them
+    modify_tensor_shapes(template_data)
+    
+    # Write the corrected model
+    with open(output_path, 'wb') as f:
+        f.write(template_data)
+    
+    print(f"✅ Created corrected protobuf: {output_path}")
+    return True
+
+def modify_tensor_shapes(data):
+    """Try to modify tensor shapes in protobuf data"""
+    # Look for shape patterns - CoreML often encodes shapes as sequences of varints
+    # We need to change from audio input shape to RNN features shape
+    
+    # Common shape patterns (as varints)
+    # [512] -> [1, 4, 128]
+    # This is very tricky without exact protobuf structure knowledge
+    
+    print("⚠️  Tensor shape modification is complex - keeping original shapes")
+
+def test_corrected_model():
+    """Test the corrected model"""
+    print("\n🧪 Testing corrected model...")
+    
+    import os
+    result = os.system("cd /Users/kikow/brandon/FluidAudioSwift && swift run fluidaudio vad-benchmark --num-files 1")
+    
+    if result == 0:
+        print("✅ Model test completed")
+        return True
+    else:
+        print("❌ Model test failed")
+        return False
+
+def main():
+    print("🚀 COREML PROTOBUF DEBUGGING")
+    print("=" * 50)
+    
+    try:
+        # Step 1: Debug existing models
+        debug_protobuf_structure()
+        
+        # Step 2: Create corrected protobuf
+        if create_corrected_protobuf():
+            print("\n✅ Corrected protobuf created")
+            
+            # Step 3: Test it
+            if test_corrected_model():
+                print("\n🎉 SUCCESS! Corrected model works")
+            else:
+                print("\n⚠️ Model created but needs further refinement")
+        else:
+            print("\n❌ Failed to create corrected protobuf")
+            
+    except Exception as e:
+        print(f"❌ Error: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    main()