Method _embed_image

sdk/python/feast/embedder.py:208–241

(self, inputs: List[Any])

        return self._image_processor

    def _embed_image(self, inputs: List[Any]) -> "np.ndarray":
        from pathlib import Path

        import numpy as np
        from PIL import Image

        all_embeddings: List["np.ndarray"] = []
        batch_size = self.config.batch_size

        for start in range(0, len(inputs), batch_size):
            batch = inputs[start : start + batch_size]
            images = []
            opened: List[Image.Image] = []
            try:
                for inp in batch:
                    if isinstance(
                        inp, (str, Path)
                    ):  # If the input string path is too large that It gives error and we could not open the image.
                        img = Image.open(inp)
                        opened.append(img)
                        images.append(img)
                    else:
                        images.append(inp)

                processed = self.image_processor(images=images, return_tensors="pt")
            finally:
                for opened_img in opened:
                    opened_img.close()

            embeddings = self.image_model.get_image_features(**processed)
            embeddings = embeddings / embeddings.norm(p=2, dim=-1, keepdim=True)
            all_embeddings.append(embeddings.detach().numpy())

        return np.concatenate(all_embeddings, axis=0)