File size: 6,173 Bytes
b7d2408 d032bfc b7d2408 03a45bc b7d2408 d032bfc b7d2408 03a45bc b7d2408 d032bfc |
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 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
"""CVAT API task methods."""
from __future__ import annotations
from typing import TYPE_CHECKING
from metrics_evaluation.schema.cvat import CvatApiTaskDetails, CvatApiTaskMediasMetainformation
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 TasksMethods, 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 TasksMethods:
"""Task-level operations for CVAT API."""
def __init__(self, client: "CvatApiClient"):
"""Initialize task methods with client reference.
Args:
client: Parent CvatApiClient instance
"""
self.client = client
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def list(self, project_id: int | None = None, token: str | None = None) -> list[CvatApiTaskDetails]:
"""List all tasks, optionally filtered by project.
Args:
project_id: Filter by project ID (optional)
token: Authentication token (optional)
Returns:
List of task details objects
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/tasks?page_size=1000"
if project_id is not None:
url += f"&project_id={project_id}"
response = self.client._make_request(
method="GET",
url=url,
headers=headers,
resource_name="tasks list",
)
response_data = response.json()
return [
CvatApiTaskDetails.model_validate(task)
for task in response_data.get("results", [])
]
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def get_task_details(
self, task_id: int, token: str | None = None
) -> CvatApiTaskDetails:
"""Fetch task details.
Args:
task_id: The ID of the task
token: Authentication token (optional)
Returns:
Task details object
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/tasks/{task_id}"
return self.client._make_request(
method="GET",
url=url,
headers=headers,
resource_name="task",
resource_id=task_id,
response_model=CvatApiTaskDetails,
)
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def get_task_media_metainformation(
self, task_id: int, token: str | None = None
) -> CvatApiTaskMediasMetainformation:
"""Fetch task media metadata.
Args:
task_id: The ID of the task
token: Authentication token (optional)
Returns:
Task media metadata object
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/tasks/{task_id}/data/meta"
return self.client._make_request(
method="GET",
url=url,
headers=headers,
resource_name="task media metadata",
resource_id=task_id,
response_model=CvatApiTaskMediasMetainformation,
)
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def get_task_job_ids(self, task_id: int, token: str | None = None) -> list[int]:
"""Fetch all job IDs for a task.
Args:
task_id: The ID of the task
token: Authentication token (optional)
Returns:
List of job IDs
"""
from metrics_evaluation.schema.cvat import CvatApiJobsListResponse
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/jobs?task_id={task_id}&page_size=1000"
response = self.client._make_request(
method="GET",
url=url,
headers=headers,
resource_name="task jobs",
resource_id=task_id,
response_model=CvatApiJobsListResponse,
)
return [job.id for job in response.results]
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def update_task(
self, task_id: int, task_data: CvatApiTaskDetails, token: str | None = None
) -> CvatApiTaskDetails:
"""Update task details (e.g., name, assignee, labels).
Args:
task_id: The ID of the task
task_data: Updated task details
token: Authentication token (optional)
Returns:
Updated task details object
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/tasks/{task_id}"
return self.client._make_request(
method="PATCH",
url=url,
headers=headers,
json_data=task_data.model_dump(exclude_unset=True),
resource_name="task",
resource_id=task_id,
response_model=CvatApiTaskDetails,
)
@retry_with_backoff(max_retries=3, initial_delay=1.0)
def get_frame(self, task_id: int, frame_number: int, token: str | None = None) -> bytes:
"""Download a single frame from a task.
Args:
task_id: The ID of the task
frame_number: The frame number to download
token: Authentication token (optional)
Returns:
Raw image bytes
"""
headers = self.client._get_headers(token)
url = f"{self.client.cvat_host}/api/tasks/{task_id}/data?type=frame&number={frame_number}&quality=original"
response = self.client._make_request(
method="GET",
url=url,
headers=headers,
resource_name="task frame",
resource_id=task_id,
)
return response.content
|