(
image: Union[str, Image.Image],
format: str = "pt",
device: str = "cpu")
| 148 | |
| 149 | |
| 150 | async def async_load_image( |
| 151 | image: Union[str, Image.Image], |
| 152 | format: str = "pt", |
| 153 | device: str = "cpu") -> Union[Image.Image, torch.Tensor]: |
| 154 | assert format in ["pt", "pil"], "format must be either Pytorch or PIL" |
| 155 | |
| 156 | if isinstance(image, Image.Image): |
| 157 | return image.convert('RGB') |
| 158 | |
| 159 | parsed_url = urlparse(image) |
| 160 | |
| 161 | if parsed_url.scheme in ["http", "https"]: |
| 162 | async with aiohttp.ClientSession() as session: |
| 163 | async with session.get(image) as response: |
| 164 | content = await response.read() |
| 165 | image = _load_and_convert_image(BytesIO(content)) |
| 166 | elif parsed_url.scheme == "data": |
| 167 | image = load_base64_image(parsed_url) |
| 168 | else: |
| 169 | image = _load_and_convert_image(Path(parsed_url.path)) |
| 170 | |
| 171 | if format == "pt": |
| 172 | return ToTensor()(image).to(device=device) |
| 173 | else: |
| 174 | return image |
| 175 | |
| 176 | |
| 177 | def _load_video_by_cv2(video: str, |
no test coverage detected