def donwload_evals()

in utils/taskcluster_downloader.py [0:0]

• 48 lines of code
• 10 McCabe index (conditional complexity)

def donwload_evals(group_id, output):
    options = {"rootUrl": ("%s" % TC_MOZILLA)}
    queue = taskcluster.Queue(options=options)
    group: Any = queue.listTaskGroup(group_id)

    results = []

    for task in group["tasks"]:
        if task["status"]["state"] != "completed":
            continue

        label = task["task"]["tags"]["kind"]
        if "evaluate" not in label:
            continue

        task_id = task["status"]["taskId"]

        task_obj: Any = queue.task(task_id)
        task["status"]["runs"][-1]["runId"]
        task_obj_label = task_obj["tags"]["label"].replace("/", "_")

        artifacts_response: Any = queue.listLatestArtifacts(task_id)
        artifacts = artifacts_response["artifacts"]
        artifact_name = [
            artifact["name"] for artifact in artifacts if artifact["name"].endswith(".metrics")
        ][0]

        print(f"Downloading {artifact_name} for {task_obj_label}")
        content, _ = downloadArtifactToBuf(
            taskId=task["status"]["taskId"],
            name=artifact_name,
            queueService=queue,
        )
        bleu, chrf, _ = content.tobytes().decode().split("\n")

        match = eval_regex.match(task_obj_label)
        if not match:
            print(f"Cannot match {task_obj_label}")
            raise ValueError(f"Cannot match {task_obj_label}")

        groups = match.groupdict()
        model = groups["model"]
        importer = groups["importer"]
        augmentation = groups.get("aug", "") or ""
        dataset = groups["dataset"]
        groups["lang"]
        suffix = groups.get("suffix", "") or ""

        result = (model + suffix, f"{importer}_{dataset}", augmentation, bleu, chrf)
        print(f"Result: {result}")
        results.append(result)

    os.makedirs(output, exist_ok=True)
    output_path = os.path.join(output, f"{group_id}-metrics.csv")
    with open(output_path, "w") as csvfile:
        csv_writer = csv.writer(csvfile)
        csv_writer.writerow(["Model", "Dataset", "Augmentation", "BLEU", "chrF"])
        for res in results:
            csv_writer.writerow(res)