[api] Strengthen type hints for detectors

Author: Breakthrough

docs/api/migration_guide.rst

@@ -160,7 +160,37 @@ All backends now return presentation timestamp (PTS) backed values from ``VideoS
 ``StatsManager`` Changes
 =======================================================================
 
-The ``StatsManager`` methods ``get_metrics()``, ``set_metrics()``, and ``metrics_exist()`` now take a ``FrameTimecode`` instead of ``int`` for the frame identifier, matching the detector interface change.
+The ``StatsManager`` methods ``get_metrics()``, ``set_metrics()``, and ``metrics_exist()`` now formally accept either a ``FrameTimecode`` or a plain ``int`` frame number for the timecode argument. Passing a ``FrameTimecode`` is preferred and matches the detector interface; the ``int`` form is retained for compatibility with the deprecated ``load_from_csv()`` path, which keys metrics by integer frame number.
+
+``StatsManager.load_from_csv()`` also accepts ``os.PathLike`` (e.g. ``pathlib.Path``) in addition to ``str`` / ``bytes`` / file handles.
+
+
+=======================================================================
+``SceneDetector`` Annotation Fixes
+=======================================================================
+
+``SceneDetector.post_process()`` now declares its parameter as ``timecode: FrameTimecode`` (previously typed as ``int``). The method already received a ``FrameTimecode`` at runtime and concrete detectors (e.g. ``ThresholdDetector``, ``ContentDetector``) already used the ``FrameTimecode`` type — only the abstract-base-class annotation was inconsistent. No call-site changes are needed; this just brings the signature into agreement with the documented and actual behavior.
+
+
+=======================================================================
+``SceneManager.detect_scenes()`` Time Arguments
+=======================================================================
+
+The ``duration`` and ``end_time`` arguments now formally accept ``int`` (frames), ``float`` (seconds), ``str`` (timecode string, e.g. ``"00:00:05.000"``), or ``FrameTimecode``. The internal code already validated these forms; the annotation was previously narrower than the documented behavior.
+
+.. code:: python
+
+    # All of these were always supported at runtime; now they type-check too:
+    scene_manager.detect_scenes(video=video, end_time=15.0)         # seconds
+    scene_manager.detect_scenes(video=video, end_time=1500)         # frames
+    scene_manager.detect_scenes(video=video, end_time="00:01:00")   # timecode
+
+
+=======================================================================
+``save_images()`` Path Handling
+=======================================================================
+
+The ``output_dir`` argument of :func:`scenedetect.output.save_images` now accepts ``os.PathLike`` (e.g. ``pathlib.Path``) in addition to ``str``. No changes are required for existing string-based callers.
 
 
 =======================================================================

pyproject.toml

@@ -61,6 +61,9 @@ unfixable = []
 
 [tool.pyright]
 include = ["scenedetect", "tests"]
+# Vendored third-party code: don't lint upstream source (mirrors the ruff
+# per-file-ignores convention).
+exclude = ["scenedetect/_thirdparty"]
 # Run pyright from inside an activated venv (or pass `--pythonpath` /
 # configure your editor's interpreter) so cv2 / av / numpy / moviepy
 # resolve. Without this, pyright uses its bundled Python and the report

scenedetect/_cli/context.py

@@ -206,6 +206,7 @@ def handle_options(
 
         # The `scenedetect` command was just started, let's initialize logging and try to load any
         # config files that were specified.
+        init_log: list = []
         try:
             init_failure = not self.config.initialized
             init_log = self.config.get_init_log()
@@ -305,9 +306,9 @@ def handle_options(
             scene_manager.auto_downscale = True
         else:
             scene_manager.auto_downscale = False
-            downscale = self.config.get_value("global", "downscale", downscale)
+            downscale_value: int = self.config.get_value("global", "downscale", downscale)
             try:
-                scene_manager.downscale = downscale
+                scene_manager.downscale = downscale_value
             except ValueError as ex:
                 logger.debug(str(ex))
                 raise click.BadParameter(str(ex), param_hint="downscale factor") from ex
@@ -532,11 +533,13 @@ def _open_video_stream(
                     framerate=framerate,
                     backend=backend,
                 )
+            duration = self.video_stream.duration
+            duration_str = f"{duration} ({duration.frame_num} frames)" if duration else "unknown"
             logger.debug(f"""Video information:
   Backend:      {type(self.video_stream).__name__}
   Resolution:   {self.video_stream.frame_size}
   Framerate:    {self.video_stream.frame_rate}
-  Duration:     {self.video_stream.duration} ({self.video_stream.duration.frame_num} frames)""")
+  Duration:     {duration_str}""")
 
         except FrameRateUnavailable as ex:
             if __debug__:

scenedetect/detector.py

@@ -61,7 +61,7 @@ def process_frame(
 
     # Optional Methods
 
-    def post_process(self, timecode: int) -> list[FrameTimecode]:
+    def post_process(self, timecode: FrameTimecode) -> list[FrameTimecode]:
         """Called after there are no more frames to process.
 
         Args:

scenedetect/detectors/adaptive_detector.py

@@ -100,6 +100,10 @@ def get_metrics(self) -> list[str]:
     def process_frame(self, timecode: FrameTimecode, frame_img: np.ndarray) -> list[FrameTimecode]:
         super().process_frame(timecode=timecode, frame_img=frame_img)
 
+        # If the parent could not calculate a frame score, there's nothing to buffer.
+        if self._frame_score is None:
+            return []
+
         # Initialize last scene cut point at the beginning of the frames of interest.
         if self._last_cut is None:
             self._last_cut = timecode

scenedetect/detectors/content_detector.py

@@ -169,7 +169,9 @@ def _calculate_frame_score(self, timecode: FrameTimecode, frame_img: numpy.ndarr
             delta_sat=_mean_pixel_distance(sat, self._last_frame.sat),
             delta_lum=_mean_pixel_distance(lum, self._last_frame.lum),
             delta_edges=(
-                0.0 if edges is None else _mean_pixel_distance(edges, self._last_frame.edges)
+                0.0
+                if edges is None or self._last_frame.edges is None
+                else _mean_pixel_distance(edges, self._last_frame.edges)
             ),
         )
 

scenedetect/detectors/transnet_v2.py

@@ -107,6 +107,8 @@ def push(self, pixels: np.ndarray, time: np.ndarray):
                 ),
             )
         else:
