# -*- coding: utf-8 -*-
"""TridentModel.ipynb
Automatically generated by Colaboratory.
Original file is located at
https://colab.research.google.com/drive/1u07dSU0DoKnNzGzySXMTisXnaloqpUEO
TRIDENT MODEL IMPLEMENTATION
Date: 14 January 2023
Authors: Egheosa Ogbomo & Amran Mohammed (The Polymer Guys)
Description: This script combines three ML-based models to identify whether an input text is related to green plastics or not.
"""
#pip install transformers
########## IMPORTING REQUIRED PYTHON PACKAGES ##########
import pandas as pd
import tensorflow as tf
import numpy as np
from transformers import AutoTokenizer, AutoModel
#import torch
import math
import time
import csv
import pandas as pd
import nltk
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
nltk.download('stopwords')
nltk.download('punkt')
import string
########## DEFINING FUNCTIONS FOR MODEL IMPLEMENTATIONS ##########
### Input data cleaner
all_stopwords = stopwords.words('english') # Making sure to only use English stopwords
extra_stopwords = ['ii', 'iii'] # Can add extra stopwords to be removed from dataset/input abstracts
all_stopwords.extend(extra_stopwords)
def clean_data(input, type='Dataframe'):
"""
As preparation for use with the text similarity model, this function removes superfluous data from either a dataframe full of
classifications, or an input string, in order for embeddings to be calculated for them. Removes:
• Entries with missing abstracts/descriptions/classifications/typos
• Duplicate entries
• Unnecessary punctuation
• Stop words (e.g., by, a , an, he, she, it)
• URLs
• All entries are in the same language
:param input: Either a dataframe or an individual string
:param type: Tells fucntion whether input is a dataframe or an individual string
:return: (if dataframe), returns a dataframe containing CPC classfication codes and their associated 'cleaned' description
:return: (if string), returns a 'cleaned' version of the input string
"""
if type == 'Dataframe':
cleaneddf = pd.DataFrame(columns=['Class', 'Description'])
for i in range(0, len(input)):
row_list = input.loc[i, :].values.flatten().tolist()
noNaN_row = [x for x in row_list if str(x) != 'nan']
listrow = []
if len(noNaN_row) > 0:
row = noNaN_row[:-1]
row = [x.strip() for x in row]
row = (" ").join(row)
text_tokens = word_tokenize(row) # splits abstracts into individual tokens to allow removal of stopwords by list comprehension
Stopword_Filtered_List = [word for word in text_tokens if not word in all_stopwords] # removes stopwords
row = (" ").join(Stopword_Filtered_List) # returns abstract to string form
removechars = ['[', ']', '{', '}', ';', '(', ')', ',', '.', ':', '/', '-', '#', '?', '@', '£', '$']
for char in removechars:
row = list(map(lambda x: x.replace(char, ''), row))
row = ''.join(row)
wnum = row.split(' ')
wnum = [x.lower() for x in wnum]
#remove duplicate words
wnum = list(dict.fromkeys(wnum))
#removing numbers
wonum = []
for x in wnum:
xv = list(x)
xv = [i.isnumeric() for i in xv]
if True in xv:
continue
else:
wonum.append(x)
row = ' '.join(wonum)
l = [noNaN_row[-1], row]
cleaneddf.loc[len(cleaneddf)] = l
cleaneddf = cleaneddf.drop_duplicates(subset=['Description'])
cleaneddf.to_csv('E:/Users/eeo21/Startup/CPC_Classifications_List/additionalcleanedclasses.csv', index=False)
return cleaneddf
elif type == 'String':
text_tokens = word_tokenize(input) # splits abstracts into individual tokens to allow removal of stopwords by list comprehension
Stopword_Filtered_List = [word for word in text_tokens if not word in all_stopwords] # removes stopwords
row = (" ").join(Stopword_Filtered_List) # returns abstract to string form
removechars = ['[', ']', '{', '}', ';', '(', ')', ',', '.', ':', '/', '-', '#', '?', '@', '£', '$']
for char in removechars:
row = list(map(lambda x: x.replace(char, ''), row))
row = ''.join(row)
wnum = row.split(' ')
wnum = [x.lower() for x in wnum]
# remove duplicate words
wnum = list(dict.fromkeys(wnum))
# removing numbers
wonum = []
for x in wnum:
xv = list(x)
xv = [i.isnumeric() for i in xv]
if True in xv:
continue
else:
wonum.append(x)
row = ' '.join(wonum)
return row
### Mean Pooler
"""
Performs a mean pooling to reduce dimension of embedding
"""
def mean_pooling(model_output, attention_mask):
token_embeddings = model_output[0] #First element of model_output contains all token embeddings
input_mask_expanded = attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float()
return tf.reduce_sum(token_embeddings * input_mask_expanded, 1) / tf.clip_by_value(input_mask_expanded.sum(1), clip_value_min=1e-9, clip_value_max=math.inf)
### Sentence Embedder
# def sentence_embedder(sentences, model_path):
# """
# Calling the sentence similarity model to generate embeddings on input text.
# :param sentences: takes input text in the form of a string
# :param model_path: path to the text similarity model
# :return returns a (1, 384) embedding of the input text
# """
# tokenizer = AutoTokenizer.from_pretrained(model_path) #instantiating the sentence embedder using HuggingFace library
# model = AutoModel.from_pretrained(model_path, from_tf=True) #making a model instance
# encoded_input = tokenizer(sentences, padding=True, truncation=True, return_tensors='pt')
# # Compute token embeddings
# with torch.no_grad():
# model_output = model(**encoded_input)
# sentence_embeddings = mean_pooling(model_output, encoded_input['attention_mask']) #outputs a (1, 384) tensor representation of input text
# return sentence_embeddings
# ### Sentence Embedding Preparation Function
# def convert_saved_embeddings(embedding_string):
# """
# Preparing pre-computed embeddings for use for comparison with new abstract embeddings .
# Pre-computed embeddings are saved as tensors in string format so need to be converted back to numpy arrays in order to calculate cosine similarity.
# :param embedding_string:
# :return: Should be a single tensor with dims (,384) in string formate
# """
# embedding = embedding_string.replace('(', '')
# embedding = embedding.replace(')', '')
# embedding = embedding.replace('[', '')
# embedding = embedding.replace(']', '')
# embedding = embedding.replace('tensor', '')
# embedding = embedding.replace(' ', '')
# embedding = embedding.split(',')
# embedding = [float(x) for x in embedding]
# embedding = np.array(embedding)
# embedding = np.expand_dims(embedding, axis=0)
# embedding = torch.from_numpy(embedding)
# return embedding
# ### Generating Class Embeddings
# Model_Path = 'Model_bert' ### Insert Path to MODEL DIRECTORY here
# def class_embbedding_generator(classes):
# """
# This function is to be used to generate and save class embeddings
# Takes an input of 'cleaned' classes, generated by clean_data function, and computes vector representations of these classes (the embeddings) and saves them to csv
# :classes: Classes should be a dataframe including all of broad scope classes that are intended to be used to make comparisons with
# """
# class_embeddings = pd.DataFrame(columns=['Class', 'Description', 'Embedding'])
# for i in range(len(classes)):
# class_name = classes.iloc[i, 0]
# print(class_name)
# class_description = classes.iloc[i, 1]
# class_description_embedding = sentence_embedder(class_description, Model_Path)
# class_description_embedding = class_description_embedding.numpy()
# class_description_embedding = torch.from_numpy(class_description_embedding)
# embedding_entry = [class_name, class_description, class_description_embedding]
# class_embeddings.loc[len(class_embeddings)] = embedding_entry
# ### Broad Scope Classifier
# Model_Path = 'Model_bert' ### Insert Path to MODEL DIRECTORY here
# def broad_scope_class_predictor(class_embeddings, abstract_embedding, N=5, Sensitivity='Medium'):
# """
# Takes in pre-computed class embeddings and abstract texts, converts abstract text into
# :param class_embeddings: dataframe of class embeddings
# :param abstract: a single abstract embedding
# :param N: N highest matching classes to return, from highest to lowest, default is 5
# :return: predictions: a full dataframe of all the predictions on the 9500+ classes, HighestSimilarity: Dataframe of the N most similar classes
# """
# predictions = pd.DataFrame(columns=['Class Name', 'Score'])
# for i in range(len(class_embeddings)):
# class_name = class_embeddings.iloc[i, 0]
# embedding = class_embeddings.iloc[i, 2]
# embedding = convert_saved_embeddings(embedding)
# abstract_embedding = abstract_embedding.numpy()
# abstract_embedding = torch.from_numpy(abstract_embedding)
# cos = torch.nn.CosineSimilarity(dim=1)
# score = cos(abstract_embedding, embedding).numpy().tolist()
# result = [class_name, score[0]]
# predictions.loc[len(predictions)] = result
# greenpredictions = predictions.tail(52)
# if Sensitivity == 'High':
# Threshold = 0.5
# elif Sensitivity == 'Medium':
# Threshold = 0.40
# elif Sensitivity == 'Low':
# Threshold = 0.35
# GreenLikelihood = 'False'
# for i in range(len(greenpredictions)):
# score = greenpredictions.iloc[i, 1]
# if float(score) >= Threshold:
# GreenLikelihood = 'True'
# break
# else:
# continue
# HighestSimilarity = predictions.nlargest(N, ['Score'])
# print(HighestSimilarity)
# print(GreenLikelihood)
# return predictions, HighestSimilarity, GreenLikelihood