[lint] Fix more type hint errors

Author: Breakthrough

scenedetect/__init__.py

@@ -105,7 +105,7 @@ def open_video(
         :class:`VideoOpenFailure`: Constructing the VideoStream fails. If multiple backends have
             been attempted, the error from the first backend will be returned.
     """
-    last_error: Exception = None
+    last_error: Exception | None = None
     # If `backend` is available, try to open the video at `path` using it.
     if backend in AVAILABLE_BACKENDS:
         backend_type = AVAILABLE_BACKENDS[backend]
@@ -183,6 +183,6 @@ def detect(
         show_progress=show_progress,
         end_time=end_timecode,
     )
-    if scene_manager.stats_manager is not None:
+    if scene_manager.stats_manager is not None and stats_file_path is not None:
         scene_manager.stats_manager.save_to_csv(csv_file=stats_file_path)
     return scene_manager.get_scene_list(start_in_scene=start_in_scene)

scenedetect/_cli/__init__.py

@@ -66,6 +66,7 @@ def _click_range(section: str, key: str) -> "click.IntRange | click.FloatRange":
     assert isinstance(val, RangeValue), f"Expected RangeValue at {section}/{key}, got {type(val)}"
     return val.click_range
 
+
 # About & copyright message string shown for the 'about' CLI command (scenedetect about).
 ABOUT_STRING = """
 Site: http://scenedetect.com/

scenedetect/_cli/controller.py

@@ -209,11 +209,13 @@ def calculate_timecode(value: str) -> FrameTimecode:
             start_time = context.start_time
             cut_list = [cut for cut in cut_list if cut > context.start_time]
 
-        end_time = context.video_stream.duration
+        video_duration = context.video_stream.duration
+        assert video_duration is not None
+        end_time = video_duration
         if context.end_time is not None:
-            end_time = min(context.end_time, context.video_stream.duration)
+            end_time = min(context.end_time, video_duration)
         elif context.duration is not None:
-            end_time = min(start_time + context.duration, context.video_stream.duration)
+            end_time = min(start_time + context.duration, video_duration)
 
         cut_list = [cut for cut in cut_list if cut < end_time]
         scene_list = get_scenes_from_cuts(cut_list=cut_list, start_pos=start_time, end_pos=end_time)

scenedetect/backends/moviepy.py

@@ -267,10 +267,8 @@ def read(self, decode: bool = True) -> np.ndarray | bool:
                 return False
             self._eof = True
         self._frame_number += 1
-        if decode:
-            last_frame_valid = self._last_frame is not None and self._last_frame is not False
-            if last_frame_valid:
-                self._last_frame_rgb = cv2.cvtColor(self._last_frame, cv2.COLOR_BGR2RGB)
-                assert self._last_frame_rgb is not None
-                return self._last_frame_rgb
+        if decode and isinstance(self._last_frame, np.ndarray):
+            self._last_frame_rgb = cv2.cvtColor(self._last_frame, cv2.COLOR_BGR2RGB)
+            assert self._last_frame_rgb is not None
+            return self._last_frame_rgb
         return not self._eof

scenedetect/backends/pyav.py

@@ -74,7 +74,7 @@ def __init__(
             VideoOpenFailure: video could not be opened (may be corrupted)
             ValueError: specified framerate is invalid
         """
-        self._container: av.container.InputContainer | None = None
+        self._container: av.container.InputContainer | None = None  # type: ignore[name-defined]
 
         # TODO(https://scenedetect.com/issues/258): See what
         # `self._container.discard_corrupt = True` does with corrupt videos.
@@ -93,15 +93,15 @@ def __init__(
 
         if threading_mode:
             try:
-                threading_mode = av.codec.context.ThreadType[threading_mode.upper()]
+                threading_mode = av.codec.context.ThreadType[threading_mode.upper()]  # type: ignore[attr-defined]
             except KeyError as _:
                 raise ValueError(
                     f"Invalid threading mode! Must be one of: {VALID_THREAD_MODES}"
                 ) from None
 
         if not suppress_output:
             logger.debug("Restoring default ffmpeg log callbacks.")
-            av.logging.restore_default_callback()
+            av.logging.restore_default_callback()  # type: ignore[attr-defined]
 
         try:
             if isinstance(path_or_io, (str, os.PathLike)):
@@ -305,7 +305,7 @@ def read(self, decode: bool = True) -> np.ndarray | bool:
             assert self._decoder is not None
             self._frame = next(self._decoder)
             self._decode_count += 1
-        except av.error.EOFError:
+        except av.error.EOFError:  # type: ignore[attr-defined]
             self._frame = last_frame
             if self._handle_eof():
                 return self.read(decode)

scenedetect/detector.py

@@ -134,10 +134,10 @@ def __init__(self, mode: Mode, length: int | float | str):
             self._filter_secs = FrameTimecode(timecode=length, fps=100.0).seconds
         else:
             self._filter_length = int(length)
-        self._last_above = None  # Last frame above threshold.
+        self._last_above: FrameTimecode | None = None  # Last frame above threshold.
         self._merge_enabled = False  # Used to disable merging until at least one cut was found.
         self._merge_triggered = False  # True when the merge filter is active.
-        self._merge_start = None  # Frame number where we started the merge filter.
+        self._merge_start: FrameTimecode | None = None  # Frame where we started merging.
 
     @property
     def max_behind(self) -> int:
@@ -165,12 +165,16 @@ def filter(self, timecode: FrameTimecode, above_threshold: bool) -> list[FrameTi
             return self._filter_suppress(timecode=timecode, above_threshold=above_threshold)
         raise RuntimeError("Unhandled FlashFilter mode.")
 
-    def _filter_suppress(self, timecode: FrameTimecode, above_threshold: bool) -> list[int]:
-        assert timecode.framerate >= 0
+    def _filter_suppress(
+        self, timecode: FrameTimecode, above_threshold: bool
+    ) -> list[FrameTimecode]:
+        framerate = timecode.framerate
+        assert framerate is not None and framerate >= 0
+        assert self._last_above is not None
         # Compute the threshold in seconds once from the first frame's framerate. This avoids
         # using an incorrect average fps (e.g. OpenCV on VFR video) on subsequent frames.
         if self._filter_secs is None:
-            self._filter_secs = self._filter_length / timecode.framerate
+            self._filter_secs = self._filter_length / framerate
         min_length_met: bool = (timecode - self._last_above) >= self._filter_secs
         if not (above_threshold and min_length_met):
             return []
@@ -179,17 +183,20 @@ def _filter_suppress(self, timecode: FrameTimecode, above_threshold: bool) -> li
         self._last_above = timecode
         return [timecode]
 
-    def _filter_merge(self, timecode: FrameTimecode, above_threshold: bool) -> list[int]:
-        assert timecode.framerate >= 0
+    def _filter_merge(self, timecode: FrameTimecode, above_threshold: bool) -> list[FrameTimecode]:
+        framerate = timecode.framerate
+        assert framerate is not None and framerate >= 0
+        assert self._last_above is not None
         # Compute the threshold in seconds once from the first frame's framerate.
         if self._filter_secs is None:
-            self._filter_secs = self._filter_length / timecode.framerate
+            self._filter_secs = self._filter_length / framerate
         min_length_met: bool = (timecode - self._last_above) >= self._filter_secs
         # Ensure last frame is always advanced to the most recent one that was above the threshold.
         if above_threshold:
             self._last_above = timecode
         if self._merge_triggered:
             # This frame was under the threshold, see if enough frames passed to disable the filter.
+            assert self._merge_start is not None
             if (
                 min_length_met
                 and not above_threshold

scenedetect/detectors/hash_detector.py

@@ -57,8 +57,8 @@ def __init__(
         self._size = size
         self._size_sq = float(size * size)
         self._factor = lowpass
-        self._last_frame: numpy.ndarray = None
-        self._last_scene_cut: FrameTimecode = None
+        self._last_frame: numpy.ndarray | None = None
+        self._last_scene_cut: FrameTimecode | None = None
         self._last_hash = numpy.array([])
         self._metric_key = f"hash_dist [size={self._size} lowpass={self._factor}]"
 
@@ -136,7 +136,7 @@ def hash_frame(frame_img, hash_size, factor) -> numpy.ndarray:
             max_value = 1
 
         # Calculate discrete cosine tranformation of the image
-        resized_img = numpy.float32(resized_img) / max_value
+        resized_img = numpy.asarray(numpy.float32(resized_img) / max_value)
         dct_complete = cv2.dct(resized_img)
 
         # Only keep the low frequency information

scenedetect/detectors/threshold_detector.py

@@ -15,6 +15,7 @@
 This detector is available from the command-line as the `detect-threshold` command.
 """
 
+import typing as ty
 import warnings
 from enum import Enum
 from logging import getLogger
@@ -82,12 +83,12 @@ def __init__(
         self.fade_bias = fade_bias
         self.min_scene_len = min_scene_len
         self.processed_frame = False
-        self.last_scene_cut = None
+        self.last_scene_cut: FrameTimecode | None = None
         # Whether to add an additional scene or not when ending on a fade out
         # (as cuts are only added on fade ins; see post_process() for details).
         self.add_final_scene = add_final_scene
         # Where the last fade (threshold crossing) was detected.
-        self.last_fade = {
+        self.last_fade: dict[str, ty.Any] = {
             "frame": None,  # FrameTimecode where the last detected fade is
             "type": None,  # type of fade, can be either 'in' or 'out'
         }
@@ -174,14 +175,12 @@ def post_process(self, timecode: FrameTimecode) -> list[FrameTimecode]:
         # scene break to indicate the end of the scene.  This is only done for
         # fade-outs, as a scene cut is already added when a fade-in is found.
         cuts: list[FrameTimecode] = []
+        elapsed = timecode if self.last_scene_cut is None else timecode - self.last_scene_cut
         if (
             self.last_fade["type"] == "out"
             and self.add_final_scene
             and self.last_fade["frame"] is not None
-            and (
-                (self.last_scene_cut is None and timecode >= self.min_scene_len)
-                or (timecode - self.last_scene_cut) >= self.min_scene_len
-            )
+            and elapsed >= self.min_scene_len
         ):
             cuts.append(self.last_fade["frame"])
         return cuts

scenedetect/output/__init__.py

@@ -244,10 +244,12 @@ def write_scene_list_html(
 def _edl_timecode(timecode: FrameTimecode) -> str:
     """Format `timecode` as ``HH:MM:SS:FF`` for a CMX 3600 EDL entry."""
     total_seconds = timecode.seconds
+    framerate = timecode.framerate
+    assert framerate is not None
     hours = int(total_seconds // 3600)
     minutes = int((total_seconds % 3600) // 60)
     seconds = int(total_seconds % 60)
-    frames_part = int((total_seconds * timecode.framerate) % timecode.framerate)
+    frames_part = int((total_seconds * framerate) % framerate)
     return f"{hours:02d}:{minutes:02d}:{seconds:02d}:{frames_part:02d}"
 
 

scenedetect/output/image.py

@@ -44,6 +44,7 @@ def _generate_timecode_list(
     `frame_margin` accepts an int (frames), float (seconds), or str (e.g. ``"0.1s"``).
     """
     framerate = scene_list[0][0].framerate
+    assert framerate is not None
     margin_secs = FrameTimecode(timecode=frame_margin, fps=framerate).seconds
     result = []
     for start, end in scene_list:
@@ -71,7 +72,7 @@ def _generate_timecode_list(
 
 def _scale_image(
     image: np.ndarray,
-    aspect_ratio: float,
+    aspect_ratio: float | None,
     height: int | None,
     width: int | None,
     scale: float | None,
@@ -90,9 +91,11 @@ def _scale_image(
         if height and not width:
             factor = height / float(image_height)
             width = int(factor * image_width)
-        if width and not height:
+        elif width and not height:
             factor = width / float(image_width)
             height = int(factor * image_height)
+        assert height is not None
+        assert width is not None
         assert height > 0 and width > 0
         image = cv2.resize(image, (width, height), interpolation=interpolation.value)
     elif scale:
@@ -106,7 +109,7 @@ def __init__(
         num_images: int = 3,
         frame_margin: int | float | str = 1,
         image_extension: str = "jpg",
-        imwrite_param: dict[str, int | None] | None = None,
+        imwrite_param: list[int] | None = None,
         image_name_template: str = "$VIDEO_NAME-Scene-$SCENE_NUMBER-$IMAGE_NUMBER",
         scale: float | None = None,
         height: int | None = None,
@@ -153,7 +156,7 @@ def __init__(
         self._height = height
         self._width = width
         self._interpolation = interpolation
-        self._imwrite_param = imwrite_param if imwrite_param else {}
+        self._imwrite_param: list[int] = imwrite_param if imwrite_param is not None else []
 
     def run(
         self,
@@ -337,7 +340,7 @@ def generate_timecode_list(self, scene_list: SceneList) -> list[list[FrameTimeco
     def resize_image(
         self,
         image: np.ndarray,
-        aspect_ratio: float,
+        aspect_ratio: float | None,
     ) -> np.ndarray:
         return _scale_image(
             image, aspect_ratio, self._height, self._width, self._scale, self._interpolation
@@ -436,7 +439,7 @@ def save_images(
             width,
             interpolation,
         )
-        return extractor.run(video, scene_list, output_dir, show_progress)
+        return extractor.run(video, scene_list, output_dir, bool(show_progress))
 
     # Setup flags and init progress bar if available.
     completed = True
@@ -467,7 +470,7 @@ def save_images(
         for j, image_timecode in enumerate(scene_timecodes):
             video.seek(image_timecode)
             frame_im = video.read()
-            if frame_im is not None and frame_im is not False:
+            if isinstance(frame_im, np.ndarray):
                 # TODO: Add extension to template.
                 # TODO: Allow NUM to be a valid suffix in addition to NUMBER.
                 file_path = "{}.{}".format(
@@ -495,9 +498,11 @@ def save_images(
                     if height and not width:
                         factor = height / float(frame_height)
                         width = int(factor * frame_width)
-                    if width and not height:
+                    elif width and not height:
                         factor = width / float(frame_width)
                         height = int(factor * frame_height)
+                    assert height is not None
+                    assert width is not None
                     assert height > 0 and width > 0
                     frame_im = cv2.resize(
                         frame_im, (width, height), interpolation=interpolation.value