File size: 5,970 Bytes
71e110b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
#!/usr/bin/env python3
"""
TTS Synchronization Tool
Compares word timestamps between espeak and ElevenLabs
"""

import subprocess
import asyncio
import json
import os
import sys
import wave

WHISPER_CLI = "/workspace/MuseTalk1.5/vendor/whisper/build/bin/whisper-cli"
WHISPER_MODEL = "/workspace/MuseTalk1.5/vendor/whisper/models/ggml-large-v3-turbo-q5_0.bin"

def get_audio_duration(filepath):
    """Get audio duration in seconds"""
    if filepath.endswith('.wav'):
        with wave.open(filepath, 'r') as f:
            return f.getnframes() / float(f.getframerate())
    else:
        result = subprocess.run([
            'ffprobe', '-v', 'error', '-show_entries', 'format=duration',
            '-of', 'default=noprint_wrappers=1:nokey=1', filepath
        ], capture_output=True, text=True)
        return float(result.stdout.strip())

def extract_word_timestamps(audio_path):
    """Extract word timestamps using local Whisper"""
    # Convert to wav if needed
    if not audio_path.endswith('.wav'):
        wav_path = audio_path.replace('.mp3', '.wav')
        subprocess.run([
            'ffmpeg', '-y', '-v', 'quiet', '-i', audio_path, '-ar', '16000', '-ac', '1', wav_path
        ], capture_output=True)
        audio_path = wav_path
    
    # Run whisper-cli
    result = subprocess.run([
        WHISPER_CLI, '-m', WHISPER_MODEL, '-f', audio_path,
        '--output-json', '--max-len', '1', '-l', 'pt', '--no-prints'
    ], capture_output=True, text=True)
    
    # Read JSON output
    json_path = audio_path + '.json'
    if os.path.exists(json_path):
        with open(json_path) as f:
            data = json.load(f)
        os.remove(json_path)
        
        words = []
        for item in data.get('transcription', []):
            text = item['text'].strip()
            if text and text not in [',', '.', '?', '!']:
                words.append({
                    'text': text,
                    'start': item['offsets']['from'],
                    'end': item['offsets']['to']
                })
        return words
    return []

def generate_espeak(text, output_path, speed=175):
    """Generate espeak audio"""
    subprocess.run([
        "espeak-ng", "-v", "pt-br", "-s", str(speed), "-w", output_path, text
    ], capture_output=True)
    return get_audio_duration(output_path)

async def generate_elevenlabs(text, output_path):
    """Generate ElevenLabs audio"""
    import httpx
    
    ELEVENLABS_API_KEY = "sk_857e9e6f2412ddf3ff5334b736e4b571641d26225c0d8d62"
    ELEVENLABS_VOICE_ID = "21m00Tcm4TlvDq8ikWAM"
    
    async with httpx.AsyncClient(timeout=60.0) as client:
        response = await client.post(
            f"https://api.elevenlabs.io/v1/text-to-speech/{ELEVENLABS_VOICE_ID}/stream",
            headers={"xi-api-key": ELEVENLABS_API_KEY, "Content-Type": "application/json"},
            json={"text": text, "model_id": "eleven_flash_v2_5", "voice_settings": {"stability": 0.5, "similarity_boost": 0.75}}
        )
        with open(output_path, "wb") as f:
            f.write(response.content)
    return get_audio_duration(output_path)

def generate_espeak_synced(text, output_path, target_duration, base_speed=175):
    """Generate espeak with adjusted speed to match target duration"""
    # Get base duration
    temp_path = "/tmp/espeak_temp.wav"
    subprocess.run([
        "espeak-ng", "-v", "pt-br", "-s", str(base_speed), "-w", temp_path, text
    ], capture_output=True)
    base_duration = get_audio_duration(temp_path)
    
    # Calculate required speed
    required_speed = int(base_speed * (base_duration / target_duration))
    required_speed = max(80, min(450, required_speed))
    
    # Generate with adjusted speed
    subprocess.run([
        "espeak-ng", "-v", "pt-br", "-s", str(required_speed), "-w", output_path, text
    ], capture_output=True)
    
    return get_audio_duration(output_path), required_speed

async def compare_timestamps(text):
    """Compare word timestamps between espeak and ElevenLabs"""
    print(f"\n{'='*70}")
    print(f"COMPARAÇÃO DE TIMESTAMPS: \"{text}\"")
    print('='*70)
    
    # Generate ElevenLabs
    eleven_path = "/tmp/compare_eleven.mp3"
    print("\n1. Gerando ElevenLabs...")
    eleven_dur = await generate_elevenlabs(text, eleven_path)
    print(f"   Duração: {eleven_dur:.2f}s")
    
    # Generate synced espeak
    espeak_path = "/tmp/compare_espeak.wav"
    print("\n2. Gerando espeak sincronizado...")
    espeak_dur, speed = generate_espeak_synced(text, espeak_path, eleven_dur)
    print(f"   Duração: {espeak_dur:.2f}s (speed={speed})")
    
    # Extract timestamps
    print("\n3. Extraindo timestamps com Whisper...")
    eleven_words = extract_word_timestamps(eleven_path)
    espeak_words = extract_word_timestamps(espeak_path)
    
    # Compare
    print("\n" + "-"*70)
    print(f"{'PALAVRA':<15} {'ELEVENLABS':<20} {'ESPEAK':<20} {'DIFF':<10}")
    print("-"*70)
    
    total_diff = 0
    count = 0
    
    for i, ew in enumerate(eleven_words):
        if i < len(espeak_words):
            sw = espeak_words[i]
            diff = abs(ew['start'] - sw['start'])
            total_diff += diff
            count += 1
            
            print(f"{ew['text']:<15} {ew['start']:>5}-{ew['end']:<5} ms    {sw['start']:>5}-{sw['end']:<5} ms    {diff:>5} ms")
    
    print("-"*70)
    
    if count > 0:
        avg_diff = total_diff / count
        print(f"\nMÉDIA DE DIFERENÇA: {avg_diff:.0f} ms")
        
        if avg_diff < 50:
            print("✓ EXCELENTE SINCRONIZAÇÃO!")
        elif avg_diff < 100:
            print("✓ BOA SINCRONIZAÇÃO")
        elif avg_diff < 200:
            print("⚠ SINCRONIZAÇÃO ACEITÁVEL")
        else:
            print("✗ SINCRONIZAÇÃO RUIM")
    
    return eleven_words, espeak_words

if __name__ == "__main__":
    text = sys.argv[1] if len(sys.argv) > 1 else "Olá, tudo bem? Como posso te ajudar?"
    asyncio.run(compare_timestamps(text))