closed-captioning/web_caption_app.py at main · LBBL96/closed-captioning · GitHub

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
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
#!/usr/bin/env python3
"""
Web-based Real-time Closed Captioning App
Uses microphone input to transcribe speech and display captions in real-time.
Integrates Claude API for enhanced transcription accuracy.
"""

import speech_recognition as sr
import threading
import queue
import time
import os
import json
import tempfile
from anthropic import Anthropic
from flask import Flask, render_template_string, jsonify
from flask_socketio import SocketIO, emit
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

def setup_google_credentials():
    """Create credentials file from environment variables"""
    # Check if credentials are in env vars
    project_id = os.getenv('GOOGLE_PROJECT_ID')
    private_key = os.getenv('GOOGLE_PRIVATE_KEY')
    client_email = os.getenv('GOOGLE_CLIENT_EMAIL')

    if project_id and private_key and client_email:
        # Create temporary credentials file
        creds = {
            "type": "service_account",
            "project_id": project_id,
            "private_key": private_key.replace('\\n', '\n'),
            "client_email": client_email,
            "token_uri": "https://oauth2.googleapis.com/token"
        }

        # Write to temp file
        temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
        json.dump(creds, temp_file)
        temp_file.close()

        # Set environment variable for Google SDK
        os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = temp_file.name
        print(f"Google credentials loaded from environment variables")
        return True
    return False

# Try to set up Google credentials
setup_google_credentials()

