| 1399 | self._path.unlink() |
| 1400 | |
| 1401 | def append_data(self, rgb: Any) -> None: |
| 1402 | import cv2 |
| 1403 | import numpy as np |
| 1404 | |
| 1405 | array = np.asarray(rgb) |
| 1406 | if array.ndim != 3 or array.shape[2] != 3: |
| 1407 | raise RuntimeError(f"Expected RGB frame with shape HxWx3, got {array.shape}.") |
| 1408 | height, width = int(array.shape[0]), int(array.shape[1]) |
| 1409 | if self._writer is None: |
| 1410 | fourcc = cv2.VideoWriter_fourcc(*"mp4v") |
| 1411 | self._writer = cv2.VideoWriter( |
| 1412 | str(self._path), |
| 1413 | fourcc, |
| 1414 | self._fps, |
| 1415 | (width, height), |
| 1416 | ) |
| 1417 | if not self._writer.isOpened(): |
| 1418 | raise RuntimeError(f"OpenCV failed to open MP4 writer: {self._path}") |
| 1419 | self._frame_shape = (height, width) |
| 1420 | elif self._frame_shape != (height, width): |
| 1421 | raise RuntimeError( |
| 1422 | f"Video frame size changed for {self._path}: " |
| 1423 | f"expected {self._frame_shape}, got {(height, width)}." |
| 1424 | ) |
| 1425 | |
| 1426 | frame_bgr = np.ascontiguousarray(array[:, :, ::-1]) |
| 1427 | self._writer.write(frame_bgr) |
| 1428 | |
| 1429 | def close(self) -> None: |
| 1430 | if self._writer is not None: |