Annotation recipes

python task_import_annotations.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --annotations-file 'annotations.zip' --import-format 'COCO 1.0'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Import annotations from a local file into an existing task, e.g. to load
predictions of a model or work made on another server.

Steps:
  1. Retrieve the task and count the objects it already has.
  2. Fetch the server's import format list and validate --import-format.
  3. Upload the annotations file and wait for the server to process it.
  4. Count the objects again to show what the import added.

Usage (run ``python task_import_annotations.py --help`` for the full list of options):
  python task_import_annotations.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --annotations-file 'annotations.zip' --import-format 'COCO 1.0'
"""

import argparse
import sys
from pathlib import Path

from cvat_sdk import make_client


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--task-id", type=int, required=True, help="id of an existing task, e.g. 42"
    )
    parser.add_argument(
        "--annotations-file",
        type=Path,
        required=True,
        help="file to import, e.g. 'annotations.zip'",
    )
    parser.add_argument(
        "--import-format",
        default="COCO 1.0",
        help="importer name, e.g. 'COCO 1.0' (default: '%(default)s')",
    )
    return parser.parse_args()


def count_objects(task) -> int:
    """All annotation objects of a task: tags, shapes, and tracks."""
    annotations = task.get_annotations()
    return len(annotations.tags) + len(annotations.shapes) + len(annotations.tracks)


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        task = client.tasks.retrieve(args.task_id)
        print(f"Task {task.id}: {count_objects(task)} objects before import")

        formats, _ = client.api_client.server_api.retrieve_annotation_formats()
        names = [f.name for f in formats.importers]
        if args.import_format not in names:
            sys.exit(
                f"Unknown import format {args.import_format!r}. Choose one of: {', '.join(names)}"
            )

        task.import_annotations(args.import_format, args.annotations_file)
        print(f"Imported {args.annotations_file} as {args.import_format!r}")

        print(f"Task {task.id}: {count_objects(task)} objects after import")


if __name__ == "__main__":
    main()

# explicit bucket
python task_import_annotations_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --cloud-storage-id 7 --filename 'annotations/task_42.zip' \
    --import-format 'COCO 1.0'

# the bucket configured as the task's source storage, replacing what the task has
python task_import_annotations_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --filename 'predictions/task_42.zip' --import-mode replace

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Import annotations into an existing task straight from a registered cloud
storage: the server pulls the file out of the bucket itself, nothing is
uploaded from this machine. Handy when a model writes its predictions to a
bucket, or when the annotation archive is too big to push through your own
connection.

The high-level Task.import_annotations() always uploads a local file, so the
import request is made with the low-level API (client.api_client.tasks_api)
and awaited with client.wait_for_completion.

Steps:
  1. Retrieve the task and count the objects it already has.
  2. Fetch the server's import format list and validate --import-format.
  3. Resolve the storage to read from: --cloud-storage-id, or the task's own
     source storage when the flag is omitted.
  4. Start the import and wait for the background request to finish.
  5. Count the objects again to show what the import added.

Register a bucket first with cloud_storage_register.py to get the storage id.

Usage (run ``python task_import_annotations_from_cloud.py --help`` for the full list of options):
  python task_import_annotations_from_cloud.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --cloud-storage-id 7 --filename 'annotations/task_42.zip' \\
      --import-format 'COCO 1.0'
"""

import argparse
import json
import sys

from cvat_sdk import make_client
from cvat_sdk.core.exceptions import BackgroundRequestException
from cvat_sdk.core.proxies.types import Location


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--task-id", type=int, required=True, help="id of an existing task, e.g. 42"
    )
    parser.add_argument(
        "--filename",
        required=True,
        help="object key of the annotation file in the bucket, e.g. 'annotations/task_42.zip'",
    )
    parser.add_argument(
        "--cloud-storage-id",
        type=int,
        help="a registered cloud storage id (see cloud_storage_register.py); "
        "omit to use the source storage configured in the task",
    )
    parser.add_argument(
        "--import-format",
        default="COCO 1.0",
        help="importer name, e.g. 'COCO 1.0' (default: '%(default)s')",
    )
    parser.add_argument(
        "--import-mode",
        choices=["append", "replace"],
        default="append",
        help="add to the task's annotations or replace them; the default keeps what "
        "the task already has (default: '%(default)s')",
    )
    return parser.parse_args()


