Update video_processing_moss_vl.py

video_processing_moss_vl.py

@@ -39,6 +39,39 @@ from transformers.video_utils import VideoMetadata, group_videos_by_shape, reord
 logger = logging.get_logger(__name__)
 
 
+TORCHCODEC_TIMESTAMP_EPSILON = 1e-6
+
+
+def clamp_timestamps_for_torchcodec(timestamps: List[float], torchcodec_metadata) -> List[float]:
+    if not timestamps:
+        return timestamps
+
+    min_pts = torchcodec_metadata.begin_stream_seconds_from_content
+    if min_pts is None:
+        min_pts = 0.0
+    # TorchCodec can reject timestamps exactly equal to the reported stream
+    # begin due to tiny metadata/decoder precision differences.
+    safe_min_pts = min_pts + TORCHCODEC_TIMESTAMP_EPSILON
+
+    max_pts_candidates = []
+    if torchcodec_metadata.num_frames_from_content and torchcodec_metadata.average_fps:
+        max_pts_candidates.append(
+            (torchcodec_metadata.num_frames_from_content - 1) / torchcodec_metadata.average_fps + min_pts
+        )
+    if torchcodec_metadata.end_stream_seconds_from_content is not None:
+        # TorchCodec requires requested PTS to be strictly smaller than the content end.
+        max_pts_candidates.append(torchcodec_metadata.end_stream_seconds_from_content - TORCHCODEC_TIMESTAMP_EPSILON)
+    if not max_pts_candidates and torchcodec_metadata.duration_seconds is not None:
+        max_pts_candidates.append(torchcodec_metadata.duration_seconds - TORCHCODEC_TIMESTAMP_EPSILON)
+
+    if max_pts_candidates:
+        max_pts = max(safe_min_pts, min(max_pts_candidates))
+        return [max(safe_min_pts, min(float(t), max_pts)) for t in timestamps]
+    if safe_min_pts > 0:
+        return [max(safe_min_pts, float(t)) for t in timestamps]
+    return [float(t) for t in timestamps]
+
+
 # -----------------------------------------------------------------------------
 # Torchcodec video frame extraction utilities
 # -----------------------------------------------------------------------------
@@ -198,7 +231,17 @@ def split_indices(indices: List[Union[int, float]], num_chunks: int) -> List[Lis
         
     Returns:
         List of index chunks.
+
+    Raises:
+        ValueError: If indices is empty or num_chunks is not positive.
     """
+    if len(indices) == 0:
+        raise ValueError("indices must not be empty")
+    if num_chunks <= 0:
+        raise ValueError("num_chunks must be positive")
+
+    # Never create empty decode jobs when there are fewer frames than workers.
+    num_chunks = min(num_chunks, len(indices))
     chunk_size = len(indices) // num_chunks
     chunks = []
     for i in range(num_chunks - 1):
@@ -239,7 +282,7 @@ def decode_with_multithreading(indices: List[int], num_threads: int, video_path:
         dict: Contains 'data', 'duration_seconds', 'pts_seconds' tensors.
     """
     chunks = split_indices(indices, num_chunks=num_threads)
-    results = Parallel(n_jobs=num_threads, prefer="threads", verbose=0)(
+    results = Parallel(n_jobs=len(chunks), prefer="threads", verbose=0)(
         delayed(decode_sequentially)(chunk, video_path) for chunk in chunks
     )
     
@@ -266,22 +309,7 @@ def decode_sequentially_timestamp(timestamp_list: List[float], video_path: str,
     try:
         metadata = decoder.metadata
 
-        min_pts = metadata.begin_stream_seconds_from_content
-        if min_pts is None:
-            min_pts = 0.0
-
-        max_pts = None
-        if metadata.num_frames_from_content and metadata.average_fps:
-            max_pts = (metadata.num_frames_from_content - 1) / metadata.average_fps + min_pts
-        elif metadata.end_stream_seconds_from_content is not None:
-            max_pts = metadata.end_stream_seconds_from_content
-        else:
-            max_pts = metadata.duration_seconds
-
-        if max_pts is not None and max_pts > 0:
-            timestamp_list = [max(min_pts, min(t, max_pts)) for t in timestamp_list]
-        elif min_pts > 0:
-            timestamp_list = [max(min_pts, t) for t in timestamp_list]
+        timestamp_list = clamp_timestamps_for_torchcodec(timestamp_list, metadata)
 
         return decoder.get_frames_played_at(timestamp_list)
     finally:
@@ -301,7 +329,7 @@ def timestamp_decode_with_multithreading(timestamp_list: List[float], num_thread
         dict: Contains 'data', 'duration_seconds', 'pts_seconds' tensors.
     """
     chunks = split_indices(timestamp_list, num_chunks=num_threads)
-    results = Parallel(n_jobs=num_threads, prefer="threads", verbose=0)(
+    results = Parallel(n_jobs=len(chunks), prefer="threads", verbose=0)(
         delayed(decode_sequentially_timestamp)(chunk, video_path) for chunk in chunks
     )
     
@@ -702,10 +730,9 @@ class MossVLVideoProcessor(BaseVideoProcessor):
             
             if len(segment) == 1:
                 # Single frame at specified time
-                timestamp = segment[0]
-                frame_batch = decoder.get_frames_played_at([timestamp])
+                actual_timestamps = clamp_timestamps_for_torchcodec([segment[0]], torchcodec_metadata)
+                frame_batch = decoder.get_frames_played_at(actual_timestamps)
                 video_tensor = frame_batch.data
-                actual_timestamps = [timestamp]
                 sample_count = 1
             else:
                 # Segment [start, end) - left-closed, right-open interval
@@ -724,6 +751,8 @@ class MossVLVideoProcessor(BaseVideoProcessor):
                     # Sample uniformly within [start, end), endpoint=False for left-closed right-open
                     actual_timestamps = np.linspace(start_time, end_time, target_frames, endpoint=False).tolist()
                 
+                actual_timestamps = clamp_timestamps_for_torchcodec(actual_timestamps, torchcodec_metadata)
+
                 # Use multithreading for extraction
                 result = timestamp_decode_with_multithreading(actual_timestamps, self.num_extract_threads, video_path)
                 video_tensor = result["data"]
@@ -1129,4 +1158,3 @@ class MossVLVideoProcessor(BaseVideoProcessor):
 
 
 __all__ = ["MossVLVideoProcessor"]
-