Upload 5 files

Dockerfile

@@ -0,0 +1,20 @@
+FROM python:3.9
+
+WORKDIR /app
+
+COPY . /app
+
+RUN pip3 install fastapi uvicorn transformers==4.42.3 pillow protobuf==4.25.3 fastapi-health pinecone_text
+
+RUN useradd -m -u 1000 user
+
+USER user
+
+ENV HOME=/home/user \
+ PATH=/home/user/.local/bin:$PATH
+
+WORKDIR $HOME/app
+
+COPY --chown=user . $HOME/app
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -0,0 +1,73 @@
+from fastapi import FastAPI, Depends, HTTPException
+from pydantic import BaseModel
+from fastapi_health import health
+from PIL import Image
+import logging
+import sys
+from io import BytesIO
+import base64
+
+from process_img import Image_Processor
+from vector_emb import EmbeddingModels
+
+logger = logging.getLogger(__name__)
+
+logging.basicConfig(
+    level=logging.getLevelName("INFO"),
+    handlers=[logging.StreamHandler(sys.stdout)],
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+logging.info('Logging module started')
+
+def get_session():
+    return True
+
+def is_database_online(session: bool = Depends(get_session)):
+    return session
+
+app = FastAPI()
+app.add_api_route("/healthz", health([is_database_online]))
+
+
+
+model = EmbeddingModels()
+img_Processor = Image_Processor()
+
+class ImageBase64(BaseModel):
+    base64_string: str
+
+class TextInput(BaseModel):
+    text: str
+
+@app.post("/design-dense/")
+async def embed_image(data: ImageBase64):
+    base64_string = data.base64_string
+    image_data = base64.b64decode(base64_string)
+    image = Image.open(BytesIO(image_data))
+    final_image = img_Processor.get_processed_img(image)
+    embeddings = model.get_single_image_embedding(final_image)
+    return embeddings
+
+
+@app.post("/sparse/")
+async def embed_text(item: TextInput):
+    try:
+        logging.info(f'Received text for embedding: {item.text}')
+        embeddings = model.get_single_sparse_text_embedding(item.text)
+        logging.info('Embedding process completed')
+        return embeddings
+    except Exception as e:
+        logging.error(f'Error during embedding process: {e}')
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.post("/design-sparse/")
+async def embed_text(item: TextInput):
+    try:
+        logging.info(f'Received text for embedding: {item.text}')
+        embeddings = model.get_single_sparse_text_embedding(item.text)
+        embeddings = model.normalize_sparse_vector_values(embeddings)
+        logging.info('Embedding process completed')
+        return embeddings
+    except Exception as e:
+        logging.error(f'Error during embedding process: {e}')
+        raise HTTPException(status_code=500, detail=str(e))

process_img.py

@@ -0,0 +1,48 @@
+import numpy as np
+from PIL import Image, ImageOps
+import logging
+
+class Image_Processor:
+    def __init__(self):
+        pass
+    def is_image_white_by_percentage(self,image_path, white_threshold):
+        image = image_path.convert('RGB')
+        image_np = np.array(image)
+        white_pixel = np.array([255, 255, 255])
+        white_pixels_count = np.sum(np.all(image_np == white_pixel, axis=-1))
+        total_pixels = image_np.shape[0] * image_np.shape[1]
+        white_pixel_percentage = (white_pixels_count / total_pixels) * 100
+        return white_pixel_percentage > white_threshold
+
+    def padding_white(self,image, output_size=(224, 224)):
+        # Ensure image is in RGB mode before padding
+        if image.mode != 'RGB':
+            image = image.convert('RGB')
+        new_image = ImageOps.pad(image, output_size, method=Image.Resampling.LANCZOS, color=(255, 255, 255))
+        return new_image
+
+    def resize_image_with_aspect_ratio(self,img):
+        target_size=224
+        width, height = img.size
+        original_aspect_ratio = width / height
+        if width > height:
+            new_width = target_size
+            new_height = int(target_size / original_aspect_ratio)
+        else:
+            new_height = target_size
+            new_width = int(target_size * original_aspect_ratio)
+        resized_img = img.resize((new_width, new_height))
+        return resized_img
+    
+    def get_processed_img(self,image):
+        white_thresh = self.is_image_white_by_percentage(image,50)
+        if white_thresh == True:
+            resized_image = self.resize_image_with_aspect_ratio(image)
+            final_image = self.padding_white(resized_image)
+            logging.info('Resized and Padded Image')
+        else:
+            final_image = self.resize_image_with_aspect_ratio(image)
+            logging.info('Resized Image')
+        
+        final_image = final_image.convert('L') if final_image.mode != 'L' else final_image
+        return final_image

requirements.txt

@@ -0,0 +1,6 @@
+fastapi
+uvicorn
+pillow
+torch
+transformers
+fastapi-health

vector_emb.py

@@ -0,0 +1,115 @@
+
+from pinecone_text.sparse import SpladeEncoder
+import re
+import torch
+import torch.nn.functional as F
+from transformers import CLIPModel, CLIPProcessor, CLIPTokenizer
+import logging
+
+class EmbeddingModels:
+    def __init__(self, device="cuda" if torch.cuda.is_available() else "cpu"):
+        self.device = device
+        logging.info(f'Using Device {self.device}')
+        self.sparse_model = SpladeEncoder(device=self.device)
+        self.img_model_ID = "openai/clip-vit-large-patch14"
+        self.img_model, self.img_processor, self.img_tokenizer = self.get_image_model_info(self.img_model_ID)
+        logging.info("Model Loaded")
+
+    def get_image_model_info(self, model_ID):
+        model = CLIPModel.from_pretrained(model_ID).to(self.device)
+        processor = CLIPProcessor.from_pretrained(model_ID)
+        tokenizer = CLIPTokenizer.from_pretrained(model_ID)
+        return model, processor, tokenizer
+
+    def get_single_image_embedding(self, my_image):
+        image = self.img_processor(
+            text=None,
+            images=my_image,
+            return_tensors="pt"
+        )["pixel_values"].to(self.device)
+
+        embedding = self.img_model.get_image_features(image)
+        logging.info("Embeddings Created")
+        embeddings = F.normalize(embedding, p=2, dim=1)
+        logging.info("Embeddings Normalized")
+        values = embeddings[0].tolist()
+        return values
+    
+    def preprocessing_patent_data(self,text):
+    # Removing Common tags in patent
+        pattern0 =  r'\b(SUBSTITUTE SHEET RULE 2 SUMMARY OF THE INVENTION|BRIEF DESCRIPTION OF PREFERRED EMBODIMENTS|BRIEF DESCRIPTION OF THE DRAWINGS/FIGURES|BEST MODE FOR CARRYING OUT THE INVENTION|BACKGROUND AND SUMMARY OF THE INVENTION|FIELD AND BACKGROUND OF THE INVENTION|BACKGROUND OF THE PRESENT INVENTION|FIELD AND BACKGROUND OF INVENTION|STAND DER TECHNIK- BACKGROUND ART|BRIEF DESCRIPTION OF THE DRAWINGS|DESCRIPTION OF THE RELATED ART|BRIEF SUMMARY OF THE INVENTION|UTILITY MODEL CLAIMS A CONTENT|DESCRIPTION OF BACKGROUND ART|BRIEF DESCRIPTION OF DRAWINGS|BACKGROUND OF THE INVENTION|BACKGROUND TO THE INVENTION|TÉCNICA ANTERIOR- PRIOR ART|DISCLOSURE OF THE INVENTION|BRIEF SUMMARY OF INVENTION|BACKGROUND OF RELATED ART|SUMMARY OF THE DISCLOSURE|SUMMARY OF THE INVENTIONS|SUMMARY OF THE INVENTION|OBJECTS OF THE INVENTION|THE CONTENT OF INVENTION|DISCLOSURE OF INVENTION|Disclosure of Invention|Complete Specification|RELATED BACKGROUND ART|BACKGROUND INFORMATION|BACKGROUND TECHNOLOGY|DETAILED DESCRIPTION|SUMMARY OF INVENTION|DETAILED DESCRIPTION|PROBLEM TO BE SOLVED|EFFECT OF INVENTION|WHAT IS CLAIMED IS|What is claimed is|What is Claim is|SUBSTITUTE SHEET|SELECTED DRAWING|BACK GROUND ART|BACKGROUND ART|Background Art|JPO&INPIT|CONSTITUTION|DEFINITIONS|Related Art|BACKGROUND|JPO&INPIT|JPO&NCIPI|COPYRIGHT|SOLUTION|SUMMARY)\b'
+        text = re.sub(pattern0, '[SEP]', text, flags=re.IGNORECASE)
+        text = ' '.join(text.split())
+        # Removing all tags between Heading to /Heading and id= 
+        regex = r'<\s*heading[^>]*>(.*?)<\s*/\s*heading>|<[^<]+>|id=\"p-\d+\"|:'
+        result = re.sub(regex, '[SEP]', text, flags=re.IGNORECASE)
+        # find_formula_names from pat text to exclude it from below logic regex
+        chemical_list = []
+        pattern1 = r'\b((?:(?:H|He|Li|Be|B|C|N|O|F|Ne|Na|Mg|Al|Si|P|S|Cl|Ar|K|Ca|Sc|Ti|V|Cr|Mn|Fe|Co|Ni|Cu|Zn|Ga|Ge|As|Se|Br|Kr|Rb|Sr|Y|Zr|Nb|Mo|Tc|Ru|Rh|Pd|Ag|Cd|In|Sn|Sb|Te|I|Xe|Cs|Ba|La|Hf|Ta|W|Re|Os|Ir|Pt|Au|Hg|Tl|Pb|Bi|Po|At|Rn|Fr|Ra|Ac|Rf|Db|Sg|Bh|Hs|Mt|Ds|Rg|Cn|Nh|Fl|Mc|Lv|Ts|Og|Ce|Pr|Nd|Pm|Sm|Eu|Gd|Tb|Dy|Ho|Er|Tm|Yb|Lu|Th|Pa|U|Np|Pu|Am|Cm|Bk|Cf|Es|Fm|Md|No|Lr)\d*)+)\b'
+        
+        formula_names = re.findall(pattern1, result)
+        for formula in formula_names:
+            if len(formula)>=2:
+                chemical_list.append(formula)
+        # print("chemical_list:", chemical_list)
+
+        # Remove numbers and alphanum inside brackets excluding chemical forms
+        pattern2 = r"\((?![A-Za-z]+\))[\w\d\s,-]+\)|\([A-Za-z]\)"
+        def keep_strings(text):
+            matched = text.group(0)
+            if any(item in matched for item in chemical_list):
+                return matched
+            return ' '
+        cleaned_text = re.sub(pattern2, keep_strings, result)
+        cleaned_text = ' '.join(cleaned_text.split())
+        cleaned_text= re.sub("(\[SEP\]+\s*)+", ' ', cleaned_text, flags=re.IGNORECASE)
+        # below new logic to remove chemical compounds (eg.chemical- polymerizable compounds)
+        p_text2=re.sub('[\—\-\═\=]', ' ', cleaned_text)
+        pattern1 = r'\b((?:(?:H|He|Li|Be|B|C|N|O|F|Ne|Na|Mg|Al|Si|P|S|Cl|Ar|K|Ca|Sc|Ti|V|Cr|Mn|Fe|Co|Ni|Cu|Zn|Ga|Ge|As|Se|Br|Kr|Rb|Sr|Y|Zr|Nb|Mo|Tc|Ru|Rh|Pd|Ag|Cd|In|Sn|Sb|Te|I|Xe|Cs|Ba|La|Hf|Ta|W|Re|Os|Ir|Pt|Au|Hg|Tl|Pb|Bi|Po|At|Rn|Fr|Ra|Ac|Rf|Db|Sg|Bh|Hs|Mt|Ds|Rg|Cn|Nh|Fl|Mc|Lv|Ts|Og|Ce|Pr|Nd|Pm|Sm|Eu|Gd|Tb|Dy|Ho|Er|Tm|Yb|Lu|Th|Pa|U|Np|Pu|Am|Cm|Bk|Cf|Es|Fm|Md|No|Lr)\d*)+)\b'
+        cleaned_text = re.sub(pattern1, "", p_text2)
+        cleaned_text = re.sub('  ,+|,  +', ' ', cleaned_text)
+        cleaned_text = re.sub(' +', ' ', cleaned_text)
+        cleaned_text = re.sub('\.+', '.', cleaned_text)
+        cleaned_text = re.sub('[0-9] [0-9] +', ' ', cleaned_text)
+        cleaned_text = re.sub('( )', ' ', cleaned_text)
+        cleaned_text=cleaned_text.strip()
+        return cleaned_text 
+
+    def get_single_sparse_text_embedding(self, df_chunk):
+        df_chunk = self.preprocessing_patent_data(df_chunk)
+        txt_sp = self.sparse_model.encode_documents(df_chunk)
+
+        # tensor = torch.tensor(txt_sp['values'])
+        # normalized_tensor = F.normalize(tensor, p=2.0, dim=0, eps=1e-12)
+        # values = normalized_tensor.tolist()
+
+        # # Update the sparse_vector with normalized values
+        # normalized_sparse_vector = {
+        #     'indices': txt_sp['indices'],
+        #     'values': values
+        # }
+        return txt_sp
+    
+    def normalize_sparse_vector_values(self,sparse_vector):
+        """
+        Normalize the values of a sparse vector to a 0-1 range using min-max scaling,
+        considering a known range of sparse scores.
+        Args:
+        sparse_vector: A dict representing a sparse vector with 'indices' and 'values'
+        min_score: The minimum score in the range of sparse scores (default is 0)
+        max_score: The maximum score in the range of sparse scores (default is 6000)
+        Returns:
+        A dict representing the sparse vector with normalized 'values'.
+        """
+        # normalized_values = [(value - min_score) / (max_score - min_score) for value in sparse_vector['values']]
+        self.tensor = torch.tensor(sparse_vector['values'])
+        self.normalized_tensor = F.normalize(self.tensor, p=2.0, dim=0, eps=1e-12)
+        values = self.normalized_tensor.tolist()
+
+        # Update the sparse_vector with normalized values
+        self.normalized_sparse_vector = {
+            'indices': sparse_vector['indices'],
+            'values': values
+        }
+        return self.normalized_sparse_vector
+