def count_objects(task) -> int:
    """All annotation objects of a task: tags, shapes, and tracks."""
    annotations = task.get_annotations()
    return len(annotations.tags) + len(annotations.shapes) + len(annotations.tracks)


def resolve_cloud_storage_id(task, requested_id: int | None) -> int:
    """The explicitly requested storage, or the one configured in the task."""
    if requested_id is not None:
        return requested_id

    storage = task.source_storage
    if not storage or storage.location.value != Location.CLOUD_STORAGE.value:
        sys.exit(f"Task {task.id} has no cloud source storage configured; pass --cloud-storage-id")
    return storage.cloud_storage_id


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        # 1. The state before the import, to compare against.
        task = client.tasks.retrieve(args.task_id)
        print(f"Task {task.id}: {count_objects(task)} objects before import")

        # 2. Validate the format against the server's list.
        formats, _ = client.api_client.server_api.retrieve_annotation_formats()
        names = [f.name for f in formats.importers]
        if args.import_format not in names:
            sys.exit(
                f"Unknown import format {args.import_format!r}. Choose one of: {', '.join(names)}"
            )

        # 3. Where to read from.
        cloud_storage_id = resolve_cloud_storage_id(task, args.cloud_storage_id)
        print(f"Reading {args.filename!r} from cloud storage {cloud_storage_id}")

        # 4. location=cloud_storage makes the server fetch the file itself; the
        # response only starts a background request, whose id is awaited below.
        _, response = client.api_client.tasks_api.create_annotations(
            task.id,
            format=args.import_format,
            filename=args.filename,
            location=Location.CLOUD_STORAGE,
            cloud_storage_id=cloud_storage_id,
            import_mode=args.import_mode,
        )
        rq_id = json.loads(response.data).get("rq_id") if response.data else None
        if not rq_id:
            sys.exit("The server did not return a request id (rq_id) for the import")

        try:
            client.wait_for_completion(rq_id, log_prefix=f"Task {task.id} annotation import")
        except BackgroundRequestException as error:
            sys.exit(
                f"Import of {args.filename!r} from cloud storage {cloud_storage_id} failed: {error}"
            )

        print(f"Imported {args.filename} as {args.import_format!r} ({args.import_mode})")

        # 5. Re-read the annotations, so the count is the server's state.
        print(f"Task {task.id}: {count_objects(task)} objects after import")


if __name__ == "__main__":
    main()

python task_edit_annotations.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --relabel 'car' 'vehicle'
python task_edit_annotations.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 42 --delete-label 'draft'

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Read a task's annotations, edit them, and write the edit back: move every
object from one label to another (--relabel) or delete every object with a
given label (--delete-label).

Steps:
  1. Retrieve the task and map its label names to ids.
  2. Read all annotations (tags, shapes, and tracks) and count objects per label.
  3. Write the edit back with a partial annotation update, so the objects that
     are not affected by it are not re-uploaded.
  4. Re-read the annotations and print the per-label object counts diff.

Usage (run ``python task_edit_annotations.py --help`` for the full list of options):
  python task_edit_annotations.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --relabel 'car' 'vehicle'
  python task_edit_annotations.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 42 --delete-label 'draft'
