Update utils/vulnerability_classifier.py

utils/vulnerability_classifier.py

@@ -1,307 +1,153 @@
-from haystack.nodes.base import BaseComponent
-from haystack.schema import Document
-from haystack.nodes import ImageToTextConverter, PDFToTextConverter
-from haystack.nodes import TextConverter, DocxToTextConverter, PreProcessor
-from pdf2image import convert_from_path
-from typing import Callable, Dict, List, Optional, Text, Tuple, Union
+from typing import List, Tuple
 from typing_extensions import Literal
-import pandas as pd
 import logging
-import re
-import string
-from haystack.pipelines import Pipeline
+import pandas as pd
+from pandas import DataFrame, Series
+from utils.config import getconfig
+from utils.preprocessing import processingpipeline
 import streamlit as st
+from transformers import pipeline
+from setfit import SetFitModel
+
+label_dict= {0: 'Agricultural communities',
+ 1: 'Children',
+ 2: 'Coastal communities',
+ 3: 'Ethnic, racial or other minorities',
+ 4: 'Fishery communities',
+ 5: 'Informal sector workers',
+ 6: 'Members of indigenous and local communities',
+ 7: 'Migrants and displaced persons',
+ 8: 'Older persons',
+ 9: 'Other',
+ 10: 'Persons living in poverty',
+ 11: 'Persons with disabilities',
+ 12: 'Persons with pre-existing health conditions',
+ 13: 'Residents of drought-prone regions',
+ 14: 'Rural populations',
+ 15: 'Sexual minorities (LGBTQI+)',
+ 16: 'Urban populations',
+ 17: 'Women and other genders'}
+
+def get_vulnerability_labels(preds):
 
-@st.cache_data
-def useOCR(file_path: str)-> Text:
     """
-    Converts image pdfs into text, Using the Farm-haystack[OCR]
-    
-    Params
-    ----------
-    file_path: file_path of uploade file, returned by add_upload function in 
-    uploadAndExample.py
+    Function that takes the numerical predictions as an input and returns a list of the labels.
     
-    Returns the text file as string.
     """
-    # we need pdf file to be first converted into image file
-    # this will create each page as image file
-    images = convert_from_path(pdf_path = file_path)
-    list_ = []
-    # save image file in cache and read them one by one to pass it to OCR
-    for i, pdf in enumerate(images):
-        # Save pages as images in the pdf
-        pdf.save(f'PDF\image_converted_{i+1}.png', 'PNG')
-        list_.append(f'PDF\image_converted_{i+1}.png')
     
-    converter = ImageToTextConverter(remove_numeric_tables=True,
-                                      valid_languages=["eng"])
-    # placeholder to collect  the text from each page
-    placeholder = []
-    for file in list_:
-        document = converter.convert(
-                            file_path=file, meta=None,
-                            )[0]
-
-        text = document.content
-        placeholder.append(text)
-    # join the text from each page by page separator
-    text = '\x0c'.join(placeholder)
-    return text
-
-
-
-class FileConverter(BaseComponent):
-    """
-    Wrapper class to convert uploaded document into text by calling appropriate 
-    Converter class, will use internally haystack PDFToTextOCR in case of image 
-    pdf. Cannot use the FileClassifier from haystack as its doesnt has any 
-    label/output class for image.
-    1. https://haystack.deepset.ai/pipeline_nodes/custom-nodes
-    2. https://docs.haystack.deepset.ai/docs/file_converters
-    3. https://github.com/deepset-ai/haystack/tree/main/haystack/nodes/file_converter
-    4. https://docs.haystack.deepset.ai/reference/file-converters-api
-    """
-
-    outgoing_edges = 1
-
-    def run(self, file_name: str , file_path: str, encoding: Optional[str]=None,
-            id_hash_keys: Optional[List[str]] = None,
-            ) -> Tuple[dict,str]:
-        """ this is required method to invoke the component in 
-            the pipeline implementation. 
-            
-        Params
-        ----------
-        file_name: name of file
-        file_path: file_path of uploade file, returned by add_upload function in 
-                    uploadAndExample.py
-        
-        See the links provided in Class docstring/description to see other params
-        
-        Return
-        ---------
-        output: dictionary, with key as identifier and value could be anything 
-                we need to return. In this case its the List of Hasyatck Document
-        
-        output_1: As there is only one outgoing edge, we pass 'output_1' string
-        """
-        try:
-            if file_name.endswith('.pdf'):
-                converter = PDFToTextConverter(remove_numeric_tables=True)
-            if file_name.endswith('.txt'):
-                converter = TextConverter(remove_numeric_tables=True)
-            if file_name.endswith('.docx'):
-                converter = DocxToTextConverter() 
-        except Exception as e:
-            logging.error(e)
-            return 
-
-
-
-        documents = []
-
-        document = converter.convert(
-                      file_path=file_path, meta=None, 
-                      encoding=encoding, id_hash_keys=id_hash_keys
-                      )[0]
-
-        text = document.content
-
-        # in case of scanned/images only PDF the content might contain only
-        # the page separator (\f or \x0c). We check if is so and use
-        # use the OCR to get the text.
-        filtered = re.sub(r'\x0c', '', text)
-
-        if filtered == "":
-            logging.info("Using OCR")
-            text = useOCR(file_path)
-            
-        documents.append(Document(content=text, 
-                              meta={"name": file_name}, 
-                              id_hash_keys=id_hash_keys))
+    # Get label names
+    preds_list = preds.tolist()
+    
+    # Get the name of the group where the prediction is equal to "1"
+    result = []
 
