src/utils/vector_utils.py

import logging
import cv2
from src.core.dependencies import get_img_transformer, image_encoder_eval, text_encoder_eval, text_tokenizer
import torch
import os
def _get_image_feat(image_path, config, device):
    logging.info(f"entering _get_image_feat with image_path: {image_path}, device: {device}")
    if not image_path or not os.path.exists(image_path):
        logging.warning(f"image path not provided or does not exist: {image_path}. returning zero features.")
        feat = torch.zeros((1, config.image_feature_output), device=device)
        logging.info(f"exiting _get_image_feat with zero features tensor shape: {feat.shape}")
        return feat
    img = cv2.imread(image_path)
    if img is None:
        logging.warning(f"failed to read image at path: {image_path}. returning zero features.")
        feat = torch.zeros((1, config.image_feature_output), device=device)
        logging.info(f"exiting _get_image_feat with zero features tensor shape: {feat.shape}")
        return feat
    logging.info(f"successfully loaded image from path: {image_path}. image shape: {img.shape}")
    img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
    
    transforms = get_img_transformer()
    img_tensor = transforms(img).unsqueeze(0).to(device)
    logging.info(f"transformed image into tensor. shape: {img_tensor.shape}, device: {img_tensor.device}")
    image_encoder = image_encoder_eval()
    logging.info("running image encoder forward pass")
    with torch.no_grad():
        img_feat = image_encoder(img_tensor)
    logging.info(f"exiting _get_image_feat with shape: {img_feat.shape}")
    return img_feat

def _get_text_feat(text, config, device):
    logging.info(f"entering _get_text_feat with text: {text}, device: {device}")
    
    text_encoder = text_encoder_eval()
    tokenizer = text_tokenizer()
    tokens = tokenizer(
        text,
        padding="max_length",
        max_length=config.max_len,
        truncation=True,
        return_tensors="pt"
    ).to(device)
    logging.info(f"tokenized text. tokens shape: {tokens['input_ids'].shape}")
    with torch.no_grad():
        txt_feat = text_encoder(tokens["input_ids"], tokens["attention_mask"])
    logging.info(f"exiting _get_text_feat with shape: {txt_feat.shape}")
    return txt_feat