Dataset recipes

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

with make_client("https://app.cvat.ai", access_token="<your token>") as client:
    task = client.tasks.retrieve(10)
    task.export_dataset("COCO 1.0", "task_10.zip", include_images=False, location=Location.LOCAL)

    project = client.projects.retrieve(7)
    project.export_dataset("COCO 1.0", "project_7.zip", include_images=False, location=Location.LOCAL)

from cvat_sdk.datasets import TaskDataset, UpdatePolicy

dataset = TaskDataset(client, 10, update_policy=UpdatePolicy.IF_MISSING_OR_STALE)
for sample in dataset.samples:
    image = sample.media.load_image()   # PIL.Image, from the cache
    shapes = sample.annotations.shapes

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

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

"""Download a project's task data incrementally: the SDK keeps a local cache and
re-downloads only what the server has changed since the last run.

``cvat_sdk.datasets.TaskDataset`` stores each task under the client's cache
directory. On every construction it compares the cached copy's ``updated_date``
with the server's: an unchanged task is served entirely from disk, a changed one
is fetched again. Media chunks already on disk are never downloaded twice.

Two things worth knowing before building a pipeline on this:

* Staleness is tracked per task, not per frame. Any change to a task - an
  annotation edit included - invalidates that task's whole cache entry, so its
  media comes down again.
* Every run still asks the server for the task's metadata and labels; that is
  how it notices a change. Only the bulky parts - chunks and annotations - are
  skipped when the cache is fresh.

Steps:
  1. Resolve the tasks to download, from --project-id or from --task-id.
  2. Build a TaskDataset for each one, which fills or reuses the cache.
  3. Report the samples found and how much the cache grew.

Usage (run ``python dataset_incremental_download.py --help`` for the full list of options):
  python dataset_incremental_download.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --project-id 7 --cache-dir ./cvat-cache
  # run the same command again: the cache does not grow and no media is fetched
  python dataset_incremental_download.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 10 11 --cache-dir ./cvat-cache --offline
"""

import argparse
import logging
import sys
from pathlib import Path

from cvat_sdk import Client, make_client
from cvat_sdk.core.client import AccessTokenCredentials
from cvat_sdk.datasets import TaskDataset, UnsupportedDatasetError, UpdatePolicy


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)",
    )
    selection = parser.add_mutually_exclusive_group(required=True)
    selection.add_argument(
        "--project-id", type=int, help="download every task of this project, e.g. 7"
    )
    selection.add_argument(
        "--task-id", type=int, nargs="+", metavar="ID", help="download these task ids"
    )
    parser.add_argument(
        "--cache-dir",
        type=Path,
        help="where to keep the downloaded data (default: the SDK's per-user cache directory)",
    )
    parser.add_argument(
        "--offline",
        action="store_true",
        help="read the cache without contacting the server; fails on anything not cached",
    )
    parser.add_argument(
        "--quiet", action="store_true", help="hide the SDK's per-file cache and download messages"
    )
    return parser.parse_args()


def connect(args: argparse.Namespace) -> Client:
    """The client to work through.

    An --offline run must make no requests at all, so it skips the server
    version handshake that Client performs on construction. Applying an access
    token needs no round trip either, which is what makes this possible.
    """
    if not args.offline:
        return make_client(args.host, access_token=args.token)

    client = Client(url=args.host, check_server_version=False)
    client.login(AccessTokenCredentials(args.token))
    return client


def cache_size(path: Path) -> int:
    """Bytes currently cached under path, which need not exist yet."""
    return sum(item.stat().st_size for item in path.rglob("*") if item.is_file())


def main() -> None:
    args = parse_args()
    if args.offline and not args.task_id:
        sys.exit("--offline needs --task-id: listing a project's tasks is itself a server call")

    logging.basicConfig(level=logging.WARNING if args.quiet else logging.INFO, format="%(message)s")

    with connect(args) as client:
        if args.cache_dir:
            client.config.cache_dir = args.cache_dir
        cache_dir = client.config.cache_dir
        print(f"Cache: {cache_dir}")
        size_before = cache_size(cache_dir)

        if args.task_id:
            task_ids = args.task_id
        else:
            task_ids = [task.id for task in client.tasks.list(project_id=args.project_id)]
            if not task_ids:
                sys.exit(f"Project {args.project_id} has no tasks to download")

        policy = UpdatePolicy.NEVER if args.offline else UpdatePolicy.IF_MISSING_OR_STALE
        downloaded = 0
        for task_id in task_ids:
            try:
                dataset = TaskDataset(client, task_id, update_policy=policy)
            except UnsupportedDatasetError as error:
                # A video task, or a task with no data. One such task must not
                # stop the rest of the selection from being downloaded.
                print(f"Skipped task {task_id}: {error}")
                continue
            except FileNotFoundError:
                print(f"Skipped task {task_id}: not in the cache, run without --offline first")
                continue
            downloaded += 1
            print(
                f"Task {task_id}: {len(dataset.samples)} sample(s), {len(dataset.labels)} label(s)"
            )

        grew = cache_size(cache_dir) - size_before
        print(f"{downloaded} of {len(task_ids)} task(s) available locally; cache grew by {grew} B")

    if not downloaded:
        sys.exit(1)


if __name__ == "__main__":
    main()

python dataset_bulk_export.py --host 'https://app.cvat.ai' --token '<your token>' \
    --task-id 10 11 12 --output-dir datasets

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

