# from typing import List, Tuple # from typing_extensions import Literal # import logging # 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 # ## Labels dictionary ### # _lab_dict = { # '0':'NO', # '1':'YES', # } # def get_target_labels(preds): # """ # Function that takes the numerical predictions as an input and returns a list of the labels. # """ # # Get label names # preds_list = preds.tolist() # predictions_names=[] # # loop through each prediction # for ele in preds_list: # # see if there is a value 1 and retrieve index # try: # index_of_one = ele.index(1) # except ValueError: # index_of_one = "NA" # # Retrieve the name of the label (if no prediction made = NA) # if index_of_one != "NA": # name = label_dict[index_of_one] # else: # name = "Other" # # Append name to list # predictions_names.append(name) # return predictions_names # @st.cache_resource # def load_targetClassifier(config_file:str = None, classifier_name:str = None): # """ # 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 # -------- # 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 # """ # if not classifier_name: # if not config_file: # logging.warning("Pass either model name or config file") # return # else: # config = getconfig(config_file) # classifier_name = config.get('target','MODEL') # logging.info("Loading classifier") # doc_classifier = pipeline("text-classification", # model=classifier_name, # top_k =1) # return doc_classifier # @st.cache_data # def target_classification(haystack_doc:pd.DataFrame, # threshold:float = 0.5, # classifier_model:pipeline= None # )->Tuple[DataFrame,Series]: # """ # Text-Classification on the list of texts provided. Classifier provides the # most appropriate label for each text. There labels indicate whether the paragraph # references a specific action, target or measure in the paragraph. # --------- # 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 target/action identification") # haystack_doc['Vulnerability Label'] = 'NA' # if not classifier_model: # classifier_model = st.session_state['target_classifier'] # # Get predictions # predictions = classifier_model(list(haystack_doc.text)) # # Get labels for predictions # pred_labels = getlabels(predictions) # # Save labels # haystack_doc['Target Label'] = pred_labels # # logging.info("Working on action/target extraction") # # if not classifier_model: # # classifier_model = st.session_state['target_classifier'] # # results = classifier_model(list(haystack_doc.text)) # # labels_= [(l[0]['label'], # # l[0]['score']) for l in results] # # df1 = DataFrame(labels_, columns=["Target Label","Target Score"]) # # df = pd.concat([haystack_doc,df1],axis=1) # # df = df.sort_values(by="Target Score", ascending=False).reset_index(drop=True) # # df['Target Score'] = df['Target Score'].round(2) # # df.index += 1 # # # df['Label_def'] = df['Target Label'].apply(lambda i: _lab_dict[i]) # return haystack_doc