Getting Started with the API Client

First, import the Api class and create an instance:

from datamint import Api

api = Api()  # Uses API key from environment or config

The Api class provides access to different endpoint handlers:

Most day-to-day workflows can stay object-based. Endpoint handlers return entity objects such as Resource, Project, and Annotation, and those entities expose convenience methods for you to use.

Working with Resources

Upload resource files

Use api.resources.upload_resource() to upload any resource type, such as DICOMs, videos, and image files:

# Upload a single file
api.resources.upload_resource("/path/to/dicom.dcm")

# Upload multiple files at once
api.resources.upload_resources([
    "/path/to/dicom.dcm",
    "/path/to/video.mp4",
])

List and filter resources

You can see the list of all uploaded resources by calling api.resources.get_list():

# Get resources with different filters
inbox_resources = api.resources.get_list(status="inbox")
dicom_resources = api.resources.get_list(mimetype="application/dicom")
ct_resources = api.resources.get_list(channel="CT scans")

for resource in ct_resources:
    print(resource.filename, resource.status)

Upload with options

You can customize the upload with various parameters:

# Upload with channel organization
api.resources.upload_resource(
    "/path/to/dicom.dcm",
    channel="CT scans",
)

# Upload with anonymization and labels
api.resources.upload_resource(
    "/path/to/dicom.dcm",
    anonymize=True,
    tags=["baseline", "ct"],
)

# Upload and publish directly to a project
project = api.projects.get_by_name("Liver Review")
api.resources.upload_resource(
    "/path/to/dicom.dcm",
    publish_to=project,
)

Download resources

To download a resource, use api.resources.download_resource_file():

# Get a resource
resources = api.resources.get_list(status="inbox", mimetype="application/dicom")
resource = resources[0]

# Download as bytes through the entity helper
bytes_obj = resource.fetch_file_data(auto_convert=False)

# Auto-convert to the appropriate object (e.g., pydicom.Dataset)
dicom_obj = resource.fetch_file_data(auto_convert=True)

# Save directly to file
resource.fetch_file_data(save_path="path/to/dicomfile.dcm")

With auto_convert=True, the function uses the resource mimetype to automatically convert to the appropriate object type (pydicom.Dataset for DICOM, etc.).

Publishing resources

To publish a resource, use api.resources.publish_resources():

resources = api.resources.get_list(status="inbox")
resource = resources[0]  # assuming there is at least one resource in the inbox

# Change status from 'inbox' to 'published'
api.resources.publish_resources(resource)

# Add the published resource to a project
project = api.projects.get_by_name("Liver Review")
api.projects.add_resources(resource, project)

If you want the resource to land directly in a project, prefer upload_resource(..., publish_to=project) during upload.

Deleting resources

To delete a resource:

resource = api.resources.get_list(filename="temp_file.dcm")[0]
api.resources.delete(resource)

# Delete multiple resources at once
api.resources.bulk_delete(resources_to_delete)

Ranking unlabeled resources

When deciding which unlabeled resources to send for annotation next, use api.resources.rank_resources() to order them by any scoring function you provide:

unlabeled = api.resources.get_not_annotated(limit=200)

ranked = api.resources.rank_resources(unlabeled, my_score_fn, top_k=20)
for resource, score in ranked:
    print(resource.filename, score)

rank_resources sorts highest score first by default (descending=True) and skips any resource for which my_score_fn returns None. Pass top_k to keep only the highest-ranked resources.

A common scoring function is model uncertainty – see Command-line tools and datamint.utils.uncertainty for how to compute it.

Working with Annotations

Inspect annotations from a resource

Every Resource can fetch its own annotations:

resource = api.resources.get_list(project_name="Liver Review")[0]
annotations = resource.fetch_annotations()

for annotation in annotations:
    print(annotation.name, annotation.annotation_type)

Upload segmentations

To upload a segmentation, use api.annotations.upload_segmentations():