-        
-        
-        logging.info('file conversion succesful')
-        output = {'documents': documents}
-        return output, 'output_1'
+    for sublist in preds_list:
+        names = [label_dict[key] for key, value in enumerate(sublist) if value == 1]
+        result.append(names)
 
-    def run_batch():
-        """
-        we dont have requirement to process the multiple files in one go
-        therefore nothing here, however to use the custom node we need to have
-        this method for the class.
-        """
+    return result
         
-        return
-
-
-def basic(s:str, remove_punc:bool = False):
-
+@st.cache_resource
+def load_vulnerabilityClassifier(config_file:str = None, classifier_name:str = None):
     """
-    Performs basic cleaning of text.
+    loads the document classifier using haystack, where the name/path of model
+    in HF-hub as string is used to fetch the model object.Either configfile or 
+    model should be passed.
+    1. https://docs.haystack.deepset.ai/reference/document-classifier-api
+    2. https://docs.haystack.deepset.ai/docs/document_classifier
     Params
-    ----------
-    s: string to be processed
-    removePunc: to remove all Punctuation including ',' and '.' or not
-    
-    Returns: processed string: see comments in the source code for more info
+    --------
+    config_file: config file path from which to read the model name
+    classifier_name: if modelname is passed, it takes a priority if not \
+    found then will look for configfile, else raise error.
+    Return: document classifier model
     """
-    
-    # Remove URLs
-    s = re.sub(r'^https?:\/\/.*[\r\n]*', ' ', s, flags=re.MULTILINE)
-    s = re.sub(r"http\S+", " ", s)
-
-    # Remove new line characters
-    s = re.sub('\n', ' ', s) 
 
-    # Remove punctuations
-    if remove_punc == True:
-      translator = str.maketrans(' ', ' ', string.punctuation) 
-      s = s.translate(translator)
-    # Remove distracting single quotes and dotted pattern
-    s = re.sub("\'", " ", s)
-    s = s.replace("..","") 
-    
-    return s.strip()
-
-
-def paraLengthCheck(paraList, max_len = 100):
-    """
-    There are cases where preprocessor cannot respect word limit, when using 
-    respect sentence boundary flag due to missing sentence boundaries.
-    Therefore we run one more round of split here for those paragraphs
+    # If no classifier given
     
-    Params
-    ---------------
-    paraList : list of paragraphs/text
-    max_len : max length to be respected by sentences which bypassed 
-              preprocessor strategy
-              
-    """
-    new_para_list = []
-    for passage in paraList:
-        # check if para exceeds words limit
-        if len(passage.content.split()) > max_len:
-          # we might need few iterations example if para = 512 tokens
-          # we need to iterate 5 times to reduce para to size limit of '100'
-            iterations = int(len(passage.content.split())/max_len)
-            for i in range(iterations):
-                temp  = " ".join(passage.content.split()[max_len*i:max_len*(i+1)])
-                new_para_list.append((temp,passage.meta['page']))
-            temp  = " ".join(passage.content.split()[max_len*(i+1):])
-            new_para_list.append((temp,passage.meta['page']))
+    if not classifier_name:
+        if not config_file:
+            logging.warning("Pass either model name or config file")
+            return
         else:
