(self, decode: bool = True)
| 278 | self._open_capture(float(self._frame_rate)) |
| 279 | |
| 280 | def read(self, decode: bool = True) -> np.ndarray | bool: |
| 281 | if not self._cap.isOpened(): |
| 282 | return False |
| 283 | has_grabbed = self._cap.grab() |
| 284 | # If we failed to grab the frame, retry a few times if required. |
| 285 | if not has_grabbed: |
| 286 | duration = self.duration |
| 287 | if duration is not None and duration > 0 and self.position < (duration - 1): |
| 288 | for _ in range(self._max_decode_attempts): |
| 289 | has_grabbed = self._cap.grab() |
| 290 | if has_grabbed: |
| 291 | break |
| 292 | # Report previous failure in debug mode. |
| 293 | if has_grabbed: |
| 294 | self._decode_failures += 1 |
| 295 | logger.debug("Frame failed to decode.") |
| 296 | if not self._warning_displayed and self._decode_failures > 1: |
| 297 | logger.warning("Failed to decode some frames, results may be inaccurate.") |
| 298 | # We didn't manage to grab a frame even after retrying, so just return. |
| 299 | if not has_grabbed: |
| 300 | return False |
| 301 | self._has_grabbed = True |
| 302 | # Need to make sure we actually grabbed a frame before calling retrieve. |
| 303 | if decode and self._has_grabbed: |
| 304 | _, frame = self._cap.retrieve() |
| 305 | return frame |
| 306 | return self._has_grabbed |
| 307 | |
| 308 | # |
| 309 | # Private Methods |
no outgoing calls