resource = api.resources.get_list(filename="dicom.dcm")[0]

# Upload segmentation
api.annotations.upload_segmentations(
    resource,
    "path/to/segmentation.png",
    name="SegmentationName",
)

Multi-class segmentations

If your segmentation has multiple classes, you can pass a dictionary mapping pixel values to class names:

class_names = {
    # Background (0) is automatic, don't specify it
    1: "tumor",
    2: "vessel",
}

api.annotations.upload_segmentations(
    resource,
    "path/to/segmentation.png",
    name=class_names,
)

Volume segmentations

Use api.annotations.upload_volume_segmentation() for NIfTI masks and other 3D segmentations:

volume_resource = api.resources.get_list(filename="volume.nii.gz")[0]

api.annotations.upload_volume_segmentation(
    volume_resource,
    "path/to/segmentation.nii.gz",
    {1: "liver", 2: "tumor"},
)

Upload geometry annotations (bounding boxes, lines)

from datamint.entities.annotations import BoxAnnotation, LineAnnotation, CoordinateSystem

# Upload a bounding box
api.annotations.upload_segmentations(
    resource,
    "path/to/box.json",
    name="tumor_box",
    annotation_type="box",
    coordinate_system=CoordinateSystem.PIXEL,
)

Upload classification annotations

# Upload image classification labels
api.annotations.upload_segmentations(
    resource,
    labels=["normal", "pathology"],
    name="diagnosis",
    annotation_type="category",
)

Inspect annotation entities

Annotation entities can fetch their own files and lazily resolve the source resource:

resource = api.resources.get_list(project_name="Liver Review")[0]
annotation = resource.fetch_annotations(annotation_type="segmentation")[0]

mask = annotation.fetch_file_data(use_cache=True)
source_resource = annotation.resource

print(annotation.name, source_resource.filename)

Measuring inter-annotator agreement

When a worklist assigns 2+ annotators to the same resources, use compute_agreement() to quantify how well they agree, and flag resources that need adjudication:

from datamint.utils.annotation_agreement import compute_agreement

# Fetch annotations for a worklist, filtered to a single annotation type
annotations = api.annotations.get_list(
    worklist_id=worklist.id,
    annotation_type="segmentation",
)

result = compute_agreement(annotations, threshold=0.7)

print(result.overall)            # summary agreement score
print(result.per_resource_mean)   # mean score per (resource_id, identifier)
print(result.flagged)             # resources below the threshold

The metric is picked automatically based on the annotation type: Dice for segmentations, IoU for bounding boxes, and Cohen’s/Fleiss’ kappa for category/label annotations. Pass metric="dice" (or "iou", "cohen_kappa", "fleiss_kappa") to override the automatic choice.

With 3+ annotators, Fleiss’ kappa requires a consistent count of raters per item (not the same rater identities every time, so a pool of 5 annotators rotating in groups of 3 per resource works fine). If rater counts vary across items, the most common count is used for overall and items with a different count are excluded from it, though they still appear in per_pair/per_resource_mean (raw pairwise agreement, useful for flagging) marked with used_in_overall=False. This means a resource can show up as low-agreement in the table even when overall looks high.

Inter-annotator agreement metrics for Datamint annotations.

Computes how well multiple annotators agree on the same resources, using a metric appropriate to the annotation type: Dice for segmentations, IoU for boxes, Cohen’s/Fleiss’ kappa for category/label annotations.

class datamint.utils.annotation_agreement.AgreementResult(per_pair, per_resource_mean, overall, flagged)

Bases: object

Result of compute_agreement().

per_pair: One row per compared annotator pair per item. For cohen_kappa/fleiss_kappa, includes a used_in_overall column. For fleiss_kappa, an item is used only if its rater count matches the most common rater count across all items (Fleiss’ kappa requires a consistent count, not the same rater identities every time), so a low-scoring row with used_in_overall=False may not be reflected in overall.

