[common] Allow FrameTimecode to be created from Timecode

This allows us to keep the existing VideoStream interface but still have backends
provide precise timing information. For now this is gated under a hard-coded feature
flag as it still needs to be integrated with the SceneDetector interface.

Author: Breakthrough

docs/api.rst

@@ -46,7 +46,7 @@ Most types/functions are also available directly from the `scenedetect` package
 
     .. code:: python
 
-        scenedetect<0.7
+        scenedetect<0.8
 
 
 .. _scenedetect-quickstart:

scenedetect/backends/opencv.py

@@ -32,7 +32,6 @@
 from scenedetect.video_stream import (
     FrameRateUnavailable,
     SeekError,
-    VideoFrame,
     VideoOpenFailure,
     VideoStream,
 )
@@ -47,6 +46,8 @@
     " ! ",  # gstreamer pipe
 )
 
+_USE_PTS_IN_DEVELOPMENT = False
+
 
 def _get_aspect_ratio(cap: cv2.VideoCapture, epsilon: float = 0.0001) -> float:
     """Display/pixel aspect ratio of the VideoCapture as a float (1.0 represents square pixels)."""
@@ -195,6 +196,16 @@ def aspect_ratio(self) -> float:
         """Display/pixel aspect ratio as a float (1.0 represents square pixels)."""
         return _get_aspect_ratio(self._cap)
 
+    @property
+    def timecode(self) -> Timecode:
+        """Current position within stream as a Timecode. This is not frame accurate."""
+        # *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)
+
     @property
     def position(self) -> FrameTimecode:
         """Current position within stream as FrameTimecode.
@@ -204,6 +215,8 @@ def position(self) -> FrameTimecode:
 
         This method will always return 0 (e.g. be equal to `base_timecode`) if no frames
         have been `read`."""
+        if _USE_PTS_IN_DEVELOPMENT:
+            return FrameTimecode(timecode=self.timecode, fps=self.frame_rate)
         if self.frame_number < 1:
             return self.base_timecode
         return self.base_timecode + (self.frame_number - 1)
@@ -272,30 +285,6 @@ def reset(self):
         self._cap.release()
         self._open_capture(self._frame_rate)
 
-    def __next__(self):
-        # NOTE: POS_FRAMES starts from 0 before any frames are read.
-        read, image = self._cap.read()
-        if not read:
-            raise StopIteration()
-        # We can only query CAP_PROP_PTS if this uses the ffmpeg backend,  however it doesn't seem
-        # to work correctly. Quite frequently consecutive frames return the same PTS. We might need
-        # to just abandon using PTS with OpenCV and rely on milliseconds. This will still result
-        # in occasional off-by-one errors for VFR videos, but better than the status quo.
-        #
-        # We should also add a config option so users can specify if OpenCV should use fixed or
-        # variable timing (i.e. if we should use CAP_PROP_POS_MSEC or CAP_PROP_POS_FRAMES for
-        # timestamp calculation).
-        USE_PTS = False
-        if USE_PTS:
-            pts = self._cap.get(cv2.CAP_PROP_PTS)
-            time_base = Fraction.from_float(self._cap.get(cv2.CAP_PROP_FPS))
-            time_base = Fraction(numerator=time_base.denominator, denominator=time_base.numerator)
-        else:
-            pts = self._cap.get(cv2.CAP_PROP_POS_MSEC)
-            time_base = Fraction(1, 1000)
-        timecode = Timecode(pts=round(pts), time_base=time_base)
-        return VideoFrame(image=image, timecode=timecode)
-
     def read(self, decode: bool = True, advance: bool = True) -> ty.Union[np.ndarray, bool]:
         """Read and decode the next frame as a np.ndarray. Returns False when video ends,
         or the maximum number of decode attempts has passed.
@@ -490,6 +479,8 @@ 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(v0.7): 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
@@ -508,6 +499,7 @@ def position(self) -> FrameTimecode:
 
         This method will always return 0 (e.g. be equal to `base_timecode`) if no frames
         have been `read`."""
+
         if self.frame_number < 1:
             return self.base_timecode
         return self.base_timecode + (self.frame_number - 1)

scenedetect/backends/pyav.py

@@ -20,12 +20,14 @@
 from scenedetect.common import Timecode
 from scenedetect.frame_timecode import MAX_FPS_DELTA, FrameTimecode
 from scenedetect.platform import get_file_name
-from scenedetect.video_stream import FrameRateUnavailable, VideoFrame, VideoOpenFailure, VideoStream
+from scenedetect.video_stream import FrameRateUnavailable, VideoOpenFailure, VideoStream
 
 logger = getLogger("pyscenedetect")
 
 VALID_THREAD_MODES = ["NONE", "SLICE", "FRAME", "AUTO"]
 
+_USE_PTS_IN_DEVELOPMENT = False
+
 
 class VideoStreamAv(VideoStream):
     """PyAV `av.InputContainer` backend."""
@@ -80,7 +82,7 @@ def __init__(
 
         self._name = "" if name is None else name
         self._path = ""
-        self._frame = None
+        self._frame: ty.Optional[av.VideoFrame] = None
         self._reopened = True
 
         if threading_mode:
@@ -183,6 +185,9 @@ def position(self) -> FrameTimecode:
 
         This can be interpreted as presentation time stamp, thus frame 1 corresponds
         to the presentation time 0.  Returns 0 even if `frame_number` is 1."""
+        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)
         if self._frame is None:
             return self.base_timecode
         return FrameTimecode(round(self._frame.time * self.frame_rate), self.frame_rate)
@@ -264,19 +269,6 @@ def reset(self):
         except Exception as ex:
             raise VideoOpenFailure() from ex
 
-    def __next__(self) -> VideoFrame:
-        # TODO: On the VFR test video, we seem to only decode 1979 frames instead of 1980. See what
-        # the issue could be.
-        try:
-            frame = next(self._container.decode(video=0))
-        except av.error.EOFError as ex:
-            if not self._handle_eof():
-                raise StopIteration() from ex
-            return next(self)  # *NOTE*: self._handle_eof must ensure we won't recurse again.
-        image = frame.to_ndarray(format="bgr24")
-        timecode = Timecode(pts=frame.pts, time_base=frame.time_base)
-        return VideoFrame(image=image, timecode=timecode)
-
     def read(self, decode: bool = True, advance: bool = True) -> ty.Union[np.ndarray, bool]:
         """Read and decode the next frame as a np.ndarray. Returns False when video ends.