leaderboard_server.py

import json
import os

import meeteval.io
import pandas as pd
from utils import calc_wer, aggregate_wer_metrics
from txt_norm import get_text_norm

# Constants
REFERENCE_BASE_PATH = os.path.abspath("references")  # e.g. ./references/single_channel/dataset1.json
TASKS_METADATA_PATH = os.path.abspath("tasks_metadata.json")

def list_files(startpath):
    for root, dirs, files in os.walk(startpath):
        level = root.replace(startpath, '').count(os.sep)
        indent = ' ' * 4 * (level)
        print('{}{}/'.format(indent, os.path.basename(root)))
        subindent = ' ' * 4 * (level + 1)
        for f in files:
            print('{}{}'.format(subindent, f))

class LeaderboardServer:
    def __init__(self):
        self.local_leaderboard = os.path.abspath("/data/submissions")
        self.submisssion_id_to_file = {}  # Maps model_id to filepath
        self.tasks_metadata = json.load(open(TASKS_METADATA_PATH))["tasks"]
        self.submission_ids = set()
        self.results_file = os.path.join(self.local_leaderboard, "results.json")
        os.makedirs(self.local_leaderboard, exist_ok=True)
        self.fetch_existing_models()
        self.text_normalizer = get_text_norm("whisper_nsf")

    def fetch_existing_models(self):
        self.submisssion_id_to_file.clear()
        self.submission_ids.clear()

        results_path = os.path.join(self.local_leaderboard, "results.json")
        if not os.path.exists(results_path):
            return
        with open(results_path) as f:
            results = json.load(f)
        for model_id in results:
            self.submission_ids.add(model_id)
            hyp_path = os.path.join(self.local_leaderboard, f"{model_id}_hyp.json")
            self.submisssion_id_to_file[model_id] = hyp_path

    def prepare_model_for_submission(self, file, metadata, task, datasets, normalize=False):
        submitted_by = metadata["submitted_by"]
        model_id = metadata["model_id"]
        original_id = f"{submitted_by}_{model_id}"
        # Run WER eval
        results = {}
        hyp_seglst = meeteval.io.load(file)

        for dataset in datasets:
            ref_path = os.path.join(REFERENCE_BASE_PATH, task, f"{dataset}.json")
            ref_seglst = meeteval.io.load(ref_path)
            sessions = ref_seglst.unique('session_id')
            local_hyps = hyp_seglst.filter(lambda seg: seg['session_id'] in sessions)
            ref_seglst = ref_seglst.map(lambda seg: {**seg, "words":self.text_normalizer(seg["words"]) if normalize else seg["words"]})
            local_hyps = local_hyps.map(lambda seg: {**seg, "words":self.text_normalizer(seg["words"]) if normalize else seg["words"]})
            per_session_wers = calc_wer(tcp_hyp_seglst=local_hyps, ref_seglst=ref_seglst, collar=5, metrics_list=["tcp_wer"])
            metrics = aggregate_wer_metrics(per_session_wers, ["tcp_wer"])
            results[dataset] = metrics

        # Update results file
        results_path = os.path.join(self.local_leaderboard, "results.json")
        if os.path.exists(results_path):
            with open(results_path) as f:
                all_results = json.load(f)
        else:
            all_results = {}

        all_results[original_id] = {
            "model_id": model_id,
            "submitted_by": submitted_by,
            "results": results
        }

        with open(results_path, "w") as f:
            json.dump(all_results, f, indent=2)

        with open(f"{self.local_leaderboard}/{original_id}_hyp.json", "w") as out_f:
            with open(file, "r") as in_f:
                out_f.write(in_f.read())

    def update_leaderboard(self):
        self.fetch_existing_models()

    def get_leaderboard(self):
        results_path = os.path.join(self.local_leaderboard, "results.json")
        if not os.path.exists(results_path):
            return pd.DataFrame(columns=["No submissions yet"])

        with open(results_path) as f:
            results = json.load(f)

        rows = []
        for content in results.values():
            row = {"Model ID": content["model_id"], "Submitted by": content["submitted_by"]}
            for k, v in content["results"].items():
                row[k] = v.get("tcp_wer", None)
            rows.append(row)

        df =  pd.DataFrame(rows)
        df.iloc[:, 2:] *= 100.0 # Convert WER to percentage, first two columns are metadata
        df = df.fillna("-")
        df = df.round(2)

        return df