Update app_parallel.py

app_parallel.py

@@ -26,7 +26,6 @@ import pickle
 from concurrent.futures import ProcessPoolExecutor, as_completed, ThreadPoolExecutor
 from stream_server import add_video,HLS_DIR, generate_m3u8
 import math
-import re
 
 # Load environment variables from .env file
 # load_dotenv()
@@ -287,35 +286,38 @@ def split_audio(audio_path, TEMP_DIR, chunk_duration):
     
     audio_clip.close()  # Close the audio clip to release resources
     return audio_chunks, total_duration
-def extract_order_from_path(temp_file_path):
-    match = re.search(r'videostream(\d+)', temp_file_path)
-    return int(match.group(1)) if match else -1  # Return -1 if no match is found, handle appropriately.
+    
+# def extract_order_from_path(temp_file_path):
+#     match = re.search(r'videostream(\d+)', temp_file_path)
+#     return int(match.group(1)) if match else -1  # Return -1 if no match is found, handle appropriately.
 # Generator function to yield chunk results as they are processed
+
 def generate_chunks(audio_chunks, preprocessed_data, args, m3u8_path, audio_duration):
     global TEMP_DIR
-    next_expected_order = 1 
-    completed_chunks = {}
     future_to_chunk = {executor.submit(process_chunk, chunk[1], preprocessed_data, args): chunk[0] for chunk in audio_chunks}
-    
+    processed_chunks = {chunk[0]: None for chunk in audio_chunks}
+    print("processed_chunks:",processed_chunks)
+    yielded_count = 1
     try:
         for chunk_idx, future in enumerate(as_completed(future_to_chunk)):
-            idx = future_to_chunk[future]  # Get the original chunk that was processed
+            idx = future_to_chunk[future]
             try:
-                base64_video, temp_file_path = future.result()  # Get the result of the completed task
-                # Extract the order number from the temp_file_path
-                order_number = extract_order_from_path(temp_file_path)
-                completed_chunks[order_number] = (temp_file_path)
-                # add video for streaming
-                while next_expected_order in completed_chunks:
-                    temp_file_path = completed_chunks.pop(next_expected_order)
-                    asyncio.run(add_video(temp_file_path, m3u8_path, audio_duration))
-                    yield json.dumps({'start_time': next_expected_order, 'video_index': next_expected_order}).encode('utf-8')
-                    next_expected_order += 1
+                base64_video, temp_file_path = future.result()
+                processed_chunks[idx] = temp_file_path
+                for expected_start_time in sorted(processed_chunks.keys()):
+                    if processed_chunks[expected_start_time] is not None:
+                        asyncio.run(add_video(processed_chunks[expected_start_time], m3u8_path, audio_duration))
+                        yield json.dumps({'start_time': expected_start_time, 'video_index': yielded_count}).encode('utf-8')
+                        processed_chunks[expected_start_time] = None
+                        yielded_count += 1
+                    else:
+                        break
             except Exception as e:
                 yield f"Task for chunk {idx} failed: {e}\n"
     finally:
         if TEMP_DIR:
             custom_cleanup(TEMP_DIR.name)
+
             
 @app.route("/run", methods=['POST'])
 def parallel_processing():