"""Export many task datasets in one run: name the tasks by id, optionally
narrowed by status, and write them locally and/or straight to a cloud storage.

A failing task does not stop the run - its error is printed and the script
exits with code 1 at the end, so a pipeline still sees the failure.

Steps:
  1. Check --cloud-storage-id once, so a wrong id fails before any export, and
     resolve the selection (--task-id, --status). Check that every selected
     task belongs to the cloud storage's workspace before exporting anything.
  2. Export each task to --output-dir and/or to --cloud-storage-id, skipping the
     ones already exported when --skip-existing is passed.
  3. Report how many tasks were exported, skipped, and failed.

Usage (run ``python dataset_bulk_export.py --help`` for the full list of options):
  python dataset_bulk_export.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 10 11 12 --output-dir datasets
  python dataset_bulk_export.py --host 'https://app.cvat.ai' --token '<your token>' \\
      --task-id 10 11 12 --cloud-storage-id 3 --output-dir datasets
"""

import argparse
import sys
from pathlib import Path

from cvat_sdk import make_client, models
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,
        nargs="+",
        metavar="ID",
        required=True,
        help="export these task ids",
    )
    parser.add_argument(
        "--status",
        choices=["annotation", "validation", "completed"],
        help="export only the tasks in this status",
    )
    parser.add_argument(
        "--output-dir", type=Path, help="directory to write the exported datasets to"
    )
    parser.add_argument(
        "--cloud-storage-id",
        type=int,
        help="also export straight to this registered cloud storage, checked before "
        "the run starts (see cloud_storage_register.py)",
    )
    parser.add_argument(
        "--export-format",
        default="COCO 1.0",
        help="exporter name, e.g. 'COCO 1.0' (default: '%(default)s')",
    )
    parser.add_argument(
        "--skip-existing",
        action="store_true",
        help="skip tasks whose output file is already in --output-dir (resume a run); "
        "local exports only",
    )
    parser.add_argument("--with-images", action="store_true", help="include images in the exports")
    return parser.parse_args()


def select_tasks(
    client, args: argparse.Namespace, storage: models.CloudStorageRead | None = None
) -> list:
    """The tasks to export, as (id, name) pairs.

    --status is applied after the ids are resolved, so a task that exists but is
    in another status is reported as filtered out rather than as missing - two
    different mistakes.
    """
    selected = []
    for task_id in args.task_id:
        try:
            task = client.tasks.retrieve(task_id)
        except Exception:
            selected.append((task_id, ""))
            continue
        if args.status and str(task.status) != args.status:
            print(f"Skipping task {task_id}: status is {task.status}, not {args.status}")
            continue
        if storage is not None and task.organization_id != storage.organization:
            sys.exit(
                f"Task {task_id} and cloud storage {storage.id} belong to different workspaces"
            )
        selected.append((task.id, task.name))
    return selected


def export_one(client, args: argparse.Namespace, task_id: int, name: str) -> str:
    local_path = args.output_dir / f"task_{task_id}.zip" if args.output_dir else None

    if args.skip_existing and local_path and local_path.exists():
        print(f"Skipped task {task_id} ({local_path} exists)")
        return "skipped"

    try:
        task = client.tasks.retrieve(task_id)
        destinations = []

        if local_path:
            task.export_dataset(
                args.export_format,
                local_path,
                include_images=args.with_images,
                location=Location.LOCAL,
            )
            destinations.append("local")

        if args.cloud_storage_id:
            task.export_dataset(
                args.export_format,
                f"task_{task_id}.zip",
                include_images=args.with_images,
                location=Location.CLOUD_STORAGE,
                cloud_storage_id=args.cloud_storage_id,
            )
            destinations.append(f"cloud storage {args.cloud_storage_id}")

        print(f"Exported task {task_id} {name!r} -> {', '.join(destinations)}")
    except Exception as error:  # one bad task must not abort the whole run
        print(f"FAILED task {task_id} {name!r}: {type(error).__name__}: {error}")
        return "failed"

    return "exported"


def main() -> None:
    args = parse_args()
    if not args.output_dir and not args.cloud_storage_id:
        sys.exit("Select a destination: pass --output-dir and/or --cloud-storage-id")
    if args.skip_existing and not args.output_dir:
        sys.exit("--skip-existing needs --output-dir: a cloud export leaves nothing local to check")
    if args.skip_existing and args.cloud_storage_id:
        sys.exit("--skip-existing cannot resume a cloud export; drop it or drop --cloud-storage-id")
    if args.output_dir:
        args.output_dir.mkdir(parents=True, exist_ok=True)

    with make_client(args.host, access_token=args.token) as client:
        storage = None
        if args.cloud_storage_id:
            try:
                storage, _ = client.api_client.cloudstorages_api.retrieve(args.cloud_storage_id)
            except Exception as error:
                sys.exit(
                    f"Cloud storage {args.cloud_storage_id} is not available to this user: {error}"
                )
            print(f"Exporting to cloud storage {storage.id} {storage.display_name!r}")

        selection = select_tasks(client, args, storage)
        if not selection:
            sys.exit("The selection is empty; nothing to export")
        print(f"Exporting {len(selection)} task(s)")

        results = [export_one(client, args, *task) for task in selection]

    failed = results.count("failed")
    skipped = results.count("skipped")
    exported = results.count("exported")
    print(f"Exported {exported} of {len(results)} task(s); {skipped} skipped, {failed} failed")
    if failed:
        sys.exit(1)


if __name__ == "__main__":
    main()