""" OpenCLIP ViT-B/32 visual encoder (ONNX). Produces 512-d image features consumed by the content classifier. Not exposed as a standalone service; the classifier owns the lifecycle. """ import logging from pathlib import Path import numpy as np import onnxruntime as ort # noqa: F401 (provider plumbing relies on this) from app.config import VisionSettings logger = logging.getLogger(__name__) _MEAN = np.array([0.48145466, 0.4578275, 0.40821073], dtype=np.float32) _STD = np.array([0.26862954, 0.26130258, 0.27577711], dtype=np.float32) _SIZE = 224 def _preprocess(image: np.ndarray) -> np.ndarray: from PIL import Image img = Image.fromarray(image).convert("RGB") w, h = img.size scale = _SIZE / min(w, h) img = img.resize((int(w * scale), int(h * scale)), Image.BICUBIC) w, h = img.size left = (w - _SIZE) // 2 top = (h - _SIZE) // 2 img = img.crop((left, top, left + _SIZE, top + _SIZE)) arr = np.array(img, dtype=np.float32) / 255.0 arr = (arr - _MEAN) / _STD arr = arr.transpose(2, 0, 1) return arr[np.newaxis] class CLIPVisualEncoder: """OpenCLIP ViT-B/32 image encoder, 512-d normalized output.""" def __init__(self, settings: VisionSettings): model_path = Path(settings.models_dir) / "embed" / "visual.onnx" from app.services.vision.providers import create_session from app.config import settings as app_settings logger.info("Loading CLIP visual encoder from %s", model_path) self._session = create_session( str(model_path), configured_providers=app_settings.vision.execution_providers, ) def encode(self, image: np.ndarray) -> np.ndarray: inp = _preprocess(image) name = self._session.get_inputs()[0].name out = self._session.run(None, {name: inp})[0][0] out = out / np.linalg.norm(out) return out.astype(np.float32)