INNER CODE UNIT · Python
close
yurijmikhalevich/rclip · rclip/main.py:125
def close(self) -> None:
self._shutdown_image_loading_executor()
def _load_images(
self, items: Iterable[Tuple[str, ImageMeta]]
) -> Iterator[Tuple[str, ImageMeta, str, npt.NDArray[np.float32]]]:
helpers._ensure_image_loading_configured()
executor = self._get_image_loading_executor()
# keep a few full batches in flight so they preprocess while the model
# processes the current one.
max_in_flight = max(self.LOOKAHEAD_BATCHES * self._indexing_batch_size, self._image_loading_workers)
items_iter = iter(items)
in_flight: Deque[Tuple[str, ImageMeta, Future[Tuple[str, str, npt.NDArray[np.float32]]]]] = deque()
def submit_next() -> None:
item = next(items_iter, None)
if item is not None:
path, meta = item