Spaces:
Running
Running
| import pandas as pd | |
| import numpy as np | |
| import tensorflow as tf | |
| import tensorflow_hub as hub | |
| import sys | |
| import random | |
| import os | |
| sys.path.append('models') | |
| from official.nlp.data import classifier_data_lib | |
| from official.nlp.bert import tokenization | |
| from official.nlp import optimization | |
| tf.get_logger().setLevel('ERROR') | |
| import math | |
| from datetime import datetime | |
| import gradio as gr | |
| config = tf.compat.v1.ConfigProto( | |
| device_count = {'cpu': 0} | |
| ) | |
| sess = tf.compat.v1.Session(config=config) | |
| num_warmup_steps=1 | |
| num_train_steps=1 | |
| init_lr = 3e-5 | |
| optimizer = optimization.create_optimizer(init_lr=init_lr, | |
| num_train_steps=num_train_steps, | |
| num_warmup_steps=num_warmup_steps, | |
| optimizer_type='adamw') | |
| ### Load Model | |
| checkpoint_filepath=r'./Checkpoint' | |
| model = tf.keras.models.load_model(checkpoint_filepath, custom_objects={'KerasLayer':hub.KerasLayer , 'AdamWeightDecay': optimizer}) | |
| df_report = pd.read_csv('./CTH_Description.csv') | |
| df_report['CTH Code'] = df_report['CTH Code'].astype(str).str.zfill(8) | |
| df_report_DUTY = pd.read_csv('./CTH_WISE_DUTY_RATE.csv') | |
| df_report_DUTY['CTH'] = df_report_DUTY['CTH'].astype(str).str.zfill(8) | |
| df = pd.read_csv("./CTH_CODE_MAP.csv") | |
| df['CTH'] = df['CTH'].astype(str).str.zfill(8) | |
| df = df[['CTH', 'code']] | |
| class_names=df[['CTH','code']].drop_duplicates(subset='CTH').sort_values(by='code',ignore_index=True)['CTH'].values.tolist() | |
| label_list=list(range(0,len(class_names))) | |
| max_seq_length = 200 # maximum length of (token) input sequences . it can be any number | |
| train_batch_size = 32 # batch size ( 16 choosen to avoid Out-Of-Memory errors) | |
| # Get BERT layer and tokenizer: | |
| # More details here: https://tfhub.dev/tensorflow/bert_en_uncased_L-12_H-768_A-12/4 | |
| bert_layer = hub.KerasLayer("https://tfhub.dev/tensorflow/bert_en_uncased_L-12_H-768_A-12/4" , trainable = True) | |
| vocab_file = bert_layer.resolved_object.vocab_file.asset_path.numpy() | |
| do_lower_case = bert_layer.resolved_object.do_lower_case.numpy() | |
| tokenizer = tokenization.FullTokenizer(vocab_file , do_lower_case) | |
| # This provides a function to convert each row to input features and label ( as required by BERT) | |
| max_seq_length = 200 # maximum length of (token) input sequences . it can be any number | |
| def to_feature(text, label, label_list=label_list, max_seq_length=max_seq_length, tokenizer=tokenizer): | |
| example = classifier_data_lib.InputExample(guid = None, | |
| text_a = text.numpy(), | |
| text_b = None, | |
| label = label.numpy()) | |
| feature = classifier_data_lib.convert_single_example(0 , example , label_list , max_seq_length , tokenizer) | |
| return (feature.input_ids , feature.input_mask , feature.segment_ids , feature.label_id) | |
| def to_feature_map(text, label): | |
| input_ids , input_mask , segment_ids , label_id = tf.py_function(to_feature , inp = [text , label], | |
| Tout = [tf.int32 , tf.int32 , tf.int32 , tf.int32]) | |
| input_ids.set_shape([max_seq_length]) | |
| input_mask.set_shape([max_seq_length]) | |
| segment_ids.set_shape([max_seq_length]) | |
| label_id.set_shape([]) | |
| x = { | |
| "input_word_ids": input_ids, | |
| "input_mask": input_mask, | |
| "input_type_ids": segment_ids | |
| } | |
| return(x,label_id) | |
| def print3largest(arr, arr_size): | |
| third = first = second = -sys.maxsize | |
| for i in range(0, arr_size): | |
| if (arr[i] > first): | |
| third = second | |
| second = first | |
| first = arr[i] | |
| elif (arr[i] > second): | |
| third = second | |
| second = arr[i] | |
| elif (arr[i] > third): | |
| third = arr[i] | |
| pred_value_max_three=[first, second, third] | |
| return pred_value_max_three | |
| def count_special_character(string): | |
| special_char= 0 | |
| for i in range(len(string)): | |
| ch = string[i] | |
| if (string[i].isalpha()): | |
| continue | |
| else: | |
| special_char += 1 | |
| if len(string)==special_char: | |
| return False | |
| else: | |
| return True | |
| def predict_CTH(txt): | |
| print('Desc: ',txt) | |
| if (txt!='') and len(txt)>=3 and (count_special_character(txt)): | |
| valid_data = tf.data.Dataset.from_tensor_slices(([txt] , [1])) # 1 refers to 'entertainment' and 2 refers to 'sport' | |
| valid_data = (valid_data.map(to_feature_map).batch(1)) | |
| preds = model.predict(valid_data) | |
| predicted_values = tf.nn.softmax(preds) | |
| arr = predicted_values.numpy().tolist()[0] | |
| n = len(arr) | |
| pred_value_max_three=print3largest(arr, n) | |
| now = datetime.now() | |
| print("Time =", now) | |
| sum_all = pred_value_max_three[0] + pred_value_max_three[1] + pred_value_max_three[2] | |
| val_1 = pred_value_max_three[0]/sum_all | |
| val_2 = pred_value_max_three[1]/sum_all | |
| val_3 = pred_value_max_three[2]/sum_all | |
| if pred_value_max_three[0]<=0.000131: | |
| Var_CTH=[] | |
| Var_desc=[] | |
| Var_duty=[] | |
| pred_duty='' | |
| pred_desc='' | |
| pred_CTH='' | |
| return{'Not a adequate description':float(1.0)} | |
| else: | |
| Var_CTH=[] | |
| Var_desc=[] | |
| Var_duty=[] | |
| pred_duty='' | |
| pred_desc='' | |
| pred_CTH='' | |
| for i in pred_value_max_three: | |
| #i=pred_value_max_three[0] | |
| predicted_code=np.where(predicted_values.numpy()==i)[1][0] | |
| pred_CTH=df[df['code'] == predicted_code]['CTH'].iloc[0] | |
| try: | |
| pred_duty=df_report_DUTY[df_report_DUTY['CTH']==str(pred_CTH)]['DUTY_RATE'].iloc[0] | |
| except: | |
| pred_duty='' | |
| pass | |
| try: | |
| pred_desc=df_report[df_report['CTH Code']==str(pred_CTH)]['Concat Description'].iloc[0] | |
| except: | |
| pred_desc='' | |
| pass | |
| Var_CTH.append(pred_CTH) | |
| Var_desc.append(pred_desc) | |
| Var_duty.append(pred_duty) | |
| P1 ='CTH: '+str(Var_CTH[0])+' Duty Rate(%): '+ str(Var_duty[0]) | |
| P2 ='CTH: '+str(Var_CTH[1])+' Duty Rate(%): '+ str(Var_duty[1]) | |
| P3 ='CTH: '+str(Var_CTH[2])+' Duty Rate(%): '+ str(Var_duty[2]) | |
| Q1='Desc: '+str(Var_desc[0]) | |
| Q2='Desc: '+str(Var_desc[1]) | |
| Q3='Desc: '+str(Var_desc[2]) | |
| return {str(P1):float(val_1),str(Q1):float(val_1), | |
| str(P2):float(val_2),str(Q2):float(val_2), | |
| str(P3):float(val_3),str(Q3):float(val_3),} | |
| else: | |
| return{'Enter Correct Description':float(1.0)} | |
| input_txt=gr.Textbox( | |
| label='Enter Your Product Descrption', | |
| lines=3, | |
| ) | |
| description="<p style='color:blue;text-align:justify;font-size:1vw;'>AdvaitBERT is modified version of BERT (Bidirectional Encoder Representation for Transformers), \ | |
| finetuned on the Text corpus of Indian Customs Declarations. It is trained for performing \ | |
| downstream tasks like automating the tariff classification and validation process of Customs \ | |
| declarations in realtime. This model may help Customs administration to efficiently use AI assisted \ | |
| NLP in realtime Customs process like Assessment, Post Clearance Audit, thereby highlighting classification \ | |
| inconsistencies and help in revenue augmentation.</a></p>" | |
| title="<h1 style='color:green;text-align:center;font-size:2vw;'>AdvaitBERT </a></h1>" | |
| article="<p style='color:black;text-align:right;font-size:1vw;'>Powered by NCTC </a></p>" | |
| #css=".gradio-container {background-color: papayawhip}", | |
| path_2='./CTH_CODE_MAP.csv' | |
| # Get the absolute path by combining the current working directory with the relative path | |
| absolute_path_1 = os.path.abspath(checkpoint_filepath) | |
| absolute_path_2 = os.path.abspath(path_2) | |
| # Print the absolute path | |
| print("Absolute path:", absolute_path_1) | |
| blocked_files=[absolute_path_1,absolute_path_2] | |
| gr.Interface( | |
| predict_CTH, | |
| inputs=input_txt, | |
| outputs="label", | |
| interpretation="default", | |
| description=description, | |
| #live=True, | |
| examples = ['200 SI/SI/SI LPO ALUMINIUM LIDS (QTY: 8820000 PCS/PRICE: 21.'], | |
| title=title, | |
| article=article, | |
| blocked_paths=blocked_files, | |
| ).launch(debug=True) |