class WebCaptionApp:
    def __init__(self):
        # Initialize Flask app
        self.app = Flask(__name__)
        self.app.config['SECRET_KEY'] = 'caption-app-secret'
        self.socketio = SocketIO(self.app, cors_allowed_origins="*")

        # Speech recognition setup
        self.recognizer = sr.Recognizer()
        self.microphone = sr.Microphone()

        # Thread-safe queue for communication
        self.text_queue = queue.Queue()

        # Claude client (will be initialized if API key is provided)
        self.claude_client = None
        self.use_claude = False

        # Audio processing thread
        self.is_running = False
        self.audio_thread = None

        # Setup routes and events
        self.setup_routes()

        # Adjust for ambient noise
        self.calibrate_microphone()

    def setup_routes(self):
        """Setup Flask routes and SocketIO events"""

        @self.app.route('/')
        def index():
            return render_template_string('''
<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Live Closed Captioning</title>
    <style>
        body {
            font-family: Arial, sans-serif;
            background-color: #000;
            color: #fff;
            margin: 0;
            padding: 20px;
            height: 100vh;
            overflow: hidden;
        }
        .container {
            max-width: 800px;
            margin: 0 auto;
            height: 100%;
            display: flex;
            flex-direction: column;
        }
        .header {
            text-align: center;
            margin-bottom: 20px;
        }
        .header h1 {
            font-size: 24px;
            font-weight: bold;
            margin: 0;
        }
        .caption-area {
            height: 300px;
            min-height: 200px;
            max-height: 50vh;
            background-color: #111;
            border: 1px solid #333;
            border-radius: 8px;
            padding: 20px;
            overflow-y: auto;
            font-size: 24px;
            line-height: 1.5;
            margin-bottom: 20px;
        }
        .caption-item {
            margin-bottom: 15px;
            padding: 10px;
            background-color: #222;
            border-radius: 4px;
        }
        .timestamp {
            color: #888;
            font-size: 16px;
            margin-bottom: 5px;
        }
        .controls {
            display: flex;
            gap: 10px;
            justify-content: center;
            align-items: center;
        }
        .btn {
            padding: 12px 24px;
            border: none;
            border-radius: 4px;
            font-size: 16px;
            cursor: pointer;
            transition: background-color 0.3s;
        }
        .btn-start {
            background-color: #2ecc71;
            color: white;
        }
        .btn-stop {
            background-color: #e74c3c;
            color: white;
        }
        .btn-clear {
            background-color: #95a5a6;
            color: white;
        }
        .status {
            color: #95a5a6;
            font-size: 14px;
        }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>Live Captions</h1>
        </div>

        <div class="caption-area" id="captionArea">
            <!-- Captions will appear here -->
        </div>

        <div class="controls">
            <button id="toggleBtn" class="btn btn-start">Start Captioning</button>
            <button id="clearBtn" class="btn btn-clear">Clear</button>
            <span id="status" class="status">Ready</span>
        </div>
    </div>

    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
    <script>
        const socket = io();
        const toggleBtn = document.getElementById('toggleBtn');
        const clearBtn = document.getElementById('clearBtn');
        const status = document.getElementById('status');
        const captionArea = document.getElementById('captionArea');

        let isRunning = false;

        // Socket events
        socket.on('caption', function(data) {
            const captionItem = document.createElement('div');
            captionItem.className = 'caption-item';
            captionItem.innerHTML = `
                <div class="timestamp">[${data.timestamp}]</div>
                <div>${data.text}</div>
            `;
            captionArea.appendChild(captionItem);
            captionArea.scrollTop = captionArea.scrollHeight;
        });

        socket.on('status', function(data) {
            status.textContent = data.status;
        });

        // Button events
        toggleBtn.addEventListener('click', function() {
            if (!isRunning) {
                socket.emit('start');
                toggleBtn.textContent = 'Stop Captioning';
                toggleBtn.className = 'btn btn-stop';
                isRunning = true;
            } else {
                socket.emit('stop');
                toggleBtn.textContent = 'Start Captioning';
                toggleBtn.className = 'btn btn-start';
                isRunning = false;
            }
        });

        clearBtn.addEventListener('click', function() {
            captionArea.innerHTML = '';
            socket.emit('clear');
        });
    </script>
</body>
</html>
            ''')

        @self.socketio.on('start')
        def handle_start():
            self.start_captioning()

        @self.socketio.on('stop')
        def handle_stop():
            self.stop_captioning()

        @self.socketio.on('clear')
        def handle_clear():
            self.clear_captions()

        # Background thread to emit captions
        def emit_captions():
            while True:
                try:
                    text = self.text_queue.get_nowait()
                    timestamp = time.strftime("%H:%M:%S")
                    self.socketio.emit('caption', {'text': text, 'timestamp': timestamp})
                except queue.Empty:
                    pass
                self.socketio.sleep(0.1)

        self.socketio.start_background_task(emit_captions)

    def calibrate_microphone(self):
        """Calibrate microphone for ambient noise"""
        print("Calibrating microphone...")
        with self.microphone as source:
            self.recognizer.adjust_for_ambient_noise(source, duration=1)
        # Wait longer before considering silence as end of phrase
        self.recognizer.pause_threshold = 1.5  # seconds of silence before phrase ends
        self.recognizer.non_speaking_duration = 0.5  # minimum silence length
        print("Microphone calibrated")

    def start_captioning(self):
        """Start the captioning process"""
        if not self.is_running:
            self.is_running = True
            self.socketio.emit('status', {'status': 'Listening...'})

            # Start audio processing thread
            self.audio_thread = threading.Thread(target=self.process_audio, daemon=True)
            self.audio_thread.start()

    def stop_captioning(self):
        """Stop the captioning process"""
        self.is_running = False
        self.socketio.emit('status', {'status': 'Stopped'})

    def clear_captions(self):
        """Clear all captions"""
        # This is handled on the client side
        pass

    def process_audio(self):
        """Process audio in separate thread"""
        print("Audio processing started...")
        while self.is_running:
            try:
                with self.microphone as source:
                    # Listen for audio - 6 seconds captures most sentences
                    audio = self.recognizer.listen(source, timeout=2, phrase_time_limit=6)
                    print(f"Audio captured, length: {len(audio.frame_data)} bytes")

                # Recognize speech using Google's free API
                try:
                    print("Sending to Google Speech API...")
                    text = self.recognizer.recognize_google(audio)
                    print(f"Recognized: {text}")

                    # Put text in queue immediately for real-time display
                    self.text_queue.put(text)

                    # Note: Claude enhancement disabled for speed
                    # To enable, uncomment below (adds ~1s delay per caption)
                    # if self.use_claude and self.claude_client:
                    #     enhanced_text = self.enhance_with_claude(text)
                    #     if enhanced_text:
                    #         self.text_queue.put(f"[Enhanced] {enhanced_text}")

                except sr.UnknownValueError:
                    # Speech was unintelligible
                    print("Could not understand audio")
                except sr.RequestError as e:
                    # API error
                    print(f"Google API error: {e}")
                    self.text_queue.put(f"API Error: {e}")

            except sr.WaitTimeoutError:
                # Timeout, continue listening
                print("Listening timeout, retrying...")
            except Exception as e:
                # Other errors
                print(f"Error in audio processing: {e}")
                self.text_queue.put(f"Error: {e}")

    def enhance_with_claude(self, text):
        """Enhance transcription using Claude API"""
        try:
            message = self.claude_client.messages.create(
                model="claude-haiku-4-5-20250214",
                max_tokens=100,
                messages=[{
                    "role": "user",
                    "content": f"Please correct any transcription errors in this text and return only the corrected version: {text}"
                }]
            )
            return message.content[0].text.strip()
        except Exception as e:
            print(f"Claude API error: {e}")
            return text

    def setup_claude(self, api_key):
        """Setup Claude API client"""
        try:
            self.claude_client = Anthropic(api_key=api_key)
            self.use_claude = True
            print("Claude API initialized successfully")
        except Exception as e:
            print(f"Failed to initialize Claude API: {e}")
            self.use_claude = False

    def run(self, host='127.0.0.1', port=5000, debug=False):
        """Start the web application"""
        # Try to load Claude API key from environment variable
        api_key = os.getenv('ANTHROPIC_API_KEY')
        if api_key:
            self.setup_claude(api_key)
        else:
            print("ANTHROPIC_API_KEY environment variable not found. Running without Claude enhancement.")

        print(f"Starting web captioning app at http://{host}:{port}")
        self.socketio.run(self.app, host=host, port=port, debug=debug)

if __name__ == "__main__":
    app = WebCaptionApp()
    app.run(port=5050, debug=True)