+            # `self.time` is set in lockstep with `self.pixels` above, so it is non-None here.
+            assert self.time is not None
             c1 = self.pixels
             c2 = pixels
 

scenedetect/output/image.py

@@ -28,7 +28,7 @@
     Interpolation,
     SceneList,
 )
-from scenedetect.platform import get_and_create_path, get_cv2_imwrite_params, tqdm
+from scenedetect.platform import StrPath, get_and_create_path, get_cv2_imwrite_params, tqdm
 from scenedetect.video_stream import VideoStream
 
 logger = logging.getLogger("pyscenedetect")
@@ -162,7 +162,7 @@ def run(
         self,
         video: VideoStream,
         scene_list: SceneList,
-        output_dir: str | None = None,
+        output_dir: StrPath | None = None,
         show_progress=False,
     ) -> dict[int, list[str]]:
         """Run image extraction on `video` using the current parameters. Thread-safe.
@@ -355,7 +355,7 @@ def save_images(
     image_extension: str = "jpg",
     encoder_param: int = 95,
     image_name_template: str = "$VIDEO_NAME-Scene-$SCENE_NUMBER-$IMAGE_NUMBER",
-    output_dir: str | None = None,
+    output_dir: StrPath | None = None,
     show_progress: bool | None = False,
     scale: float | None = None,
     height: int | None = None,

scenedetect/scene_manager.py

@@ -419,8 +419,8 @@ def stop(self) -> None:
     def detect_scenes(
         self,
         video: VideoStream | None = None,
-        duration: FrameTimecode | None = None,
-        end_time: FrameTimecode | None = None,
+        duration: "int | float | str | FrameTimecode | None" = None,
+        end_time: "int | float | str | FrameTimecode | None" = None,
         frame_skip: int = 0,
         show_progress: bool = False,
         callback: ty.Callable[[np.ndarray, FrameTimecode], None] | None = None,

scenedetect/stats_manager.py

@@ -96,19 +96,22 @@ class StatsManager:
     Only metrics consisting of `float` or `int` should be used currently.
     """
 
-    def __init__(self, base_timecode: FrameTimecode | None = None):
+    def __init__(self, base_timecode: int | FrameTimecode | None = None):
         """Initialize a new StatsManager.
 
         Arguments:
             base_timecode: Timecode associated with this object. Must not be None (default value
                 will be removed in a future release).
         """
-        # Frame metrics is a dict of frame (int): metric_dict (Dict[str, float])
-        # of each frame metric key and the value it represents (usually float).
-        self._frame_metrics: dict[FrameTimecode, dict[str, float]] = dict()
+        # Frame metrics keyed by either an `int` frame number or a `FrameTimecode`. Both forms
+        # hash/compare to the same dict slot (`FrameTimecode.__hash__` returns `frame_num`), so
+        # public methods accept both interchangeably for the same frame.
+        self._frame_metrics: dict[int | FrameTimecode, dict[str, float]] = dict()
         self._metric_keys: set[str] = set()
         self._metrics_updated: bool = False  # Flag indicating if metrics require saving.
-        self._base_timecode: FrameTimecode | None = base_timecode  # Used for timing calculations.
+        self._base_timecode: int | FrameTimecode | None = (
+            base_timecode  # Used for timing calculations.
+        )
 
     @property
     def metric_keys(self) -> ty.Iterable[str]:
@@ -120,7 +123,9 @@ def register_metrics(self, metric_keys: ty.Iterable[str]) -> None:
 
     # TODO(https://scenedetect.com/issues/507): We should support the dictionary protocol instead
     # of using this bespoke interface. It would be useful for Pandas compatibility as well.
-    def get_metrics(self, timecode: FrameTimecode, metric_keys: ty.Iterable[str]) -> list[ty.Any]:
+    def get_metrics(
+        self, timecode: int | FrameTimecode, metric_keys: ty.Iterable[str]
+    ) -> list[ty.Any]:
         """Return the requested statistics/metrics for a given timecode.
 
         Returns:
@@ -129,7 +134,7 @@ def get_metrics(self, timecode: FrameTimecode, metric_keys: ty.Iterable[str]) ->
         """
         return [self._get_metric(timecode, metric_key) for metric_key in metric_keys]
 
-    def set_metrics(self, timecode: FrameTimecode, metric_kv_dict: dict[str, ty.Any]) -> None:
+    def set_metrics(self, timecode: int | FrameTimecode, metric_kv_dict: dict[str, ty.Any]) -> None:
         """Set Metrics: Sets the provided statistics/metrics for a given frame.
 
         Arguments:
@@ -139,7 +144,7 @@ def set_metrics(self, timecode: FrameTimecode, metric_kv_dict: dict[str, ty.Any]
         for metric_key in metric_kv_dict:
             self._set_metric(timecode, metric_key, metric_kv_dict[metric_key])
 
-    def metrics_exist(self, timecode: FrameTimecode, metric_keys: ty.Iterable[str]) -> bool:
+    def metrics_exist(self, timecode: int | FrameTimecode, metric_keys: ty.Iterable[str]) -> bool:
         """Metrics Exist: Checks if the given metrics/stats exist for the given frame.
 
         Returns:
@@ -188,6 +193,10 @@ def save_to_csv(
         frame_keys = sorted(self._frame_metrics.keys())
         logger.info("Writing %d frames to CSV...", len(frame_keys))
         for frame_key in frame_keys:
+            # `frame_key` may be a bare `int` if the deprecated `load_from_csv` populated the dict.
+            # Skip such rows since we cannot recover a timecode without a base framerate.
+            if not isinstance(frame_key, FrameTimecode):
+                continue
             csv_writer.writerow(
                 [frame_key.frame_num + 1, frame_key.get_timecode()]
                 + [str(metric) for metric in self.get_metrics(frame_key, metric_keys)]
@@ -209,7 +218,7 @@ def valid_header(row: list[str]) -> bool:
 
     # TODO(v1.0): Create a replacement for a calculation cache that functions like load_from_csv
     # did, but is better integrated with detectors for cached calculations instead of statistics.
-    def load_from_csv(self, csv_file: str | bytes | ty.TextIO) -> int | None:
+    def load_from_csv(self, csv_file: StrPath | bytes | ty.TextIO) -> int | None:
         """[DEPRECATED] DO NOT USE
 
         Load all metrics stored in a CSV file into the StatsManager instance. Will be removed in a
@@ -233,7 +242,7 @@ def load_from_csv(self, csv_file: str | bytes | ty.TextIO) -> int | None:
 
         # If we get a path instead of an open file handle, check that it exists, and if so,
         # recursively call ourselves again but with file set instead of path.
-        if isinstance(csv_file, (str, bytes, Path)):
+        if isinstance(csv_file, (str, bytes, os.PathLike)):
             if os.path.exists(csv_file):
                 with open(csv_file) as file:
                     return self.load_from_csv(csv_file=file)
@@ -288,16 +297,18 @@ def load_from_csv(self, csv_file: str | bytes | ty.TextIO) -> int | None:
 
     # TODO: Get rid of these functions and simplify the implementation of this class.
 
-    def _get_metric(self, timecode: FrameTimecode, metric_key: str) -> ty.Any | None:
+    def _get_metric(self, timecode: int | FrameTimecode, metric_key: str) -> ty.Any | None:
         if self._metric_exists(timecode, metric_key):
             return self._frame_metrics[timecode][metric_key]
         return None
 
-    def _set_metric(self, timecode: FrameTimecode, metric_key: str, metric_value: ty.Any) -> None:
+    def _set_metric(
+        self, timecode: int | FrameTimecode, metric_key: str, metric_value: ty.Any
+    ) -> None:
         self._metrics_updated = True
         if timecode not in self._frame_metrics:
             self._frame_metrics[timecode] = dict()
         self._frame_metrics[timecode][metric_key] = metric_value
 
-    def _metric_exists(self, timecode: FrameTimecode, metric_key: str) -> bool:
+    def _metric_exists(self, timecode: int | FrameTimecode, metric_key: str) -> bool:
         return timecode in self._frame_metrics and metric_key in self._frame_metrics[timecode]