1
0
Fork 0
mirror of https://github.com/ldericher/fftcgtool synced 2025-01-15 23:03:00 +00:00
fftcgtool/fftcg/imageloader.py

73 lines
2 KiB
Python
Raw Normal View History

2021-08-04 01:39:19 +00:00
import io
import logging
import queue
2021-08-04 01:41:14 +00:00
import threading
2021-08-04 01:39:19 +00:00
import requests
2021-08-04 01:41:14 +00:00
from PIL import Image
2021-08-04 01:39:19 +00:00
class ImageLoader(threading.Thread):
2021-08-09 03:32:05 +00:00
def __init__(self, url_queue: queue.Queue, resolution: tuple[int, int], language: str):
2021-08-09 04:47:59 +00:00
super().__init__()
2021-08-04 01:39:19 +00:00
self.__queue = url_queue
2021-08-04 01:39:19 +00:00
self.__resolution = resolution
self.__language = language
self.__images = {}
def run(self) -> None:
logger = logging.getLogger(__name__)
2021-08-04 16:36:23 +00:00
2021-08-04 01:39:19 +00:00
while not self.__queue.empty():
# take next url
url = self.__queue.get()
2021-08-04 01:39:19 +00:00
# fetch image (retry on fail)
2021-08-04 01:39:19 +00:00
while True:
logger.info(f"downloading image {url}")
2021-08-04 01:39:19 +00:00
try:
res = requests.get(url)
2021-08-04 01:39:19 +00:00
image = Image.open(io.BytesIO(res.content))
2021-08-04 18:15:38 +00:00
# unify images
2021-08-04 01:39:19 +00:00
image.convert("RGB")
image = image.resize(self.__resolution, Image.BICUBIC)
break
2021-08-09 03:44:12 +00:00
except requests.exceptions.RequestException:
2021-08-04 01:39:19 +00:00
pass
# put image in correct position
self.__images[url] = image
2021-08-04 01:39:19 +00:00
# image is processed
self.__queue.task_done()
@classmethod
2021-08-09 03:32:05 +00:00
def load(cls, urls: list[str], resolution: tuple[int, int], language: str, num_threads: int) -> list[Image.Image]:
url_queue = queue.Queue()
for url in urls:
url_queue.put(url)
2021-08-04 01:39:19 +00:00
2021-08-04 16:36:23 +00:00
loaders = []
2021-08-04 01:39:19 +00:00
for _ in range(num_threads):
loader = cls(url_queue, resolution, language)
2021-08-04 16:36:23 +00:00
loaders.append(loader)
loader.start()
url_queue.join()
2021-08-04 16:36:23 +00:00
# stitch all "images" dicts together
2021-08-04 16:36:23 +00:00
images = {}
for loader in loaders:
2021-08-09 03:32:05 +00:00
images |= loader.images
2021-08-04 01:39:19 +00:00
2021-08-09 02:03:24 +00:00
# sort images to match the initial "urls" list
images = [images[url] for url in urls]
2021-08-04 18:15:38 +00:00
2021-08-04 16:36:23 +00:00
return images
2021-08-04 01:39:19 +00:00
2021-08-04 16:36:23 +00:00
@property
2021-08-09 03:32:05 +00:00
def images(self) -> dict[str, Image.Image]:
2021-08-04 01:39:19 +00:00
return self.__images