-            # paragraphs which dont need any splitting
-            new_para_list.append((passage.content, passage.meta['page']))
+            config = getconfig(config_file)
+            classifier_name = config.get('vulnerability','MODEL')
     
-    logging.info("New paragraphs length {}".format(len(new_para_list)))
-    return new_para_list
-
-class UdfPreProcessor(BaseComponent):
-    """
-    class to preprocess the document returned by FileConverter. It will check
-    for splitting strategy and splits the document by word or sentences and then
-    synthetically create the paragraphs. 
-    1. https://docs.haystack.deepset.ai/docs/preprocessor
-    2. https://docs.haystack.deepset.ai/reference/preprocessor-api
-    3. https://github.com/deepset-ai/haystack/tree/main/haystack/nodes/preprocessor
-    """
-    outgoing_edges = 1
-
-    def run(self, documents:List[Document], remove_punc:bool=False, apply_clean = True,
-            split_by: Literal["sentence", "word"] = 'sentence',
-            split_length:int = 2, split_respect_sentence_boundary:bool = False,
-            split_overlap:int = 0):
-
-        """ this is required method to invoke the component in 
-        the pipeline implementation. 
-            
-        Params
-        ----------
-        documents: documents from the output dictionary returned by Fileconverter
-        remove_punc: to remove all Punctuation including ',' and '.' or not
-        split_by: document splitting strategy either as word or sentence
-        split_length: when synthetically creating the paragrpahs from document,
-                      it defines the length of paragraph.
-        split_respect_sentence_boundary: Used when using 'word' strategy for 
-        splititng of text.
-        split_overlap: Number of words or sentences that overlap when creating
-        the paragraphs. This is done as one sentence or 'some words' make sense
-        when  read in together with others. Therefore the overlap is used.
-        
-        Return
-        ---------
-        output: dictionary, with key as identifier and value could be anything 
-                we need to return. In this case the output will contain 4 objects
-                the paragraphs text list as List, Haystack document, Dataframe and 
-                one raw text file.
-        
-        output_1: As there is only one outgoing edge, we pass 'output_1' string
-      
-        """
-        
-        if split_by == 'sentence':
-            split_respect_sentence_boundary = False
+    logging.info("Loading vulnerability classifier")
     
-        else:
-            split_respect_sentence_boundary = split_respect_sentence_boundary
-      
-        preprocessor = PreProcessor(
-            clean_empty_lines=True,
-            clean_whitespace=True,
-            clean_header_footer=True,
-            split_by=split_by,
-            split_length=split_length,
-            split_respect_sentence_boundary= split_respect_sentence_boundary,
-            split_overlap=split_overlap,
+    # we are using the pipeline as the model is multilabel and DocumentClassifier 
+    # from Haystack doesnt support multilabel
+    # in pipeline we use 'sigmoid' to explicitly tell pipeline to make it multilabel
+    # if not then it will automatically use softmax, which is not a desired thing.
+    # doc_classifier = TransformersDocumentClassifier(
+    #                     model_name_or_path=classifier_name,
+    #                     task="text-classification",
+    #                     top_k = None)
+
+    # Download model from HF Hub
+    doc_classifier = SetFitModel.from_pretrained(classifier_name)
+    
+    
+    # doc_classifier = pipeline("text-classification", 
+    #                         model=classifier_name, 
+    #                         return_all_scores=True, 
+    #                         function_to_apply= "sigmoid")
 
-            # will add page number only in case of PDF not for text/docx file.
-            add_page_number=True
-            )
-        
-        for i in documents:
-            # # basic cleaning before passing it to preprocessor.
-            # i = basic(i)
-            docs_processed = preprocessor.process([i])
-            if apply_clean:
-                for item in docs_processed:
-                    item.content = basic(item.content, remove_punc= remove_punc)
-            else:
-                pass
+    return doc_classifier
 