per_resource_mean: Mean score per (resource_id, identifier).

overall: Summary agreement score for the whole input (mean of per-resource means for Dice/IoU; a single kappa value computed over all items for cohen_kappa/fleiss_kappa).

flagged: Rows of per_resource_mean below the requested threshold. Empty (but present) when no threshold was given.

Parameters:

per_pair (DataFrame)
per_resource_mean (DataFrame)
overall (float)
flagged (DataFrame)

flagged: DataFrame

overall: float

per_pair: DataFrame

per_resource_mean: DataFrame

datamint.utils.annotation_agreement.cohen_kappa(labels_a, labels_b)

Cohen’s kappa between two annotators’ labels over the same items.

Parameters:

labels_a (Sequence[str])
labels_b (Sequence[str])

Return type:

float

datamint.utils.annotation_agreement.compute_agreement(annotations, metric='auto', threshold=None)

Compute inter-annotator agreement over a set of annotations.

Groups annotations by (resource_id, identifier, frame_index), then compares every pair of annotators on each group with a metric appropriate to the annotation type.

Parameters:

annotations (Sequence[Annotation]) – Annotations to compare, typically fetched with api.annotations.get_list(worklist_id=..., annotation_type=...). All annotations must share the same “kind” (all segmentations, all boxes, or all category/label) unless metric is given explicitly, since a fair comparison metric can’t be picked automatically across mixed types.
metric (Literal['auto', 'dice', 'iou', 'cohen_kappa', 'fleiss_kappa']) – 'auto' picks Dice for segmentations, IoU for boxes, and Cohen’s kappa (2 annotators) or Fleiss’ kappa (3+) for category/label annotations. Pass one explicitly to override.
threshold (float | None) – Optional score cutoff. Rows of per_resource_mean below it are returned in AgreementResult.flagged for adjudication.

Return type:

AgreementResult

Returns:

AgreementResult with per-pair scores, per-resource means, an overall summary score, and any flagged low-agreement resources.

datamint.utils.annotation_agreement.dice_coefficient(mask_a, mask_b)

Dice similarity coefficient between two binary masks of identical shape.

Parameters:

mask_a (ndarray)
mask_b (ndarray)

Return type:

float

datamint.utils.annotation_agreement.fleiss_kappa(ratings)

Fleiss’ kappa across 3+ annotators.

Parameters:: ratings (Sequence[Sequence[str]]) – One entry per item, each a sequence of category labels (one per annotator). Every item must have the same number of annotators.
Return type:: float

datamint.utils.annotation_agreement.iou_boxes(box_a, box_b)

IoU between the axis-aligned bounding rectangles of two box geometries.

Both boxes must use the same coordinate_system.

Parameters:

box_a (BoxGeometry)
box_b (BoxGeometry)

Return type:

float

Working with Projects

Create and manage projects

# Create a new project
project = api.projects.create(
    name="My Project",
    description="Project description",
)

# Add existing resources to it
resources = api.resources.get_list(channel="CT scans")
api.projects.add_resources(resources, project)

# Work with project resources through the entity
for resource in project.fetch_resources():
    print(resource.filename)

Project helper methods

The Project entity provides shortcuts for common project workflows:

project = api.projects.get_by_name("My Project")

# Cache all resource files locally for faster follow-up access
project.cache_resources()

resource = project.fetch_resources()[0]
project.set_work_status(resource, "annotated")

# Pin the metrics that matter most for this project (replaces the full list)
project.set_pinned_metrics(["val/accuracy", "val/f1"])

specs = project.get_annotations_specs()
print([spec.identifier for spec in specs])

Project-scoped dataset splits

The project split endpoints return ProjectResourceSplit records, which contain:

Field	Description
`split_name`	Logical split name such as `train`, `val`, or `test`.
`project_id`	Project that owns the assignment.
`resource_id`	Resource assigned within that project.
`created_at` / `created_by`	Audit metadata present when an assignment has been created.
`deleted_at` / `deleted_by`	Audit metadata present when an assignment has been deleted.

