[backends] Tighten None safety, reduce unnecessary None checks

Author: Breakthrough

pyproject.toml

@@ -61,15 +61,35 @@ unfixable = []
 
 [tool.pyright]
 include = ["scenedetect", "tests"]
+# 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
+# fills with cascading "import could not be resolved" noise.
+# Per-developer venv paths are deliberately NOT pinned here.
+#
+# Modes: "off" | "basic" | "standard" | "strict".
+# We're at "basic" for 0.7. Bumping requires the cleanup in TODO(0.8) below.
 typeCheckingMode = "basic"
-# cv2, av, and moviepy ship without type stubs; these reports generate
-# unactionable noise without catching real issues in this codebase.
+
+# Third-party noise: cv2, av, moviepy ship without (or with partial) type
+# stubs; these reports are unactionable in this codebase.
 reportMissingTypeStubs = "none"
 reportUnknownMemberType = "none"
 reportUnknownArgumentType = "none"
 reportUnknownVariableType = "none"
 reportUnknownParameterType = "none"
+
+# Click + pytest decorators are conventionally untyped.
+reportUntypedFunctionDecorator = "none"
+
+# TODO(0.8): Audit the rules below. They were added globally with a
+# third-party justification, but they affect the whole codebase. After the
+# remaining first-party type-debt is cleared (FrameTimecode/TimecodeLike in
+# common.py, _cli/config.py CropValue overloads, Click CLI typing in
+# _cli/__init__.py), re-enable each in isolation and decide keep/drop:
+#   reportMissingParameterType  — first-party coverage looks complete
+#   reportMissingTypeArgument   — modern Python defaults absorb most
+#   reportPrivateUsage          — no cross-module _private access seen
 reportMissingParameterType = "none"
 reportMissingTypeArgument = "none"
-reportUntypedFunctionDecorator = "none"
 reportPrivateUsage = "none"

scenedetect/backends/opencv.py

