PolevalFastAPI/main.py

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,
        df_expected_text,
        average='macro',
        zero_division=0.0
        )
    f1_text_score = f1_text_score * 100

    f1_sentence_score = f1_score(
        df_expected_sentence,
        df_prediction_sentence,
        average='macro',
        zero_division=0.0
        ) 
    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