[timecode] Finalize VFR support #168

Author: Breakthrough

scenedetect/_cli/commands.py

@@ -405,12 +405,13 @@ def _save_xml_fcp(
     ElementTree.SubElement(sequence, "duration").text = f"{duration.frame_num}"
 
     rate = ElementTree.SubElement(sequence, "rate")
-    ElementTree.SubElement(rate, "timebase").text = str(context.video_stream.frame_rate)
+    fps = float(context.video_stream.frame_rate)
+    ElementTree.SubElement(rate, "timebase").text = str(round(fps))
     ElementTree.SubElement(rate, "ntsc").text = "False"
 
     timecode = ElementTree.SubElement(sequence, "timecode")
     tc_rate = ElementTree.SubElement(timecode, "rate")
-    ElementTree.SubElement(tc_rate, "timebase").text = str(context.video_stream.frame_rate)
+    ElementTree.SubElement(tc_rate, "timebase").text = str(round(fps))
     ElementTree.SubElement(tc_rate, "ntsc").text = "False"
     ElementTree.SubElement(timecode, "frame").text = "0"
     ElementTree.SubElement(timecode, "displayformat").text = "NDF"
@@ -427,7 +428,7 @@ def _save_xml_fcp(
         ElementTree.SubElement(clip, "name").text = f"Shot {i + 1}"
         ElementTree.SubElement(clip, "enabled").text = "TRUE"
         ElementTree.SubElement(clip, "rate").append(
-            ElementTree.fromstring(f"<timebase>{context.video_stream.frame_rate}</timebase>")
+            ElementTree.fromstring(f"<timebase>{round(fps)}</timebase>")
         )
         # TODO: Are these supposed to be frame numbers or another format?
         ElementTree.SubElement(clip, "start").text = str(start.frame_num)
@@ -501,7 +502,7 @@ def save_otio(
     video_name = context.video_stream.name
     video_path = os.path.abspath(context.video_stream.path)
     video_base_name = os.path.basename(context.video_stream.path)
-    frame_rate = context.video_stream.frame_rate
+    frame_rate = float(context.video_stream.frame_rate)
 
     # List of track mapping to resource type.
     # TODO(https://scenedetect.com/issues/497): Allow OTIO export without an audio track.

scenedetect/backends/moviepy.py

@@ -17,14 +17,15 @@
 """
 
 import typing as ty
+from fractions import Fraction
 from logging import getLogger
 
 import cv2
 import numpy as np
 from moviepy.video.io.ffmpeg_reader import FFMPEG_VideoReader
 
 from scenedetect.backends.opencv import VideoStreamCv2
-from scenedetect.common import _USE_PTS_IN_DEVELOPMENT, FrameTimecode
+from scenedetect.common import FrameTimecode, Timecode, framerate_to_fraction
 from scenedetect.platform import get_file_name
 from scenedetect.video_stream import SeekError, VideoOpenFailure, VideoStream
 
@@ -83,9 +84,9 @@ def __init__(
     """Unique name used to identify this backend."""
 
     @property
-    def frame_rate(self) -> float:
-        """Framerate in frames/sec."""
-        return self._reader.fps
+    def frame_rate(self) -> Fraction:
+        """Framerate in frames/sec as a rational Fraction."""
+        return framerate_to_fraction(self._reader.fps)
 
     @property
     def path(self) -> ty.Union[bytes, str]:
@@ -135,7 +136,14 @@ def position(self) -> FrameTimecode:
         calling `read`. This will always return 0 (e.g. be equal to `base_timecode`) if no frames
         have been `read` yet."""
         frame_number = max(self._frame_number - 1, 0)
-        return FrameTimecode(frame_number, self.frame_rate)
+        # Synthesize a Timecode from the frame count and rational framerate.
+        # MoviePy assumes CFR, so this is equivalent to frame-based timing.
+        # Use the framerate denominator as the time_base denominator for exact timing.
+        fps = self.frame_rate
+        time_base = Fraction(1, fps.numerator)
+        pts = frame_number * fps.denominator
+        timecode = Timecode(pts=pts, time_base=time_base)
+        return FrameTimecode(timecode=timecode, fps=fps)
 
     @property
     def position_ms(self) -> float:
@@ -173,10 +181,6 @@ def seek(self, target: ty.Union[FrameTimecode, float, int]):
             ValueError: `target` is not a valid value (i.e. it is negative).
         """
         success = False
-        if _USE_PTS_IN_DEVELOPMENT:
-            # TODO(https://scenedetect.com/issue/168): Need to handle PTS here.
-            raise NotImplementedError()
-
         if not isinstance(target, FrameTimecode):
             target = FrameTimecode(target, self.frame_rate)
         try:

scenedetect/backends/opencv.py

@@ -27,7 +27,7 @@
 import cv2
 import numpy as np
 
-from scenedetect.common import _USE_PTS_IN_DEVELOPMENT, MAX_FPS_DELTA, FrameTimecode, Timecode
+from scenedetect.common import MAX_FPS_DELTA, FrameTimecode, Timecode, framerate_to_fraction
 from scenedetect.platform import get_file_name
 from scenedetect.video_stream import (
     FrameRateUnavailable,
@@ -111,7 +111,7 @@ def __init__(
         self._cap: ty.Optional[cv2.VideoCapture] = (
             None  # Reference to underlying cv2.VideoCapture object.
         )
-        self._frame_rate: ty.Optional[float] = None
+        self._frame_rate: ty.Optional[Fraction] = None
 
         # VideoCapture state
         self._has_grabbed = False
@@ -144,7 +144,7 @@ def capture(self) -> cv2.VideoCapture:
     """Unique name used to identify this backend."""
 
     @property
-    def frame_rate(self) -> float:
+    def frame_rate(self) -> Fraction:
         assert self._frame_rate
         return self._frame_rate
 
@@ -196,30 +196,25 @@ def aspect_ratio(self) -> float:
 
     @property
     def timecode(self) -> Timecode:
-        """Current position within stream as a Timecode. This is not frame accurate."""
+        """Current position within stream as a Timecode."""
         # *NOTE*: Although OpenCV has `CAP_PROP_PTS`, it doesn't seem to be reliable. For now, we
-        # use `CAP_PROP_POS_MSEC` instead, with a time base of 1/1000. Unfortunately this means that
-        # rounding errors will affect frame accuracy with this backend.
-        pts = self._cap.get(cv2.CAP_PROP_POS_MSEC)
-        time_base = Fraction(1, 1000)
-        return Timecode(pts=round(pts), time_base=time_base)
+        # use `CAP_PROP_POS_MSEC` instead, converting to microseconds for sufficient precision to
+        # avoid frame-boundary rounding errors at common framerates like 24000/1001.
+        ms = self._cap.get(cv2.CAP_PROP_POS_MSEC)
+        time_base = Fraction(1, 1000000)
+        return Timecode(pts=round(ms * 1000), time_base=time_base)
 
     @property
     def position(self) -> FrameTimecode:
-        # TODO(https://scenedetect.com/issue/168): See if there is a better way to do this, or
-        # add a config option before landing this.
-        if _USE_PTS_IN_DEVELOPMENT:
-            timecode = self.timecode
-            # If PTS is 0 but we've read frames, derive from frame number.
-            # This handles image sequences and cases where CAP_PROP_POS_MSEC is unreliable.
-            if timecode.pts == 0 and self.frame_number > 0:
-                time_sec = (self.frame_number - 1) / self.frame_rate
-                pts = round(time_sec * 1000)
-                timecode = Timecode(pts=pts, time_base=Fraction(1, 1000))
-            return FrameTimecode(timecode=timecode, fps=self.frame_rate)
-        if self.frame_number < 1:
-            return self.base_timecode
-        return self.base_timecode + (self.frame_number - 1)
+        timecode = self.timecode
+        # If PTS is 0 but we've read frames, derive from frame number.
+        # This handles image sequences and cases where CAP_PROP_POS_MSEC is unreliable.
+        if timecode.pts == 0 and self.frame_number > 0:
+            fps = self.frame_rate
+            time_base = Fraction(1, fps.numerator)
+            pts = (self.frame_number - 1) * fps.denominator
+            timecode = Timecode(pts=pts, time_base=time_base)
+        return FrameTimecode(timecode=timecode, fps=self.frame_rate)
 
     @property
     def position_ms(self) -> float:
@@ -235,8 +230,9 @@ def seek(self, target: ty.Union[FrameTimecode, float, int]):
         if target < 0:
             raise ValueError("Target seek position cannot be negative!")
 
-        # TODO(https://scenedetect.com/issue/168): Shouldn't use frames for VFR video here.
-        # Have to seek one behind and call grab() after to that the VideoCapture
+        # Seeking is done via frame number since OpenCV doesn't support PTS-based seeking.
+        # After seeking, position returns actual PTS from CAP_PROP_POS_MSEC.
+        # Have to seek one behind and call grab() after so that the VideoCapture
         # returns a valid timestamp when using CAP_PROP_POS_MSEC.
         target_frame_cv2 = (self.base_timecode + target).frame_num
         if target_frame_cv2 > 0:
@@ -329,14 +325,11 @@ def _open_capture(self, framerate: ty.Optional[float] = None):
                 raise FrameRateUnavailable()
 
         self._cap = cap
-        self._frame_rate = framerate
+        self._frame_rate = framerate_to_fraction(framerate)
         self._has_grabbed = False
         cap.set(cv2.CAP_PROP_ORIENTATION_AUTO, 1.0)  # https://github.com/opencv/opencv/issues/26795
 
 
-# TODO(https://scenedetect.com/issues/168): Support non-monotonic timing for `position`. VFR timecode
-# support is a prerequisite for this. Timecodes are currently calculated by multiplying the
-# framerate by number of frames. Actual elapsed time can be obtained via `position_ms` for now.
 class VideoCaptureAdapter(VideoStream):
     """Adapter for existing VideoCapture objects. Unlike VideoStreamCv2, this class supports
     VideoCaptures which may not support seeking.
@@ -378,7 +371,7 @@ def __init__(
                 raise FrameRateUnavailable()
 
         self._cap = cap
-        self._frame_rate: float = framerate
+        self._frame_rate: Fraction = framerate_to_fraction(framerate)
         self._num_frames = 0
         self._max_read_attempts = max_read_attempts
         self._decode_failures = 0
@@ -408,7 +401,7 @@ def capture(self) -> cv2.VideoCapture:
     """Unique name used to identify this backend."""
 
     @property
-    def frame_rate(self) -> float:
+    def frame_rate(self) -> Fraction:
         """Framerate in frames/sec."""
         assert self._frame_rate
         return self._frame_rate
@@ -439,8 +432,6 @@ def frame_size(self) -> ty.Tuple[int, int]:
     @property
     def duration(self) -> ty.Optional[FrameTimecode]:
         """Duration of the stream as a FrameTimecode, or None if non terminating."""
-        # TODO(https://scenedetect.com/issue/168): This will be incorrect for VFR. See if there is
-        # another property we can use to estimate the video length correctly.
         frame_count = math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_COUNT))
         if frame_count > 0:
             return self.base_timecode + frame_count
@@ -455,7 +446,12 @@ def aspect_ratio(self) -> float:
     def position(self) -> FrameTimecode:
         if self.frame_number < 1:
             return self.base_timecode
-        return self.base_timecode + (self.frame_number - 1)
+        # Synthesize a Timecode from frame count and rational framerate.
+        fps = self.frame_rate
+        time_base = Fraction(1, fps.numerator)
+        pts = (self.frame_number - 1) * fps.denominator
+        timecode = Timecode(pts=pts, time_base=time_base)
+        return FrameTimecode(timecode=timecode, fps=fps)
 
     @property
     def position_ms(self) -> float:

scenedetect/backends/pyav.py

@@ -18,7 +18,7 @@
 import av
 import numpy as np
 
-from scenedetect.common import _USE_PTS_IN_DEVELOPMENT, MAX_FPS_DELTA, FrameTimecode, Timecode
+from scenedetect.common import MAX_FPS_DELTA, FrameTimecode, Timecode
 from scenedetect.platform import get_file_name
 from scenedetect.video_stream import FrameRateUnavailable, VideoOpenFailure, VideoStream
 
@@ -172,8 +172,8 @@ def duration(self) -> FrameTimecode:
         return self.base_timecode + self._duration_frames
 
     @property
-    def frame_rate(self) -> float:
-        """Frame rate in frames/sec."""
+    def frame_rate(self) -> Fraction:
+        """Frame rate in frames/sec as a rational Fraction."""
         return self._frame_rate
 
     @property
@@ -184,10 +184,8 @@ def position(self) -> FrameTimecode:
         to the presentation time 0.  Returns 0 even if `frame_number` is 1."""
         if self._frame is None:
             return self.base_timecode
-        if _USE_PTS_IN_DEVELOPMENT:
-            timecode = Timecode(pts=self._frame.pts, time_base=self._frame.time_base)
-            return FrameTimecode(timecode=timecode, fps=self.frame_rate)
-        return FrameTimecode(round(self._frame.time * self.frame_rate), self.frame_rate)
+        timecode = Timecode(pts=self._frame.pts, time_base=self._frame.time_base)
+        return FrameTimecode(timecode=timecode, fps=self.frame_rate)
 
     @property
     def position_ms(self) -> float:
@@ -204,10 +202,8 @@ def frame_number(self) -> int:
         Will return 0 until the first frame is `read`."""
 
         if self._frame:
-            if _USE_PTS_IN_DEVELOPMENT:
-                # frame_number is 1-indexed, so add 1 to the 0-based frame position.
-                return round(self._frame.time * self.frame_rate) + 1
-            return self.position.frame_num + 1
+            # frame_number is 1-indexed, so add 1 to the 0-based frame position.
+            return round(self._frame.time * self.frame_rate) + 1
         return 0
 
     @property
@@ -258,7 +254,6 @@ def seek(self, target: ty.Union[FrameTimecode, float, int]) -> None:
             raise ValueError("Target cannot be negative!")
         beginning = target == 0
 
-        # TODO(https://scenedetect.com/issues/168): This breaks with PTS mode enabled.
         target = self.base_timecode + target
         if target >= 1:
             target = target - 1