[timecode] Expand PTS support #168

Allows much more tests to pass now, down to just a few.

Author: Breakthrough

scenedetect/backends/opencv.py

@@ -209,7 +209,14 @@ 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:
-            return FrameTimecode(timecode=self.timecode, fps=self.frame_rate)
+            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)

scenedetect/backends/pyav.py

@@ -182,11 +182,11 @@ 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 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)
-        if self._frame is None:
-            return self.base_timecode
         return FrameTimecode(round(self._frame.time * self.frame_rate), self.frame_rate)
 
     @property
@@ -205,9 +205,8 @@ def frame_number(self) -> int:
 
         if self._frame:
             if _USE_PTS_IN_DEVELOPMENT:
-                return FrameTimecode(
-                    round(self._frame.time * self.frame_rate), self.frame_rate
-                ).frame_num
+                # 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
         return 0
 

scenedetect/common.py

@@ -252,18 +252,22 @@ def frame_num(self) -> ty.Optional[int]:
                 stacklevel=2,
                 category=UserWarning,
             )
-            # We can calculate the approx. # of frames by taking the presentation time and the
-            # time base itself.
-            (num, den) = (self._time.time_base * self._time.pts).as_integer_ratio()
-            return num / den
+            # Calculate approximate frame number from seconds and framerate.
+            if self._rate is not None:
+                return round(self._time.seconds * float(self._rate))
+            # No framerate available - return estimate based on time.
+            return round(self._time.seconds)
         if isinstance(self._time, _Seconds):
             return self._seconds_to_frames(self._time.value)
         return self._time.value
 
     @property
-    def framerate(self) -> float:
+    def framerate(self) -> ty.Optional[float]:
         """The framerate to use for distance between frames and to calculate frame numbers.
-        For a VFR video, this may just be the average framerate."""
+        For a VFR video, this may just be the average framerate. Returns None if framerate
+        is unknown (e.g. when working with pure Timecode representations)."""
+        if self._rate is None:
+            return None
         return float(self._rate)
 
     @property
@@ -499,6 +503,9 @@ def __eq__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> bool:
             return False
         if _compare_as_fixed(self, other):
             return self.frame_num == other.frame_num
+        # For integer comparison, use frame numbers to avoid floating point precision issues.
+        if isinstance(other, int):
+            return self.frame_num == other
         if isinstance(self._time, (Timecode, _Seconds)):
             return self.seconds == self._get_other_as_seconds(other)
         return self.frame_num == self._get_other_as_frames(other)
@@ -508,34 +515,49 @@ def __ne__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> bool:
             return True
         if _compare_as_fixed(self, other):
             return self.frame_num != other.frame_num
+        # For integer comparison, use frame numbers to avoid floating point precision issues.
+        if isinstance(other, int):
+            return self.frame_num != other
         if isinstance(self._time, (Timecode, _Seconds)):
             return self.seconds != self._get_other_as_seconds(other)
         return self.frame_num != self._get_other_as_frames(other)
 
     def __lt__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> bool:
         if _compare_as_fixed(self, other):
             return self.frame_num < other.frame_num
+        # For integer comparison, use frame numbers to avoid floating point precision issues.
+        if isinstance(other, int):
+            return self.frame_num < other
         if isinstance(self._time, (Timecode, _Seconds)):
             return self.seconds < self._get_other_as_seconds(other)
         return self.frame_num < self._get_other_as_frames(other)
 
     def __le__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> bool:
         if _compare_as_fixed(self, other):
             return self.frame_num <= other.frame_num
+        # For integer comparison, use frame numbers to avoid floating point precision issues.
+        if isinstance(other, int):
+            return self.frame_num <= other
         if isinstance(self._time, (Timecode, _Seconds)):
             return self.seconds <= self._get_other_as_seconds(other)
         return self.frame_num <= self._get_other_as_frames(other)
 
     def __gt__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> bool:
         if _compare_as_fixed(self, other):
             return self.frame_num > other.frame_num
+        # For integer comparison, use frame numbers to avoid floating point precision issues.
+        if isinstance(other, int):
+            return self.frame_num > other
         if isinstance(self._time, (Timecode, _Seconds)):
             return self.seconds > self._get_other_as_seconds(other)
         return self.frame_num > self._get_other_as_frames(other)
 
     def __ge__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> bool:
         if _compare_as_fixed(self, other):
             return self.frame_num >= other.frame_num
+        # For integer comparison, use frame numbers to avoid floating point precision issues.
+        if isinstance(other, int):
+            return self.frame_num >= other
         if isinstance(self._time, (Timecode, _Seconds)):
             return self.seconds >= self._get_other_as_seconds(other)
         return self.frame_num >= self._get_other_as_frames(other)
@@ -565,7 +587,9 @@ def __iadd__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> "FrameT
                 pts=max(0, timecode.pts + round(seconds / timecode.time_base)),
                 time_base=timecode.time_base,
             )
-            self._rate = None
+            # Preserve rate if available from self or other.
+            if self._rate is None and isinstance(other, FrameTimecode):
+                self._rate = other._rate
             return self
 
         other_is_seconds = isinstance(other, FrameTimecode) and isinstance(other._time, _Seconds)
@@ -610,7 +634,9 @@ def __isub__(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> "FrameT
                 pts=max(0, timecode.pts - round(seconds / timecode.time_base)),
                 time_base=timecode.time_base,
             )
-            self._rate = None
+            # Preserve rate if available from self or other.
+            if self._rate is None and isinstance(other, FrameTimecode):
+                self._rate = other._rate
             return self
 
         other_is_seconds = isinstance(other, FrameTimecode) and isinstance(other._time, _Seconds)
@@ -652,21 +678,20 @@ def __repr__(self) -> str:
         return f"{self.get_timecode()} [frame_num={self._time.value}, fps={self._rate}]"
 
     def __hash__(self) -> int:
-        if isinstance(self._time, Timecode):
-            return hash(self._time)
+        # Use frame_num for consistent hashing regardless of internal representation.
+        # This ensures that FrameTimecodes representing the same frame have the same hash,
+        # enabling proper dictionary lookups in StatsManager.
         return self.frame_num
 
     def _get_other_as_seconds(self, other: ty.Union[int, float, str, "FrameTimecode"]) -> float:
         """Get the time in seconds from `other` for arithmetic operations."""
         if isinstance(other, int):
-            if isinstance(self._time, Timecode):
-                # TODO(https://scenedetect.com/issue/168): We need to convert every place that uses
-                # frame numbers with timestamps to convert to a non-frame based way of temporal
-                # logic and instead use seconds-based.
-                if _USE_PTS_IN_DEVELOPMENT and other == 1:
-                    return self.seconds
-                raise NotImplementedError()
-            return float(other) / self._rate
+            # Convert frame number to seconds using framerate.
+            if self._rate is None:
+                raise NotImplementedError(
+                    "Cannot convert frame number to seconds without framerate"
+                )
+            return float(other) / float(self._rate)
         if isinstance(other, float):
             return other
         if isinstance(other, str):

scenedetect/scene_manager.py

@@ -404,7 +404,7 @@ def _process_frame(
                 for cut in cuts:
                     for position, frame in self._frame_buffer:
                         if cut == position:
-                            callback(frame, position)
+                            callback(frame, int(position))
         return new_cuts
 
     def _post_process(self, timecode: FrameTimecode) -> None: