Update app.py

app.py

@@ -11,6 +11,10 @@ from concurrent.futures import ThreadPoolExecutor
 from typing import List, Tuple, Optional, Dict, Any
 import math
 from dataclasses import dataclass
+import multiprocessing
+import psutil
+import concurrent.futures
+import gc
 
 class TimingManager:
     def __init__(self):
@@ -189,8 +193,27 @@ class TTSError(Exception):
     """Custom exception for TTS processing errors"""
     pass
 
+class ResourceOptimizer:
+    @staticmethod
+    def get_optimal_workers():
+        cpu_count = multiprocessing.cpu_count()
+        return max(cpu_count - 1, 1)  # Leave one core for system
+    
+    @staticmethod
+    def get_memory_limit():
+        # Use up to 70% of available RAM
+        return int(psutil.virtual_memory().available * 0.7)
+    
+    @staticmethod
+    def get_batch_size(total_segments):
+        # Calculate optimal batch size based on CPU cores
+        return min(total_segments, ResourceOptimizer.get_optimal_workers() * 2)
+
 async def process_segment_with_timing(segment: Segment, voice: str, rate: str, pitch: str) -> Segment:
     """Process a complete segment as a single TTS unit with improved error handling"""
+    # Pre-allocate memory for audio processing
+    gc.collect()  # Force garbage collection before processing
+    
     audio_file = os.path.join(tempfile.gettempdir(), f"temp_segment_{segment.id}_{uuid.uuid4()}.wav")
     try:
         # Process the entire segment text as one unit, replacing newlines with spaces
@@ -207,7 +230,8 @@ async def process_segment_with_timing(segment: Segment, voice: str, rate: str, p
         
         try:
             segment.audio = AudioSegment.from_file(audio_file)
-            # Reduced silence to 30ms for more natural flow
+            # Optimize memory usage for audio processing
+            segment.audio = segment.audio.set_channels(1)  # Convert to mono for memory efficiency
             silence = AudioSegment.silent(duration=30)
             segment.audio = silence + segment.audio + silence
             segment.duration = len(segment.audio)
@@ -215,16 +239,12 @@ async def process_segment_with_timing(segment: Segment, voice: str, rate: str, p
             raise TTSError(f"Failed to process audio file for segment {segment.id}: {str(e)}")
         
         return segment
-    except Exception as e:
-        if not isinstance(e, TTSError):
-            raise TTSError(f"Unexpected error processing segment {segment.id}: {str(e)}")
-        raise
     finally:
         if os.path.exists(audio_file):
             try:
                 os.remove(audio_file)
             except Exception:
-                pass  # Ignore deletion errors
+                pass
 
 # IMPROVEMENT 2: Better File Management with cleanup
 class FileManager:
@@ -294,56 +314,45 @@ async def generate_accurate_srt(
     lines_per_segment: int,
     progress_callback=None,
     parallel: bool = True,
-    max_workers: int = 4
+    max_workers: Optional[int] = None
 ) -> Tuple[str, str]:
-    """Generate accurate SRT with parallel processing option"""
+    """Generate accurate SRT with optimized resource utilization"""
     processor = TextProcessor(words_per_line, lines_per_segment)
     segments = processor.split_into_segments(text)
-    
     total_segments = len(segments)
-    processed_segments = []
     
-    # Update progress to show segmentation is complete
-    if progress_callback:
-        progress_callback(0.1, "Text segmentation complete")
+    # Optimize worker count based on system resources
+    if max_workers is None:
+        max_workers = ResourceOptimizer.get_optimal_workers()
     
     if parallel and total_segments > 1:
-        # Process segments in parallel
-        processed_count = 0
-        segment_tasks = []
-        
-        # Create a semaphore to limit concurrent tasks
+        # Enhanced parallel processing with resource optimization
+        batch_size = ResourceOptimizer.get_batch_size(total_segments)
         semaphore = asyncio.Semaphore(max_workers)
+        processed_segments = []
+        processed_count = 0
         
-        async def process_with_semaphore(segment):
-            async with semaphore:
-                nonlocal processed_count
-                try:
-                    result = await process_segment_with_timing(segment, voice, rate, pitch)
-                    processed_count += 1
-                    if progress_callback:
-                        progress = 0.1 + (0.8 * processed_count / total_segments)
-                        progress_callback(progress, f"Processed {processed_count}/{total_segments} segments")
-                    return result
-                except Exception as e:
-                    # Handle errors in individual segments
-                    processed_count += 1
-                    if progress_callback:
-                        progress = 0.1 + (0.8 * processed_count / total_segments)
-                        progress_callback(progress, f"Error in segment {segment.id}: {str(e)}")
-                    raise
-        
-        # Create tasks for all segments
-        for segment in segments:
-            segment_tasks.append(process_with_semaphore(segment))
-        
-        # Run all tasks and collect results
-        try:
-            processed_segments = await asyncio.gather(*segment_tasks)
-        except Exception as e:
+        # Process in batches for better resource utilization
+        for i in range(0, total_segments, batch_size):
+            batch = segments[i:i + batch_size]
+            batch_tasks = []
+            
+            for segment in batch:
+                batch_tasks.append(
+                    process_with_semaphore(segment, voice, rate, pitch, semaphore)
+                )
+            
+            # Process batch with maximum resource utilization
+            batch_results = await asyncio.gather(*batch_tasks)
+            processed_segments.extend(batch_results)
+            
+            # Force garbage collection between batches
+            gc.collect()
+            
             if progress_callback:
-                progress_callback(0.9, f"Error during parallel processing: {str(e)}")
-            raise TTSError(f"Failed during parallel processing: {str(e)}")
+                processed_count += len(batch)
+                progress = 0.1 + (0.8 * processed_count / total_segments)
+                progress_callback(progress, f"Processed {processed_count}/{total_segments} segments")
     else:
         # Process segments sequentially (original method)
         for i, segment in enumerate(segments):
@@ -417,6 +426,10 @@ async def generate_accurate_srt(
     
     return srt_path, audio_path
 
+async def process_with_semaphore(segment, voice, rate, pitch, semaphore):
+    async with semaphore:
+        return await process_segment_with_timing(segment, voice, rate, pitch)
+
 # IMPROVEMENT 4: Progress Reporting with proper error handling for older Gradio versions
 async def process_text_with_progress(
     text, 
@@ -601,4 +614,10 @@ with gr.Blocks(title="Advanced TTS with Configurable SRT Generation") as app:
     )
 
 if __name__ == "__main__":
+    # Set process priority to high
+    p = psutil.Process()
+    try:
+        p.nice(psutil.BELOW_NORMAL_PRIORITY_CLASS if os.name == 'nt' else 10)
+    except Exception:
+        pass
     app.launch()