[project] Use import typing as ty everywhere

Author: Breakthrough

scenedetect/backends/__init__.py

@@ -83,7 +83,7 @@
 # TODO: Future VideoStream implementations under consideration:
 #  - Nvidia VPF: https://developer.nvidia.com/blog/vpf-hardware-accelerated-video-processing-framework-in-python/
 
-from typing import Dict, Type
+import typing as ty
 
 # OpenCV must be available at minimum.
 from scenedetect.backends.opencv import VideoCaptureAdapter, VideoStreamCv2
@@ -100,7 +100,7 @@
 
 # TODO: Lazy-loading backends would improve startup performance. However, this requires removing
 # some of the re-exported types above from the public API.
-AVAILABLE_BACKENDS: Dict[str, Type] = {
+AVAILABLE_BACKENDS: ty.Dict[str, ty.Type] = {
     backend.BACKEND_NAME: backend
     for backend in filter(
         None,
@@ -114,5 +114,5 @@
 """All available backends that :func:`scenedetect.open_video` can consider for the `backend`
 parameter. These backends must support construction with the following signature:
 
-    BackendType(path: str, framerate: Optional[float])
+    BackendType(path: str, framerate: ty.Optional[float])
 """

scenedetect/backends/moviepy.py

@@ -16,8 +16,8 @@
 image sequences or AviSynth scripts are supported as inputs.
 """
 
+import typing as ty
 from logging import getLogger
-from typing import AnyStr, Optional, Tuple, Union
 
 import cv2
 import numpy as np
@@ -34,7 +34,9 @@
 class VideoStreamMoviePy(VideoStream):
     """MoviePy `FFMPEG_VideoReader` backend."""
 
-    def __init__(self, path: AnyStr, framerate: Optional[float] = None, print_infos: bool = False):
+    def __init__(
+        self, path: ty.AnyStr, framerate: ty.Optional[float] = None, print_infos: bool = False
+    ):
         """Open a video or device.
 
         Arguments:
@@ -64,8 +66,8 @@ def __init__(self, path: AnyStr, framerate: Optional[float] = None, print_infos:
         # This will always be one behind self._reader.lastread when we finally call read()
         # as MoviePy caches the first frame when opening the video. Thus self._last_frame
         # will always be the current frame, and self._reader.lastread will be the next.
-        self._last_frame: Union[bool, np.ndarray] = False
-        self._last_frame_rgb: Optional[np.ndarray] = None
+        self._last_frame: ty.Union[bool, np.ndarray] = False
+        self._last_frame_rgb: ty.Optional[np.ndarray] = None
         # Older versions don't track the video position when calling read_frame so we need
         # to keep track of the current frame number.
         self._frame_number = 0
@@ -86,7 +88,7 @@ def frame_rate(self) -> float:
         return self._reader.fps
 
     @property
-    def path(self) -> Union[bytes, str]:
+    def path(self) -> ty.Union[bytes, str]:
         """Video path."""
         return self._path
 
@@ -101,12 +103,12 @@ def is_seekable(self) -> bool:
         return True
 
     @property
-    def frame_size(self) -> Tuple[int, int]:
+    def frame_size(self) -> ty.Tuple[int, int]:
         """Size of each video frame in pixels as a tuple of (width, height)."""
         return tuple(self._reader.infos["video_size"])
 
     @property
-    def duration(self) -> Optional[FrameTimecode]:
+    def duration(self) -> ty.Optional[FrameTimecode]:
         """Duration of the stream as a FrameTimecode, or None if non terminating."""
         assert isinstance(self._reader.infos["duration"], float)
         return self.base_timecode + self._reader.infos["duration"]
@@ -155,7 +157,7 @@ def frame_number(self) -> int:
         This method will always return 0 if no frames have been `read`."""
         return self._frame_number
 
-    def seek(self, target: Union[FrameTimecode, float, int]):
+    def seek(self, target: ty.Union[FrameTimecode, float, int]):
         """Seek to the given timecode. If given as a frame number, represents the current seek
         pointer (e.g. if seeking to 0, the next frame decoded will be the first frame of the video).
 
@@ -207,7 +209,7 @@ def reset(self, print_infos=False):
         self._eof = False
         self._reader = FFMPEG_VideoReader(self._path, print_infos=print_infos)
 
-    def read(self, decode: bool = True, advance: bool = True) -> Union[np.ndarray, bool]:
+    def read(self, decode: bool = True, advance: bool = True) -> ty.Union[np.ndarray, bool]:
         """Read and decode the next frame as a np.ndarray. Returns False when video ends.
 
         Arguments:

scenedetect/backends/opencv.py

@@ -19,8 +19,8 @@
 
 import math
 import os.path
+import typing as ty
 from logging import getLogger
-from typing import AnyStr, Optional, Tuple, Union
 
 import cv2
 import numpy as np
@@ -58,10 +58,10 @@ class VideoStreamCv2(VideoStream):
 
     def __init__(
         self,
-        path: AnyStr = None,
-        framerate: Optional[float] = None,
+        path: ty.AnyStr = None,
+        framerate: ty.Optional[float] = None,
         max_decode_attempts: int = 5,
-        path_or_device: Union[bytes, str, int] = None,
+        path_or_device: ty.Union[bytes, str, int] = None,
     ):
         """Open a video file, image sequence, or network stream.
 
@@ -98,10 +98,10 @@ def __init__(
         self._is_device = isinstance(self._path_or_device, int)
 
         # Initialized in _open_capture:
-        self._cap: Optional[cv2.VideoCapture] = (
+        self._cap: ty.Optional[cv2.VideoCapture] = (
             None  # Reference to underlying cv2.VideoCapture object.
         )
-        self._frame_rate: Optional[float] = None
+        self._frame_rate: ty.Optional[float] = None
 
         # VideoCapture state
         self._has_grabbed = False
@@ -140,7 +140,7 @@ def frame_rate(self) -> float:
         return self._frame_rate
 
     @property
-    def path(self) -> Union[bytes, str]:
+    def path(self) -> ty.Union[bytes, str]:
         """Video or device path."""
         if self._is_device:
             assert isinstance(self._path_or_device, (int))
@@ -168,15 +168,15 @@ def is_seekable(self) -> bool:
         return not self._is_device
 
     @property
-    def frame_size(self) -> Tuple[int, int]:
+    def frame_size(self) -> ty.Tuple[int, int]:
         """Size of each video frame in pixels as a tuple of (width, height)."""
         return (
             math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
             math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
         )
 
     @property
-    def duration(self) -> Optional[FrameTimecode]:
+    def duration(self) -> ty.Optional[FrameTimecode]:
         """Duration of the stream as a FrameTimecode, or None if non terminating."""
         if self._is_device:
             return None
@@ -218,7 +218,7 @@ def frame_number(self) -> int:
         This method will always return 0 if no frames have been `read`."""
         return math.trunc(self._cap.get(cv2.CAP_PROP_POS_FRAMES))
 
-    def seek(self, target: Union[FrameTimecode, float, int]):
+    def seek(self, target: ty.Union[FrameTimecode, float, int]):
         """Seek to the given timecode. If given as a frame number, represents the current seek
         pointer (e.g. if seeking to 0, the next frame decoded will be the first frame of the video).
 
@@ -264,7 +264,7 @@ def reset(self):
         self._cap.release()
         self._open_capture(self._frame_rate)
 
-    def read(self, decode: bool = True, advance: bool = True) -> Union[np.ndarray, bool]:
+    def read(self, decode: bool = True, advance: bool = True) -> ty.Union[np.ndarray, bool]:
         """Read and decode the next frame as a np.ndarray. Returns False when video ends,
         or the maximum number of decode attempts has passed.
 
@@ -308,7 +308,7 @@ def read(self, decode: bool = True, advance: bool = True) -> Union[np.ndarray, b
     # Private Methods
     #
 
-    def _open_capture(self, framerate: Optional[float] = None):
+    def _open_capture(self, framerate: ty.Optional[float] = None):
         """Opens capture referenced by this object and resets internal state."""
         if self._is_device and self._path_or_device < 0:
             raise ValueError("Invalid/negative device ID specified.")
@@ -364,7 +364,7 @@ class VideoCaptureAdapter(VideoStream):
     def __init__(
         self,
         cap: cv2.VideoCapture,
-        framerate: Optional[float] = None,
+        framerate: ty.Optional[float] = None,
         max_read_attempts: int = 5,
     ):
         """Create from an existing OpenCV VideoCapture object. Used for webcams, live streams,
@@ -448,15 +448,15 @@ def is_seekable(self) -> bool:
         return False
 
     @property
-    def frame_size(self) -> Tuple[int, int]:
+    def frame_size(self) -> ty.Tuple[int, int]:
         """Reported size of each video frame in pixels as a tuple of (width, height)."""
         return (
             math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
             math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
         )
 
     @property
-    def duration(self) -> Optional[FrameTimecode]:
+    def duration(self) -> ty.Optional[FrameTimecode]:
         """Duration of the stream as a FrameTimecode, or None if non terminating."""
         frame_count = math.trunc(self._cap.get(cv2.CAP_PROP_FRAME_COUNT))
         if frame_count > 0:
@@ -500,15 +500,15 @@ def frame_number(self) -> int:
         This method will always return 0 if no frames have been `read`."""
         return self._num_frames
 
-    def seek(self, target: Union[FrameTimecode, float, int]):
+    def seek(self, target: ty.Union[FrameTimecode, float, int]):
         """The underlying VideoCapture is assumed to not support seeking."""
         raise NotImplementedError("Seeking is not supported.")
 
     def reset(self):
         """Not supported."""
         raise NotImplementedError("Reset is not supported.")
 
-    def read(self, decode: bool = True, advance: bool = True) -> Union[np.ndarray, bool]:
+    def read(self, decode: bool = True, advance: bool = True) -> ty.Union[np.ndarray, bool]:
         """Read and decode the next frame as a np.ndarray. Returns False when video ends,
         or the maximum number of decode attempts has passed.
 

scenedetect/backends/pyav.py

@@ -11,8 +11,8 @@
 #
 """:class:`VideoStreamAv` provides an adapter for the PyAV av.InputContainer object."""
 
+import typing as ty
 from logging import getLogger
-from typing import AnyStr, BinaryIO, Optional, Tuple, Union
 
 import av
 import numpy as np
@@ -35,10 +35,10 @@ class VideoStreamAv(VideoStream):
     # calculates the end time.
     def __init__(
         self,
-        path_or_io: Union[AnyStr, BinaryIO],
-        framerate: Optional[float] = None,
-        name: Optional[str] = None,
-        threading_mode: Optional[str] = None,
+        path_or_io: ty.Union[ty.AnyStr, ty.BinaryIO],
+        framerate: ty.Optional[float] = None,
+        name: ty.Optional[str] = None,
+        threading_mode: ty.Optional[str] = None,
         suppress_output: bool = False,
     ):
         """Open a video by path.
@@ -147,12 +147,12 @@ def __del__(self):
     """Unique name used to identify this backend."""
 
     @property
-    def path(self) -> Union[bytes, str]:
+    def path(self) -> ty.Union[bytes, str]:
         """Video path."""
         return self._path
 
     @property
-    def name(self) -> Union[bytes, str]:
+    def name(self) -> ty.Union[bytes, str]:
         """Name of the video, without extension."""
         return self._name
 
@@ -162,7 +162,7 @@ def is_seekable(self) -> bool:
         return self._io.seekable()
 
     @property
-    def frame_size(self) -> Tuple[int, int]:
+    def frame_size(self) -> ty.Tuple[int, int]:
         """Size of each video frame in pixels as a tuple of (width, height)."""
         return (self._codec_context.width, self._codec_context.height)
 
@@ -219,7 +219,7 @@ def aspect_ratio(self) -> float:
         frame_aspect_ratio = self.frame_size[0] / self.frame_size[1]
         return display_aspect_ratio / frame_aspect_ratio
 
-    def seek(self, target: Union[FrameTimecode, float, int]) -> None:
+    def seek(self, target: ty.Union[FrameTimecode, float, int]) -> None:
         """Seek to the given timecode. If given as a frame number, represents the current seek
         pointer (e.g. if seeking to 0, the next frame decoded will be the first frame of the video).
 
@@ -263,7 +263,7 @@ def reset(self):
         except Exception as ex:
             raise VideoOpenFailure() from ex
 
-    def read(self, decode: bool = True, advance: bool = True) -> Union[np.ndarray, bool]:
+    def read(self, decode: bool = True, advance: bool = True) -> ty.Union[np.ndarray, bool]:
         """Read and decode the next frame as a np.ndarray. Returns False when video ends.
 
         Arguments:

scenedetect/detectors/adaptive_detector.py

@@ -16,8 +16,8 @@
 This detector is available from the command-line as the `detect-adaptive` command.
 """
 
+import typing as ty
 from logging import getLogger
-from typing import List, Optional
 
 import numpy as np
 
@@ -42,9 +42,9 @@ def __init__(
         min_content_val: float = 15.0,
         weights: ContentDetector.Components = ContentDetector.DEFAULT_COMPONENT_WEIGHTS,
         luma_only: bool = False,
-        kernel_size: Optional[int] = None,
+        kernel_size: ty.Optional[int] = None,
         video_manager=None,
-        min_delta_hsv: Optional[float] = None,
+        min_delta_hsv: ty.Optional[float] = None,
     ):
         """
         Arguments:
@@ -98,7 +98,7 @@ def __init__(
         self._first_frame_num = None
 
         # NOTE: This must be different than `self._last_scene_cut` which is used by the base class.
-        self._last_cut: Optional[int] = None
+        self._last_cut: ty.Optional[int] = None
 
         self._buffer = []
 
@@ -107,15 +107,15 @@ def event_buffer_length(self) -> int:
         """Number of frames any detected cuts will be behind the current frame due to buffering."""
         return self.window_width
 
-    def get_metrics(self) -> List[str]:
+    def get_metrics(self) -> ty.List[str]:
         """Combines base ContentDetector metric keys with the AdaptiveDetector one."""
         return super().get_metrics() + [self._adaptive_ratio_key]
 
     def stats_manager_required(self) -> bool:
         """Not required for AdaptiveDetector."""
         return False
 
-    def process_frame(self, frame_num: int, frame_img: Optional[np.ndarray]) -> List[int]:
+    def process_frame(self, frame_num: int, frame_img: ty.Optional[np.ndarray]) -> ty.List[int]:
         """Process the next frame. `frame_num` is assumed to be sequential.
 
         Args:
@@ -124,7 +124,7 @@ def process_frame(self, frame_num: int, frame_img: Optional[np.ndarray]) -> List
             frame_img (numpy.ndarray or None): Video frame corresponding to `frame_img`.
 
         Returns:
-            List[int]: List of frames where scene cuts have been detected. There may be 0
+           ty.List[int]: List of frames where scene cuts have been detected. There may be 0
             or more frames in the list, and not necessarily the same as frame_num.
         """
 
@@ -168,7 +168,7 @@ def process_frame(self, frame_num: int, frame_img: Optional[np.ndarray]) -> List
             return [target_frame]
         return []
 
-    def get_content_val(self, frame_num: int) -> Optional[float]:
+    def get_content_val(self, frame_num: int) -> ty.Optional[float]:
         """Returns the average content change for a frame."""
         # TODO(v0.7): Add DeprecationWarning that `get_content_val` will be removed in v0.7.
         logger.error(