metrics_evaluation/cvat_api/USAGE_GUIDE.md

# CVAT API Usage Guide

Complete guide for working with CVAT tasks, jobs, and annotations.

## Table of Contents

- [Installation](#installation)
- [Initialization](#initialization)
- [Common Workflows](#common-workflows)
  - [Download and Update Job Annotations](#download-and-update-job-annotations)
  - [Download and Update Task Annotations](#download-and-update-task-annotations)
  - [Update Job Metadata](#update-job-metadata)
  - [Update Task Metadata](#update-task-metadata)
- [Complete Examples](#complete-examples)

## Installation

```bash
pip install -r requirements.txt
```

## Initialization

```python
from cvat_api import CvatApiClient
from pathlib import Path

# Initialize client
client = CvatApiClient(
    cvat_host="https://app.cvat.ai",
    cvat_username="your_username",
    cvat_password="your_password",
    cvat_organization="your_organization"
)
```

## Common Workflows

### Download and Update Job Annotations

Complete workflow for downloading annotations, modifying them, and uploading back:

```python
# 1. Download annotations
job_id = 2355063
annotations = client.annotations.get_job_annotations(job_id)

# 2. Modify annotations
# Add a new shape
from shared.schema.cvat import CvatApiShape

new_shape = CvatApiShape(
    id=0,  # Will be assigned by server
    type="rectangle",
    frame=0,
    label_id=1,
    points=[100.0, 100.0, 200.0, 200.0],
    occluded=False,
    z_order=0,
    source="manual",
    attributes=[]
)
annotations.shapes.append(new_shape)

# Or modify existing shapes
for shape in annotations.shapes:
    if shape.label_id == 5:
        shape.occluded = True  # Mark as occluded

# 3. Upload modified annotations
updated_annotations = client.annotations.put_job_annotations(
    job_id=job_id,
    annotations=annotations
)

print(f"✅ Updated {len(updated_annotations.shapes)} shapes")
```

### Download and Update Task Annotations

Same workflow but at task level:

```python
# 1. Download task annotations
task_id = 1322143
annotations = client.annotations.get_task_annotations(task_id)

# 2. Modify annotations (same as above)
# ... make your changes ...

# 3. Upload modified annotations
updated_annotations = client.annotations.put_task_annotations(
    task_id=task_id,
    annotations=annotations
)
```

### Update Job Metadata

Update job status, assignee, stage, etc.:

```python
# 1. Get current job details
job_id = 2355063
job_details = client.jobs.get_job_details(job_id)

# 2. Modify job metadata
job_details.state = "completed"  # Change state
job_details.stage = "acceptance"  # Change stage

# 3. Update job
updated_job = client.jobs.update_job(
    job_id=job_id,
    job_data=job_details
)

print(f"✅ Job {job_id} updated: {updated_job.state} / {updated_job.stage}")
```

### Update Task Metadata

Update task name, assignee, labels, etc.:

```python
# 1. Get current task details
task_id = 1322143
task_details = client.tasks.get_task_details(task_id)

# 2. Modify task metadata
task_details.name = "Updated Task Name"
# Note: Use model_dump(exclude_unset=True) in update method to only send changed fields

# 3. Update task
updated_task = client.tasks.update_task(
    task_id=task_id,
    task_data=task_details
)

print(f"✅ Task {task_id} updated: {updated_task.name}")
```

## Complete Examples

### Example 1: Batch Process Job Annotations

Process annotations for all jobs in a task:

```python
from cvat_api import CvatApiClient

# Initialize
client = CvatApiClient(...)

# Get all jobs in task
task_id = 1322143
job_ids = client.tasks.get_task_job_ids(task_id)

# Process each job
for job_id in job_ids:
    # Download annotations
    annotations = client.annotations.get_job_annotations(job_id)
    
    # Apply your modifications
    for shape in annotations.shapes:
        if shape.label_id == 5:  # Example: filter by label
            # Modify shape properties
            shape.occluded = True
    
    # Upload modified annotations
    client.annotations.put_job_annotations(job_id, annotations)
    print(f"✅ Processed job {job_id}")
```

### Example 2: Download All Images and Annotations

Download complete dataset from a task:

```python
from pathlib import Path

task_id = 1322143
output_dir = Path("./output")

# 1. Download all images
images = client.downloads.download_task_images(
    task_id=task_id,
    output_dir=output_dir / "images"
)
print(f"Downloaded {len(images)} images")

# 2. Download annotations
annotations = client.annotations.get_task_annotations(task_id)

# 3. Save annotations to file
import json
output_file = output_dir / "annotations.json"
output_file.write_text(annotations.model_dump_json(indent=2))
print(f"Saved annotations to {output_file}")
```

### Example 3: Quality Control Workflow

Implement a QC workflow that marks reviewed jobs:

```python
from shared.schema.cvat import CvatApiJobsListRequest

# List all jobs in "annotation" stage
request = CvatApiJobsListRequest(
    task_id=1322143,
    stage="annotation",
    state="completed"
)
response = client.jobs.list_jobs(request)

for job in response.results:
    # Download and review annotations
    annotations = client.annotations.get_job_annotations(job.id)
    
    # Run your QC checks
    issues_found = run_quality_checks(annotations)
    
    # Update job based on QC results
    job_details = client.jobs.get_job_details(job.id)
    
    if not issues_found:
        job_details.stage = "acceptance"
        job_details.state = "completed"
    else:
        job_details.state = "rejected"
    
    client.jobs.update_job(job.id, job_details)
    print(f"✅ QC processed job {job.id}")
```

### Example 4: Export Annotations with Metadata

Export complete annotation dataset with metadata:

```python
# Get project details
project_id = 239220
project = client.projects.get_project_details(project_id)
labels = client.projects.get_project_labels(project_id)

# Get all tasks
task_ids = client.projects.get_project_tasks(project_id)

# Build export structure
export_data = {
    "project": project.model_dump(),
    "labels": [label.model_dump() for label in labels],
    "tasks": []
}

# Export each task
for task_id in task_ids:
    task_details = client.tasks.get_task_details(task_id)
    annotations = client.annotations.get_task_annotations(task_id)
    
    export_data["tasks"].append({
        "task": task_details.model_dump(),
        "annotations": annotations.model_dump()
    })

# Save export
import json
Path("export.json").write_text(json.dumps(export_data, indent=2))
print(f"✅ Exported {len(task_ids)} tasks")
```

## Tips and Best Practices

### 1. Use Partial Updates

When updating jobs or tasks, use only the fields you want to change:

```python
# Good - only update specific fields
job_details = client.jobs.get_job_details(job_id)
job_details.state = "completed"
client.jobs.update_job(job_id, job_details)
```

### 2. Handle Errors

Always handle potential errors:

```python
try:
    annotations = client.annotations.get_job_annotations(job_id)
except requests.HTTPError as e:
    print(f"Failed to get annotations: {e}")
    # Handle error appropriately
```

### 3. Batch Operations

For multiple jobs/tasks, use batch operations:

```python
job_ids = client.tasks.get_task_job_ids(task_id)
for job_id in job_ids:
    try:
        # Process each job
        process_job(client, job_id)
    except Exception as e:
        print(f"Error processing job {job_id}: {e}")
        continue  # Continue with next job
```

### 4. Validate Before Upload

Validate annotations before uploading:

```python
annotations = client.annotations.get_job_annotations(job_id)

# Validate
for shape in annotations.shapes:
    assert shape.label_id in valid_label_ids
    assert len(shape.points) >= 4  # For rectangles

# Upload after validation
client.annotations.put_job_annotations(job_id, annotations)
```

## Error Handling

The client includes automatic retry logic for transient errors (500/502/503/504):

```python
# This will automatically retry up to 10 times for server errors
annotations = client.annotations.get_job_annotations(job_id)
```

For custom error handling:

```python
import requests
from requests.exceptions import Timeout, ConnectionError

try:
    annotations = client.annotations.get_job_annotations(job_id)
except Timeout:
    print("Request timed out")
except ConnectionError:
    print("Connection failed")
except requests.HTTPError as e:
    if e.response.status_code == 404:
        print("Job not found")
    elif e.response.status_code == 403:
        print("Access denied")
    else:
        print(f"HTTP error: {e}")
```

## See Also

- [API Documentation](./client.py) - Complete API reference
- [Test Suite](../tests/test_cvat_api/) - Usage examples in tests
- [CVAT API Docs](https://app.cvat.ai/api/docs/) - Official CVAT API documentation