-        df = pd.DataFrame(docs_processed)
-        all_text = " ".join(df.content.to_list())
-        para_list = df.content.to_list()
-        logging.info('document split into {} paragraphs'.format(len(para_list)))
-        output = {'documents': docs_processed,
-                  'dataframe': df,
-                  'text': all_text,
-                  'paraList': para_list
-                 }
-        return output, "output_1"
-    def run_batch():
-        """
-            we dont have requirement to process the multiple files in one go
-            therefore nothing here, however to use the custom node we need to have
-            this method for the class.
-        """
-        return
 
-def processingpipeline():
+@st.cache_data
+def vulnerability_classification(haystack_doc:pd.DataFrame,
+                        threshold:float = 0.5, 
+                        classifier_model:pipeline= None
+                        )->Tuple[DataFrame,Series]:
     """
-    Returns the preprocessing pipeline. Will use FileConverter and UdfPreProcesor 
-    from utils.preprocessing
+    Text-Classification on the list of texts provided. Classifier provides the 
+    most appropriate label for each text. these labels are in terms of if text 
+    reference a group in a vulnerable situation.
+    ---------
+    haystack_doc: List of haystack Documents. The output of Preprocessing Pipeline 
+    contains the list of paragraphs in different format,here the list of 
+    Haystack Documents is used.
+    threshold: threshold value for the model to keep the results from classifier
+    classifiermodel: you can pass the classifier model directly,which takes priority
+    however if not then looks for model in streamlit session.
+    In case of streamlit avoid passing the model directly.
+    Returns
+    ----------
+    df: Dataframe with two columns['SDG:int', 'text']
+    x: Series object with the unique SDG covered in the document uploaded and 
+    the number of times it is covered/discussed/count_of_paragraphs. 
     """
+    logging.info("Working on vulnerability Identification")
+    haystack_doc['Vulnerability Label'] = 'NA'
+    # haystack_doc['PA_check'] = haystack_doc['Policy-Action Label'].apply(lambda x: True if len(x) != 0 else False)
+                            
+    # df1 = haystack_doc[haystack_doc['PA_check'] == True]
+    # df = haystack_doc[haystack_doc['PA_check'] == False]
+    if not classifier_model:
+        classifier_model = st.session_state['vulnerability_classifier']
+    
+        predictions = classifier_model(list(haystack_doc.text))
 
-    preprocessing_pipeline = Pipeline()
-    file_converter = FileConverter()
-    custom_preprocessor = UdfPreProcessor()
-
-    preprocessing_pipeline.add_node(component=file_converter, 
-                                    name="FileConverter", inputs=["File"])
-    preprocessing_pipeline.add_node(component = custom_preprocessor, 
-                            name ='UdfPreProcessor', inputs=["FileConverter"])
+        
 
-    return preprocessing_pipeline
+        pred_labels = get_vulnerability_labels(predictions)
+      
+        haystack_doc['Vulnerability Label'] = pred_labels
+    #   placeholder = {}
+    #   for j in range(len(temp)):
+    #     placeholder[temp[j]['label']] = temp[j]['score']
+    #   list_.append(placeholder)
+    # labels_ = [{**list_[l]} for l in range(len(predictions))]
+    # truth_df = DataFrame.from_dict(labels_)
+    # truth_df = truth_df.round(2)
+    # truth_df = truth_df.astype(float) >= threshold
+    # truth_df = truth_df.astype(str)
+    # categories = list(truth_df.columns)
+    # truth_df['Vulnerability Label'] = truth_df.apply(lambda x: {i if x[i]=='True' else 
+    #                                           None for i in categories}, axis=1)
+    # truth_df['Vulnerability Label'] = truth_df.apply(lambda x: list(x['Vulnerability Label'] 
+    #                                                         -{None}),axis=1)
+    # haystack_doc['Vulnerability Label'] = list(truth_df['Vulnerability Label'])
+    return haystack_doc