video_stream.py 7.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218
  1. # -*- coding: utf-8 -*-
  2. #
  3. # PySceneDetect: Python-Based Video Scene Detector
  4. # -------------------------------------------------------------------
  5. # [ Site: https://scenedetect.com ]
  6. # [ Docs: https://scenedetect.com/docs/ ]
  7. # [ Github: https://github.com/Breakthrough/PySceneDetect/ ]
  8. #
  9. # Copyright (C) 2014-2024 Brandon Castellano <http://www.bcastell.com>.
  10. # PySceneDetect is licensed under the BSD 3-Clause License; see the
  11. # included LICENSE file, or visit one of the above pages for details.
  12. #
  13. """``scenedetect.video_stream`` Module
  14. This module contains the :class:`VideoStream` class, which provides a library agnostic
  15. interface for video input. To open a video by path, use :func:`scenedetect.open_video`:
  16. .. code:: python
  17. from scenedetect import open_video
  18. video = open_video('video.mp4')
  19. while True:
  20. frame = video.read()
  21. if frame is False:
  22. break
  23. print("Read %d frames" % video.frame_number)
  24. You can also optionally specify a framerate and a specific backend library to use. Unless specified,
  25. OpenCV will be used as the video backend. See :mod:`scenedetect.backends` for a detailed example.
  26. New :class:`VideoStream <scenedetect.video_stream.VideoStream>` implementations can be
  27. tested by adding it to the test suite in `tests/test_video_stream.py`.
  28. """
  29. from abc import ABC, abstractmethod
  30. from typing import Tuple, Optional, Union
  31. import numpy as np
  32. from scenedetect.frame_timecode import FrameTimecode
  33. ##
  34. ## VideoStream Exceptions
  35. ##
  36. class SeekError(Exception):
  37. """Either an unrecoverable error happened while attempting to seek, or the underlying
  38. stream is not seekable (additional information will be provided when possible).
  39. The stream is guaranteed to be left in a valid state, but the position may be reset."""
  40. class VideoOpenFailure(Exception):
  41. """Raised by a backend if opening a video fails."""
  42. # pylint: disable=useless-super-delegation
  43. def __init__(self, message: str = "Unknown backend error."):
  44. """
  45. Arguments:
  46. message: Additional context the backend can provide for the open failure.
  47. """
  48. super().__init__(message)
  49. # pylint: enable=useless-super-delegation
  50. class FrameRateUnavailable(VideoOpenFailure):
  51. """Exception instance to provide consistent error messaging across backends when the video frame
  52. rate is unavailable or cannot be calculated. Subclass of VideoOpenFailure."""
  53. def __init__(self):
  54. super().__init__('Unable to obtain video framerate! Specify `framerate` manually, or'
  55. ' re-encode/re-mux the video and try again.')
  56. ##
  57. ## VideoStream Interface (Base Class)
  58. ##
  59. class VideoStream(ABC):
  60. """ Interface which all video backends must implement. """
  61. #
  62. # Default Implementations
  63. #
  64. @property
  65. def base_timecode(self) -> FrameTimecode:
  66. """FrameTimecode object to use as a time base."""
  67. return FrameTimecode(timecode=0, fps=self.frame_rate)
  68. #
  69. # Abstract Static Methods
  70. #
  71. @staticmethod
  72. @abstractmethod
  73. def BACKEND_NAME() -> str:
  74. """Unique name used to identify this backend. Should be a static property in derived
  75. classes (`BACKEND_NAME = 'backend_identifier'`)."""
  76. raise NotImplementedError
  77. #
  78. # Abstract Properties
  79. #
  80. @property
  81. @abstractmethod
  82. def path(self) -> Union[bytes, str]:
  83. """Video or device path."""
  84. raise NotImplementedError
  85. @property
  86. @abstractmethod
  87. def name(self) -> Union[bytes, str]:
  88. """Name of the video, without extension, or device."""
  89. raise NotImplementedError
  90. @property
  91. @abstractmethod
  92. def is_seekable(self) -> bool:
  93. """True if seek() is allowed, False otherwise."""
  94. raise NotImplementedError
  95. @property
  96. @abstractmethod
  97. def frame_rate(self) -> float:
  98. """Frame rate in frames/sec."""
  99. raise NotImplementedError
  100. @property
  101. @abstractmethod
  102. def duration(self) -> Optional[FrameTimecode]:
  103. """Duration of the stream as a FrameTimecode, or None if non terminating."""
  104. raise NotImplementedError
  105. @property
  106. @abstractmethod
  107. def frame_size(self) -> Tuple[int, int]:
  108. """Size of each video frame in pixels as a tuple of (width, height)."""
  109. raise NotImplementedError
  110. @property
  111. @abstractmethod
  112. def aspect_ratio(self) -> float:
  113. """Pixel aspect ratio as a float (1.0 represents square pixels)."""
  114. raise NotImplementedError
  115. @property
  116. @abstractmethod
  117. def position(self) -> FrameTimecode:
  118. """Current position within stream as FrameTimecode.
  119. This can be interpreted as presentation time stamp, thus frame 1 corresponds
  120. to the presentation time 0. Returns 0 even if `frame_number` is 1."""
  121. raise NotImplementedError
  122. @property
  123. @abstractmethod
  124. def position_ms(self) -> float:
  125. """Current position within stream as a float of the presentation time in
  126. milliseconds. The first frame has a PTS of 0."""
  127. raise NotImplementedError
  128. @property
  129. @abstractmethod
  130. def frame_number(self) -> int:
  131. """Current position within stream as the frame number.
  132. Will return 0 until the first frame is `read`."""
  133. raise NotImplementedError
  134. #
  135. # Abstract Methods
  136. #
  137. @abstractmethod
  138. def read(self, decode: bool = True, advance: bool = True) -> Union[np.ndarray, bool]:
  139. """Read and decode the next frame as a np.ndarray. Returns False when video ends.
  140. Arguments:
  141. decode: Decode and return the frame.
  142. advance: Seek to the next frame. If False, will return the current (last) frame.
  143. Returns:
  144. If decode = True, the decoded frame (np.ndarray), or False (bool) if end of video.
  145. If decode = False, a bool indicating if advancing to the the next frame succeeded.
  146. """
  147. raise NotImplementedError
  148. @abstractmethod
  149. def reset(self) -> None:
  150. """ Close and re-open the VideoStream (equivalent to seeking back to beginning). """
  151. raise NotImplementedError
  152. @abstractmethod
  153. def seek(self, target: Union[FrameTimecode, float, int]) -> None:
  154. """Seek to the given timecode. If given as a frame number, represents the current seek
  155. pointer (e.g. if seeking to 0, the next frame decoded will be the first frame of the video).
  156. For 1-based indices (first frame is frame #1), the target frame number needs to be converted
  157. to 0-based by subtracting one. For example, if we want to seek to the first frame, we call
  158. seek(0) followed by read(). If we want to seek to the 5th frame, we call seek(4) followed
  159. by read(), at which point frame_number will be 5.
  160. May not be supported on all backend types or inputs (e.g. cameras).
  161. Arguments:
  162. target: Target position in video stream to seek to.
  163. If float, interpreted as time in seconds.
  164. If int, interpreted as frame number.
  165. Raises:
  166. SeekError: An error occurs while seeking, or seeking is not supported.
  167. ValueError: `target` is not a valid value (i.e. it is negative).
  168. """
  169. raise NotImplementedError