| 265 | |
| 266 | |
| 267 | def load_video(video: str, |
| 268 | num_frames: int = 10, |
| 269 | fps: int = 30, |
| 270 | format: str = "pt", |
| 271 | device: str = "cpu") -> VideoData: |
| 272 | parsed_url = urlparse(video) |
| 273 | results = None |
| 274 | if parsed_url.scheme in ["http", "https", ""]: |
| 275 | results = _load_video_by_cv2(video, num_frames, fps, format, device) |
| 276 | elif parsed_url.scheme == "data": |
| 277 | decoded_video = load_base64_video(video) |
| 278 | # TODO: any ways to read videos from memory, instead of writing to a tempfile? |
| 279 | with tempfile.NamedTemporaryFile(delete=True, |
| 280 | suffix='.mp4') as tmp_file: |
| 281 | tmp_file.write(decoded_video) |
| 282 | tmp_file.flush() |
| 283 | results = _load_video_by_cv2(tmp_file.name, num_frames, fps, format, |
| 284 | device) |
| 285 | else: |
| 286 | raise ValueError(f"Unsupported video scheme: {parsed_url.scheme}") |
| 287 | |
| 288 | return results |
| 289 | |
| 290 | |
| 291 | async def async_load_video(video: str, |