44
55import cv2
66import concurrent .futures
7+ import os
8+ import sys
9+ import time
10+ import threading
711from multiprocessing import cpu_count
812from typing import List , Callable , Optional
913from tqdm import tqdm
@@ -27,6 +31,20 @@ def __init__(self, max_workers: Optional[int] = None):
2731 max_workers: Maximum number of worker threads. If None, uses cpu_count().
2832 """
2933 self .max_workers = max_workers or cpu_count ()
34+ self ._is_windows = os .name == 'nt' or sys .platform .startswith ('win' )
35+ self ._cancellation_event = threading .Event ()
36+
37+ # Windows-specific settings
38+ if self ._is_windows :
39+ # Reduce thread count on Windows to avoid thread pool issues
40+ self .max_workers = min (self .max_workers , 4 )
41+ # Add timeouts for Windows operations
42+ self ._operation_timeout = 300 # 5 minutes
43+ self ._progress_timeout = 30 # 30 seconds
44+
45+ def cancel_processing (self ):
46+ """Cancel ongoing sharpness analysis."""
47+ self ._cancellation_event .set ()
3048
3149 def calculate_sharpness (self , extraction_result : ExtractionResult , progress_callback = None ) -> ExtractionResult :
3250 """
@@ -110,7 +128,7 @@ def chunk_progress_callback(phase, chunk_completed, chunk_total, description):
110128
111129 def _process_chunk_parallel (self , frame_paths : List [str ],
112130 progress_callback : Optional [Callable ] = None ) -> List [float ]:
113- """Process a chunk of frames in parallel."""
131+ """Process a chunk of frames in parallel with Windows-specific error handling ."""
114132 scores = [0.0 ] * len (frame_paths ) # Initialize with default scores
115133 num_workers = min (self .max_workers , len (frame_paths )) if frame_paths else 1
116134
@@ -120,39 +138,115 @@ def _process_chunk_parallel(self, frame_paths: List[str],
120138
121139 futures = {}
122140 try :
123- with concurrent .futures .ThreadPoolExecutor (max_workers = num_workers ) as executor :
141+ # Windows-specific executor configuration
142+ executor_kwargs = {'max_workers' : num_workers }
143+ if self ._is_windows :
144+ # Use context manager with explicit timeout on Windows
145+ executor_kwargs .update ({
146+ 'thread_name_prefix' : 'SharpFrames_' ,
147+ })
148+
149+ with concurrent .futures .ThreadPoolExecutor (** executor_kwargs ) as executor :
124150 # Submit tasks
125151 for idx , path in enumerate (frame_paths ):
152+ # Check for cancellation before submitting
153+ if self ._cancellation_event .is_set ():
154+ print ("Sharpness analysis cancelled" )
155+ return scores
156+
126157 future = executor .submit (self ._calculate_single_frame_sharpness , path )
127158 futures [future ] = {"index" : idx , "path" : path }
128159
129- # Process completed futures
130- for future in concurrent .futures .as_completed (futures ):
131- task_info = futures [future ]
132- idx = task_info ["index" ]
133- path = task_info ["path" ]
134-
135- try :
136- score = future .result ()
137- scores [idx ] = score
160+ # Process completed futures with timeout on Windows
161+ timeout = self ._operation_timeout if self ._is_windows else None
162+
163+ try :
164+ for future in concurrent .futures .as_completed (futures , timeout = timeout ):
165+ # Check for cancellation
166+ if self ._cancellation_event .is_set ():
167+ print ("Sharpness analysis cancelled" )
168+ # Cancel remaining futures
169+ for remaining_future in futures :
170+ remaining_future .cancel ()
171+ return scores
172+
173+ task_info = futures [future ]
174+ idx = task_info ["index" ]
175+ path = task_info ["path" ]
138176
139- except Exception as e :
140- print (f"Warning: Failed to process { path } { e } )
141- scores [idx ] = 0.0 # Use default score for failed frames
142-
143- # Update progress after each frame
144- completed_count += 1
145- if progress_callback :
146- progress_callback ("sharpness" , completed_count , total_frames ,
147- f"Analyzed { completed_count } { total_frames } )
177+ try :
178+ # Add timeout for individual result retrieval on Windows
179+ result_timeout = 10 if self ._is_windows else None
180+ score = future .result (timeout = result_timeout )
181+ scores [idx ] = score
182+
183+ except concurrent .futures .TimeoutError :
184+ print (f"Warning: Timeout processing { path } )
185+ scores [idx ] = 0.0
186+ except Exception as e :
187+ print (f"Warning: Failed to process { path } { e } )
188+ scores [idx ] = 0.0 # Use default score for failed frames
189+
190+ # Update progress after each frame with timeout protection
191+ completed_count += 1
192+ if progress_callback :
193+ try :
194+ progress_callback ("sharpness" , completed_count , total_frames ,
195+ f"Analyzed { completed_count } { total_frames } )
196+ except Exception as e :
197+ print (f"Warning: Progress callback failed: { e } )
198+
199+ except concurrent .futures .TimeoutError :
200+ print (f"Warning: Overall processing timeout after { timeout } )
201+ # Cancel remaining futures and return partial results
202+ for future in futures :
203+ future .cancel ()
148204
149205 except Exception as e :
150- print (f"Error in parallel processing: { e } )
206+ error_msg = f"Error in parallel processing: { e }
207+ print (error_msg )
208+
209+ # On Windows, try fallback to sequential processing for small datasets
210+ if self ._is_windows and len (frame_paths ) <= 10 :
211+ print ("Attempting fallback to sequential processing on Windows..." )
212+ return self ._process_chunk_sequential (frame_paths , progress_callback )
151213 # Return zeros for all frames on complete failure
152214 return [0.0 ] * len (frame_paths )
153215
154216 return scores
155217
218+ def _process_chunk_sequential (self , frame_paths : List [str ],
219+ progress_callback : Optional [Callable ] = None ) -> List [float ]:
220+ """Fallback sequential processing for Windows when threading fails."""
221+ scores = []
222+ total_frames = len (frame_paths )
223+
224+ for idx , path in enumerate (frame_paths ):
225+ if self ._cancellation_event .is_set ():
226+ print ("Sequential processing cancelled" )
227+ break
228+
229+ try :
230+ score = self ._calculate_single_frame_sharpness (path )
231+ scores .append (score )
232+ except Exception as e :
233+ print (f"Warning: Failed to process { path } { e } )
234+ scores .append (0.0 )
235+
236+ # Update progress
237+ if progress_callback :
238+ try :
239+ progress_callback ("sharpness" , idx + 1 , total_frames ,
240+ f"Analyzed { idx + 1 } { total_frames } )
241+ except Exception as e :
242+ print (f"Warning: Progress callback failed: { e } )
243+
244+ # Pad with zeros if processing was cancelled early
245+ while len (scores ) < total_frames :
246+ scores .append (0.0 )
247+
248+ return scores
249+
156250 def _calculate_single_frame_sharpness (self , frame_path : str ) -> float :
157251 """
158252 Calculate sharpness score for a single frame using Laplacian variance.
@@ -167,6 +261,10 @@ def _calculate_single_frame_sharpness(self, frame_path: str) -> float:
167261 ImageProcessingError: If frame processing fails
168262 """
169263 try :
264+ # Check for cancellation
265+ if self ._cancellation_event .is_set ():
266+ return 0.0
267+
170268 # Read image in grayscale
171269 img_gray = cv2 .imread (frame_path , cv2 .IMREAD_GRAYSCALE )
172270 if img_gray is None :
0 commit comments