"""

import argparse
import sys
from collections import Counter

from cvat_sdk import make_client, models
from cvat_sdk.core.proxies.annotations import AnnotationUpdateAction


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--task-id", type=int, required=True, help="id of an existing task, e.g. 42"
    )
    action = parser.add_mutually_exclusive_group(required=True)
    action.add_argument(
        "--relabel",
        nargs=2,
        metavar=("FROM", "TO"),
        help="move all objects from label FROM to label TO",
    )
    action.add_argument("--delete-label", metavar="NAME", help="delete all objects with this label")
    return parser.parse_args()


def label_counts(annotations, label_names: dict[int, str]) -> Counter:
    """Objects per label name, over tags, shapes, and tracks alike."""
    return Counter(
        label_names[obj.label_id]
        for obj in [*annotations.tags, *annotations.shapes, *annotations.tracks]
    )


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        task = client.tasks.retrieve(args.task_id)
        label_ids = {label.name: label.id for label in task.get_labels()}
        label_names = {id: name for name, id in label_ids.items()}

        source = args.relabel[0] if args.relabel else args.delete_label
        affected = set(args.relabel) if args.relabel else {source}
        for name in affected:
            if name not in label_ids:
                sys.exit(f"Label {name!r} not found in task {task.id}")

        annotations = task.get_annotations()
        before = label_counts(annotations, label_names)

        tags = [tag for tag in annotations.tags if tag.label_id == label_ids[source]]
        shapes = [shape for shape in annotations.shapes if shape.label_id == label_ids[source]]
        tracks = [track for track in annotations.tracks if track.label_id == label_ids[source]]
        matched = len(tags) + len(shapes) + len(tracks)

        if args.relabel:
            target_id = label_ids[args.relabel[1]]
            task.update_annotations(
                models.PatchedLabeledDataRequest(
                    tags=[
                        models.LabeledImageRequest(**{**tag.to_dict(), "label_id": target_id})
                        for tag in tags
                    ],
                    shapes=[
                        models.LabeledShapeRequest(**{**shape.to_dict(), "label_id": target_id})
                        for shape in shapes
                    ],
                    tracks=[
                        models.LabeledTrackRequest(**{**track.to_dict(), "label_id": target_id})
                        for track in tracks
                    ],
                ),
                action=AnnotationUpdateAction.UPDATE,
            )
            print(f"Moved {matched} objects from {source!r} to {args.relabel[1]!r}")
        else:
            # An empty id list would make remove_annotations() drop *all* the task
            # annotations, so skip the request when the label has no objects.
            if matched:
                task.remove_annotations(ids=[obj.id for obj in [*tags, *shapes, *tracks]])
            print(f"Deleted {matched} objects with label {source!r}")

        after = label_counts(task.get_annotations(), label_names)
        for name in sorted(affected):
            print(f"  {name}: {before[name]} -> {after[name]}")


if __name__ == "__main__":
    main()

python project_annotation_stats.py --host 'https://app.cvat.ai' --token '<your token>' \
    --project-id 7

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Aggregate what was annotated across a project: object counts per label and
per object type for every task, printed and written to a CSV report.

Steps:
  1. Retrieve the project and its label names.
  2. Walk the project's tasks and read each task's annotations.
  3. Count objects per (label, type), where the type is a shape type such as
     'rectangle' or 'polygon', 'tag' for tags, or 'track' for tracks.
  4. Print a per-task breakdown with per-label project totals, and write
     the CSV report to --output.

Usage (run ``python project_annotation_stats.py --help`` for the full list of options):
  python project_annotation_stats.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 7 --output annotation_stats.csv
"""

import argparse
import csv
from collections import Counter
from pathlib import Path

from cvat_sdk import make_client


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--project-id", type=int, required=True, help="id of an existing project, e.g. 7"
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("annotation_stats.csv"),
        help="path to write the CSV report to (default: %(default)s)",
    )
    return parser.parse_args()


def task_counts(task, label_names: dict[int, str]) -> Counter:
    """Objects per (label name, object type) in one task."""
    annotations = task.get_annotations()
    counts = Counter()
    for tag in annotations.tags:
        counts[(label_names[tag.label_id], "tag")] += 1
    for shape in annotations.shapes:
        counts[(label_names[shape.label_id], str(shape.type))] += 1
    for track in annotations.tracks:
        counts[(label_names[track.label_id], "track")] += 1
    return counts


