metrics_evaluation/cvat_api/tasks.py

"""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