[lint] None checks for backends

Author: Breakthrough

scenedetect/backends/moviepy.py

@@ -102,7 +102,7 @@ def __init__(self, path: ty.AnyStr, framerate: float | None = None, print_infos:
         self._frame_number = 0
         # We need to manually keep track of EOF as duration may not be accurate.
         self._eof = False
-        self._aspect_ratio: float = None
+        self._aspect_ratio: float | None = None
 
     #
     # VideoStream Methods/Properties
@@ -211,11 +211,13 @@ def seek(self, target: FrameTimecode | float | int):
         success = False
         if not isinstance(target, FrameTimecode):
             target = FrameTimecode(target, self.frame_rate)
+        duration = self.duration
+        assert duration is not None
         try:
             self._last_frame = _retry_on_oserror(
                 "seek", lambda: self._reader.get_frame(target.seconds)
             )
-            if hasattr(self._reader, "last_read") and target >= self.duration:
+            if hasattr(self._reader, "last_read") and target >= duration:
                 raise SeekError("MoviePy > 2.0 does not have proper EOF semantics (#461).")
             self._frame_number = min(
                 target.frame_num,
@@ -228,7 +230,7 @@ def seek(self, target: FrameTimecode | float | int):
             #
             # We need to ensure consistency for seeking past end of video with respect to errors and
             # behaviour, and should probably gracefully stop at the last frame instead of throwing.
-            if target >= self.duration:
+            if target >= duration:
                 raise SeekError("Target frame is beyond end of video!") from ex
             raise
         finally:
@@ -263,5 +265,6 @@ def read(self, decode: bool = True) -> np.ndarray | bool:
             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
         return not self._eof

scenedetect/backends/opencv.py

@@ -178,6 +178,7 @@ 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)),
@@ -188,11 +189,13 @@ 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
@@ -201,6 +204,7 @@ 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)
@@ -219,18 +223,22 @@ 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: FrameTimecode | float | int):
         if self._is_device:
             raise SeekError("Cannot seek if input is a device!")
         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:
@@ -260,16 +268,20 @@ def seek(self, target: FrameTimecode | float | int):
 
     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(self._frame_rate)
+        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()
         # If we failed to grab the frame, retry a few times if required.
         if not has_grabbed:
-            if self.duration > 0 and self.position < (self.duration - 1):
+            duration = self.duration
+            if duration is not None and duration > 0 and self.position < (duration - 1):
                 for _ in range(self._max_decode_attempts):
                     has_grabbed = self._cap.grab()
                     if has_grabbed:

scenedetect/backends/pyav.py

@@ -68,7 +68,7 @@ def __init__(
             VideoOpenFailure: video could not be opened (may be corrupted)
             ValueError: specified framerate is invalid
         """
-        self._container = None
+        self._container: av.container.InputContainer | None = None
 
         # TODO(https://scenedetect.com/issues/258): See what
         # `self._container.discard_corrupt = True` does with corrupt videos.
@@ -213,7 +213,7 @@ def rate(self) -> Fraction:
         return self._video_stream.guessed_rate
 
     @property
-    def time_base(self) -> Fraction:
+    def time_base(self) -> Fraction | None:
         if self._frame:
             return self._frame.time_base
         return None
@@ -265,6 +265,7 @@ def seek(self, target: FrameTimecode | float | int) -> 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)
@@ -274,6 +275,7 @@ def seek(self, target: FrameTimecode | float | int) -> 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
@@ -288,9 +290,11 @@ 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
+            assert self._decoder is not None
             self._frame = next(self._decoder)
             self._decode_count += 1
         except av.error.EOFError:
@@ -300,6 +304,7 @@ def read(self, decode: bool = True) -> np.ndarray | bool:
             return False
         except StopIteration:
             return False
+        assert self._frame is not None
         return self._frame.to_ndarray(format="bgr24") if decode else True
 
     #
@@ -309,6 +314,7 @@ 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
@@ -365,6 +371,7 @@ 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

scenedetect/common.py

@@ -247,7 +247,7 @@ def __init__(
             raise TypeError("Timecode format/type unrecognized.")
 
     @property
-    def frame_num(self) -> int | None:
+    def frame_num(self) -> int:
         """The frame number. For VFR video or Timecode-backed objects, this is an approximation
         based on the average framerate. Prefer using `pts` and `time_base` for precise timing."""
         if isinstance(self._time, Timecode):