MogensR commited on
Commit
b157fef
·
1 Parent(s): a00a1ac

Update video_processor.py

Browse files
Files changed (1) hide show
  1. video_processor.py +18 -17
video_processor.py CHANGED
@@ -12,10 +12,11 @@
12
  from typing import Optional, Tuple, Dict, Any, Callable
13
  from pathlib import Path
14
 
15
- from app_config import ProcessingConfig
16
- from memory_manager import MemoryManager
17
- from progress_tracker import ProgressTracker
18
- from exceptions import VideoProcessingError, VideoFileError, BackgroundProcessingError, SegmentationError
 
19
 
20
  # Import utilities
21
  from utilities import (
@@ -35,11 +36,11 @@ class CoreVideoProcessor:
35
  """
36
 
37
  def __init__(self, sam2_predictor: Any, matanyone_model: Any,
38
- config: ProcessingConfig, memory_manager: MemoryManager):
39
  self.sam2_predictor = sam2_predictor
40
  self.matanyone_model = matanyone_model
41
  self.config = config
42
- self.memory_manager = memory_manager
43
 
44
  # Processing state
45
  self.processing_active = False
@@ -245,7 +246,7 @@ def _process_video_frames(
245
  """Process all video frames"""
246
 
247
  # Initialize progress tracking
248
- progress_tracker = ProgressTracker(
249
  total_frames=video_info['total_frames'],
250
  callback=progress_callback,
251
  track_performance=True
@@ -260,7 +261,7 @@ def _process_video_frames(
260
  self.frame_cache.clear()
261
 
262
  try:
263
- progress_tracker.set_stage("Processing frames")
264
 
265
  while True:
266
  # Check for cancellation
@@ -280,7 +281,7 @@ def _process_video_frames(
280
 
281
  try:
282
  # Update progress
283
- progress_tracker.update(frame_count, "Processing frame")
284
 
285
  # Process frame
286
  processed_frame = self._process_single_frame(
@@ -315,7 +316,7 @@ def _process_video_frames(
315
  frame_count += 1
316
 
317
  # Finalize progress tracking
318
- final_stats = progress_tracker.finalize()
319
 
320
  return {
321
  'success': successful_frames > 0,
@@ -376,13 +377,13 @@ def _segment_person(self, frame: np.ndarray, frame_number: int) -> np.ndarray:
376
  mask = segment_person_hq(frame, self.sam2_predictor)
377
 
378
  if mask is None or mask.size == 0:
379
- raise SegmentationError(frame_number, "Segmentation returned empty mask")
380
 
381
  return mask
382
 
383
  except Exception as e:
384
  self.stats['segmentation_errors'] += 1
385
- raise SegmentationError(frame_number, f"Segmentation failed: {str(e)}")
386
 
387
  def _should_refine_mask(self, frame_number: int) -> bool:
388
  """Determine if mask should be refined for this frame"""
@@ -524,18 +525,18 @@ def prepare_background(
524
  try:
525
  if background_choice == "custom" and custom_background_path:
526
  if not os.path.exists(custom_background_path):
527
- raise BackgroundProcessingError("custom", f"File not found: {custom_background_path}")
528
 
529
  background = cv2.imread(custom_background_path)
530
  if background is None:
531
- raise BackgroundProcessingError("custom", "Could not read custom background image")
532
 
533
  logger.info(f"Loaded custom background: {custom_background_path}")
534
 
535
  else:
536
  # Use professional background
537
  if background_choice not in PROFESSIONAL_BACKGROUNDS:
538
- raise BackgroundProcessingError(background_choice, "Unknown professional background")
539
 
540
  bg_config = PROFESSIONAL_BACKGROUNDS[background_choice]
541
  background = create_professional_background(bg_config, width, height)
@@ -548,12 +549,12 @@ def prepare_background(
548
 
549
  # Validate background
550
  if background is None or background.size == 0:
551
- raise BackgroundProcessingError(background_choice, "Background image is empty")
552
 
553
  return background
554
 
555
  except Exception as e:
556
- if isinstance(e, BackgroundProcessingError):
557
  logger.error(str(e))
558
  return None
559
  else:
 
12
  from typing import Optional, Tuple, Dict, Any, Callable
13
  from pathlib import Path
14
 
15
+ # Import modular components
16
+ import app_config
17
+ import memory_manager
18
+ import progress_tracker
19
+ import exceptions
20
 
21
  # Import utilities
22
  from utilities import (
 
36
  """
37
 
38
  def __init__(self, sam2_predictor: Any, matanyone_model: Any,
39
+ config: app_config.ProcessingConfig, memory_mgr: memory_manager.MemoryManager):
40
  self.sam2_predictor = sam2_predictor
41
  self.matanyone_model = matanyone_model
42
  self.config = config
43
+ self.memory_manager = memory_mgr
44
 
45
  # Processing state
46
  self.processing_active = False
 
246
  """Process all video frames"""
247
 
248
  # Initialize progress tracking
249
+ prog_tracker = progress_tracker.ProgressTracker(
250
  total_frames=video_info['total_frames'],
251
  callback=progress_callback,
252
  track_performance=True
 
261
  self.frame_cache.clear()
262
 
263
  try:
264
+ prog_tracker.set_stage("Processing frames")
265
 
266
  while True:
267
  # Check for cancellation
 
281
 
282
  try:
283
  # Update progress
284
+ prog_tracker.update(frame_count, "Processing frame")
285
 
286
  # Process frame
287
  processed_frame = self._process_single_frame(
 
316
  frame_count += 1
317
 
318
  # Finalize progress tracking
319
+ final_stats = prog_tracker.finalize()
320
 
321
  return {
322
  'success': successful_frames > 0,
 
377
  mask = segment_person_hq(frame, self.sam2_predictor)
378
 
379
  if mask is None or mask.size == 0:
380
+ raise exceptions.SegmentationError(frame_number, "Segmentation returned empty mask")
381
 
382
  return mask
383
 
384
  except Exception as e:
385
  self.stats['segmentation_errors'] += 1
386
+ raise exceptions.SegmentationError(frame_number, f"Segmentation failed: {str(e)}")
387
 
388
  def _should_refine_mask(self, frame_number: int) -> bool:
389
  """Determine if mask should be refined for this frame"""
 
525
  try:
526
  if background_choice == "custom" and custom_background_path:
527
  if not os.path.exists(custom_background_path):
528
+ raise exceptions.BackgroundProcessingError("custom", f"File not found: {custom_background_path}")
529
 
530
  background = cv2.imread(custom_background_path)
531
  if background is None:
532
+ raise exceptions.BackgroundProcessingError("custom", "Could not read custom background image")
533
 
534
  logger.info(f"Loaded custom background: {custom_background_path}")
535
 
536
  else:
537
  # Use professional background
538
  if background_choice not in PROFESSIONAL_BACKGROUNDS:
539
+ raise exceptions.BackgroundProcessingError(background_choice, "Unknown professional background")
540
 
541
  bg_config = PROFESSIONAL_BACKGROUNDS[background_choice]
542
  background = create_professional_background(bg_config, width, height)
 
549
 
550
  # Validate background
551
  if background is None or background.size == 0:
552
+ raise exceptions.BackgroundProcessingError(background_choice, "Background image is empty")
553
 
554
  return background
555
 
556
  except Exception as e:
557
+ if isinstance(e, exceptions.BackgroundProcessingError):
558
  logger.error(str(e))
559
  return None
560
  else: