PalmOilRipenessDetection-Python/src/infrastructure/vision_service.py

import os
import vertexai
from vertexai.vision_models import Image, MultiModalEmbeddingModel
import base64
import io
from PIL import Image as PILImage
from typing import List

class VertexVisionService:
    def __init__(self, project_id: str, location: str):
        self.project_id = project_id
        self.location = location
        self._model = None

    def _get_model(self):
        """Lazy load the model and catch billing/connection errors."""
        if self._model is None:
            try:
                # Ensure credentials are set before init if using service account key
                # (This is now handled globally in main.py)
                vertexai.init(project=self.project_id, location=self.location)
                self._model = MultiModalEmbeddingModel.from_pretrained("multimodalembedding@001")
            except Exception as e:
                # Log the specific error (e.g., Billing Disabled)
                print(f"Vertex AI Initialization Failed: {e}")
                raise RuntimeError("Cloud services (Vectorization/Search) are currently unavailable.")
        return self._model

    def get_image_embedding(self, image_path: str) -> List[float]:
        if not os.path.exists(image_path):
            raise FileNotFoundError(f"Image not found at {image_path}")
        
        model = self._get_model() # This will raise the RuntimeError if billing is down
        image = Image.load_from_file(image_path)
        # Standardizing to 1408 dimensions for consistency
        embeddings = model.get_embeddings(image=image, dimension=1408)
        return embeddings.image_embedding

    def get_text_embedding(self, text: str) -> List[float]:
        """Converts text query into a 1408-d vector."""
        model = self._get_model()
        embeddings = model.get_embeddings(
            contextual_text=text, 
            dimension=1408
        )
        return embeddings.text_embedding

    def encode_image_to_base64(self, image_path: str) -> str:
        """Resizes image to 640x640 and encodes to Base64."""
        with PILImage.open(image_path) as img:
            img.thumbnail((640, 640))
            buffered = io.BytesIO()
            img.save(buffered, format="JPEG", quality=85)
            return base64.b64encode(buffered.getvalue()).decode('utf-8')