File size: 4,281 Bytes
b7d2408 03a45bc b7d2408 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 |
"""CVAT API annotation methods."""
from typing import TYPE_CHECKING
from metrics_evaluation.schema.cvat import CvatApiJobList, CvatApiTaskList
from .retry import retry_with_backoff
# TYPE_CHECKING is False at runtime but True during static type checking.
# This allows us to import CvatApiClient for type hints without creating a circular
# import (client.py imports AnnotationsMethods, and we need CvatApiClient for type hints).
# Benefits:
# - Avoids circular import errors at runtime
# - Provides proper type checking during development
# - No performance overhead (import only happens during type checking, not at runtime)
# This is a recommended pattern in modern Python (PEP 484, PEP 563).
# -- Claude Code
if TYPE_CHECKING:
from .client import CvatApiClient
class AnnotationsMethods:
"""Annotation GET/PUT operations for CVAT API."""
def __init__(self, client: "CvatApiClient"):
"""Initialize annotation methods with client reference.
Args:
client: Parent CvatApiClient instance
"""
self.client = client
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def get_job_annotations(
self, job_id: int, token: str | None = None
) -> CvatApiJobList:
"""Fetch annotations for a job.
Args:
job_id: The ID of the job
token: Authentication token (optional)
Returns:
Job annotations object with tags, shapes, and tracks
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/jobs/{job_id}/annotations"
return self.client._make_request(
method="GET",
url=url,
headers=headers,
resource_name="job annotations",
resource_id=job_id,
response_model=CvatApiJobList,
)
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def get_task_annotations(
self, task_id: int, token: str | None = None
) -> CvatApiTaskList:
"""Fetch annotations for a task.
Args:
task_id: The ID of the task
token: Authentication token (optional)
Returns:
Task annotations object with tags, shapes, and tracks
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/tasks/{task_id}/annotations"
return self.client._make_request(
method="GET",
url=url,
headers=headers,
resource_name="task annotations",
resource_id=task_id,
response_model=CvatApiTaskList,
)
def put_job_annotations(
self, job_id: int, annotations: CvatApiJobList, token: str | None = None
) -> CvatApiJobList:
"""Update annotations for a job.
Args:
job_id: The ID of the job
annotations: Job annotations to upload
token: Authentication token (optional)
Returns:
Updated job annotations
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/jobs/{job_id}/annotations"
return self.client._make_request(
method="PUT",
url=url,
headers=headers,
json_data=annotations.model_dump(),
resource_name="job annotations",
resource_id=job_id,
response_model=CvatApiJobList,
)
def put_task_annotations(
self, task_id: int, annotations: CvatApiTaskList, token: str | None = None
) -> CvatApiTaskList:
"""Update annotations for a task.
Args:
task_id: The ID of the task
annotations: Task annotations to upload
token: Authentication token (optional)
Returns:
Updated task annotations
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/tasks/{task_id}/annotations"
return self.client._make_request(
method="PUT",
url=url,
headers=headers,
json_data=annotations.model_dump(),
resource_name="task annotations",
resource_id=task_id,
response_model=CvatApiTaskList,
)
|