Use api.projects.assign_splits() to write assignments, api.projects.get_splits() to list them, and api.projects.get_resource_split() to inspect one resource within a project:

from datamint import Api

api = Api()
project = api.projects.get_by_name("FracAtlas")
resources = list(project.fetch_resources())

train_resources = resources[:100]
val_resources = resources[100:120]

# Note: assign_splits(resources, split_name, project) — project is the third argument
api.projects.assign_splits(train_resources, "train", project)
api.projects.assign_splits(val_resources, "val", project)

assignments = api.projects.get_splits(project)
train_assignments = api.projects.get_splits(project, split_name="train")
first_resource_assignment = api.projects.get_resource_split(project, resources[0])

For project-backed datasets, split() now prefers project-scoped assignments automatically when you do not pass ratio kwargs:

from datamint.dataset import ImageDataset

dataset = ImageDataset(project=project, include_unannotated=True)

parts = dataset.split()
snapshot = parts["train"].split_as_of_timestamp

# Reuse the exact assignment snapshot later.
replayed_parts = dataset.split(as_of_timestamp=snapshot)

Each returned subset records split_name, split_source, and split_as_of_timestamp for reproducibility. Local ratio splits remain available with calls such as dataset.split(train=0.8, val=0.2, seed=42). Legacy split:* tag-based splitting is still supported for backwards compatibility, but it is deprecated in favor of project-scoped splits.

Working with Channels

Organize resources with channels

# List all channels
channels = api.channels.get_list()

# Create a new channel
api.channels.create(name="CT Scans", description="CT scan images")

# List channels with resources
for channel in channels:
    print(channel.name, channel.resource_count)

# Delete a channel
api.channels.delete(channel)

See also the tutorial notebooks: upload_data.ipynb

Working with Models

api.models is a thin facade over Datamint’s MLflow-backed model registry: it wraps MLflow’s RegisteredModel/ModelVersion objects in Model / ModelVersion, so you can register, list, and inspect models without knowing MLflow’s object model.

Register and list models

# Create a model (or fetch it if it already exists, the default behavior)
model = api.models.create("my-model", description="Segmentation model")

# Look up a model by name; returns None if it doesn't exist
model = api.models.get_by_name("my-model")

# List every registered model
all_models = api.models.get_list()

# Only models with a deployed image
deployed_models = api.models.get_list(only_deployed=True)

Models are also created automatically when you pass --ai-model to Command-line tools (datamint-upload) with a name that doesn’t exist yet.

Inspect versions and metrics

Each Model can list its ModelVersion objects, and each version exposes what it was trained for and how it performed:

model = api.models.get_by_name("my-model")

versions = model.get_versions()
latest = model.get_latest_version()          # highest version number
champion = model.get_latest_version(alias="champion")

print(latest.get_task_type())                 # e.g. "segmentation"
print(latest.get_supported_modes())            # e.g. ["auto", "interactive"]
print(latest.get_metrics())                    # e.g. {"val/dice": 0.87}

get_metrics() returns {} for versions with no training run behind them (for example, a model registered externally rather than trained through a Datamint trainers), rather than raising. Model.get_supported_modes()/get_metrics() are shortcuts that delegate to the latest version when you don’t need a specific one.

Deploy a registered model

Use api.deploy.start() to deploy a model:

# Deploy a registered model
deploy_job = api.deploy.start(
    model_name="my-model",
    model_alias="latest",
)
print(deploy_job.status)

# Wait for deployment to complete
deploy_job = deploy_job.wait()
print("Deployment complete:", deploy_job.status)

# Check whether a model has a deployed image
model.is_deployed()

Working with Users

User management operations:

# List all users
users = api.users.get_all()

# Get user by email (email serves as the entity ID)
user = api.users.get_by_email("user@example.com")