Update app_parallel.py

app_parallel.py

@@ -72,6 +72,7 @@ app = Flask(__name__)
 
 TEMP_DIR = None
 start_time = None
+future_to_chunk = {}
 
 app.config['temp_response'] = None
 app.config['generation_thread'] = None
@@ -273,7 +274,7 @@ def split_audio(audio_path, chunk_duration):
 
 # Generator function to yield chunk results as they are processed
 def generate_chunks(audio_chunks, preprocessed_data, args):
-    future_to_chunk = {executor.submit(process_chunk, chunk[1], preprocessed_data, args): chunk[0] for chunk in audio_chunks}
+    # future_to_chunk = {executor.submit(process_chunk, chunk[1], preprocessed_data, args): chunk[0] for chunk in audio_chunks}
     
     for future in as_completed(future_to_chunk):
         idx = future_to_chunk[future]  # Get the original chunk that was processed
@@ -349,13 +350,28 @@ def parallel_processing():
     print(f"Splitting the audio into {chunk_duration}-second chunks...")
     audio_chunks = split_audio(driven_audio_path, chunk_duration=chunk_duration)
     print(f"Audio has been split into {len(audio_chunks)} chunks: {audio_chunks}")
-  
-    try:
-        return stream_with_context(generate_chunks(audio_chunks, preprocessed_data, args))
-        # base64_video, temp_file_path, duration = process_chunk(driven_audio_path, preprocessed_data, args)
-    except Exception as e:
-        return jsonify({'status': 'error', 'message': str(e)}), 500
 
+    future_to_chunk = {executor.submit(process_chunk, chunk[1], preprocessed_data, args): chunk[0] for chunk in audio_chunks}
+    return jsonify({"status": "processing started"}), 200
+    # try:
+    #     return stream_with_context(generate_chunks(audio_chunks, preprocessed_data, args))
+    #     # base64_video, temp_file_path, duration = process_chunk(driven_audio_path, preprocessed_data, args)
+    # except Exception as e:
+    #     return jsonify({'status': 'error', 'message': str(e)}), 500
+
+@app.route("/stream", methods=["GET"])
+def stream_results():
+    global future_to_chunk
+     def generate():
+        for future in as_completed(future_results):
+            idx = future_results.index(future)
+            try:
+                base64_video, temp_file_path = future.result()
+                yield json.dumps({'start_time': idx, 'path': temp_file_path}).encode('utf-8')
+            except Exception as e:
+                yield json.dumps({'start_time': idx, 'error': str(e)}).encode('utf-8')
+
+    return stream_with_context(generate())
 
 @app.route("/health", methods=["GET"])
 def health_status():