INNER CODE UNIT · Python

_load_images

yurijmikhalevich/rclip · rclip/main.py:128

  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
        in_flight.append((path, meta, executor.submit(_read_and_preprocess, path)))

    for _ in range(max_in_flight):

View source record →