(video: str,
num_frames: int = 10,
fps: int = 30,
format: str = "pt",
device: str = "cpu")
| 289 | |
| 290 | |
| 291 | async def async_load_video(video: str, |
| 292 | num_frames: int = 10, |
| 293 | fps: int = 30, |
| 294 | format: str = "pt", |
| 295 | device: str = "cpu") -> VideoData: |
| 296 | assert format in ["pt", "pil"], "format must be either Pytorch or PIL" |
| 297 | |
| 298 | parsed_url = urlparse(video) |
| 299 | |
| 300 | if parsed_url.scheme in ["http", "https"]: |
| 301 | async with aiohttp.ClientSession() as session: |
| 302 | async with session.get(video) as response: |
| 303 | with tempfile.NamedTemporaryFile(delete=True, |
| 304 | suffix='.mp4') as tmp: |
| 305 | tmp.write(await response.content.read()) |
| 306 | tmp.flush() |
| 307 | results = _load_video_by_cv2(tmp.name, num_frames, fps, |
| 308 | format, device) |
| 309 | elif parsed_url.scheme == "data": |
| 310 | decoded_video = load_base64_video(video) |
| 311 | # TODO: any ways to read videos from memory, instead of writing to a tempfile? |
| 312 | with tempfile.NamedTemporaryFile(delete=True, |
| 313 | suffix='.mp4') as tmp_file: |
| 314 | tmp_file.write(decoded_video) |
| 315 | tmp_file.flush() |
| 316 | results = _load_video_by_cv2(tmp_file.name, num_frames, fps, format, |
| 317 | device) |
| 318 | else: |
| 319 | results = _load_video_by_cv2(video, num_frames, fps, format, device) |
| 320 | return results |
| 321 | |
| 322 | |
| 323 | def load_audio( |
no test coverage detected