metadata_annotations.py

# Handle annotation analysis for the paper (last updated for LREC2026)

import glob
import json
import numpy as np
import pandas as pd
import seaborn as sns
from nltk import agreement
import matplotlib.pyplot as plt
from itertools import combinations
from sklearn.metrics import confusion_matrix


def openJson(path):
    "Open a json file."
    with open(path,'r',encoding='utf-8') as f:
        data = json.load(f)
    return data


def writeJson(path,data):
	"Create a json file."
	with open(path,"w",encoding='utf-8') as f:
		json.dump(data,f,indent=4,ensure_ascii=False)


def openPandasCSV(path):
	"Open a csv file with pandas."
	return pd.read_csv(path,encoding="utf-8",sep="\t")


def getAnnotationRes():
    "save all annotation in a simple format ID:LABEL for each annotator."
    annotation_res = {}
    for path in glob.glob(f"annotations/snowclone_1000_*.csv"):
        if "merge" not in path:
            annotator = path.split("_")[-1].replace(".csv","")
            if annotator not in annotation_res:
                annotation_res[annotator] = {}
            csv = openPandasCSV(path)
            labels = {}
            for i,lab in enumerate(csv["label"]):
                labels[csv["snowclone_candidate"][i]] = lab
            for iid,label in labels.items():
                    if iid not in annotation_res[annotator]:
                        annotation_res[annotator][iid] = label
    writeJson("annotations/snowclone_annotation_matrix.json",annotation_res)


def getInterAnnotatorAgreement():
    "Compute inter-annotator agreement score from an annotation matrix in .json format."
    data = openJson("annotations/snowclone_annotation_matrix.json")
    data_inter = []
    for key,value in data.items():
        for k,v in value.items():
            data_inter.append([key,k,v])
    ratingtask = agreement.AnnotationTask(data=data_inter)
    results = {
    #"kappa "  : ratingtask.kappa()#,
    #"fleiss " : ratingtask.multi_kappa()#,
    "alpha "  : ratingtask.alpha()#,
    #"scotts " : ratingtask.pi()
    }
    print(json.dumps(results, indent = 2))


def computeConfusionMatrix(df,annotator_a,annotator_b):
    "Create a confusion matrix for a pair of annotators."
    a_labels = df.loc[annotator_a].values
    b_labels = df.loc[annotator_b].values
    labels = np.unique(np.concatenate((a_labels, b_labels)))  #get all unique labels
    cm = confusion_matrix(a_labels,b_labels,labels=labels)
    return cm,labels


def saveConfusionMatrix(cm,labels,annotator_a,annotator_b,max=300,merge=False):
    "Save the confusion matrix in a .png file."
    sns.heatmap(cm, annot=True,fmt='d',cmap='Blues',xticklabels=labels,yticklabels=labels,vmin=0,vmax=max)
    plt.xlabel(annotator_a)
    plt.ylabel(annotator_b)
    if merge:
        plt.savefig(f"annotations/merged_confusion_matrix.png")
    else:
        plt.savefig(f"annotations/confusion_matrix_{annotator_a}_{annotator_b}.png")
    plt.close()


def getConfusionMatrix():
    "Get confusion matrices between each possible pairs of annotators."
    data = openJson("annotations/snowclone_annotation_matrix.json")
    df = pd.DataFrame.from_dict(data, orient='index')
    annotators = list(data.keys())
    pairs = [list(i) for i in combinations(annotators,2)]
    for pair in pairs:
        cm,labels = computeConfusionMatrix(df,pair[0],pair[1])
        saveConfusionMatrix(cm,labels,pair[0],pair[1],max=1000)


def getDisagreements():
    "Get every instance of disaagreement between annotators."
    disagreements = {}
    data = openJson("annotations/snowclone_annotation_matrix.json")
    for i in data["A1"].keys(): #to change
        test = []
        for annotator, values in data.items():
            test.append(values[str(i)])
        if len(set(test)) != 1:
            disagreements[str(i)] = {}
            for a,v in data.items():
                disagreements[str(i)][a] = v[str(i)]
    print(f"{len(disagreements)} disagreements")
    writeJson("annotations/disagreements.json",disagreements)


def plotResults(annotator):
    """
    Plot the results of the annotation for each annotator, 
    a.k.a the repartition of annotations he mades according to computed cosine similarity.
    """

    thresholds = np.arange(0.99,-0.01,-0.01).tolist()
    data = openPandasCSV(f"annotations/snowclone_1000_{annotator}.csv")
    label_col_name = 'label'
    if annotator == "merge":
         label_col_name = "label_final"

    value_1 = [0 for i in thresholds]
    value_0 = [0 for i in thresholds]

    for j,threshold in enumerate(thresholds):
        
        for i,label in enumerate(data[label_col_name]):
            if data["score"][i] > threshold:
                if label == 0:
                    value_0[j] += 1
                else:
                    value_1[j] += 1

    plt.plot(thresholds, value_1, linestyle='-', color="royalblue", alpha=0.8, label="snowclone")
    plt.plot(thresholds, value_0, linestyle='-', color="limegreen", alpha=0.8, label="not_snowclone")
    plt.axvline(x=0.78,color="red",linestyle='--')
    plt.axvline(x=0.81,color="black",linestyle='--')

    plt.xlabel('Computed cosine similarity')
    plt.ylabel('Occurrences')

    plt.legend()
    plt.title(annotator)
    #plt.xlim(thresholds[0],thresholds[-1])

    plt.ylim(top=700)
    plt.tight_layout()
    plt.savefig(f"annotations/threshold_labels_{annotator}.png")
    plt.close()


getAnnotationRes()
getInterAnnotatorAgreement()
getConfusionMatrix()
getDisagreements()
for annotator in ["A1","A2","A3","merge"]:
    plotResults(annotator)