src/components/vectorizing_data.py

import os
import logging
from pydoc import text
from typing import List
import pandas as pd

import numpy as np
import torch
from PIL import Image
from torch.utils.data import Dataset, DataLoader
from src.models.muti_model import ImageEncoder, TextEncoder
from transformers import AutoTokenizer
from torchvision.transforms import v2
import mlflow.pytorch
import torch
from src.utils.asyncHandler import asyncHandler
from src.retreiver.custom_vec_db import CustomVectorDb
from src.entity.data_access import Connect_data
import requests
from src.entity.model import MyModel
from src.entity.config_entity import ModelTrainingConfig, DataIngestionConfig
from io import BytesIO
from src.config.app_config import app_config
from src.constants import TEXT_MODEL_NAME, MODEL_URI,INDEX_NAME,NUM_WORKERS

from warnings import deprecated

class InferenceDataSet(Dataset):
    def __init__(self, config, data_path=None, data_frame=None):
        logging.info(f"InferenceDataSet - initializing with config: {config}, data_path: {data_path}")
        self.config = config
        if data_frame is not None:
            logging.info("InferenceDataSet - using provided data_frame")
            self.data_frame = data_frame
        elif data_path is not None:
            logging.info(f"InferenceDataSet - loading data from csv path: {data_path}")
            self.data_frame = pd.read_csv(data_path)
        else:
            logging.error("InferenceDataSet - initialization failed: no data_path or data_frame provided")
            raise ValueError("Either data_path or data_frame must be provided")
        logging.info(f"InferenceDataSet - loaded data count: {len(self.data_frame)}")

    def __len__(self):
        length = len(self.data_frame)
        logging.info(f"InferenceDataSet - checking length: {length}")
        return length

    def __getitem__(self, idx):
        logging.info(f"InferenceDataSet - fetching item at index: {idx}")
        item = self.data_frame.iloc[idx]
        img_url = item["image_url"]
        logging.info(f"InferenceDataSet - downloading image url: {img_url}")
        try:
            response = requests.get(img_url)
            response.raise_for_status()
        except Exception as e:
            logging.warning(f"InferenceDataSet - primary download failed for url {img_url}: {e}. retrying...")
            response = requests.get(img_url)
        img_path = BytesIO(response.content)
        text_data = str(item["product_search_description"])
        image = Image.open(img_path).convert("RGB")
        image_np = np.array(image)
        image_tensor = self.config.transforms(image_np).unsqueeze(0).to(self.config.device)
        logging.info(f"InferenceDataSet - image processed to tensor shape: {image_tensor.shape}")
        with torch.no_grad():
            img_feat = self.config.image_encoder(image_tensor).squeeze(0).cpu()
        tokens = self.config.tokenizer(
            text_data,
            padding="max_length",
            truncation=True,
            max_length=self.config.config.max_len,
            return_tensors="pt"
        ).to(self.config.device)
        logging.info(f"InferenceDataSet - text tokenized. tokens shape: {tokens['input_ids'].shape}")
        with torch.no_grad():
            txt_feat = self.config.text_encoder(tokens['input_ids'], tokens['attention_mask']).squeeze(0).cpu()
        logging.info(f"InferenceDataSet - extraction complete for item index {idx}")
        return (
            img_feat,
            txt_feat,
            item.to_dict()
        )