@@ -119,18 +119,13 @@ def __init__(
         self._path_or_device: str | int = resolved
         self._is_device = isinstance(self._path_or_device, int)
 
-        # Initialized in _open_capture:
-        self._cap: cv2.VideoCapture | None = (
-            None  # Reference to underlying cv2.VideoCapture object.
-        )
-        self._frame_rate: Fraction | None = None
-
         # VideoCapture state
         self._has_grabbed = False
         self._max_decode_attempts = max_decode_attempts
         self._decode_failures = 0
         self._warning_displayed = False
 
+        # `_open_capture` populates `_cap` and `_frame_rate`.
         self._open_capture(framerate)
 
     #
@@ -145,7 +140,6 @@ def capture(self) -> cv2.VideoCapture:
         backing this object. Seeking or using the read/grab methods through this property are
         unsupported and will leave this object in an inconsistent state.
         """
-        assert self._cap
         return self._cap
 
     #
@@ -157,7 +151,6 @@ def capture(self) -> cv2.VideoCapture:
 
     @property
     def frame_rate(self) -> Fraction:
-        assert self._frame_rate
         return self._frame_rate
 
     @property
@@ -189,7 +182,6 @@ def is_seekable(self) -> bool:
     @property
     def frame_size(self) -> tuple[int, int]:
         """Size of each video frame in pixels as a tuple of (width, height)."""
-        assert self._cap is not None
         return (
             math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
             math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
@@ -200,13 +192,11 @@ def duration(self) -> FrameTimecode | None:
         """Duration of the stream as a FrameTimecode, or None if non terminating."""
         if self._is_device:
             return None
-        assert self._cap is not None
         return self.base_timecode + math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_COUNT))
 
     @property
     def aspect_ratio(self) -> float:
         """Display/pixel aspect ratio as a float (1.0 represents square pixels)."""
-        assert self._cap is not None
         return _get_aspect_ratio(self._cap)
 
     @property
@@ -215,7 +205,6 @@ def timecode(self) -> Timecode:
         # *NOTE*: Although OpenCV has `CAP_PROP_PTS`, it doesn't seem to be reliable. For now, we
         # use `CAP_PROP_POS_MSEC` instead, converting to microseconds for sufficient precision to
         # avoid frame-boundary rounding errors at common framerates like 24000/1001.
-        assert self._cap is not None
         ms = self._cap.get(cv2.CAP_PROP_POS_MSEC)
         time_base = Fraction(1, 1000000)
         return Timecode(pts=round(ms * 1000), time_base=time_base)
@@ -234,12 +223,10 @@ def position(self) -> FrameTimecode:
 
     @property
     def position_ms(self) -> float:
-        assert self._cap is not None
         return self._cap.get(cv2.CAP_PROP_POS_MSEC)
 
     @property
     def frame_number(self) -> int:
-        assert self._cap is not None
         return math.trunc(self._cap.get(cv2.CAP_PROP_POS_FRAMES))
 
     def seek(self, target: TimecodeLike):
@@ -249,9 +236,6 @@ def seek(self, target: TimecodeLike):
             target = FrameTimecode(target, self.frame_rate)
         if target < 0:
             raise ValueError("Target seek position cannot be negative!")
-        assert self._cap is not None
-
-        assert self._frame_rate is not None
         target_secs = (self.base_timecode + target).seconds
         self._has_grabbed = False
         if target_secs > 0:
@@ -281,13 +265,10 @@ def seek(self, target: TimecodeLike):
 
     def reset(self):
         """Close and re-open the VideoStream (should be equivalent to calling `seek(0)`)."""
-        assert self._cap is not None
-        assert self._frame_rate is not None
         self._cap.release()
         self._open_capture(float(self._frame_rate))
 
     def read(self, decode: bool = True) -> np.ndarray | bool:
-        assert self._cap is not None
         if not self._cap.isOpened():
             return False
         has_grabbed = self._cap.grab()
@@ -364,8 +345,8 @@ def _open_capture(self, framerate: float | None = None):
             if framerate < MAX_FPS_DELTA:
                 raise FrameRateUnavailable()
 
-        self._cap = cap
-        self._frame_rate = framerate_to_fraction(framerate)
+        self._cap: cv2.VideoCapture = cap
+        self._frame_rate: Fraction = framerate_to_fraction(framerate)
         self._has_grabbed = False
         cap.set(cv2.CAP_PROP_ORIENTATION_AUTO, 1.0)  # https://github.com/opencv/opencv/issues/26795
 
@@ -430,7 +411,6 @@ def capture(self) -> cv2.VideoCapture:
         backing this object. Using the read/grab methods through this property are unsupported and
         will leave this object in an inconsistent state.
         """
-        assert self._cap
         return self._cap
 
     #
@@ -443,7 +423,6 @@ def capture(self) -> cv2.VideoCapture:
     @property
     def frame_rate(self) -> Fraction:
         """Framerate in frames/sec."""
-        assert self._frame_rate
         return self._frame_rate
 
     @property

scenedetect/backends/pyav.py

@@ -74,8 +74,6 @@ def __init__(
             VideoOpenFailure: video could not be opened (may be corrupted)
             ValueError: specified framerate is invalid
         """
-        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.
         super().__init__()
@@ -113,7 +111,7 @@ def __init__(
             else:
                 self._io = path_or_io
 
-            self._container = av.open(self._io)
+            self._container: av.container.InputContainer = av.open(self._io)  # type: ignore[attr-defined]
             if threading_mode is not None:
                 self._video_stream.thread_type = threading_mode
                 self._reopened = False
@@ -145,8 +143,10 @@ def __init__(
         self._duration_frames = self._get_duration()
 
     def __del__(self):
-        if self._container is not None:
-            self._container.close()
+        # `_container` is unset if `__init__` raised before `av.open()` succeeded.
+        container = getattr(self, "_container", None)
+        if container is not None:
+            container.close()
 
     #
     # VideoStream Methods/Properties
@@ -273,7 +273,6 @@ def seek(self, target: TimecodeLike) -> None:
         self._frame = None
         self._decoder = None
         self._decode_count = 0
-        assert self._container is not None
         self._container.seek(target_pts, stream=self._video_stream)
         if not beginning:
             self.read(decode=False)
@@ -283,7 +282,6 @@ def seek(self, target: TimecodeLike) -> None:
 
     def reset(self):
         """Close and re-open the VideoStream (should be equivalent to calling `seek(0)`)."""
-        assert self._container is not None
         self._container.close()
         self._frame = None
         self._decoder = None
@@ -298,7 +296,6 @@ def read(self, decode: bool = True) -> np.ndarray | bool:
         # B-frame reordering) is never flushed prematurely. Creating a new generator each call
         # caused the last buffered frame to be lost at EOF.
         if self._decoder is None:
-            assert self._container is not None
             self._decoder = self._container.decode(video=0)
         try:
             last_frame = self._frame
@@ -322,7 +319,6 @@ def read(self, decode: bool = True) -> np.ndarray | bool:
     @property
     def _video_stream(self):
         """PyAV `av.video.stream.VideoStream` being used."""
-        assert self._container is not None
         return self._container.streams.video[0]
 
     @property
@@ -379,7 +375,6 @@ def _handle_eof(self):
         except:
             self._io.seek(orig_pos)
             raise
-        assert self._container is not None
         self._container.close()
         self._container = container
         self._decoder = None