| 41 | |
| 42 | |
| 43 | class VideoReader: |
| 44 | def __init__(self, video_path): |
| 45 | if not os.path.isfile(video_path): |
| 46 | raise ValueError(f'Video path "{video_path}" does not point to a file.') |
| 47 | self.video_path = video_path |
| 48 | self.video = cv2.VideoCapture(video_path) |
| 49 | if not self.video.isOpened(): |
| 50 | raise OSError("Video could not be opened; it may be corrupted.") |
| 51 | self.parse_metadata() |
| 52 | self._bbox = 0, 1, 0, 1 |
| 53 | self._n_frames_robust = None |
| 54 | |
| 55 | def __repr__(self): |
| 56 | string = "Video (duration={:0.2f}, fps={}, dimensions={}x{})" |
| 57 | return string.format(self.calc_duration(), self.fps, *self.dimensions) |
| 58 | |
| 59 | def __len__(self): |
| 60 | return self._n_frames |
| 61 | |
| 62 | def check_integrity(self): |
| 63 | dest = os.path.join(self.directory, f"{self.name}.log") |
| 64 | command = f'ffmpeg -v error -i "{self.video_path}" -f null - 2>"{dest}"' |
| 65 | subprocess.call(command, shell=True) |
| 66 | if os.path.getsize(dest) != 0: |
| 67 | warnings.warn(f'Video contains errors. See "{dest}" for a detailed report.', stacklevel=2) |
| 68 | |
| 69 | def check_integrity_robust(self): |
| 70 | numframes = self.video.get(cv2.CAP_PROP_FRAME_COUNT) |
| 71 | fr = 0 |
| 72 | while fr < numframes: |
| 73 | success, frame = self.video.read() |
| 74 | if not success or frame is None: |
| 75 | warnings.warn(f"Opencv failed to load frame {fr}. Use ffmpeg to re-encode video file", stacklevel=2) |
| 76 | fr += 1 |
| 77 | |
| 78 | @property |
| 79 | def name(self): |
| 80 | return os.path.splitext(os.path.split(self.video_path)[1])[0] |
| 81 | |
| 82 | @property |
| 83 | def format(self): |
| 84 | return os.path.splitext(self.video_path)[1] |
| 85 | |
| 86 | @property |
| 87 | def directory(self): |
| 88 | return os.path.dirname(self.video_path) |
| 89 | |
| 90 | @property |
| 91 | def metadata(self): |
| 92 | return dict(n_frames=len(self), fps=self.fps, width=self.width, height=self.height) |
| 93 | |
| 94 | def get_n_frames(self, robust=False): |
| 95 | if not robust: |
| 96 | return self._n_frames |
| 97 | elif not self._n_frames_robust: |
| 98 | command = ( |
| 99 | f'ffprobe -i "{self.video_path}" -v error -count_frames ' |
| 100 | f"-select_streams v:0 -show_entries stream=nb_read_frames " |
no outgoing calls
no test coverage detected