179 lines
5.6 KiB
Python
179 lines
5.6 KiB
Python
![]() |
from fastapi import FastAPI
|
||
|
from pydantic import BaseModel
|
||
|
from sklearn.metrics import f1_score
|
||
|
|
||
|
import pandas as pd
|
||
|
import distance
|
||
|
import io
|
||
|
|
||
|
|
||
|
ERROR_RESPONSE = {
|
||
|
"status": 400
|
||
|
}
|
||
|
app = FastAPI()
|
||
|
|
||
|
|
||
|
class Data(BaseModel):
|
||
|
challenge: str
|
||
|
dev_expected: str = ""
|
||
|
dev_out: str = ""
|
||
|
testA_expected: str = ""
|
||
|
testA_out: str = ""
|
||
|
testB_expected: str = ""
|
||
|
testB_out: str = ""
|
||
|
dev_in: str= ""
|
||
|
testA_in: str=""
|
||
|
testB_in: str=""
|
||
|
|
||
|
|
||
|
def preprocess_data(out, expected):
|
||
|
out = out.split("\n")
|
||
|
expected = expected.split("\n")[:-1]
|
||
|
|
||
|
out = out[:len(expected)]
|
||
|
|
||
|
return out, expected
|
||
|
|
||
|
|
||
|
def get_levenshtein_score(trues, preds):
|
||
|
def normalize_answer(s):
|
||
|
return s.lower()
|
||
|
|
||
|
levenstein_scores = []
|
||
|
for true, pred in [(true, pred) for (true, pred) in zip(trues, preds) if true != ""]:
|
||
|
if pred == "":
|
||
|
levenstein_score = 0
|
||
|
else:
|
||
|
levenstein_score = 1 - distance.nlevenshtein(normalize_answer(true), normalize_answer(pred))
|
||
|
levenstein_scores.append(levenstein_score)
|
||
|
|
||
|
avg_levenstein_score = sum(levenstein_scores) / len(levenstein_scores) * 100
|
||
|
return avg_levenstein_score
|
||
|
|
||
|
|
||
|
def get_answerability_f1(trues, preds):
|
||
|
def get_answerability(answers):
|
||
|
return [1 if answer == "" else 0 for answer in answers]
|
||
|
|
||
|
true_answerability = get_answerability(trues)
|
||
|
predicted_answerability = get_answerability(preds)
|
||
|
answerability_f1 = f1_score(true_answerability, predicted_answerability, zero_division=0.0) * 100
|
||
|
return answerability_f1
|
||
|
|
||
|
|
||
|
def get_scores(trues, preds):
|
||
|
scores = {}
|
||
|
scores["Levenshtein"] = get_levenshtein_score(trues, preds)
|
||
|
scores["AnswerabilityF1"] = get_answerability_f1(trues, preds)
|
||
|
scores["Final"] = round((scores["Levenshtein"] + scores["AnswerabilityF1"]) / 2, 2)
|
||
|
|
||
|
scores["Levenshtein"] = round(get_levenshtein_score(trues, preds), 2)
|
||
|
scores["AnswerabilityF1"] = round(get_answerability_f1(trues, preds), 2)
|
||
|
return scores
|
||
|
|
||
|
|
||
|
def get_emotion_recognition_scores(df_in, df_expected, df_predition):
|
||
|
text_annotation = df_in['text'].apply(lambda x: x == '#' * len(x))
|
||
|
|
||
|
df_expected_text = df_expected[text_annotation]
|
||
|
df_expected_sentence = df_expected[~text_annotation]
|
||
|
|
||
|
df_prediction_text = df_predition[text_annotation]
|
||
|
df_prediction_sentence = df_predition[~text_annotation]
|
||
|
|
||
|
f1_text_score = f1_score(
|
||
|
df_prediction_text.values.flatten(),
|
||
|
df_expected_text.values.flatten(),
|
||
|
average='macro',
|
||
|
zero_division=0.0,
|
||
|
labels=[True, False]
|
||
|
)
|
||
|
f1_text_score = f1_text_score * 100
|
||
|
|
||
|
f1_sentence_score = f1_score(
|
||
|
df_expected_sentence.values.flatten(),
|
||
|
df_prediction_sentence.values.flatten(),
|
||
|
average='macro',
|
||
|
zero_division=0.0,
|
||
|
labels=[True, False]
|
||
|
)
|
||
|
f1_sentence_score = f1_sentence_score * 100
|
||
|
|
||
|
final_score = (f1_text_score + f1_sentence_score) / 2
|
||
|
|
||
|
return {
|
||
|
"SentenceF1": round(f1_sentence_score, 2),
|
||
|
"TextF1": round(f1_text_score, 2),
|
||
|
"FinalF1": round(final_score, 2)
|
||
|
}
|
||
|
|
||
|
@app.get("/")
|
||
|
async def root(data: Data):
|
||
|
challenge = data.challenge
|
||
|
dev_expected = data.dev_expected
|
||
|
dev_out = data.dev_out
|
||
|
testA_expected = data.testA_expected
|
||
|
testA_out = data.testA_out
|
||
|
testB_expected = data.testB_expected
|
||
|
testB_out = data.testB_out
|
||
|
dev_in = data.dev_in
|
||
|
testA_in = data.testA_in
|
||
|
testB_in = data.testB_in
|
||
|
|
||
|
if challenge == "QuestionAnswering":
|
||
|
results = {}
|
||
|
if len(dev_out) > 0:
|
||
|
dev_out, dev_expected = preprocess_data(dev_out, dev_expected)
|
||
|
if len(dev_out) != len(dev_expected):
|
||
|
return ERROR_RESPONSE
|
||
|
results["dev-0"] = get_scores(dev_expected, dev_out)
|
||
|
|
||
|
if len(testA_out) > 0:
|
||
|
testA_out, testA_expected = preprocess_data(testA_out, testA_expected)
|
||
|
if len(testA_out) != len(testA_expected):
|
||
|
return ERROR_RESPONSE
|
||
|
results["test-A"] = get_scores(testA_expected, testA_out)
|
||
|
|
||
|
if len(testB_out) > 0:
|
||
|
testB_out, testB_expected = preprocess_data(testB_out, testB_expected)
|
||
|
if len(testB_out) != len(testB_expected):
|
||
|
return ERROR_RESPONSE
|
||
|
results["test-B"] = get_scores(testB_expected, testB_out)
|
||
|
|
||
|
if len(results) == 0:
|
||
|
return ERROR_RESPONSE
|
||
|
else:
|
||
|
return {
|
||
|
"status": 200,
|
||
|
"results": results
|
||
|
}
|
||
|
elif challenge == "EmotionRecognition":
|
||
|
results = {}
|
||
|
if len(dev_out) > 0:
|
||
|
df_in = pd.read_table(io.StringIO(dev_in))
|
||
|
df_expected = pd.read_table(io.StringIO(dev_expected))
|
||
|
df_predition = pd.read_table(io.StringIO(dev_out))
|
||
|
|
||
|
results["dev-0"] = get_emotion_recognition_scores(df_in, df_expected, df_predition)
|
||
|
if len(testA_out) > 0:
|
||
|
df_in = pd.read_table(io.StringIO(testA_in))
|
||
|
df_expected = pd.read_table(io.StringIO(testA_expected))
|
||
|
df_predition = pd.read_table(io.StringIO(testA_out))
|
||
|
|
||
|
results["test-A"] = get_emotion_recognition_scores(df_in, df_expected, df_predition)
|
||
|
if len(testB_out) > 0:
|
||
|
df_in = pd.read_table(io.StringIO(testB_in))
|
||
|
df_expected = pd.read_table(io.StringIO(testB_expected))
|
||
|
df_predition = pd.read_table(io.StringIO(testB_out))
|
||
|
|
||
|
results["test-B"] = get_emotion_recognition_scores(df_in, df_expected, df_predition)
|
||
|
|
||
|
if len(results) == 0:
|
||
|
return ERROR_RESPONSE
|
||
|
else:
|
||
|
return {
|
||
|
"status": 200,
|
||
|
"results": results
|
||
|
}
|
||
|
else:
|
||
|
return ERROR_RESPONSE
|