| 1321 | |
| 1322 | |
| 1323 | class _VideoWriterSet: |
| 1324 | def __init__(self, writers: dict[str, Any]) -> None: |
| 1325 | self._writers = dict(writers) |
| 1326 | |
| 1327 | @classmethod |
| 1328 | def open( |
| 1329 | cls, |
| 1330 | paths: Mapping[str, Path], |
| 1331 | *, |
| 1332 | fps: float, |
| 1333 | quality: int, |
| 1334 | ) -> "_VideoWriterSet": |
| 1335 | if not paths: |
| 1336 | return cls({}) |
| 1337 | backend = _preferred_video_backend() |
| 1338 | if backend is None: |
| 1339 | raise RuntimeError( |
| 1340 | "MP4 writing requires either imageio with an ffmpeg/pyav backend " |
| 1341 | "or OpenCV. Install one of those or pass --no-video." |
| 1342 | ) |
| 1343 | if backend == "opencv": |
| 1344 | return cls( |
| 1345 | { |
| 1346 | camera_id: _OpenCvVideoWriter(path, fps=float(fps)) |
| 1347 | for camera_id, path in paths.items() |
| 1348 | } |
| 1349 | ) |
| 1350 | |
| 1351 | import imageio.v2 as imageio |
| 1352 | |
| 1353 | writers: dict[str, Any] = {} |
| 1354 | try: |
| 1355 | for camera_id, path in paths.items(): |
| 1356 | path.parent.mkdir(parents=True, exist_ok=True) |
| 1357 | if path.exists(): |
| 1358 | path.unlink() |
| 1359 | writers[camera_id] = imageio.get_writer( |
| 1360 | str(path), |
| 1361 | fps=float(fps), |
| 1362 | codec="libx264", |
| 1363 | quality=int(quality), |
| 1364 | macro_block_size=1, |
| 1365 | ffmpeg_params=["-pix_fmt", "yuv420p"], |
| 1366 | ) |
| 1367 | except BaseException: |
| 1368 | for writer in writers.values(): |
| 1369 | writer.close() |
| 1370 | raise |
| 1371 | return cls(writers) |
| 1372 | |
| 1373 | def append(self, camera_id: str, rgb: Any) -> None: |
| 1374 | writer = self._writers.get(camera_id) |
| 1375 | if writer is None: |
| 1376 | raise KeyError(f"No video writer for camera: {camera_id}") |
| 1377 | writer.append_data(rgb) |
| 1378 | |
| 1379 | def close(self) -> None: |
| 1380 | errors: list[BaseException] = [] |
nothing calls this directly
no outgoing calls
no test coverage detected