def main() -> None:
    args = parse_args()
    report_path = args.output
    with make_client(args.host, access_token=args.token) as client:
        project = client.projects.retrieve(args.project_id)
        label_names = {label.id: label.name for label in project.get_labels()}
        tasks = project.get_tasks()

        label_totals = Counter()
        total = 0
        with report_path.open("w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["task_id", "task_name", "label", "type", "count"])
            for task in tasks:
                counts = task_counts(task, label_names)
                print(f"Task {task.id} {task.name!r}: {sum(counts.values())} objects")
                for (label, type_), count in sorted(counts.items()):
                    print(f"  {label}/{type_}: {count}")
                    writer.writerow([task.id, task.name, label, type_, count])
                    label_totals[label] += count
                total += sum(counts.values())

        print(f"Project {project.id}: {total} objects across {len(tasks)} tasks")
        for label, count in sorted(label_totals.items()):
            print(f"  {label}: {count}")
        print(f"Wrote {report_path.resolve()}")


if __name__ == "__main__":
    main()

python project_find_duplicates.py --host 'https://app.cvat.ai' --token '<your token>' \
    --project-id 7

# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

"""Find objects annotated twice: objects on the same frame that have the same
label, the same shape type, and exactly the same coordinates.

Duplicates appear when an import runs twice, when two annotators' job ranges
overlap, or after a merge. Shapes whose coordinates differ are different
objects, so the comparison is an exact one and needs no similarity threshold.

The recipe only reports. It exits 1 when it finds a duplicate, so it can gate
an export pipeline; --no-fail turns that off.

Steps:
  1. Retrieve the project, its labels, and the tasks to inspect.
  2. For each task, read the annotations and the job that owns each frame.
  3. Group each frame's objects by (label, type, coordinates) and keep the
     groups with more than one member.
  4. Print the groups, write the CSV report, and set the exit code.

Usage (run ``python project_find_duplicates.py --help`` for the full list of options):
  python project_find_duplicates.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 7
"""

import argparse
import csv
import sys
from collections import defaultdict
from dataclasses import asdict, dataclass, fields
from itertools import groupby
from pathlib import Path

from cvat_sdk import make_client, models


@dataclass
class Annotated:
    """One annotated object, reduced to what the duplicate search compares."""

    frame: int
    label_id: int
    type: str
    shape_id: int | str
    track_id: int | str
    points: tuple[float, ...]
    elements: tuple[tuple[int, tuple[float, ...]], ...] = ()


@dataclass
class Duplicate:
    """One member of a duplicate group, as reported and written to the CSV."""

    task_id: int
    job_id: int | str
    frame: int
    group: int
    label: str
    type: str
    shape_id: int | str
    track_id: int | str
    points: str


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    parser.add_argument("--host", required=True, help="CVAT server URL, e.g. 'https://app.cvat.ai'")
    parser.add_argument(
        "--token",
        required=True,
        help="Personal Access Token (CVAT UI: Profile -> Security)",
    )
    parser.add_argument(
        "--project-id", type=int, required=True, help="id of an existing project, e.g. 7"
    )
    parser.add_argument(
        "--task-id",
        type=int,
        nargs="+",
        metavar="ID",
        help="inspect only these task ids (must belong to the project); "
        "omit to inspect every task in the project",
    )
    parser.add_argument(
        "--any-label",
        action="store_true",
        help="also group objects that carry different labels, e.g. the same car "
        "annotated once as 'car' and once as 'vehicle'",
    )
    parser.add_argument(
        "--output",
        type=Path,
        default=Path("duplicates.csv"),
        help="path to write the CSV report to (default: %(default)s)",
    )
    parser.add_argument(
        "--no-fail", action="store_true", help="always exit 0, even when duplicates were found"
    )
    return parser.parse_args()


def iter_objects(annotations):
    """The shapes and the track keyframes the recipe compares.

    Track keyframes and skeleton keypoints marked `outside` are skipped: they
    are intentionally out of view. Tags are skipped too - they have no
    coordinates to compare. Skeleton track frames need explicit keyframes for
    every element; this recipe does not interpolate missing keypoints.
    """
    for shape in annotations.shapes:
        elements = ()
        if str(shape.type) == "skeleton":
            elements = skeleton_coordinates(shape.elements)
            if not elements:
                continue
        yield Annotated(
            frame=shape.frame,
            label_id=shape.label_id,
            type=str(shape.type),
            shape_id=shape.id,
            track_id="",
            points=tuple(shape.points),
            elements=elements,
        )
    for track in annotations.tracks:
        for shape in track.shapes:
            if shape.outside:
                continue
            elements = ()
            if str(shape.type) == "skeleton":
                # Element tracks can have independent keyframes. Comparing a
                # partial pose would require interpolation, outside this recipe.
                keypoints = [
                    (element.label_id, keyframe)
                    for element in track.elements
                    for keyframe in element.shapes
                    if keyframe.frame == shape.frame
                ]
                if len(keypoints) != len(track.elements):
                    continue
                elements = tuple(
                    sorted(
                        (label_id, tuple(keyframe.points))
                        for label_id, keyframe in keypoints
                        if not keyframe.outside
                    )
                )
                if not elements:
                    continue
            yield Annotated(
                frame=shape.frame,
                label_id=track.label_id,
                type=str(shape.type),
                shape_id=shape.id,
                track_id=track.id,
                points=tuple(shape.points),
                elements=elements,
            )


def skeleton_coordinates(
    elements: list[models.SubLabeledShape],
) -> tuple[tuple[int, tuple[float, ...]], ...]:
    """Visible keypoints, keyed by label rather than their serialized order."""
    return tuple(
        sorted(
            (element.label_id, tuple(element.points)) for element in elements if not element.outside
        )
    )


def format_points(points: tuple[float, ...], limit: int = 8) -> str:
    """The coordinates as text, cut short: a mask's points are a whole RLE."""
    head = ",".join(f"{value:.2f}" for value in points[:limit])
    return f"{head},..." if len(points) > limit else head


def find_duplicates(task, label_names: dict[int, str], same_label: bool) -> list[Duplicate]:
    """The duplicate objects of one task, numbered by group.

    Two objects are duplicates when they sit on the same frame and share the
    shape type and the coordinates, so the objects can be bucketed by that key
    in one pass instead of compared pairwise.
    """
    job_of_frame = {}
    for job in task.get_jobs():
        for frame in range(job.start_frame, job.stop_frame + 1):
            job_of_frame.setdefault(frame, job.id)

    groups: dict[tuple, list[Annotated]] = defaultdict(list)
    for obj in iter_objects(task.get_annotations()):
        label_part = obj.label_id if same_label else None
        groups[(obj.frame, label_part, obj.type, obj.points, obj.elements)].append(obj)

    duplicates = []
    group_number = 0
    for key in sorted(groups, key=lambda key: key[0]):
        members = groups[key]
        if len(members) < 2:
            continue
        group_number += 1
        for obj in members:
            duplicates.append(
                Duplicate(
                    task_id=task.id,
                    job_id=job_of_frame.get(obj.frame, ""),
                    frame=obj.frame,
                    group=group_number,
                    label=label_names[obj.label_id],
                    type=obj.type,
                    shape_id=obj.shape_id,
                    track_id=obj.track_id,
                    points=format_points(obj.points),
                )
            )
    return duplicates


def select_tasks(client, project, task_ids: list[int] | None) -> list:
    """The project's tasks, or just the requested ones.

    With --task-id the tasks are retrieved by id: listing every task of a large
    project only to throw most of them away would be a waste.
    """
    if not task_ids:
        return list(project.get_tasks())

    tasks = []
    for task_id in task_ids:
        try:
            task = client.tasks.retrieve(task_id)
        except Exception:
            task = None
        if task is None or task.project_id != project.id:
            sys.exit(f"Task id {task_id} was not found in project {project.id}")
        tasks.append(task)
    return tasks


def main() -> None:
    args = parse_args()
    with make_client(args.host, access_token=args.token) as client:
        project = client.projects.retrieve(args.project_id)
        label_names = {label.id: label.name for label in project.get_labels()}

        tasks = select_tasks(client, project, args.task_id)
        if not tasks:
            sys.exit(f"Project {project.id} has no tasks to inspect")

        duplicates: list[Duplicate] = []
        for task in tasks:
            duplicates.extend(find_duplicates(task, label_names, not args.any_label))

    groups = 0
    for _, members in groupby(duplicates, key=lambda d: (d.task_id, d.group)):
        members = list(members)
        first = members[0]
        groups += 1
        print(
            f"duplicate group {first.group} in task {first.task_id}, frame {first.frame}: "
            f"{len(members)} objects at {first.points}"
        )
        for member in members:
            origin = (
                f"track {member.track_id}" if member.track_id != "" else f"shape {member.shape_id}"
            )
            print(f"  {member.label} {member.type} {origin}")

    with args.output.open("w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=[field.name for field in fields(Duplicate)])
        writer.writeheader()
        writer.writerows(asdict(duplicate) for duplicate in duplicates)

    print(f"Found {groups} duplicate group(s), {len(duplicates)} object(s)")
    print(f"Wrote {args.output.resolve()}")
    if duplicates and not args.no_fail:
        sys.exit(1)


if __name__ == "__main__":
    main()