class Vectorizer:
    def __init__(self, data_path=None):
        logging.info(f"Vectorizer - initializing with data_path: {data_path}")
        if data_path is None:
            data_path = DataIngestionConfig().data_path
        self.data_path = data_path
        self.config = ModelTrainingConfig()
        logging.info("Vectorizer - instantiating model wrapper")
        self.model = MyModel(config=self.config)
        try:
            logging.info("Vectorizer - initializing DagsHub/MLflow logging integration")
            import dagshub
            dagshub.auth.add_app_token(app_config.mlflow_api_key)
            dagshub.init(repo_owner=app_config.dagshub_owner, repo_name=app_config.dagshub_repo, mlflow=True)
            logging.info("Vectorizer - DagsHub MLflow integration ready")
        except Exception as ex:
            logging.warning(f"Vectorizer - DagsHub initialization failed: {ex}")
        
        
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        logging.info(f"Vectorizer - using device: {self.device}")
        self.tokenizer = AutoTokenizer.from_pretrained(TEXT_MODEL_NAME, token=app_config.huggingface_api_key)
        self.image_encoder = ImageEncoder(self.config).to(self.device).eval()
        self.text_encoder = TextEncoder(self.config).to(self.device).eval()
        self.transforms = v2.Compose([
            v2.ToImage(),
            v2.ToDtype(torch.float32, scale=True),
            v2.Resize(size=(224, 224), antialias=True),
            v2.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
        ])
        
        logging.info("Vectorizer - loading model checkpoints from MLflow registry")
        self.model.model = mlflow.pytorch.load_model(
            model_uri=MODEL_URI,
            map_location=self.device
        )
        logging.info("Vectorizer - MLflow model loaded successfully. connecting to Pinecone.")
        self.vec_db = CustomVectorDb(
            api_key=app_config.pine_cone_api_key,
            index_name=INDEX_NAME,
            img_dimension=self.config.image_feature_output,   # 2048 (ResNet-50 raw feats)
            txt_dimension=self.config.text_feature_output,    # 768  (BERT-style CLS token)
        )
        logging.info("Vectorizer - Pinecone client and dual indexes initialized")

    @asyncHandler
    @deprecated(
        "ingest_vec is deprecated — it relies on the old MyModel.predict_emb (single merged embedding). "
        "Use ingest_vectors instead, which stores image and text embeddings in separate Pinecone indexes."
    )
    async def ingest_vec(self, override: bool = False):
        logging.info(f"[DEPRECATED] Vectorizer.ingest_vec - initiating ingestion. override: {override}")
        if override:
            logging.warning("Vectorizer.ingest_vec - deleting existing vectors from Pinecone index")
            self.vec_db.delete_all()
        connector = Connect_data(data_path=self.data_path)
        df = await connector.load_data()

        # cleaning data
        df.dropna(inplace=True)

        if not override:
            logging.info("Vectorizer.ingest_vec - retrieving existing vector IDs from Pinecone")
            existing_ids = set(self.vec_db.get_all_ids())
            logging.info(f"Vectorizer.ingest_vec - retrieved {len(existing_ids)} existing IDs")
            df = df[~df["id"].astype(str).isin(existing_ids)]
        if df.empty:
            logging.warning("Vectorizer.ingest_vec - all data already exists in vector store. skipping ingestion.")
            return
        logging.info(f"Vectorizer.ingest_vec - processing dataset for {len(df)} samples")
        dataset = InferenceDataSet(
            data_frame=df,
            config=self
        )
        dataloader = DataLoader(
            dataset,
            batch_size=self.config.batch_size,
            shuffle=False,
            num_workers=NUM_WORKERS
        )
        logging.info("Vectorizer.ingest_vec - starting batch upsert into Pinecone (old single-index approach)")
        self.vec_db.batch_upsert(
            dataloader=dataloader,
            model=self.model
        )
        logging.info("Vectorizer.ingest_vec - ingestion completed successfully")

    @asyncHandler
    async def ingest_vectors(self, override: bool = False):
        """Ingest embeddings into the dual Pinecone indexes (image + text).

        ``ImageEncoder`` and ``TextEncoder`` produce the raw feature vectors which
        are stored independently, allowing ``query()`` to use RRF when both
        modalities are available at query time.
        """
        logging.info(f"Vectorizer.ingest_vectors - initiating dual-index ingestion. override: {override}")
        if override:
            logging.warning("Vectorizer.ingest_vectors - clearing both Pinecone indexes")
            self.vec_db.delete_all()
        connector = Connect_data(data_path=self.data_path)
        df = await connector.load_data()
        df.dropna(inplace=True)
        
        if not override:
            logging.info("Vectorizer.ingest_vectors - retrieving existing vector IDs from Pinecone")
            existing_ids = set(self.vec_db.get_all_ids())
            logging.info(f"Vectorizer.ingest_vectors - {len(existing_ids)} IDs already present; skipping those rows")
            df = df[~df["id"].astype(str).isin(existing_ids)]
        if df.empty:
            logging.warning("Vectorizer.ingest_vectors - nothing new to ingest. exiting.")
            return
        logging.info(f"Vectorizer.ingest_vectors - encoding {len(df)} samples into dual indexes")
        dataset = InferenceDataSet(
            data_frame=df,
            config=self   # carries image_encoder, text_encoder, transforms, tokenizer
        )
        dataloader = DataLoader(
            dataset,
            batch_size=self.config.batch_size,
            shuffle=False,
            num_workers=NUM_WORKERS
        )
        logging.info("Vectorizer.ingest_vectors - calling vec_db.batch_upload")
        self.vec_db.batch_upload(dataloader=dataloader)
        logging.info("Vectorizer.ingest_vectors - dual-index ingestion completed successfully")

    @asyncHandler
    @deprecated("this method is deprecated use invoke instead")
    async def get_similar_data(self, vector: List[float], top_k: int = 5):
        logging.info(f"Vectorizer.get_similar_data - querying Pinecone vector store for top_k: {top_k}")
        res = self.vec_db.search(vector, top_k)
        logging.info(f"Vectorizer.get_similar_data - search complete. response type: {type(res)}")
        return res
    
    @asyncHandler
    async def invoke(self, img_vec: List[float], text_vec: List[float], top_k: int = 5):
        logging.info(f"Vectorizer.invoke - querying Pinecone vector store for top_k: {top_k}")
        res = self.vec_db.query(img_vec,text_vec, top_k)
        logging.info(f"Vectorizer.invoke - search complete. response type: {type(res)}")
        return res