项目文件夹

文件
2026-07-13 13:32:23 +08:00

4934 行
189 KiB
Python

# Copyright (C) 2019-2022 Intel Corporation
# Copyright (C) CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT
from __future__ import annotations
import mimetypes
import os
import re
import shutil
import string
import textwrap
import uuid
import warnings
from collections import OrderedDict
from collections.abc import Iterable, Sequence
from contextlib import closing
from copy import copy
from datetime import datetime
from inspect import isclass
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any, cast
from urllib.parse import urlparse
import django_rq
from django.conf import settings
from django.contrib.auth.models import Group, User
from django.core.files.uploadedfile import UploadedFile
from django.db import transaction
from django.db.models import Count, Prefetch, prefetch_related_objects
from django.utils import timezone
from django.utils.functional import cached_property
from drf_spectacular.utils import OpenApiExample, extend_schema_field, extend_schema_serializer
from numpy import random
from PIL import Image
from rest_framework import exceptions, serializers
from rest_framework.reverse import reverse
from cvat.apps.dataset_manager.formats.utils import get_label_color
from cvat.apps.engine import field_validation, models
from cvat.apps.engine.cloud_provider import (
Credentials,
Status,
db_storage_to_storage_instance,
get_cloud_storage_instance,
)
from cvat.apps.engine.log import ServerLogManager
from cvat.apps.engine.media_io.frame_provider import TaskFrameProvider
from cvat.apps.engine.permissions import ProjectPermission, TaskPermission
from cvat.apps.engine.rq import RunningBackgroundProcessesError, update_org_related_data_in_rq_jobs
from cvat.apps.engine.task import ensure_task_is_initialized
from cvat.apps.engine.task_validation import HoneypotFrameSelector
from cvat.apps.engine.types import ExtendedRequest
from cvat.apps.engine.utils import (
CvatChunkTimestampMismatchError,
av_scan_paths,
build_field_filter_params,
format_list,
get_list_view_name,
get_path_size,
grouped,
parse_specific_attributes,
take_by,
)
from cvat.apps.iam.permissions import get_iam_context
from cvat.apps.organizations.models import Organization
from cvat.apps.webhooks.models import Webhook
from cvat.utils import django_database as db_utils
from cvat.utils.paths import problem_with_untrusted_path
from utils.dataset_manifest import ImageManifestManager
slogger = ServerLogManager(__name__)
class CanonicalRelativePathValidator:
def __init__(self, *, allow_trailing_slash: bool = False) -> None:
self.allow_trailing_slash = allow_trailing_slash
def __call__(self, value: str) -> None:
if problem := problem_with_untrusted_path(
value, allow_trailing_slash=self.allow_trailing_slash
):
raise serializers.ValidationError(problem)
class WriteOnceMixin:
"""
Adds support for write once fields to serializers.
To use it, specify a list of fields as `write_once_fields` on the
serializer's Meta:
```
class Meta:
model = SomeModel
fields = '__all__'
write_once_fields = ('collection', )
```
Now the fields in `write_once_fields` can be set during POST (create),
but cannot be changed afterwards via PUT or PATCH (update).
Inspired by http://stackoverflow.com/a/37487134/627411.
"""
def get_fields(self):
fields = super().get_fields()
# We're only interested in PATCH and PUT.
if "update" in getattr(self.context.get("view"), "action", ""):
fields = self._update_write_once_fields(fields)
return fields
def _update_write_once_fields(self, fields):
"""
Set all fields in `Meta.write_once_fields` to read_only.
"""
write_once_fields = getattr(self.Meta, "write_once_fields", None)
if not write_once_fields:
return fields
if not isinstance(write_once_fields, (list, tuple)):
raise TypeError(
"The `write_once_fields` option must be a list or tuple. "
"Got {}.".format(type(write_once_fields).__name__)
)
for field_name in write_once_fields:
fields[field_name].read_only = True
return fields
@extend_schema_field(serializers.URLField)
class HyperlinkedEndpointSerializer(serializers.Serializer):
key_field = "pk"
def __init__(self, view_name=None, *, filter_key=None, **kwargs):
if isclass(view_name) and issubclass(view_name, models.models.Model):
view_name = get_list_view_name(view_name)
elif not isinstance(view_name, str):
raise TypeError(view_name)
kwargs["read_only"] = True
super().__init__(**kwargs)
self.view_name = view_name
self.filter_key = filter_key
def get_attribute(self, instance):
return instance
def to_representation(self, instance):
request = self.context.get("request")
if not request:
return None
return serializers.Hyperlink(
reverse(
self.view_name,
request=request,
query=build_field_filter_params(self.filter_key, getattr(instance, self.key_field)),
),
instance,
)
class _CollectionSummarySerializer(serializers.Serializer):
# This class isn't recommended for direct use in public serializers
# because it produces too generic description in the schema.
# Consider creating a dedicated inherited class instead.
count = serializers.IntegerField(default=0)
def __init__(self, model, *, url_filter_key, **kwargs):
super().__init__(**kwargs)
self._collection_key = self.source
self._model = model
self._url_filter_key = url_filter_key
def bind(self, field_name, parent):
super().bind(field_name, parent)
self._collection_key = self._collection_key or self.source
self._model = self._model or type(self.parent)
def get_fields(self):
fields = super().get_fields()
fields["url"] = HyperlinkedEndpointSerializer(self._model, filter_key=self._url_filter_key)
if not fields["count"].source:
fields["count"].source = self._collection_key + ".count"
return fields
def get_attribute(self, instance):
return instance
class JobsSummarySerializer(_CollectionSummarySerializer):
count = serializers.IntegerField(source="total_jobs_count", default=0)
completed = serializers.IntegerField(source="completed_jobs_count", allow_null=True)
validation = serializers.IntegerField(source="validation_jobs_count", allow_null=True)
def __init__(self, *, model=models.Job, url_filter_key, **kwargs):
super().__init__(model=model, url_filter_key=url_filter_key, **kwargs)
MAX_FILENAME_LENGTH = 1024
class TasksSummarySerializer(_CollectionSummarySerializer):
pass
class CommentsSummarySerializer(_CollectionSummarySerializer):
pass
class LabelsSummarySerializer(serializers.Serializer):
url = serializers.URLField(read_only=True)
def get_url(self, request, instance):
filter_key = instance.__class__.__name__.lower() + "_id"
return reverse("label-list", request=request, query={filter_key: instance.id})
def to_representation(self, instance):
request = self.context.get("request")
if not request:
return None
return {
"url": self.get_url(request, instance),
}
class IssuesSummarySerializer(serializers.Serializer):
url = serializers.URLField(read_only=True)
count = serializers.IntegerField(read_only=True)
def get_url(self, request, instance):
return reverse("issue-list", request=request, query={"job_id": instance.id})
def get_count(self, instance):
return getattr(instance, "issue__count", 0)
def to_representation(self, instance):
request = self.context.get("request")
if not request:
return None
return {"url": self.get_url(request, instance), "count": self.get_count(instance)}
class OrgTransferableMixin:
def _validate_org_transferring(self, attrs: dict[str, Any]):
"""Method should be called explicitly"""
if disallowed_fields := set(attrs.keys()) - {
"source_storage",
"target_storage",
"organization_id",
}:
raise serializers.ValidationError(
f"Fields {disallowed_fields} cannot be updated when transferring to another workspace"
)
# FUTURE-TODO: race condition is possible here
def update_organization(
self,
instance: models.Task | models.Project,
*,
validated_data: dict[str, Any],
update_fields: list[str],
):
update_date = timezone.now()
request = cast(ExtendedRequest, self.context["request"])
organization_id = validated_data["organization_id"]
organization_slug = None
if organization_id is not None:
try:
organization_slug = (
Organization.objects.filter(pk=organization_id).values_list("slug", flat=True)
)[0]
except IndexError:
raise serializers.ValidationError("Invalid organization id")
cur_user_id = request.user.id
if instance.owner_id != cur_user_id:
instance.owner_id = cur_user_id
update_fields.append("owner_id")
if instance.assignee_id is not None:
instance.update_assignee(None, updated_date=update_date)
update_fields.extend(["assignee_id", "assignee_updated_date"])
self.update_child_objects_on_organization_update(
instance,
owner_id=cur_user_id,
organization_id=organization_id,
updated_date=update_date,
)
instance.organization_id = organization_id
update_fields.append("organization_id")
# update source and target storages
_update_related_storages(
instance,
validated_data=validated_data,
workspace_transferring=True,
dst_organization_id=organization_id,
)
id_param = f"{instance.__class__.__name__}_id".lower()
try:
update_org_related_data_in_rq_jobs( # pylint: disable=unexpected-keyword-arg
organization_id, organization_slug, **{id_param: instance.pk}
)
except RunningBackgroundProcessesError as ex:
raise serializers.ValidationError(
f"Background processes are still running in the {ex.queue_name!r} queue"
)
def update_child_objects_on_organization_update(
self,
instance: models.Task,
*,
organization_id: int | None,
owner_id: int,
updated_date: datetime,
):
raise NotImplementedError()
class BasicUserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ("url", "id", "username", "first_name", "last_name")
class UserSerializer(serializers.ModelSerializer):
groups = serializers.SlugRelatedField(
many=True, slug_field="name", queryset=Group.objects.all()
)
has_analytics_access = serializers.BooleanField(
source="profile.has_analytics_access",
required=False,
read_only=True,
)
class Meta:
model = User
fields = (
"url",
"id",
"username",
"first_name",
"last_name",
"email",
"groups",
"is_staff",
"is_superuser",
"is_active",
"last_login",
"date_joined",
"has_analytics_access",
)
read_only_fields = ("last_login", "date_joined", "has_analytics_access")
write_only_fields = ("password",)
extra_kwargs = {"last_login": {"allow_null": True}}
class DelimitedStringListField(serializers.ListField):
def to_representation(self, value):
return super().to_representation(value.split("\n"))
def to_internal_value(self, data):
return "\n".join(super().to_internal_value(data))
class AttributeSerializer(serializers.ModelSerializer):
id = serializers.IntegerField(required=False)
deleted = serializers.BooleanField(
required=False,
write_only=True,
help_text="Delete the attribute and all related annotation values.",
)
values = DelimitedStringListField(
allow_empty=True, child=serializers.CharField(allow_blank=True, max_length=200)
)
class Meta:
model = models.AttributeSpec
fields = ("id", "name", "mutable", "input_type", "default_value", "values", "deleted")
extra_kwargs = {
"default_value": {"required": False},
}
def validate(self, attrs):
if attrs.get("deleted"):
if attrs.get("id") is None:
raise serializers.ValidationError("Deleted attribute must have an ID")
return attrs
class SublabelSerializer(serializers.ModelSerializer):
id = serializers.IntegerField(required=False)
attributes = AttributeSerializer(
many=True,
source="attributespec_set",
default=[],
help_text="The list of attributes. "
"To remove an attribute, pass the full attribute body with deleted=true. "
"Related annotation attribute values will be deleted.",
)
color = serializers.CharField(
allow_blank=True,
required=False,
help_text="The hex value for the RGB color. "
"Will be generated automatically, unless specified explicitly.",
)
type = serializers.ChoiceField(
choices=models.LabelType.choices(),
required=False,
help_text="Associated annotation type for this label",
)
has_parent = serializers.BooleanField(source="has_parent_label", required=False)
class Meta:
model = models.Label
fields = (
"id",
"name",
"color",
"attributes",
"type",
"has_parent",
)
read_only_fields = ("parent",)
class SkeletonSerializer(serializers.ModelSerializer):
id = serializers.IntegerField(required=False)
svg = serializers.CharField(allow_blank=True, required=False)
class Meta:
model = models.Skeleton
fields = (
"id",
"svg",
)
class LabelSerializer(SublabelSerializer):
deleted = serializers.BooleanField(
required=False,
write_only=True,
help_text="Delete the label. Only applicable in the PATCH methods of a project or a task.",
)
sublabels = SublabelSerializer(many=True, required=False)
svg = serializers.CharField(allow_blank=True, required=False)
has_parent = serializers.BooleanField(read_only=True, source="has_parent_label", required=False)
class Meta:
model = models.Label
fields = (
"id",
"name",
"color",
"attributes",
"deleted",
"type",
"svg",
"sublabels",
"project_id",
"task_id",
"parent_id",
"has_parent",
)
read_only_fields = ("id", "svg", "project_id", "task_id")
extra_kwargs = {
"project_id": {"required": False, "allow_null": False},
"task_id": {"required": False, "allow_null": False},
"parent_id": {
"required": False,
},
}
def to_representation(self, instance):
label = super().to_representation(instance)
if label["type"] == str(models.LabelType.SKELETON):
label["svg"] = instance.skeleton.svg
# Clean mutually exclusive fields
if not label.get("task_id"):
label.pop("task_id", None)
if not label.get("project_id"):
label.pop("project_id", None)
return label
def __init__(self, *args, **kwargs):
self._local = kwargs.pop("local", False)
"""
Indicates that the operation is called from the dedicated ViewSet
and not from the parent entity, i.e. a project or task.
"""
super().__init__(*args, **kwargs)
def validate(self, attrs):
if self._local and attrs.get("deleted"):
# NOTE: Navigate clients to the right method
raise serializers.ValidationError(
"Labels cannot be deleted by updating in this endpoint. "
"Please use the DELETE method instead."
)
if attrs.get("deleted") and attrs.get("id") is None:
raise serializers.ValidationError("Deleted label must have an ID")
return attrs
@staticmethod
def check_attribute_names_unique(attrs):
encountered_names = set()
for attribute in attrs:
if attribute.get("deleted"):
continue
attr_name = attribute.get("name")
if attr_name is None:
continue
if attr_name in encountered_names:
raise serializers.ValidationError(
f"Duplicate attribute with name '{attr_name}' exists"
)
else:
encountered_names.add(attr_name)
@staticmethod
def check_attribute_names_available(
db_attributes: dict[int, str], attrs: list[dict[str, Any]]
) -> None:
requested_attribute_names = {
attr["id"]: attr["name"]
for attr in attrs
if attr.get("id") is not None and attr.get("name") is not None
}
if not requested_attribute_names:
return
current_attribute_ids = {name: attr_id for attr_id, name in db_attributes.items()}
swapped_attr_names = set()
occupied_attr_names = set()
for attr_id, attr_name in requested_attribute_names.items():
current_attr_id = current_attribute_ids.get(attr_name)
if current_attr_id is None or current_attr_id == attr_id:
continue
if current_attr_id in requested_attribute_names:
swapped_attr_names.add(attr_name)
if current_name := db_attributes.get(attr_id):
swapped_attr_names.add(current_name)
else:
occupied_attr_names.add(attr_name)
if swapped_attr_names:
attr_names = ", ".join(f'"{name}"' for name in swapped_attr_names)
raise serializers.ValidationError(f"Cannot swap attribute names {attr_names}")
if occupied_attr_names:
attr_names = ", ".join(f'"{name}"' for name in occupied_attr_names)
raise serializers.ValidationError(
f"Attribute names are already used by this label: {attr_names}"
)
@staticmethod
def _split_attribute_values(values: str) -> list[str]:
return values.split("\n") if values else []
@classmethod
def _validate_attribute_value(cls, input_type: str, value: str, values: str) -> None:
if input_type == str(models.AttributeType.TEXT):
return
if input_type == str(models.AttributeType.CHECKBOX):
valid = value.lower() in {"true", "false"}
elif input_type == str(models.AttributeType.NUMBER):
attr_values = cls._split_attribute_values(values)
try:
valid = float(attr_values[0]) <= float(value) <= float(attr_values[1])
except (IndexError, ValueError):
valid = False
else:
valid = value in cls._split_attribute_values(values)
if not valid:
raise serializers.ValidationError(
'Attribute field "default_value" is invalid for attribute input type'
)
@classmethod
def _update_attribute(cls, db_attr: models.AttributeSpec, attr: dict[str, Any]) -> None:
for field_name in ("mutable", "input_type"):
if field_name in attr and attr[field_name] != getattr(db_attr, field_name):
raise serializers.ValidationError(
f'Attribute field "{field_name}" cannot be changed'
)
new_values = attr.get("values", db_attr.values)
if new_values != db_attr.values:
if db_attr.input_type in (
str(models.AttributeType.RADIO),
str(models.AttributeType.SELECT),
):
old_values_list = cls._split_attribute_values(db_attr.values)
new_values_list = cls._split_attribute_values(new_values)
if not set(old_values_list).issubset(new_values_list):
raise serializers.ValidationError(
'Attribute field "values" can only be appended for radio and select attributes'
)
elif db_attr.input_type == str(models.AttributeType.NUMBER):
raise serializers.ValidationError(
'Attribute field "values" cannot be changed for number attributes'
)
new_default_value = attr.get("default_value", db_attr.default_value)
if new_default_value != db_attr.default_value:
cls._validate_attribute_value(db_attr.input_type, new_default_value, new_values)
db_attr.name = attr.get("name", db_attr.name)
db_attr.default_value = new_default_value
db_attr.values = new_values
db_attr.save()
@classmethod
@transaction.atomic
def update_label(
cls,
validated_data: dict[str, Any],
svg: str,
sublabels: Iterable[dict[str, Any]],
*,
parent_instance: models.Project | models.Task,
parent_label: models.Label | None = None,
) -> models.Label | None:
parent_info, logger = cls._get_parent_info(parent_instance)
attributes = validated_data.pop("attributespec_set", [])
cls.check_attribute_names_unique(attributes)
label_exists = validated_data.get("id") is not None
if label_exists:
label_filter = dict(parent_info)
if parent_label is not None:
# Enforce nested update scope: a sublabel ID must belong to the
# parent skeleton label currently being patched.
label_filter["parent"] = parent_label
try:
db_label = models.Label.objects.get(id=validated_data["id"], **label_filter)
except models.Label.DoesNotExist as exc:
raise exceptions.NotFound(
detail="Not found label with id #{} to change".format(validated_data["id"])
) from exc
updated_type = validated_data.get("type") or db_label.type
if str(models.LabelType.SKELETON) in [db_label.type, updated_type]:
# do not permit changing types from/to skeleton
logger.warning(
"Label id {} ({}): an attempt to change label type from {} to {}. "
"Changing from or to '{}' is not allowed, the type won't be changed.".format(
db_label.id,
db_label.name,
db_label.type,
updated_type,
str(models.LabelType.SKELETON),
)
)
else:
db_label.type = updated_type
db_label.name = validated_data.get("name") or db_label.name
logger.info("Label id {} ({}) was updated".format(db_label.id, db_label.name))
else:
try:
db_label = models.Label.create(
name=validated_data.get("name"),
type=validated_data.get("type", models.LabelType.ANY),
parent=parent_label,
**parent_info,
)
except models.InvalidLabel as exc:
raise exceptions.ValidationError(str(exc)) from exc
logger.info("New {} label was created".format(db_label.name))
cls.update_labels(sublabels, parent_instance=parent_instance, parent_label=db_label)
if db_label.type == str(models.LabelType.SKELETON):
for db_sublabel in list(db_label.sublabels.all()):
svg = svg.replace(
f'data-label-name="{db_sublabel.name}"', f'data-label-id="{db_sublabel.id}"'
)
db_skeleton = models.Skeleton.objects.create(root=db_label, svg=svg)
logger.info(f"label:update Skeleton id:{db_skeleton.id} for label_id:{db_label.id}")
if validated_data.get("deleted"):
assert validated_data["id"] # must be checked in the validate()
db_label.delete()
return None
if not validated_data.get("color", None):
other_label_colors = [
label.color
for label in parent_instance.label_set.exclude(id=db_label.id).order_by("id")
]
db_label.color = get_label_color(db_label.name, other_label_colors)
else:
db_label.color = validated_data.get("color", db_label.color)
try:
db_label.save()
except models.InvalidLabel as exc:
raise exceptions.ValidationError(str(exc)) from exc
if label_exists:
cls.update_labels(sublabels, parent_instance=parent_instance, parent_label=db_label)
deleted_attributes = [attr for attr in attributes if attr.get("deleted")]
upserted_attributes = [attr for attr in attributes if not attr.get("deleted")]
def get_db_attr(attr_id: int) -> models.AttributeSpec:
try:
return models.AttributeSpec.objects.get(id=attr_id, label=db_label)
except models.AttributeSpec.DoesNotExist as ex:
raise exceptions.NotFound(f"Attribute with id #{attr_id} does not exist") from ex
# Apply deletions before creates/updates. This keeps an atomic request valid
# when an attribute is renamed to a name released by another deleted attribute.
for attr in deleted_attributes:
attr_id = attr.get("id")
if attr_id is None:
raise serializers.ValidationError("Deleted attribute must have an ID")
db_attr = get_db_attr(attr_id)
logger.info("{} attribute for {} label was deleted".format(db_attr.name, db_label.name))
db_attr.delete()
if label_exists:
db_attributes = dict(db_label.attributespec_set.values_list("id", "name"))
cls.check_attribute_names_available(db_attributes, upserted_attributes)
for attr in upserted_attributes:
attr_id = attr.get("id", None)
if attr_id is not None:
db_attr = get_db_attr(attr_id)
created = False
else:
db_attr, created = models.AttributeSpec.objects.get_or_create(
label=db_label, name=attr["name"], defaults=attr
)
if created:
logger.info(
"New {} attribute for {} label was created".format(db_attr.name, db_label.name)
)
else:
logger.info(
"{} attribute for {} label was updated".format(db_attr.name, db_label.name)
)
cls._update_attribute(db_attr, attr)
return db_label
@classmethod
@transaction.atomic
def create_labels(
cls,
labels: Iterable[dict[str, Any]],
*,
parent_instance: models.Project | models.Task,
parent_label: models.Label | None = None,
):
parent_info, logger = cls._get_parent_info(parent_instance)
label_colors = list()
for label in labels:
attributes = label.pop("attributespec_set")
cls.check_attribute_names_unique(attributes)
if label.get("id", None):
del label["id"]
if not label.get("color", None):
label["color"] = get_label_color(label["name"], label_colors)
label_colors.append(label["color"])
sublabels = label.pop("sublabels", [])
svg = label.pop("svg", "")
try:
db_label = models.Label.create(**label, **parent_info, parent=parent_label)
except models.InvalidLabel as exc:
raise exceptions.ValidationError(str(exc)) from exc
logger.info(
f"label:create Label id:{db_label.id} for spec:{label} "
f"with sublabels:{sublabels}, parent_label:{parent_label}"
)
cls.create_labels(sublabels, parent_instance=parent_instance, parent_label=db_label)
if db_label.type == str(models.LabelType.SKELETON):
for db_sublabel in list(db_label.sublabels.all()):
svg = svg.replace(
f'data-label-name="{db_sublabel.name}"', f'data-label-id="{db_sublabel.id}"'
)
db_skeleton = models.Skeleton.objects.create(root=db_label, svg=svg)
logger.info(f"label:create Skeleton id:{db_skeleton.id} for label_id:{db_label.id}")
for attr in attributes:
if attr.get("id", None):
del attr["id"]
models.AttributeSpec.objects.create(label=db_label, **attr)
@classmethod
@transaction.atomic
def update_labels(
cls,
labels: Iterable[dict[str, Any]],
*,
parent_instance: models.Project | models.Task,
parent_label: models.Label | None = None,
):
_, logger = cls._get_parent_info(parent_instance)
for label in labels:
sublabels = label.pop("sublabels", [])
svg = label.pop("svg", "")
db_label = cls.update_label(
label, svg, sublabels, parent_instance=parent_instance, parent_label=parent_label
)
if db_label:
logger.info(
f"label:update Label id:{db_label.id} for spec:{label} "
f"with sublabels:{sublabels}, parent_label:{parent_label}"
)
else:
logger.info(
f"label:delete label:{label} with "
f"sublabels:{sublabels}, parent_label:{parent_label}"
)
@classmethod
def _get_parent_info(cls, parent_instance: models.Project | models.Task):
parent_info = {}
if isinstance(parent_instance, models.Project):
parent_info["project"] = parent_instance
logger = slogger.project[parent_instance.id]
elif isinstance(parent_instance, models.Task):
parent_info["task"] = parent_instance
logger = slogger.task[parent_instance.id]
else:
raise TypeError(f"Unexpected parent instance type {type(parent_instance).__name__}")
return parent_info, logger
def update(self, instance, validated_data):
if not self._local:
return super().update(instance, validated_data)
if isinstance(instance.project, models.Project):
parent_instance = instance.project
parent_serializer = ProjectWriteSerializer(parent_instance)
elif isinstance(instance.task, models.Task):
parent_instance = instance.task
parent_serializer = TaskWriteSerializer(parent_instance)
else:
raise serializers.ValidationError("Label must belong to a project or a task")
data = copy(validated_data)
data["id"] = instance.id
data.setdefault("name", instance.name)
sublabels = data.pop("sublabels", [])
svg = data.pop("svg", "")
self.update_label(data, svg, sublabels, parent_instance=parent_instance)
parent_instance.touch()
parent_serializer.update_child_objects_on_labels_update(parent_instance)
self.instance = models.Label.objects.get(pk=instance.pk)
return self.instance
class StorageSerializer(serializers.ModelSerializer):
cloud_storage_id = serializers.IntegerField(required=False, allow_null=True)
class Meta:
model = models.Storage
fields = ("id", "location", "cloud_storage_id")
def validate(self, attrs: dict[str, Any]):
new_cloud_storage_id = attrs.get("cloud_storage_id")
new_location = attrs.get("location")
if new_cloud_storage_id:
if new_location and new_location != models.Location.CLOUD_STORAGE:
raise serializers.ValidationError(
f"It is not allowed to specify '{new_location}' location together with cloud storage id"
)
elif not new_location:
raise serializers.ValidationError(
"The configuration is not full, the location=cloud_storage is missing"
)
if not models.CloudStorage.objects.filter(id=new_cloud_storage_id).exists():
raise serializers.ValidationError(
f"The specified cloud storage {new_cloud_storage_id} does not exist."
)
else:
if new_location == models.Location.CLOUD_STORAGE:
raise serializers.ValidationError(
"Cloud storage was selected as location but its id was not specified"
)
elif not new_location and "cloud_storage_id" in attrs:
raise serializers.ValidationError(
"It is not allowed to reset a cloud storage id without explicitly resetting a location"
)
return super().validate(attrs)
class JobReadListSerializer(serializers.ListSerializer):
def to_representation(self, data):
if (request := self.context.get("request")) and isinstance(data, list) and data:
# Optimized prefetch only for the current page
page: list[models.Job] = data
# Annotate page objects
# We do it explicitly here and not in the LIST queryset to avoid
# doing the same DB computations twice - one time for the page retrieval
# and another one for the COUNT(*) request to get the total count
page_task_ids = set(j.get_task_id() for j in page)
# Prefetch related object visibility
# This avoids N+1 queries when serializing
iam_context = get_iam_context(request, None)
visible_tasks_perm = TaskPermission.create_scope_list(request, iam_context)
visible_tasks_queryset = models.Task.objects.filter(id__in=page_task_ids)
visible_task_ids = set(
visible_tasks_perm.filter(visible_tasks_queryset).values_list("id", flat=True)
)
page_project_ids = set(
job.segment.task.project_id
for job in page
if job.segment.task_id in visible_task_ids
)
visible_projects_perm = ProjectPermission.create_scope_list(request, iam_context)
visible_projects_queryset = models.Project.objects.filter(id__in=page_project_ids)
visible_project_ids = set(
visible_projects_perm.filter(visible_projects_queryset).values_list("id", flat=True)
)
page_storage_ids = set(
v
for job in page
if job.segment.task_id in visible_task_ids
for v in (job.segment.task.source_storage_id, job.segment.task.target_storage_id)
)
visible_storages = {
s.id: s for s in models.Storage.objects.filter(id__in=page_storage_ids)
}
# Join the prefetched objects
# Keep in mind that the object ids fetched in the earlier queries
# might be missing in the later queries because of locks and removals,
# so should not be expected to be present and should be checked before access.
for job in page:
job.user_can_view_task = job.segment.task_id in visible_task_ids
if job.segment.task_id in visible_task_ids:
task = job.segment.task
if task.source_storage_id in visible_storages:
task.source_storage = visible_storages[task.source_storage_id]
if task.target_storage_id in visible_storages:
task.target_storage = visible_storages[task.target_storage_id]
if task.project_id:
task.user_can_view_project = task.project_id in visible_project_ids
# Fetching it here removes 1 extra join for all jobs in the COUNT(*) request,
# limiting it only for the page
job_ids = set(j.id for j in page)
issue_counts = dict(
models.Job.objects.with_issue_counts()
.filter(id__in=job_ids)
.values_list("id", "issue__count")
)
children_counts = dict(
models.Job.objects.with_child_jobs_counts()
.filter(id__in=job_ids)
.values_list("id", "child_jobs__count")
)
for job in page:
job.issue__count = issue_counts.get(job.id, 0)
job.child_jobs__count = children_counts.get(job.id, 0)
prefetch_related_objects(
page,
"segment__task__data",
"segment__task__annotation_guide",
"segment__task__project__annotation_guide",
)
return super().to_representation(data)
@extend_schema_serializer(deprecate_fields=["consensus_replicas"])
class JobReadSerializer(serializers.ModelSerializer):
task_id = serializers.ReadOnlyField(source="get_task_id")
task_name = serializers.SerializerMethodField()
project_id = serializers.ReadOnlyField(source="get_project_id", allow_null=True)
project_name = serializers.SerializerMethodField()
guide_id = serializers.ReadOnlyField(source="get_guide_id", allow_null=True)
start_frame = serializers.ReadOnlyField(source="segment.start_frame")
stop_frame = serializers.ReadOnlyField(source="segment.stop_frame")
frame_count = serializers.ReadOnlyField(source="segment.frame_count")
assignee = BasicUserSerializer(allow_null=True, read_only=True)
# We're using CharField to produce simple strings instead of enums in the generated SDK.
# SDK enums require explicit .value calls to access the string representation.
# TODO: move to ChoicesField when SDK supports seamless transition from string to enum
dimension = serializers.CharField(source="segment.task.dimension", read_only=True)
mode = serializers.CharField(source="segment.task.mode", read_only=True)
media_type = serializers.CharField(source="segment.task.media_type", read_only=True)
data_chunk_size = serializers.ReadOnlyField(source="segment.task.data.chunk_size")
organization = serializers.ReadOnlyField(source="organization_id", allow_null=True)
data_original_chunk_type = serializers.ChoiceField(
source="segment.task.data.original_chunk_type",
choices=models.DataChoice.choices(),
allow_blank=False,
read_only=True,
)
data_compressed_chunk_type = serializers.ChoiceField(
source="segment.task.data.compressed_chunk_type",
choices=models.DataChoice.choices(),
allow_blank=False,
read_only=True,
)
bug_tracker = serializers.CharField(
max_length=2000, source="get_bug_tracker", allow_null=True, read_only=True
)
labels = LabelsSummarySerializer(source="*")
issues = IssuesSummarySerializer(source="*")
target_storage = StorageSerializer(required=False, allow_null=True)
source_storage = StorageSerializer(required=False, allow_null=True)
parent_job_id = serializers.ReadOnlyField(allow_null=True)
consensus_replicas = serializers.IntegerField(read_only=True)
replicas_count = serializers.IntegerField(read_only=True)
class Meta:
model = models.Job
fields = (
"url",
"id",
"task_id",
"task_name",
"project_id",
"project_name",
"assignee",
"guide_id",
"dimension",
"mode",
"media_type",
"bug_tracker",
"status",
"stage",
"state",
"frame_count",
"start_frame",
"stop_frame",
"data_chunk_size",
"data_compressed_chunk_type",
"data_original_chunk_type",
"created_date",
"updated_date",
"issues",
"labels",
"type",
"organization",
"target_storage",
"source_storage",
"assignee_updated_date",
"parent_job_id",
"consensus_replicas",
"replicas_count",
)
read_only_fields = fields
list_serializer_class = JobReadListSerializer
def _can_see_task(self, instance: models.Job) -> bool:
request = self.context.get("request")
if not request:
return False
can_see_task = getattr(instance, "user_can_view_task", None)
if can_see_task is None:
perm = TaskPermission.create_scope_view(request, instance.segment.task)
can_see_task = perm.check_access().allow
return can_see_task
def _can_see_project(self, instance: models.Job) -> bool:
request = self.context.get("request")
if not request:
return False
can_see_project = getattr(instance.segment.task, "user_can_view_project", None)
if can_see_project is None:
if not instance.segment.task.project_id:
return False
perm = ProjectPermission.create_scope_view(request, instance.segment.task.project)
can_see_project = perm.check_access().allow
return can_see_project
@extend_schema_field(serializers.CharField(allow_null=True))
def get_task_name(self, instance: models.Job) -> str | None:
return instance.segment.task.name if self._can_see_task(instance) else None
@extend_schema_field(serializers.CharField(allow_null=True))
def get_project_name(self, instance: models.Job) -> str | None:
return instance.segment.task.project.name if self._can_see_project(instance) else None
def to_representation(self, instance: models.Job):
data = super().to_representation(instance)
if instance.segment.type == models.SegmentType.SPECIFIC_FRAMES:
data["data_compressed_chunk_type"] = models.DataChoice.IMAGESET
if "replicas_count" in self.fields:
data["replicas_count"] = getattr(instance, "child_jobs__count", 0)
data["consensus_replicas"] = data["replicas_count"]
if self._can_see_task(instance):
if task_source_storage := instance.get_source_storage():
data["source_storage"] = StorageSerializer(task_source_storage).data
if task_target_storage := instance.get_target_storage():
data["target_storage"] = StorageSerializer(task_target_storage).data
return data
class JobWriteSerializer(WriteOnceMixin, serializers.ModelSerializer):
assignee = serializers.IntegerField(allow_null=True, required=False)
# NOTE: Field sets can be expressed using serializer inheritance, but it is
# harder to use then: we need to make a manual switch in get_serializer_class()
# and create an extra serializer type in the API schema.
# Need to investigate how it can be simplified. It can also be done just internally,
# (e.g. just on the validation side), but it will complicate the implementation.
type = serializers.ChoiceField(choices=models.JobType.choices())
task_id = serializers.IntegerField()
frame_selection_method = serializers.ChoiceField(
choices=models.JobFrameSelectionMethod.choices(), required=False
)
frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
required=False,
allow_empty=False,
help_text=textwrap.dedent("""\
The list of frame ids. Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.MANUAL)),
)
frame_count = serializers.IntegerField(
min_value=1,
required=False,
help_text=textwrap.dedent("""\
The number of frames included in the GT job.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_UNIFORM)),
)
frame_share = serializers.FloatField(
required=False,
validators=[field_validation.validate_share],
help_text=textwrap.dedent("""\
The share of frames included in the GT job.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_UNIFORM)),
)
frames_per_job_count = serializers.IntegerField(
min_value=1,
required=False,
help_text=textwrap.dedent("""\
The number of frames included in the GT job from each annotation job.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_PER_JOB)),
)
frames_per_job_share = serializers.FloatField(
required=False,
validators=[field_validation.validate_share],
help_text=textwrap.dedent("""\
The share of frames included in the GT job from each annotation job.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_PER_JOB)),
)
random_seed = serializers.IntegerField(
min_value=0,
required=False,
help_text=textwrap.dedent("""\
The seed value for the random number generator.
The same value will produce the same frame sets.
Applicable only to random frame selection methods.
By default, a random value is used.
"""),
)
class Meta:
model = models.Job
random_selection_params = (
"frame_count",
"frame_share",
"frames_per_job_count",
"frames_per_job_share",
"random_seed",
)
manual_selection_params = ("frames",)
write_once_fields = (
(
"type",
"task_id",
"frame_selection_method",
)
+ random_selection_params
+ manual_selection_params
)
fields = (
"assignee",
"stage",
"state",
) + write_once_fields
def to_representation(self, instance):
serializer = JobReadSerializer(instance, context=self.context)
return serializer.data
def validate(self, attrs):
frame_selection_method = attrs.get("frame_selection_method")
if frame_selection_method == models.JobFrameSelectionMethod.RANDOM_UNIFORM:
field_validation.require_one_of_fields(attrs, ["frame_count", "frame_share"])
elif frame_selection_method == models.JobFrameSelectionMethod.RANDOM_PER_JOB:
field_validation.require_one_of_fields(
attrs, ["frames_per_job_count", "frames_per_job_share"]
)
elif frame_selection_method == models.JobFrameSelectionMethod.MANUAL:
field_validation.require_field(attrs, "frames")
if "frames" in attrs and frame_selection_method != models.JobFrameSelectionMethod.MANUAL:
raise serializers.ValidationError(
'"frames" can only be used when "frame_selection_method" is "{}"'.format(
models.JobFrameSelectionMethod.MANUAL
)
)
return super().validate(attrs)
@transaction.atomic
def create(self, validated_data):
if validated_data["type"] != models.JobType.GROUND_TRUTH:
raise serializers.ValidationError(f"Unexpected job type '{validated_data['type']}'")
task_id = validated_data.pop("task_id")
task = models.Task.objects.get(pk=task_id)
if task.data_id is None:
raise serializers.ValidationError(
"This task has no data attached yet. Please set up task data and try again"
)
ensure_task_is_initialized(task=task)
if task.data.validation_mode in (models.ValidationMode.GT_POOL, models.ValidationMode.GT):
raise serializers.ValidationError(
f'Task with validation mode "{task.data.validation_mode}" '
"cannot have more than 1 GT job"
)
task = models.Task.objects.select_for_update().get(pk=task_id)
if task.media_type == models.MediaType.AUDIO:
frames = []
elif task.media_type == models.MediaType.IMAGE:
task_size = task.data.size
valid_frame_ids = task.data.get_valid_frame_indices()
frame_selection_method = validated_data.pop("frame_selection_method")
if frame_selection_method == models.JobFrameSelectionMethod.RANDOM_UNIFORM:
if frame_count := validated_data.pop("frame_count", None):
if task_size < frame_count:
raise serializers.ValidationError(
f"The number of frames requested ({frame_count}) "
f"must not be greater than the number of the task frames ({task_size})"
)
elif frame_share := validated_data.pop("frame_share", None):
frame_count = max(1, int(frame_share * task_size))
else:
raise serializers.ValidationError(
"The number of validation frames is not specified"
)
seed = validated_data.pop("random_seed", None)
# The RNG backend must not change to yield reproducible results,
# so here we specify it explicitly
rng = random.Generator(random.MT19937(seed=seed))
frames = rng.choice(
list(valid_frame_ids), size=frame_count, shuffle=False, replace=False
).tolist()
elif frame_selection_method == models.JobFrameSelectionMethod.RANDOM_PER_JOB:
if frame_count := validated_data.pop("frames_per_job_count", None):
if task_size < frame_count:
raise serializers.ValidationError(
f"The number of frames requested ({frame_count}) "
f"must be not be greater than the segment size ({task.segment_size})"
)
elif frame_share := validated_data.pop("frames_per_job_share", None):
frame_count = min(max(1, int(frame_share * task.segment_size)), task_size)
else:
raise serializers.ValidationError(
"The number of validation frames is not specified"
)
task_frame_provider = TaskFrameProvider(task)
seed = validated_data.pop("random_seed", None)
# The RNG backend must not change to yield reproducible results,
# so here we specify it explicitly
rng = random.Generator(random.MT19937(seed=seed))
frames: list[int] = []
overlap = task.overlap
for segment in task.segment_set.all():
segment_frames = set(
map(task_frame_provider.get_rel_frame_number, segment.frame_set)
)
selected_frames = segment_frames.intersection(frames)
selected_count = len(selected_frames)
missing_count = min(len(segment_frames), frame_count) - selected_count
if missing_count <= 0:
continue
selectable_segment_frames = set(
sorted(segment_frames)[overlap * (segment.start_frame != 0) :]
).difference(selected_frames)
frames.extend(
rng.choice(
tuple(selectable_segment_frames), size=missing_count, replace=False
).tolist()
)
frames = list(map(task_frame_provider.get_abs_frame_number, frames))
elif frame_selection_method == models.JobFrameSelectionMethod.MANUAL:
frames = validated_data.pop("frames")
unique_frames = set(frames)
if len(unique_frames) != len(frames):
raise serializers.ValidationError("Frames must not repeat")
invalid_ids = unique_frames.difference(range(task_size))
if invalid_ids:
raise serializers.ValidationError(
"The following frames do not exist in the task: {}".format(
format_list(tuple(map(str, sorted(invalid_ids))))
)
)
task_frame_provider = TaskFrameProvider(task)
frames = list(map(task_frame_provider.get_abs_frame_number, frames))
else:
raise serializers.ValidationError(
f"Unexpected frame selection method '{frame_selection_method}'"
)
else:
raise serializers.ValidationError(
f"Ground Truth jobs are not available for the '{task.media_type}' media type"
)
# Save the new job
segment = models.Segment.objects.create(
start_frame=0,
stop_frame=task.data.size - 1,
frames=frames,
task=task,
type=models.SegmentType.SPECIFIC_FRAMES if frames else models.SegmentType.RANGE,
)
job_params = {
"type": validated_data.pop("type"),
"segment": segment,
"assignee_id": validated_data.pop("assignee", None),
}
if validated_data:
raise serializers.ValidationError(
"Fields {} specified, but not used.".format(
", ".join(f'"{k}"' for k in validated_data)
)
)
try:
job = super().create(job_params)
except models.TaskGroundTruthJobsLimitError as ex:
raise serializers.ValidationError(ex.message) from ex
if job_params.get("assignee_id"):
job.assignee_updated_date = job.updated_date
job.save(update_fields=["assignee_updated_date"])
job.make_dirs()
task.data.update_validation_layout(
models.ValidationLayout(mode=models.ValidationMode.GT, frames=frames)
)
return job
def update(self, instance: models.Job, validated_data: dict[str, Any]):
stage = validated_data.get("stage", instance.stage)
state = validated_data.get(
"state", models.StateChoice.NEW if stage != instance.stage else instance.state
)
if "stage" in validated_data or "state" in validated_data:
if stage == models.StageChoice.ANNOTATION:
validated_data["status"] = models.StatusChoice.ANNOTATION
elif stage == models.StageChoice.ACCEPTANCE and state == models.StateChoice.COMPLETED:
validated_data["status"] = models.StatusChoice.COMPLETED
else:
validated_data["status"] = models.StatusChoice.VALIDATION
if state != instance.state:
validated_data["state"] = state
if "assignee" in validated_data and (
(assignee_id := validated_data.pop("assignee")) != instance.assignee_id
):
instance.update_assignee(assignee_id)
instance = super().update(instance, validated_data)
return instance
class SimpleJobSerializer(serializers.ModelSerializer):
assignee = BasicUserSerializer(allow_null=True)
class Meta:
model = models.Job
fields = ("url", "id", "assignee", "status", "stage", "state", "type")
read_only_fields = fields
class JobValidationLayoutWriteSerializer(serializers.Serializer):
frame_selection_method = serializers.ChoiceField(
choices=models.JobFrameSelectionMethod.choices(),
required=True,
help_text=textwrap.dedent("""\
The method to use for frame selection of new real frames for honeypots in the job
"""),
)
honeypot_real_frames = serializers.ListSerializer(
child=serializers.IntegerField(min_value=0),
required=False,
allow_empty=False,
help_text=textwrap.dedent("""\
The list of frame ids. Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.MANUAL)),
)
def __init__(
self, *args, bulk_context: _TaskValidationLayoutBulkUpdateContext | None = None, **kwargs
):
super().__init__(*args, **kwargs)
self._bulk_context = bulk_context
def validate(self, attrs):
frame_selection_method = attrs["frame_selection_method"]
if frame_selection_method == models.JobFrameSelectionMethod.MANUAL:
field_validation.require_field(attrs, "honeypot_real_frames")
elif frame_selection_method == models.JobFrameSelectionMethod.RANDOM_UNIFORM:
pass
else:
assert False
if (
"honeypot_real_frames" in attrs
and frame_selection_method != models.JobFrameSelectionMethod.MANUAL
):
raise serializers.ValidationError(
'"honeypot_real_frames" can only be used when '
f'"frame_selection_method" is "{models.JobFrameSelectionMethod.MANUAL}"'
)
return super().validate(attrs)
@transaction.atomic
def update(self, instance: models.Job, validated_data: dict[str, Any]) -> models.Job:
from cvat.apps.engine.cache import (
Callback,
MediaCache,
enqueue_create_chunk_job,
wait_for_rq_job,
)
from cvat.apps.engine.media_io.frame_provider import JobFrameProvider
db_job = instance
db_segment = db_job.segment
db_task = db_segment.task
db_data = db_task.require_data()
if not (
hasattr(db_data, "validation_layout")
and db_data.validation_layout.mode == models.ValidationMode.GT_POOL
):
raise serializers.ValidationError(
"Honeypots can only be modified if the task "
f"validation mode is '{models.ValidationMode.GT_POOL}'"
)
if db_job.type == models.JobType.GROUND_TRUTH:
raise serializers.ValidationError(
f"Honeypots cannot exist in {models.JobType.GROUND_TRUTH} jobs"
)
assert not hasattr(db_data, "video")
frame_step = db_data.get_frame_step()
def _to_rel_frame(abs_frame: int) -> int:
return (abs_frame - db_data.start_frame) // frame_step
def _to_abs_frame(rel_frame: int) -> int:
return rel_frame * frame_step + db_data.start_frame
bulk_context = self._bulk_context
if bulk_context:
db_frames = bulk_context.all_db_frames
task_honeypot_frames = set(bulk_context.honeypot_frames)
task_all_validation_frames = set(bulk_context.all_validation_frames)
task_active_validation_frames = set(bulk_context.active_validation_frames)
else:
db_frames: dict[int, models.Image] = {
_to_rel_frame(frame.frame): frame for frame in db_data.images.all()
}
task_honeypot_frames = set(
_to_rel_frame(frame_id)
for frame_id, frame in db_frames.items()
if frame.is_placeholder
)
task_all_validation_frames = set(db_data.validation_layout.frames)
task_active_validation_frames = set(db_data.validation_layout.active_frames)
segment_frame_set = set(map(_to_rel_frame, db_segment.frame_set))
segment_honeypots = sorted(segment_frame_set & task_honeypot_frames)
segment_honeypots_count = len(segment_honeypots)
frame_selection_method = validated_data["frame_selection_method"]
if frame_selection_method == models.JobFrameSelectionMethod.MANUAL:
requested_frames: list[int] = validated_data["honeypot_real_frames"]
requested_inactive_frames: set[int] = set()
requested_normal_frames: set[int] = set()
for requested_validation_frame in requested_frames:
if requested_validation_frame not in task_all_validation_frames:
requested_normal_frames.add(requested_validation_frame)
continue
if requested_validation_frame not in task_active_validation_frames:
requested_inactive_frames.add(requested_validation_frame)
continue
if requested_normal_frames:
raise serializers.ValidationError(
"Could not update honeypot frames: "
"frames {} are not from the validation pool".format(
format_list(tuple(map(str, sorted(requested_normal_frames))))
)
)
if requested_inactive_frames:
raise serializers.ValidationError(
"Could not update honeypot frames: "
"frames {} are disabled. Restore them in the validation pool first".format(
format_list(tuple(map(str, sorted(requested_inactive_frames))))
)
)
if len(requested_frames) != segment_honeypots_count:
raise serializers.ValidationError(
"Could not update honeypot frames: "
"the number of honeypots must remain the same. "
"Requested {}, current {}".format(
len(requested_frames), segment_honeypots_count
)
)
elif frame_selection_method == models.JobFrameSelectionMethod.RANDOM_UNIFORM:
if len(task_active_validation_frames) < segment_honeypots_count:
raise serializers.ValidationError(
"Can't select validation frames: "
"the remaining number of validation frames ({}) "
"is less than the number of honeypots in a job ({}). "
"Try to restore some validation frames".format(
len(task_active_validation_frames), segment_honeypots_count
)
)
if bulk_context:
frame_selector = bulk_context.honeypot_frame_selector
else:
active_validation_frame_counts = {
validation_frame: 0 for validation_frame in task_active_validation_frames
}
for task_honeypot_frame in task_honeypot_frames:
real_frame = _to_rel_frame(db_frames[task_honeypot_frame].real_frame)
if real_frame in task_active_validation_frames:
active_validation_frame_counts[real_frame] += 1
frame_selector = HoneypotFrameSelector(active_validation_frame_counts)
requested_frames = frame_selector.select_next_frames(segment_honeypots_count)
requested_frames = list(map(_to_abs_frame, requested_frames))
else:
assert False
# Replace validation frames in the job
updated_honeypots = {}
for frame, requested_validation_frame in zip(segment_honeypots, requested_frames):
db_requested_frame = db_frames[requested_validation_frame]
db_segment_frame = db_frames[frame]
assert db_segment_frame.is_placeholder
if db_segment_frame.real_frame == db_requested_frame.frame:
continue
# Change image in the current segment honeypot frame
db_segment_frame.real_frame = db_requested_frame.frame
db_segment_frame.path = db_requested_frame.path
db_segment_frame.width = db_requested_frame.width
db_segment_frame.height = db_requested_frame.height
updated_honeypots[frame] = db_segment_frame
if updated_honeypots:
if bulk_context:
bulk_context.updated_honeypots.update(updated_honeypots)
else:
# Update image infos
models.Image.objects.bulk_update(
updated_honeypots.values(), fields=["path", "width", "height", "real_frame"]
)
models.RelatedFile.images.through.objects.filter(
image_id__in=updated_honeypots
).delete()
for updated_honeypot in updated_honeypots.values():
validation_frame = db_frames[_to_rel_frame(updated_honeypot.real_frame)]
updated_honeypot.related_files.set(validation_frame.related_files.all())
# Remove annotations on changed validation frames
self._clear_annotations_on_frames(db_segment, updated_honeypots)
# Update manifest
manifest_path = db_data.get_manifest_path()
if os.path.isfile(manifest_path):
manifest = ImageManifestManager(manifest_path)
manifest.reorder([db_frame.path for db_frame in db_frames.values()])
# Update chunks
job_frame_provider = JobFrameProvider(db_job)
updated_segment_chunk_ids = set(
job_frame_provider.get_chunk_number(updated_segment_frame_id)
for updated_segment_frame_id in updated_honeypots
)
segment_frames = sorted(segment_frame_set)
segment_frame_map = dict(zip(segment_honeypots, requested_frames))
chunks_to_be_removed = []
queue = django_rq.get_queue(settings.CVAT_QUEUES.CHUNKS.value)
for chunk_id in sorted(updated_segment_chunk_ids):
chunk_frames = segment_frames[
chunk_id * db_data.chunk_size : (chunk_id + 1) * db_data.chunk_size
]
for quality in models.FrameQuality:
if db_data.storage_method == models.StorageMethodChoice.FILE_SYSTEM:
rq_id = f"segment_{db_segment.id}_write_chunk_{chunk_id}_{quality}"
rq_job = enqueue_create_chunk_job(
queue=queue,
rq_job_id=rq_id,
create_callback=Callback(
callable=self._write_updated_static_chunk,
args=[
db_segment.id,
chunk_id,
chunk_frames,
quality,
{
chunk_frame: db_frames[chunk_frame].path
for chunk_frame in chunk_frames
},
segment_frame_map,
],
),
)
wait_for_rq_job(rq_job)
chunks_to_be_removed.append(
{"db_segment": db_segment, "chunk_number": chunk_id, "quality": quality}
)
context_image_chunks_to_be_removed = [
{"db_data": db_data, "frame_number": f} for f in updated_honeypots
]
if bulk_context:
bulk_context.chunks_to_be_removed.extend(chunks_to_be_removed)
bulk_context.context_image_chunks_to_be_removed.extend(
context_image_chunks_to_be_removed
)
bulk_context.segments_with_updated_chunks.append(db_segment.id)
else:
media_cache = MediaCache()
media_cache.remove_segments_chunks(chunks_to_be_removed)
media_cache.remove_context_images_chunks(context_image_chunks_to_be_removed)
db_segment.chunks_updated_date = timezone.now()
db_segment.save(update_fields=["chunks_updated_date"])
if updated_honeypots or (
# even if the randomly selected frames were the same as before, we should still
# consider it an update to the validation frames and restore them, if they were deleted
frame_selection_method
== models.JobFrameSelectionMethod.RANDOM_UNIFORM
):
# deleted frames that were updated in the job should be restored, as they are new now
if set(db_data.deleted_frames).intersection(updated_honeypots):
db_data.deleted_frames = sorted(
set(db_data.deleted_frames).difference(updated_honeypots)
)
db_data.save(update_fields=["deleted_frames"])
new_updated_date = timezone.now()
db_job.updated_date = new_updated_date
if bulk_context:
bulk_context.updated_segments.append(db_segment.id)
else:
db_segment.job_set.update(updated_date=new_updated_date)
db_task.touch()
if db_task.project:
db_task.project.touch()
return instance
def _clear_annotations_on_frames(self, segment: models.Segment, frames: Sequence[int]):
models.clear_annotations_on_frames_in_honeypot_task(segment.task, frames=frames)
@staticmethod
def _write_updated_static_chunk(
db_segment_id: int,
chunk_id: int,
chunk_frames: list[int],
quality: models.FrameQuality,
frame_path_map: dict[int, str],
segment_frame_map: dict[int, int],
):
from cvat.apps.engine.media_io.frame_provider import prepare_image_chunk
db_segment = models.Segment.objects.select_related("task").get(pk=db_segment_id)
initial_chunks_updated_date = db_segment.chunks_updated_date
db_task = db_segment.task
task_frame_provider = TaskFrameProvider(db_task)
db_data = db_task.require_data()
def _iterate_chunk_frames():
for chunk_frame in chunk_frames:
db_frame_path = frame_path_map[chunk_frame]
chunk_real_frame = segment_frame_map.get(chunk_frame, chunk_frame)
yield (
task_frame_provider.get_frame(chunk_real_frame, quality=quality).data,
os.path.basename(db_frame_path),
)
with closing(_iterate_chunk_frames()) as frame_iter:
chunk, _ = prepare_image_chunk(
frame_iter,
quality=quality,
db_task=db_task,
dump_unchanged=True,
)
db_segment.refresh_from_db(fields=["chunks_updated_date"])
if db_segment.chunks_updated_date > initial_chunks_updated_date:
raise CvatChunkTimestampMismatchError(
"Attempting to write an out of date static chunk, "
f"segment.chunks_updated_date: {db_segment.chunks_updated_date}, "
f"expected_ts: {initial_chunks_updated_date}"
)
chunk_path = db_data.get_static_segment_chunk_path(chunk_id, db_segment_id, quality)
with open(chunk_path, "wb") as f:
f.write(chunk.getvalue())
class JobValidationLayoutReadSerializer(serializers.Serializer):
honeypot_count = serializers.IntegerField(min_value=0, required=False)
honeypot_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
required=False,
help_text=textwrap.dedent("""\
The list of frame ids for honeypots in the job
"""),
)
honeypot_real_frames = serializers.ListSerializer(
child=serializers.IntegerField(min_value=0),
required=False,
help_text=textwrap.dedent("""\
The list of real (validation) frame ids for honeypots in the job
"""),
)
def to_representation(self, instance: models.Job):
validation_layout = getattr(instance.segment.task.data, "validation_layout", None)
if not validation_layout:
return {}
data = {}
if validation_layout.mode == models.ValidationMode.GT_POOL:
db_segment = instance.segment
segment_frame_set = db_segment.frame_set
db_data = db_segment.task.require_data()
frame_step = db_data.get_frame_step()
def _to_rel_frame(abs_frame: int) -> int:
return (abs_frame - db_data.start_frame) // frame_step
segment_honeypot_frames = []
for frame in db_segment.task.data.images.all():
if not frame.is_placeholder:
continue
if not frame.frame in segment_frame_set:
continue
segment_honeypot_frames.append(
(_to_rel_frame(frame.frame), _to_rel_frame(frame.real_frame))
)
segment_honeypot_frames.sort(key=lambda v: v[0])
data = {
"honeypot_count": len(segment_honeypot_frames),
"honeypot_frames": [v[0] for v in segment_honeypot_frames],
"honeypot_real_frames": [v[1] for v in segment_honeypot_frames],
}
return super().to_representation(data)
class _TaskValidationLayoutBulkUpdateContext:
def __init__(
self,
*,
all_db_frames: dict[int, models.Image],
honeypot_frames: list[int],
all_validation_frames: list[int],
active_validation_frames: list[int],
honeypot_frame_selector: HoneypotFrameSelector | None = None,
):
self.updated_honeypots: dict[int, models.Image] = {}
self.updated_segments: list[int] = []
self.chunks_to_be_removed: list[dict[str, Any]] = []
self.context_image_chunks_to_be_removed: list[dict[str, Any]] = []
self.segments_with_updated_chunks: list[int] = []
self.all_db_frames = all_db_frames
self.honeypot_frames = honeypot_frames
self.all_validation_frames = all_validation_frames
self.active_validation_frames = active_validation_frames
self.honeypot_frame_selector = honeypot_frame_selector
class TaskValidationLayoutWriteSerializer(serializers.Serializer):
disabled_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
required=False,
help_text=textwrap.dedent("""\
The list of frame ids to be excluded from validation
"""),
)
frame_selection_method = serializers.ChoiceField(
choices=models.JobFrameSelectionMethod.choices(),
required=False,
help_text=textwrap.dedent("""\
The method to use for frame selection of new real frames for honeypots in the task
"""),
)
honeypot_real_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
required=False,
help_text=textwrap.dedent("""\
The list of frame ids. Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.MANUAL)),
)
def validate(self, attrs):
frame_selection_method = attrs.get("frame_selection_method")
if frame_selection_method == models.JobFrameSelectionMethod.MANUAL:
field_validation.require_field(attrs, "honeypot_real_frames")
elif frame_selection_method == models.JobFrameSelectionMethod.RANDOM_UNIFORM:
pass
if (
"honeypot_real_frames" in attrs
and frame_selection_method != models.JobFrameSelectionMethod.MANUAL
):
raise serializers.ValidationError(
'"honeypot_real_frames" can only be used when '
f'"frame_selection_method" is "{models.JobFrameSelectionMethod.MANUAL}"'
)
return super().validate(attrs)
@transaction.atomic
def update(self, instance: models.Task, validated_data: dict[str, Any]) -> models.Task:
# FIXME: this operation is not atomic and it is not protected from race conditions
# (basically, as many others). Currently, it's up to the user to ensure no parallel
# calls happen. It also affects any image access, including exports with images, backups,
# automatic annotation, chunk downloading, etc.
db_validation_layout: models.ValidationLayout | None = getattr(
instance.data, "validation_layout", None
)
if not db_validation_layout:
raise serializers.ValidationError("Validation is not configured in the task")
if "disabled_frames" in validated_data:
requested_disabled_frames = validated_data["disabled_frames"]
unknown_requested_disabled_frames = set(requested_disabled_frames).difference(
db_validation_layout.frames
)
if unknown_requested_disabled_frames:
raise serializers.ValidationError(
"Unknown frames requested for exclusion from the validation set: {}".format(
format_list(tuple(map(str, sorted(unknown_requested_disabled_frames))))
)
)
gt_job_meta_serializer = JobDataMetaWriteSerializer(
instance.gt_job, {"deleted_frames": requested_disabled_frames}
)
gt_job_meta_serializer.is_valid(raise_exception=True)
gt_job_meta_serializer.save()
db_validation_layout.refresh_from_db()
instance.data.refresh_from_db()
frame_selection_method = validated_data.get("frame_selection_method")
if frame_selection_method and not (
db_validation_layout
and instance.data.validation_layout.mode == models.ValidationMode.GT_POOL
):
raise serializers.ValidationError(
"Honeypots can only be modified if the task "
f"validation mode is '{models.ValidationMode.GT_POOL}'"
)
if not frame_selection_method:
return instance
assert not hasattr(instance.data, "video")
# Populate the prefetch cache for required objects
prefetch_related_objects(
[instance],
Prefetch("data__images", queryset=models.Image.objects.order_by("frame")),
"segment_set",
"segment_set__job_set",
)
frame_provider = TaskFrameProvider(instance)
db_frames = {
frame_provider.get_rel_frame_number(db_image.frame): db_image
for db_image in instance.data.images.all()
}
honeypot_frames = sorted(f for f, v in db_frames.items() if v.is_placeholder)
all_validation_frames = db_validation_layout.frames
active_validation_frames = db_validation_layout.active_frames
bulk_context = _TaskValidationLayoutBulkUpdateContext(
all_db_frames=db_frames,
honeypot_frames=honeypot_frames,
all_validation_frames=all_validation_frames,
active_validation_frames=active_validation_frames,
)
if frame_selection_method == models.JobFrameSelectionMethod.MANUAL:
requested_honeypot_real_frames = validated_data["honeypot_real_frames"]
task_honeypot_frames_count = len(honeypot_frames)
if task_honeypot_frames_count != len(requested_honeypot_real_frames):
raise serializers.ValidationError(
"Invalid size of 'honeypot_real_frames' array, "
f"expected {task_honeypot_frames_count}"
)
elif frame_selection_method == models.JobFrameSelectionMethod.RANDOM_UNIFORM:
# Reset distribution for active validation frames
active_validation_frame_counts = {f: 0 for f in active_validation_frames}
frame_selector = HoneypotFrameSelector(active_validation_frame_counts)
bulk_context.honeypot_frame_selector = frame_selector
# Could be done using Django ORM, but using order_by() and filter()
# would result in an extra DB request
db_jobs = sorted(
(
db_job
for db_segment in instance.segment_set.all()
for db_job in db_segment.job_set.all()
if db_job.type == models.JobType.ANNOTATION
),
key=lambda j: j.segment.start_frame,
)
for db_job in db_jobs:
job_serializer_params = {"frame_selection_method": frame_selection_method}
if frame_selection_method == models.JobFrameSelectionMethod.MANUAL:
segment_frame_set = db_job.segment.frame_set
job_serializer_params["honeypot_real_frames"] = [
requested_frame
for rel_frame, requested_frame in zip(
honeypot_frames, requested_honeypot_real_frames
)
if frame_provider.get_abs_frame_number(rel_frame) in segment_frame_set
]
job_validation_layout_serializer = JobValidationLayoutWriteSerializer(
db_job, job_serializer_params, bulk_context=bulk_context
)
job_validation_layout_serializer.is_valid(raise_exception=True)
job_validation_layout_serializer.save()
self._perform_bulk_updates(instance, bulk_context=bulk_context)
return instance
def _perform_bulk_updates(
self,
db_task: models.Task,
*,
bulk_context: _TaskValidationLayoutBulkUpdateContext,
):
updated_segments = bulk_context.updated_segments
if not updated_segments:
return
self._update_frames_in_bulk(db_task, bulk_context=bulk_context)
# Import it here to avoid circular import
from cvat.apps.engine.cache import MediaCache
media_cache = MediaCache()
media_cache.remove_segments_chunks(bulk_context.chunks_to_be_removed)
media_cache.remove_context_images_chunks(bulk_context.context_image_chunks_to_be_removed)
# Update segments
updated_date = timezone.now()
for updated_segments_batch in take_by(updated_segments, chunk_size=1000):
models.Job.objects.filter(segment_id__in=updated_segments_batch).update(
updated_date=updated_date
)
for updated_segment_chunks_batch in take_by(
bulk_context.segments_with_updated_chunks, chunk_size=1000
):
models.Segment.objects.filter(id__in=updated_segment_chunks_batch).update(
chunks_updated_date=updated_date
)
# Update parent objects
db_task.touch()
if db_task.project:
db_task.project.touch()
def _update_frames_in_bulk(
self,
db_task: models.Task,
*,
bulk_context: _TaskValidationLayoutBulkUpdateContext,
):
self._clear_annotations_on_frames(db_task, bulk_context.updated_honeypots)
# The django generated bulk_update() query is too slow, so we use bulk_create() instead
# NOTE: Silk doesn't show these queries in the list of queries
# for some reason, but they can be seen in the profile
db_utils.bulk_create(
models.Image,
list(bulk_context.updated_honeypots.values()),
update_conflicts=True,
update_fields=["path", "width", "height", "real_frame"],
unique_fields=[
# required for Postgres
# https://docs.djangoproject.com/en/4.2/ref/models/querysets/#bulk-create
"id"
],
)
# Update related images in 2 steps: remove all m2m for honeypots, then add (copy) new ones
# 1. remove
for updated_honeypots_batch in take_by(
bulk_context.updated_honeypots.values(), chunk_size=1000
):
models.RelatedFile.images.through.objects.filter(
image_id__in=(db_honeypot.id for db_honeypot in updated_honeypots_batch)
).delete()
# 2. batched add (copy): collect all the new records and insert
frame_provider = TaskFrameProvider(db_task)
honeypots_by_validation_frame = grouped(
bulk_context.updated_honeypots,
key=lambda honeypot_frame: frame_provider.get_rel_frame_number(
bulk_context.updated_honeypots[honeypot_frame].real_frame
),
) # validation frame -> [honeypot_frame, ...]
new_m2m_objects = []
m2m_objects_by_validation_image_id = grouped(
models.RelatedFile.images.through.objects.filter(
image_id__in=(
bulk_context.all_db_frames[validation_frame].id
for validation_frame in honeypots_by_validation_frame
)
).all(),
key=lambda m2m_obj: m2m_obj.image_id,
)
for validation_frame, validation_frame_honeypots in honeypots_by_validation_frame.items():
validation_frame_m2m_objects = m2m_objects_by_validation_image_id.get(
bulk_context.all_db_frames[validation_frame].id
)
if not validation_frame_m2m_objects:
continue
# Copy validation frame m2m objects to corresponding honeypots
for honeypot_frame in validation_frame_honeypots:
new_m2m_objects.extend(
models.RelatedFile.images.through(
image_id=bulk_context.all_db_frames[honeypot_frame].id,
relatedfile_id=m2m_obj.relatedfile_id,
)
for m2m_obj in validation_frame_m2m_objects
)
db_utils.bulk_create(models.RelatedFile.images.through, new_m2m_objects)
# Update manifest if present
manifest_path = db_task.data.get_manifest_path()
if os.path.isfile(manifest_path):
manifest = ImageManifestManager(manifest_path)
manifest.reorder([db_frame.path for db_frame in bulk_context.all_db_frames.values()])
def _clear_annotations_on_frames(self, db_task: models.Task, frames: Sequence[int]):
models.clear_annotations_on_frames_in_honeypot_task(db_task, frames=frames)
class TaskValidationLayoutReadSerializer(serializers.ModelSerializer):
validation_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
source="frames",
required=False,
help_text=textwrap.dedent("""\
The list of frame ids to be used for validation
"""),
)
disabled_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
required=False,
help_text=textwrap.dedent("""\
The list of frame ids excluded from validation
"""),
)
honeypot_count = serializers.IntegerField(min_value=0, required=False)
honeypot_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
required=False,
help_text=textwrap.dedent("""\
The list of frame ids for all honeypots in the task
"""),
)
honeypot_real_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
required=False,
help_text=textwrap.dedent("""\
The list of real (validation) frame ids for all honeypots in the task
"""),
)
class Meta:
model = models.ValidationLayout
fields = (
"mode",
"frames_per_job_count",
"validation_frames",
"disabled_frames",
"honeypot_count",
"honeypot_frames",
"honeypot_real_frames",
)
read_only_fields = fields
extra_kwargs = {
"mode": {"allow_null": True},
}
def to_representation(self, instance: models.ValidationLayout):
if instance.mode == models.ValidationMode.GT_POOL:
db_data: models.Data = instance.task_data
frame_step = db_data.get_frame_step()
def _to_rel_frame(abs_frame: int) -> int:
return (abs_frame - db_data.start_frame) // frame_step
placeholder_queryset = models.Image.objects.filter(
data_id=instance.task_data_id, is_placeholder=True
)
honeypot_count = placeholder_queryset.count()
instance.honeypot_count = honeypot_count
# TODO: make this information optional, if there are use cases with too big responses
instance.honeypot_frames = []
instance.honeypot_real_frames = []
for frame, real_frame in (
placeholder_queryset.order_by("frame")
.values_list("frame", "real_frame")
.iterator(chunk_size=10000)
):
instance.honeypot_frames.append(_to_rel_frame(frame))
instance.honeypot_real_frames.append(_to_rel_frame(real_frame))
return super().to_representation(instance)
class SegmentSerializer(serializers.ModelSerializer):
jobs = SimpleJobSerializer(many=True, source="job_set")
frames = serializers.ListSerializer(child=serializers.IntegerField(), allow_empty=True)
class Meta:
model = models.Segment
fields = ("start_frame", "stop_frame", "jobs", "type", "frames")
read_only_fields = fields
def to_representation(self, instance: models.Segment):
data = super().to_representation(instance)
if instance.type == models.SegmentType.SPECIFIC_FRAMES:
task_frame_provider = TaskFrameProvider(instance.task)
data["frames"] = sorted(map(task_frame_provider.get_rel_frame_number, data["frames"]))
else:
data.pop("frames")
return data
class ClientFileSerializer(serializers.ModelSerializer):
class Meta:
model = models.ClientFile
fields = ("file",)
# pylint: disable=no-self-use
def to_internal_value(self, data):
return {"file": data}
# pylint: disable=no-self-use
def to_representation(self, instance):
if instance:
upload_dir = instance.data.get_upload_dirname()
return Path(instance.file.path).relative_to(upload_dir).as_posix()
else:
return instance
class ServerFileSerializer(serializers.ModelSerializer):
class Meta:
model = models.ServerFile
fields = ("file",)
extra_kwargs = {
"file": {"validators": [CanonicalRelativePathValidator(allow_trailing_slash=True)]}
}
def to_internal_value(self, data):
return super().to_internal_value({"file": data})
# pylint: disable=no-self-use
def to_representation(self, instance):
return instance.file if instance else instance
class RemoteFileSerializer(serializers.ModelSerializer):
class Meta:
model = models.RemoteFile
fields = ("file",)
# pylint: disable=no-self-use
def to_internal_value(self, data):
return {"file": data}
# pylint: disable=no-self-use
def to_representation(self, instance):
return instance.file if instance else instance
class RqStatusSerializer(serializers.Serializer):
state = serializers.ChoiceField(choices=["Queued", "Started", "Finished", "Failed"])
message = serializers.CharField(allow_blank=True, default="")
progress = serializers.FloatField(max_value=100, default=0)
def __init__(self, instance=None, data=..., **kwargs):
warnings.warn(
"RqStatusSerializer is deprecated, "
"use cvat.apps.redis_handler.serializers.RequestSerializer instead",
DeprecationWarning,
)
super().__init__(instance, data, **kwargs)
class JobFiles(serializers.ListField):
"""
Read JobFileMapping docs for more info.
"""
def __init__(self, *args, **kwargs):
kwargs.setdefault(
"child", serializers.CharField(allow_blank=False, max_length=MAX_FILENAME_LENGTH)
)
kwargs.setdefault("allow_empty", False)
super().__init__(*args, **kwargs)
class JobFileMapping(serializers.ListField):
"""
Represents a file-to-job mapping.
Useful to specify a custom job configuration during task creation.
This option is not compatible with most other job split-related options.
Files in the jobs must not overlap or repeat.
Job file mapping files must be a subset of the input files.
If directories are specified in server_files, all files obtained by recursive search
in the specified directories will be used as input files.
In case of missing items in the input files, an error will be raised.
Example:
[
["file1.jpg", "file2.jpg"], # job #1 files
["file3.png"], # job #2 files
["file4.jpg", "file5.png", "file6.bmp"], # job #3 files
]
"""
def __init__(self, *args, **kwargs):
kwargs.setdefault("child", JobFiles())
kwargs.setdefault("allow_empty", False)
kwargs.setdefault("help_text", textwrap.dedent(__class__.__doc__))
super().__init__(*args, **kwargs)
class ValidationParamsSerializer(serializers.ModelSerializer):
mode = serializers.ChoiceField(choices=models.ValidationMode.choices(), required=True)
frame_selection_method = serializers.ChoiceField(
choices=models.JobFrameSelectionMethod.choices(), required=False
)
frames = serializers.ListField(
write_only=True,
child=serializers.CharField(max_length=MAX_FILENAME_LENGTH),
required=False,
allow_empty=False,
help_text=textwrap.dedent("""\
The list of file names to be included in the validation set.
Applicable only to the "{}" frame selection method.
Can only be used for images.
""".format(models.JobFrameSelectionMethod.MANUAL)),
)
frame_count = serializers.IntegerField(
min_value=1,
required=False,
help_text=textwrap.dedent("""\
The number of frames to be included in the validation set.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_UNIFORM)),
)
frame_share = serializers.FloatField(
required=False,
validators=[field_validation.validate_share],
help_text=textwrap.dedent("""\
The share of frames to be included in the validation set.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_UNIFORM)),
)
frames_per_job_count = serializers.IntegerField(
min_value=1,
required=False,
help_text=textwrap.dedent("""\
The number of frames to be included in the validation set from each annotation job.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_PER_JOB)),
)
frames_per_job_share = serializers.FloatField(
required=False,
validators=[field_validation.validate_share],
help_text=textwrap.dedent("""\
The share of frames to be included in the validation set from each annotation job.
Applicable only to the "{}" frame selection method
""".format(models.JobFrameSelectionMethod.RANDOM_PER_JOB)),
)
random_seed = serializers.IntegerField(
min_value=0,
required=False,
help_text=textwrap.dedent("""\
The seed value for the random number generator.
The same value will produce the same frame sets.
Applicable only to random frame selection methods.
By default, a random value is used.
"""),
)
class Meta:
fields = (
"mode",
"frame_selection_method",
"random_seed",
"frames",
"frame_count",
"frame_share",
"frames_per_job_count",
"frames_per_job_share",
)
model = models.ValidationParams
def validate(self, attrs):
if attrs["mode"] == models.ValidationMode.GT:
if "frame_selection_method" in attrs:
field_validation.require_one_of_values(
attrs,
"frame_selection_method",
[
models.JobFrameSelectionMethod.MANUAL,
models.JobFrameSelectionMethod.RANDOM_UNIFORM,
models.JobFrameSelectionMethod.RANDOM_PER_JOB,
],
)
elif attrs["mode"] == models.ValidationMode.GT_POOL:
field_validation.require_one_of_values(
attrs,
"frame_selection_method",
[
models.JobFrameSelectionMethod.MANUAL,
models.JobFrameSelectionMethod.RANDOM_UNIFORM,
],
)
field_validation.require_one_of_fields(
attrs, ["frames_per_job_count", "frames_per_job_share"]
)
else:
assert False, f"Unknown validation mode {attrs['mode']}"
if "frame_selection_method" not in attrs:
return super().validate(attrs)
if attrs["frame_selection_method"] == models.JobFrameSelectionMethod.RANDOM_UNIFORM:
field_validation.require_one_of_fields(attrs, ["frame_count", "frame_share"])
elif attrs["frame_selection_method"] == models.JobFrameSelectionMethod.RANDOM_PER_JOB:
field_validation.require_one_of_fields(
attrs, ["frames_per_job_count", "frames_per_job_share"]
)
elif attrs["frame_selection_method"] == models.JobFrameSelectionMethod.MANUAL:
field_validation.require_field(attrs, "frames")
if (
"frames" in attrs
and attrs["frame_selection_method"] != models.JobFrameSelectionMethod.MANUAL
):
raise serializers.ValidationError(
'"frames" can only be used when "frame_selection_method" is "{}"'.format(
models.JobFrameSelectionMethod.MANUAL
)
)
if frames := attrs.get("frames"):
unique_frames = set(frames)
if len(unique_frames) != len(frames):
raise serializers.ValidationError("Frames must not repeat")
return super().validate(attrs)
@transaction.atomic
def create(self, validated_data: dict[str, Any]) -> models.ValidationParams:
frames = validated_data.pop("frames", None)
instance = super().create(validated_data)
if frames:
db_utils.bulk_create(
models.ValidationFrame,
[
models.ValidationFrame(validation_params=instance, path=frame)
for frame in frames
],
)
return instance
@transaction.atomic
def update(
self, instance: models.ValidationParams, validated_data: dict[str, Any]
) -> models.ValidationParams:
frames = validated_data.pop("frames", None)
instance = super().update(instance, validated_data)
if frames:
models.ValidationFrame.objects.filter(validation_params=instance).delete()
db_utils.bulk_create(
models.ValidationFrame,
[
models.ValidationFrame(validation_params=instance, path=frame)
for frame in frames
],
)
return instance
class DataSerializer(serializers.ModelSerializer):
"""
Read more about parameters here:
https://docs.cvat.ai/docs/manual/basics/create-annotation-task/#advanced-configuration
"""
image_quality = serializers.IntegerField(
min_value=1,
max_value=100,
required=False,
help_text="Image quality to use during annotation, required for image and video-based tasks",
)
use_zip_chunks = serializers.BooleanField(
default=False,
help_text=textwrap.dedent("""\
When true, video chunks will be represented as zip archives with decoded video frames.
When false, video chunks are represented as video segments
"""),
)
client_files = ClientFileSerializer(
many=True,
default=[],
help_text=textwrap.dedent("""
Uploaded files.
Must contain all files from job_file_mapping if job_file_mapping is not empty.
"""),
)
server_files = ServerFileSerializer(
many=True,
default=[],
help_text=textwrap.dedent("""
Paths to files from a file share mounted on the server, or from a cloud storage.
Must contain all files from job_file_mapping if job_file_mapping is not empty.
"""),
)
server_files_exclude = serializers.ListField(
required=False,
default=[],
child=serializers.CharField(
max_length=MAX_FILENAME_LENGTH,
validators=[CanonicalRelativePathValidator(allow_trailing_slash=True)],
),
help_text=textwrap.dedent("""\
Paths to files and directories from a file share mounted on the server, or from a cloud storage
that should be excluded from the directories specified in server_files.
This option cannot be used together with filename_pattern.
The server_files_exclude parameter cannot be used to exclude a part of dataset from an archive.
Examples:
Exclude all files from subfolder 'sub/sub_1/sub_2'and single file 'sub/image.jpg' from specified folder:
server_files = ['sub/'], server_files_exclude = ['sub/sub_1/sub_2/', 'sub/image.jpg']
Exclude all cloud storage files with prefix 'sub' from the content of manifest file:
server_files = ['manifest.jsonl'], server_files_exclude = ['sub/']
"""),
)
remote_files = RemoteFileSerializer(
many=True,
default=[],
help_text=textwrap.dedent("""
Direct download URLs for files.
Must contain all files from job_file_mapping if job_file_mapping is not empty.
"""),
)
use_cache = serializers.BooleanField(
default=False,
help_text=textwrap.dedent("""\
Enable or disable task data chunk caching for the task.
Read more: https://docs.cvat.ai/docs/manual/advanced/data_on_fly/
"""),
)
copy_data = serializers.BooleanField(
default=False,
help_text=textwrap.dedent("""\
Copy data from the server file share to CVAT during the task creation.
This will create a copy of the data, making the server independent from
the file share availability
"""),
)
cloud_storage_id = serializers.IntegerField(
write_only=True,
allow_null=True,
required=False,
help_text=textwrap.dedent("""\
If not null, the files referenced by server_files will be retrieved
from the cloud storage with the specified ID.
The cloud storages applicable depend on the context.
In the user sandbox, only the user sandbox cloud storages can be used.
In an organization, only the organization cloud storages can be used.
"""),
)
filename_pattern = serializers.CharField(
allow_null=True,
required=False,
help_text=textwrap.dedent("""\
A filename filter for cloud storage files
listed in the manifest. Supports fnmatch wildcards.
Read more: https://docs.python.org/3/library/fnmatch.html
"""),
)
job_file_mapping = JobFileMapping(required=False, write_only=True)
upload_file_order = serializers.ListField(
child=serializers.CharField(max_length=MAX_FILENAME_LENGTH),
default=list,
allow_empty=True,
write_only=True,
help_text=textwrap.dedent("""\
Allows to specify file order for client_file uploads.
Only valid with the "{}" sorting method selected.
To state that the input files are sent in the correct order,
pass an empty list.
If you want to send files in an arbitrary order
and reorder them afterwards on the server,
pass the list of file names in the required order.
""".format(models.SortingMethod.PREDEFINED)),
)
validation_params = ValidationParamsSerializer(allow_null=True, required=False)
class Meta:
model = models.Data
fields = (
"chunk_size",
"image_quality",
"start_frame",
"stop_frame",
"frame_filter",
"client_files",
"server_files",
"remote_files",
"use_zip_chunks",
"server_files_exclude",
"cloud_storage_id",
"use_cache",
"copy_data",
"storage_method",
"sorting_method",
"filename_pattern",
"job_file_mapping",
"upload_file_order",
"validation_params",
)
extra_kwargs = {
"chunk_size": {"help_text": "Maximum number of frames per chunk"},
"start_frame": {"help_text": "First frame index"},
"stop_frame": {"help_text": "Last frame index"},
"frame_filter": {"help_text": "Frame filter. The only supported syntax is: 'step=N'"},
}
def __init__(self, *args, **kwargs):
kwargs.setdefault("help_text", self.__doc__)
super().__init__(*args, **kwargs)
# pylint: disable=no-self-use
def validate_frame_filter(self, value):
match = re.search(r"step\s*=\s*([1-9]\d*)", value)
if not match:
raise serializers.ValidationError("Invalid frame filter expression")
return value
# pylint: disable=no-self-use
def validate_chunk_size(self, value):
if not value > 0:
raise serializers.ValidationError("Chunk size must be a positive integer")
return value
def validate_job_file_mapping(self, value):
existing_files = set()
for job_files in value:
for filename in job_files:
if filename in existing_files:
raise serializers.ValidationError(
f"The same file '{filename}' cannot be used multiple "
"times in the job file mapping"
)
existing_files.add(filename)
return value
def validate_remote_files(self, value: list[dict[str, str]]) -> list[dict[str, str]]:
errors: list[str] = []
for entry in value:
url = entry["file"]
parsed = urlparse(url)
if parsed.scheme not in ("http", "https") or not parsed.netloc:
errors.append(f"{url!r}: remote_files entries must be http(s) URLs.")
if errors:
raise serializers.ValidationError(errors)
return value
# pylint: disable=no-self-use
def validate(self, attrs):
if (
"start_frame" in attrs
and "stop_frame" in attrs
and attrs["start_frame"] > attrs["stop_frame"]
):
raise serializers.ValidationError("Stop frame must be more or equal start frame")
filename_pattern = attrs.get("filename_pattern")
server_files_exclude = attrs.get("server_files_exclude")
server_files = attrs.get("server_files", [])
if filename_pattern and len(
list(filter(lambda x: not x["file"].endswith(".jsonl"), server_files))
):
raise serializers.ValidationError(
"The filename_pattern can only be used with specified manifest or without server_files"
)
if filename_pattern and server_files_exclude:
raise serializers.ValidationError(
"The filename_pattern and server_files_exclude cannot be used together"
)
validation_params = attrs.pop("validation_params", None)
if validation_params:
validation_params_serializer = ValidationParamsSerializer(data=validation_params)
validation_params_serializer.is_valid(raise_exception=True)
attrs["validation_params"] = validation_params_serializer.validated_data
return attrs
@transaction.atomic
def create(self, validated_data):
files = self._pop_data(validated_data)
validation_params = validated_data.pop("validation_params", None)
db_data = models.Data.objects.create(**validated_data)
db_data.make_dirs()
self._create_files(db_data, files)
db_data.save()
if validation_params:
validation_params_serializer = ValidationParamsSerializer(data=validation_params)
validation_params_serializer.is_valid(raise_exception=True)
db_data.validation_params = validation_params_serializer.save(task_data=db_data)
return db_data
@transaction.atomic
def update(self, instance, validated_data):
validation_params = validated_data.pop("validation_params", None)
files = self._pop_data(validated_data)
for key, value in validated_data.items():
setattr(instance, key, value)
self._create_files(instance, files)
instance.save()
if validation_params:
validation_params_serializer = ValidationParamsSerializer(
instance=getattr(instance, "validation_params", None), data=validation_params
)
validation_params_serializer.is_valid(raise_exception=True)
instance.validation_params = validation_params_serializer.save(task_data=instance)
return instance
# pylint: disable=no-self-use
def _pop_data(self, validated_data):
client_files = validated_data.pop("client_files")
server_files = validated_data.pop("server_files")
remote_files = validated_data.pop("remote_files")
validated_data.pop("job_file_mapping", None) # optional, not present in Data
validated_data.pop("upload_file_order", None) # optional, not present in Data
validated_data.pop("server_files_exclude", None) # optional, not present in Data
for extra_key in {"use_zip_chunks", "use_cache", "copy_data"}:
validated_data.pop(extra_key)
files = {
"client_files": client_files,
"server_files": server_files,
"remote_files": remote_files,
}
return files
# pylint: disable=no-self-use
@transaction.atomic
def _create_files(self, instance, files):
for files_type, files_model in zip(
("client_files", "server_files", "remote_files"),
(models.ClientFile, models.ServerFile, models.RemoteFile),
):
if files_type in files:
db_utils.bulk_create(
files_model, [files_model(data=instance, **f) for f in files[files_type]]
)
class TaskReadListSerializer(serializers.ListSerializer):
def to_representation(self, data):
if (request := self.context.get("request")) and isinstance(data, list) and data:
# Optimized prefetch only for the current page
page: list[models.Task] = data
# Annotate page objects
# We do it explicitly here and not in the LIST queryset to avoid
# doing the same DB computations twice - one time for the page retrieval
# and another one for the COUNT(*) request to get the total count
page_task_ids = set(t.id for t in page)
job_summary_fields = [m.value for m in models.TaskQuerySet.JobSummaryFields]
job_counts = {
task["id"]: task
for task in models.Task.objects.filter(id__in=page_task_ids)
.with_job_summary()
.values("id", *job_summary_fields)
}
# Prefetch visible related objects
# This avoids N+1 queries when serializing
page_project_ids = set(task.project_id for task in page)
visible_projects_perm = ProjectPermission.create_scope_list(request)
visible_projects_queryset = models.Project.objects.filter(id__in=page_project_ids)
visible_projects = {
p.id: p
for p in visible_projects_perm.filter(visible_projects_queryset).only("id", "name")
}
page_storage_ids = set(
v for task in page for v in (task.source_storage_id, task.target_storage_id)
)
storages = {s.id: s for s in models.Storage.objects.filter(id__in=page_storage_ids)}
# Keep in mind that the object ids fetched in the earlier queries
# might be missing in the later queries because of locks and removals,
# so should not be expected to be present and should be checked before access.
for task in page:
if task.project_id:
task.user_can_view_project = task.project_id in visible_projects
if task.user_can_view_project:
task.project = visible_projects[task.project_id]
task_job_summary = job_counts.get(task.id)
for k in job_summary_fields:
setattr(task, k, task_job_summary[k])
if task.source_storage_id in storages:
task.source_storage = storages[task.source_storage_id]
if task.target_storage_id in storages:
task.target_storage = storages[task.target_storage_id]
prefetch_related_objects(
page,
"data",
Prefetch(
"data__validation_layout",
queryset=models.ValidationLayout.objects.only("id", "task_data_id", "mode"),
),
"annotation_guide",
)
return super().to_representation(data)
@extend_schema_serializer(deprecate_fields=["organization"])
class TaskReadSerializer(serializers.ModelSerializer):
data_chunk_size = serializers.ReadOnlyField(source="data.chunk_size", required=False)
data_compressed_chunk_type = serializers.ChoiceField(
source="data.compressed_chunk_type",
choices=models.DataChoice.choices(),
required=False,
allow_blank=False,
read_only=True,
)
data_original_chunk_type = serializers.ChoiceField(
source="data.original_chunk_type",
choices=models.DataChoice.choices(),
required=False,
allow_blank=False,
read_only=True,
)
data_cloud_storage_id = serializers.ReadOnlyField(
source="data.cloud_storage_id", required=False
)
size = serializers.ReadOnlyField(source="data.size", required=False)
image_quality = serializers.ReadOnlyField(source="data.image_quality", required=False)
data = serializers.ReadOnlyField(source="data_id", required=False)
owner = BasicUserSerializer(required=False, allow_null=True)
assignee = BasicUserSerializer(allow_null=True, required=False)
project_id = serializers.IntegerField(required=False, allow_null=True)
project_name = serializers.SerializerMethodField()
guide_id = serializers.IntegerField(
source="annotation_guide.id", required=False, allow_null=True
)
organization_id = serializers.IntegerField(required=False, read_only=True, allow_null=True)
# We're using CharField to produce simple strings instead of enums in the generated SDK.
# SDK enums require explicit .value calls to access the string representation.
# TODO: move to ChoicesField when SDK supports seamless transition from string to enum
dimension = serializers.CharField(allow_blank=True, required=False, read_only=True)
mode = serializers.CharField(allow_blank=True, required=False, read_only=True)
media_type = serializers.CharField(allow_blank=True, required=False, read_only=True)
target_storage = StorageSerializer(required=False, allow_null=True)
source_storage = StorageSerializer(required=False, allow_null=True)
jobs = JobsSummarySerializer(url_filter_key="task_id", source="segment_set")
labels = LabelsSummarySerializer(source="*")
validation_mode = serializers.CharField(
source="data.validation_mode",
required=False,
allow_null=True,
help_text="Describes how the task validation is performed. Configured at task creation",
)
consensus_enabled = serializers.BooleanField(
source="get_consensus_enabled", required=False, read_only=True
)
class Meta:
model = models.Task
fields = (
"url",
"id",
"name",
"project_id",
"project_name",
"owner",
"assignee",
"bug_tracker",
"created_date",
"updated_date",
"overlap",
"segment_size",
"status",
"data_chunk_size",
"data_original_chunk_type",
"data_compressed_chunk_type",
"data_cloud_storage_id",
"guide_id",
"size",
"image_quality",
"data",
"dimension",
"mode",
"media_type",
"subset",
"organization_id",
"organization", # deprecated field
"target_storage",
"source_storage",
"jobs",
"labels",
"assignee_updated_date",
"validation_mode",
"consensus_enabled",
)
read_only_fields = fields
extra_kwargs = {
"organization": {"allow_null": True},
"overlap": {"allow_null": True},
}
list_serializer_class = TaskReadListSerializer
def get_consensus_enabled(self, instance: models.Task) -> bool:
return instance.consensus_replicas > 0
def _can_see_project(self, instance: models.Task) -> bool:
request = self.context.get("request")
if not request:
return False
can_see_project = getattr(instance, "user_can_view_project", None)
if can_see_project is None:
if not instance.project_id:
return False
perm = ProjectPermission.create_scope_view(request, instance.project)
can_see_project = perm.check_access().allow
return can_see_project
@extend_schema_field(serializers.CharField(allow_null=True))
def get_project_name(self, instance: models.Task) -> str | None:
return instance.project.name if self._can_see_project(instance) else None
def to_representation(self, instance):
representation = super().to_representation(instance)
representation["consensus_enabled"] = self.get_consensus_enabled(instance)
if instance.media_type not in (
models.MediaType.IMAGE,
# TODO: deprecated for 3d, remove later
models.MediaType.POINT_CLOUD,
):
representation.pop("image_quality", None)
if not instance.media_type:
representation.pop("data_compressed_chunk_type", None)
representation.pop("data_original_chunk_type", None)
representation.pop("data_chunk_size", None)
return representation
class TaskWriteSerializer(WriteOnceMixin, serializers.ModelSerializer, OrgTransferableMixin):
labels = LabelSerializer(many=True, source="label_set", partial=True, required=False)
owner_id = serializers.IntegerField(write_only=True, allow_null=True, required=False)
assignee_id = serializers.IntegerField(write_only=True, allow_null=True, required=False)
project_id = serializers.IntegerField(required=False, allow_null=True)
organization_id = serializers.IntegerField(write_only=True, required=False, allow_null=True)
target_storage = StorageSerializer(required=False, allow_null=False)
source_storage = StorageSerializer(required=False, allow_null=False)
consensus_replicas = serializers.IntegerField(
required=False,
default=0,
min_value=0,
help_text=textwrap.dedent("""\
The number of consensus replica jobs for each annotation job.
Configured at task creation
"""),
)
class Meta:
model = models.Task
fields = (
"url",
"id",
"name",
"project_id",
"owner_id",
"assignee_id",
"bug_tracker",
"overlap",
"segment_size",
"labels",
"subset",
"target_storage",
"source_storage",
"consensus_replicas",
"organization_id",
)
write_once_fields = ("overlap", "segment_size", "consensus_replicas")
update_only_fields = ("organization_id",)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if getattr(self.context.get("view"), "action", "") == "create":
for field in self.Meta.update_only_fields:
self.fields.pop(field)
def to_representation(self, instance):
serializer = TaskReadSerializer(instance, context=self.context)
return serializer.data
def validate_consensus_replicas(self, value):
max_replicas = settings.MAX_CONSENSUS_REPLICAS
if value and (value == 1 or value < 0 or value > max_replicas):
raise serializers.ValidationError(
f"Consensus replicas must be 0 "
f"or a positive number more than 1 and less than {max_replicas + 1}, "
f"got {value}"
)
return value or 0
# pylint: disable=no-self-use
@transaction.atomic
def create(self, validated_data):
project_id = validated_data.get("project_id")
if validated_data.get("label_set") and project_id:
raise serializers.ValidationError("Task must have only one of Label set or project_id")
project = None
if project_id:
try:
project = models.Project.objects.get(id=project_id)
except models.Project.DoesNotExist:
raise serializers.ValidationError(
f"The specified project #{project_id} does not exist."
)
if project.organization != validated_data.get("organization"):
raise serializers.ValidationError(
"The task and its project should be in the same organization."
)
labels = validated_data.pop("label_set", [])
# configure source/target storages for import/export
storages = _configure_related_storages(
{
"source_storage": validated_data.pop("source_storage", None),
"target_storage": validated_data.pop("target_storage", None),
}
)
db_task = models.Task.objects.create(**storages, **validated_data)
task_path = db_task.get_dirname()
if os.path.isdir(task_path):
shutil.rmtree(task_path)
os.makedirs(task_path)
LabelSerializer.create_labels(labels, parent_instance=db_task)
if validated_data.get("assignee_id"):
db_task.assignee_updated_date = db_task.updated_date
db_task.save(update_fields=["assignee_updated_date"])
return db_task
@transaction.atomic
def update_base_properties(
self,
instance: models.Task,
*,
validated_data: dict[str, Any],
update_fields: list[str],
):
for field_name in ("name", "bug_tracker", "subset", "owner_id", "assignee_id"):
if field_name in validated_data and (
field_value := validated_data[field_name]
) != getattr(instance, field_name):
if field_name != "assignee_id":
setattr(instance, field_name, field_value)
else:
instance.update_assignee(field_value)
update_fields.append(field_name)
def update_labels(
self,
instance: models.Task,
*,
validated_data: dict[str, Any],
update_fields: list[str],
):
if (labels := validated_data.get("label_set")) and instance.project_id is None:
LabelSerializer.update_labels(labels, parent_instance=instance)
update_fields.append("updated_date")
def update_project(
self,
instance: models.Task,
*,
validated_data: dict[str, Any],
update_fields: list[str],
):
labels = validated_data.get("label_set", [])
validated_project_id = validated_data["project_id"]
project = models.Project.objects.get(id=validated_project_id)
if project.tasks.count() and project.tasks.first().dimension != instance.dimension:
raise serializers.ValidationError(
f"Dimension ({instance.dimension}) of the task must be the same as other tasks in project ({project.tasks.first().dimension})"
)
if instance.project_id is None:
label_set = instance.label_set.all()
else:
label_set = instance.project.label_set.all()
for old_label in label_set:
new_label_for_name = list(filter(lambda x: x.get("id", None) == old_label.id, labels))
if len(new_label_for_name):
old_label.name = new_label_for_name[0].get("name", old_label.name)
try:
if old_label.parent:
new_label = project.label_set.filter(
name=old_label.name, parent__name=old_label.parent.name
).first()
else:
new_label = project.label_set.filter(name=old_label.name).first()
except ValueError:
raise serializers.ValidationError(
f'Target project does not have label with name "{old_label.name}"'
)
for old_attr in old_label.attributespec_set.all():
new_attr = new_label.attributespec_set.filter(
name=old_attr.name, values=old_attr.values, input_type=old_attr.input_type
).first()
if new_attr is None:
raise serializers.ValidationError(
"Target project does not have "
f'"{old_label.name}" label with "{old_attr.name}" attribute'
)
for model, model_name in (
(models.LabeledTrackAttributeVal, "track"),
(models.LabeledShapeAttributeVal, "shape"),
(models.LabeledImageAttributeVal, "image"),
(models.TrackedShapeAttributeVal, "shape__track"),
):
model.objects.filter(
**{
f"{model_name}__job__segment__task": instance,
f"{model_name}__label": old_label,
"spec": old_attr,
}
).update(spec=new_attr)
for model in (models.LabeledTrack, models.LabeledShape, models.LabeledImage):
model.objects.filter(job__segment__task=instance, label=old_label).update(
label=new_label
)
if instance.project_id is None:
instance.label_set.all().delete()
instance.project = project
update_fields.append("project")
# pylint: disable=no-self-use
@transaction.atomic
def update(self, instance: models.Task, validated_data: dict):
update_fields: list[str] = []
if (
"organization_id" in validated_data
and validated_data.get("organization_id") != instance.organization_id
):
self.update_organization(
instance, validated_data=validated_data, update_fields=update_fields
)
elif (
validated_project_id := validated_data.get("project_id")
) is not None and validated_project_id != instance.project_id:
self.update_project(
instance, validated_data=validated_data, update_fields=update_fields
)
else:
self.update_base_properties(
instance, validated_data=validated_data, update_fields=update_fields
)
self.update_labels(instance, validated_data=validated_data, update_fields=update_fields)
# update source and target storages
_update_related_storages(instance, validated_data=validated_data)
if update_fields:
instance.save(update_fields=list(set(update_fields) | {"updated_date"}))
if "label_set" in validated_data and not instance.project_id:
self.update_child_objects_on_labels_update(instance)
return instance
def update_child_objects_on_labels_update(self, instance: models.Task):
models.Job.objects.filter(
updated_date__lt=instance.updated_date, segment__task=instance
).update(updated_date=instance.updated_date)
def update_child_objects_on_organization_update(
self,
instance: models.Task,
*,
organization_id: int | None,
owner_id: int,
updated_date: datetime,
):
if instance.data.cloud_storage_id:
instance.data.cloud_storage = None
instance.data.save(update_fields=["cloud_storage"])
models.Job.objects.filter(segment__task__id=instance.pk, assignee__isnull=False).update(
assignee=None, assignee_updated_date=updated_date
)
def _validate_org_transferring(self, attrs: dict[str, Any]):
if "project_id" in attrs.keys():
raise serializers.ValidationError(
"A task cannot be moved into a project and into an organization at the same time"
)
elif self.instance.project_id:
raise serializers.ValidationError(
"Only top-level resources can be moved between workspaces"
)
super()._validate_org_transferring(attrs)
def validate(self, attrs):
if self.instance and "organization_id" in attrs.keys():
self._validate_org_transferring(attrs)
# When moving task labels can be mapped to one, but when not names must be unique
if "project_id" in attrs.keys() and self.instance is not None:
project_id = attrs.get("project_id")
if project_id is not None:
project = models.Project.objects.filter(id=project_id).first()
if project is None:
raise serializers.ValidationError(f"Cannot find project with ID {project_id}")
# Check that all labels can be mapped
new_label_names = set()
old_labels = (
self.instance.project.label_set.all()
if self.instance.project_id
else self.instance.label_set.all()
)
new_sublabel_names = {}
for old_label in old_labels:
new_labels = tuple(
filter(lambda x: x.get("id") == old_label.id, attrs.get("label_set", []))
)
if len(new_labels):
parent = new_labels[0].get("parent", old_label.parent)
if parent:
if parent.name not in new_sublabel_names:
new_sublabel_names[parent.name] = set()
new_sublabel_names[parent.name].add(
new_labels[0].get("name", old_label.name)
)
else:
new_label_names.add(new_labels[0].get("name", old_label.name))
else:
parent = old_label.parent
if parent:
if parent.name not in new_sublabel_names:
new_sublabel_names[parent.name] = set()
new_sublabel_names[parent.name].add(old_label.name)
else:
new_label_names.add(old_label.name)
target_project = models.Project.objects.get(id=project_id)
target_project_label_names = set()
target_project_sublabel_names = {}
for label in target_project.label_set.all():
parent = label.parent
if parent:
if parent.name not in target_project_sublabel_names:
target_project_sublabel_names[parent.name] = set()
target_project_sublabel_names[parent.name].add(label.name)
else:
target_project_label_names.add(label.name)
if not new_label_names.issubset(target_project_label_names):
raise serializers.ValidationError(
"All task or project label names must be mapped to the target project"
)
for label, sublabels in new_sublabel_names.items():
if sublabels != target_project_sublabel_names.get(label):
raise serializers.ValidationError(
"All task or project label names must be mapped to the target project"
)
return attrs
@extend_schema_serializer(deprecate_fields=["organization"])
class ProjectReadSerializer(serializers.ModelSerializer):
owner = BasicUserSerializer(allow_null=True, required=False, read_only=True)
assignee = BasicUserSerializer(allow_null=True, required=False, read_only=True)
guide_id = serializers.IntegerField(
source="annotation_guide.id", required=False, allow_null=True
)
organization_id = serializers.IntegerField(
source="organization.id", required=False, read_only=True, allow_null=True
)
task_subsets = serializers.ListField(
child=serializers.CharField(), required=False, read_only=True
)
dimension = serializers.CharField(required=False, allow_null=True, read_only=True)
target_storage = StorageSerializer(required=False, allow_null=True, read_only=True)
source_storage = StorageSerializer(required=False, allow_null=True, read_only=True)
tasks = TasksSummarySerializer(models.Task, url_filter_key="project_id")
labels = LabelsSummarySerializer(source="*")
class Meta:
model = models.Project
fields = (
"url",
"id",
"name",
"owner",
"assignee",
"guide_id",
"bug_tracker",
"task_subsets",
"created_date",
"updated_date",
"status",
"dimension",
"organization", # deprecated field
"organization_id",
"target_storage",
"source_storage",
"tasks",
"labels",
"assignee_updated_date",
)
read_only_fields = fields
extra_kwargs = {"organization": {"allow_null": True}}
def to_representation(self, instance):
response = super().to_representation(instance)
task_subsets = {task.subset for task in instance.tasks.all() if task.subset}
task_dimension = next(
(task.dimension for task in instance.tasks.all() if task.dimension),
None, # backward compatibility; TODO: migrate to "" for consistency with tasks
)
response["task_subsets"] = list(task_subsets)
response["dimension"] = task_dimension
return response
class ProjectWriteSerializer(serializers.ModelSerializer, OrgTransferableMixin):
labels = LabelSerializer(
write_only=True, many=True, source="label_set", partial=True, default=[]
)
owner_id = serializers.IntegerField(write_only=True, allow_null=True, required=False)
assignee_id = serializers.IntegerField(write_only=True, allow_null=True, required=False)
organization_id = serializers.IntegerField(write_only=True, allow_null=True, required=False)
target_storage = StorageSerializer(write_only=True, required=False)
source_storage = StorageSerializer(write_only=True, required=False)
class Meta:
model = models.Project
fields = (
"name",
"labels",
"owner_id",
"assignee_id",
"bug_tracker",
"target_storage",
"source_storage",
"organization_id",
)
update_only_fields = ("organization_id",)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if not self.partial:
for field in self.Meta.update_only_fields:
self.fields.pop(field)
def to_representation(self, instance):
serializer = ProjectReadSerializer(instance, context=self.context)
return serializer.data
def validate(self, attrs):
if self.instance and "organization_id" in attrs.keys():
self._validate_org_transferring(attrs)
return attrs
# pylint: disable=no-self-use
@transaction.atomic
def create(self, validated_data):
labels = validated_data.pop("label_set")
# configure source/target storages for import/export
storages = _configure_related_storages(
{
"source_storage": validated_data.pop("source_storage", None),
"target_storage": validated_data.pop("target_storage", None),
}
)
db_project = models.Project.objects.create(**storages, **validated_data)
project_path = db_project.get_dirname()
if os.path.isdir(project_path):
shutil.rmtree(project_path)
os.makedirs(project_path)
LabelSerializer.create_labels(labels, parent_instance=db_project)
if validated_data.get("assignee_id"):
db_project.assignee_updated_date = db_project.updated_date
db_project.save(update_fields=["assignee_updated_date"])
return db_project
# FUTURE-TODO: move into a common interface
def update_base_properties(
self,
instance: models.Project,
*,
validated_data: dict[str, Any],
update_fields: list[str],
):
for field_name in ("name", "bug_tracker", "owner_id", "assignee_id"):
if field_name in validated_data and (
field_value := validated_data[field_name]
) != getattr(instance, field_name):
if field_name != "assignee_id":
setattr(instance, field_name, field_value)
else:
instance.update_assignee(field_value)
update_fields.append(field_name)
# pylint: disable=no-self-use
@transaction.atomic
def update(self, instance: models.Project, validated_data: dict):
update_fields: list[str] = []
if (
"organization_id" in validated_data
and validated_data.get("organization_id") != instance.organization_id
):
self.update_organization(
instance, validated_data=validated_data, update_fields=update_fields
)
else:
self.update_base_properties(
instance, validated_data=validated_data, update_fields=update_fields
)
if labels := validated_data.get("label_set"):
LabelSerializer.update_labels(labels, parent_instance=instance)
update_fields.append("updated_date")
# update source and target storages
_update_related_storages(instance, validated_data=validated_data)
if update_fields:
instance.save(update_fields=list(set(update_fields) | {"updated_date"}))
if "label_set" in validated_data:
self.update_child_objects_on_labels_update(instance)
return instance
@transaction.atomic
def update_child_objects_on_labels_update(self, instance: models.Project):
models.Task.objects.filter(updated_date__lt=instance.updated_date, project=instance).update(
updated_date=instance.updated_date
)
models.Job.objects.filter(
updated_date__lt=instance.updated_date, segment__task__project=instance
).update(updated_date=instance.updated_date)
def update_child_objects_on_organization_update(
self,
instance: models.Project,
*,
organization_id: int | None,
owner_id: int,
updated_date: datetime,
):
models.Data.objects.filter(
id__in=models.Task.objects.filter(project=instance).values("data_id"),
).update(cloud_storage_id=None)
instance.tasks.update(
organization_id=organization_id,
owner_id=owner_id,
assignee_id=None,
assignee_updated_date=updated_date,
)
models.Job.objects.filter(
segment__task__project_id=instance.pk, assignee__isnull=False
).update(assignee=None, assignee_updated_date=updated_date)
Webhook.objects.filter(project_id=instance.pk).update(organization_id=organization_id)
class AboutSerializer(serializers.Serializer):
name = serializers.CharField(max_length=128)
description = serializers.CharField(max_length=2048)
version = serializers.CharField(max_length=64)
logo_url = serializers.CharField()
subtitle = serializers.CharField(max_length=1024)
class FrameMetaSerializer(serializers.Serializer):
width = serializers.IntegerField(required=False)
height = serializers.IntegerField(required=False)
name = serializers.CharField(max_length=MAX_FILENAME_LENGTH)
related_files = serializers.IntegerField()
# for compatibility with version 2.3.0
has_related_context = serializers.SerializerMethodField()
@extend_schema_field(serializers.BooleanField)
def get_has_related_context(self, obj: dict) -> bool:
return obj["related_files"] != 0
class ChapterMetadataSerializer(serializers.Serializer):
title = serializers.CharField(required=False)
class ChapterSerializer(serializers.Serializer):
id = serializers.IntegerField()
start = serializers.IntegerField()
stop = serializers.IntegerField()
metadata = ChapterMetadataSerializer(many=False)
class PluginsSerializer(serializers.Serializer):
ANALYTICS = serializers.BooleanField()
MODELS = serializers.BooleanField()
class DataMetaReadSerializer(serializers.ModelSerializer):
frames = FrameMetaSerializer(many=True, allow_null=True)
chapters = ChapterSerializer(many=True, allow_null=True, required=False)
image_quality = serializers.IntegerField(min_value=0, max_value=100, required=False)
deleted_frames = serializers.ListField(child=serializers.IntegerField(min_value=0))
included_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0),
allow_null=True,
required=False,
help_text=textwrap.dedent("""\
A list of valid frame ids. The None value means all frames are included.
"""),
)
chunks_updated_date = serializers.DateTimeField()
class Meta:
model = models.Data
fields = (
"chapters",
"chunks_updated_date",
"chunk_size",
"size",
"image_quality",
"start_frame",
"stop_frame",
"frame_filter",
"frames",
"deleted_frames",
"included_frames",
"storage",
"cloud_storage_id",
)
read_only_fields = fields
extra_kwargs = {
"chunks_updated_date": {"help_text": textwrap.dedent("""\
The date of the last chunk data update.
Chunks downloaded before this date are outdated and should be redownloaded.
""")},
"size": {"help_text": textwrap.dedent("""\
The number of frames included. Deleted frames do not affect this value.
""")},
}
def to_representation(self, instance):
serialized = super().to_representation(instance)
if (task := self._context.get("task")) and task.media_type == models.MediaType.AUDIO:
# Can also be checked via hasattr(instance, 'audio'), but it results in extra requests
# TODO: deprecated for 3d, remove later
serialized.pop("image_quality", None) # not relevant for audio
return serialized
class DataMetaWriteSerializer(serializers.ModelSerializer):
deleted_frames = serializers.ListField(
child=serializers.IntegerField(min_value=0), required=False
)
cloud_storage_id = serializers.IntegerField(required=False, allow_null=True)
class Meta:
model = models.Data
fields = ("deleted_frames", "cloud_storage_id")
def validate_cloud_storage_id(self, cloud_storage_id: int):
try:
db_storage: models.CloudStorage = models.CloudStorage.objects.get(id=cloud_storage_id)
storage = db_storage_to_storage_instance(db_storage)
storage_status = storage.get_status()
if storage_status != Status.AVAILABLE:
raise serializers.ValidationError(
f"The specified cloud storage '{db_storage.display_name}' is not available."
)
except models.CloudStorage.DoesNotExist:
raise serializers.ValidationError(
f"The specified cloud storage {cloud_storage_id} does not exist."
)
return cloud_storage_id
def validate_deleted_frames(self, requested_deleted_frames: list[int]):
requested_deleted_frames_set = set(requested_deleted_frames)
if len(requested_deleted_frames_set) != len(requested_deleted_frames):
raise serializers.ValidationError("Deleted frames cannot repeat")
unknown_requested_deleted_frames = requested_deleted_frames_set.difference(
range(self.instance.size)
)
if unknown_requested_deleted_frames:
raise serializers.ValidationError(
"Unknown frames {} requested for removal".format(
format_list(tuple(map(str, sorted(unknown_requested_deleted_frames))))
)
)
validation_layout = getattr(self.instance, "validation_layout", None)
if validation_layout and validation_layout.mode == models.ValidationMode.GT_POOL:
gt_frame_set = set(validation_layout.frames)
changed_deleted_frames = requested_deleted_frames_set.difference(
self.instance.deleted_frames
)
if not gt_frame_set.isdisjoint(changed_deleted_frames):
raise serializers.ValidationError(
f"When task validation mode is {models.ValidationMode.GT_POOL}, "
"GT frames can only be deleted and restored via the "
"GT job's api/jobs/{id}/data/meta endpoint"
)
return requested_deleted_frames
def update(self, instance: models.Data, validated_data):
instance = super().update(instance, validated_data)
db_task = models.Task.objects.filter(data=instance).first()
if validated_data.get("cloud_storage_id"):
task_frame_provider = TaskFrameProvider(db_task)
for quality in models.FrameQuality:
task_frame_provider.invalidate_chunks(quality=quality)
if db_task:
db_task.touch()
return instance
class JobDataMetaWriteSerializer(serializers.ModelSerializer):
deleted_frames = serializers.ListField(child=serializers.IntegerField(min_value=0))
class Meta:
model = models.Job
fields = ("deleted_frames",)
@transaction.atomic
def update(self, instance: models.Job, validated_data: dict[str, Any]) -> models.Job:
db_segment = instance.segment
db_task = db_segment.task
db_data = db_task.require_data()
deleted_frames = validated_data["deleted_frames"]
task_frame_provider = TaskFrameProvider(db_task)
segment_rel_frame_set = set(
map(task_frame_provider.get_rel_frame_number, db_segment.frame_set)
)
unknown_deleted_frames = set(deleted_frames) - segment_rel_frame_set
if unknown_deleted_frames:
raise serializers.ValidationError(
"Frames {} do not belong to the job".format(
format_list(list(map(str, unknown_deleted_frames)))
)
)
updated_deleted_validation_frames = None
updated_deleted_task_frames = None
if instance.type == models.JobType.GROUND_TRUTH:
updated_deleted_validation_frames = deleted_frames + [
f
for f in db_data.validation_layout.disabled_frames
if f not in segment_rel_frame_set
]
if db_data.validation_layout.mode == models.ValidationMode.GT_POOL:
# GT pool owns its frames, so we exclude them from the task
# Them and the related honeypots in jobs
updated_validation_abs_frame_set = set(
map(task_frame_provider.get_abs_frame_number, updated_deleted_validation_frames)
)
excluded_placeholder_frames = [
task_frame_provider.get_rel_frame_number(frame)
for frame, real_frame in (
models.Image.objects.filter(data=db_data, is_placeholder=True)
.values_list("frame", "real_frame")
.iterator(chunk_size=10000)
)
if real_frame in updated_validation_abs_frame_set
]
updated_deleted_task_frames = deleted_frames + excluded_placeholder_frames
elif db_data.validation_layout.mode == models.ValidationMode.GT:
# Regular GT jobs only refer to the task frames, without data ownership
pass
else:
assert False
else:
updated_deleted_task_frames = deleted_frames + [
f for f in db_data.deleted_frames if f not in segment_rel_frame_set
]
if updated_deleted_validation_frames is not None:
db_data.validation_layout.disabled_frames = updated_deleted_validation_frames
db_data.validation_layout.save(update_fields=["disabled_frames"])
if updated_deleted_task_frames is not None:
db_data.deleted_frames = updated_deleted_task_frames
db_data.save(update_fields=["deleted_frames"])
db_task.touch()
if db_task.project:
db_task.project.touch()
return instance
class AttributeValSerializer(serializers.Serializer):
spec_id = serializers.IntegerField()
value = serializers.CharField(max_length=4096, allow_blank=True)
def to_internal_value(self, data):
data["value"] = str(data["value"])
return super().to_internal_value(data)
class AttributedAnnotationSerializer(serializers.Serializer):
attributes = AttributeValSerializer(many=True, default=[])
class ScoredAnnotationSerializer(serializers.Serializer):
score = serializers.FloatField(min_value=0, max_value=1, default=1)
class FrameAnnotationSerializer(serializers.Serializer):
frame = serializers.IntegerField(min_value=0)
class AnnotationSerializer(serializers.Serializer):
id = serializers.IntegerField(default=None, allow_null=True)
label_id = serializers.IntegerField(min_value=0)
group = serializers.IntegerField(
min_value=0,
default=0,
allow_null=True, # backward compatibility; TODO: disallow on the DB level
)
source = serializers.CharField(
# TODO: change the field type to ChoiceField,
# when SDK can compare string enum values without explicit .value access
default=models.SourceType.MANUAL
)
def _validate_id_absent(self, value):
if value is not None:
raise serializers.ValidationError("must be absent")
return value
def _validate_id_present(self, value):
if value is None:
raise serializers.ValidationError("must be present and not null")
return value
def validate_group(self, value):
return value or 0 # backward compatibility; TODO: disallow on the DB level
@cached_property
def validate_id(self):
# avoid circular import
from cvat.apps.dataset_manager.task import PatchAction
# It would've been better to determine the validator in `__init__`,
# but in a nested serializer the top-level context doesn't actually become
# accessible until after initialization.
if action := self.context.get("annotation_action"):
if action == PatchAction.CREATE:
return self._validate_id_absent
elif action == PatchAction.UPDATE:
# Logically, we should return _validate_id_present here.
# However, due to the way the implementation historically worked,
# passing annotations without IDs would work as a "create" operation.
# There are almost certainly clients relying on this, so keep allowing it.
return None
elif action == PatchAction.DELETE:
return self._validate_id_present
else:
assert False, f"Unknown action {action!r}"
return None
def validate(self, attrs):
attrs = super().validate(attrs)
source = attrs.get("source")
try:
models.SourceType(source)
except ValueError:
if attrs.get("id"):
# Workaround for the DB records that could have been introduced by the UI before
# https://github.com/cvat-ai/cvat/issues/8874 was fixed.
# We allow the DB to store the old invalid annotations and return them from
# the server API, but disallow saving new ones. If the annotations were used
# for updating the existing annotations, we silently fix the input annotations.
# This is done this way to avoid heavy DB migrations.
source = str(models.SourceType.MANUAL)
else:
raise serializers.ValidationError(
{
"source": "must be one of {}, got '{}'".format(
format_list([f"'{v[0]}'" for v in models.SourceType.choices()]), source
)
}
)
attrs["source"] = source
return attrs
class LabeledImageSerializer(
AnnotationSerializer, FrameAnnotationSerializer, AttributedAnnotationSerializer
):
pass
class OptimizedFloatListField(serializers.ListField):
"""Default ListField is extremely slow when try to process long lists of points"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs, child=serializers.FloatField())
def to_internal_value(self, data):
return self.run_child_validation(data)
def to_representation(self, data):
return data
def run_child_validation(self, data):
errors = OrderedDict()
for idx, item in enumerate(data):
if type(item) not in [int, float]:
errors[idx] = exceptions.ValidationError("Value must be a float or an integer")
if not errors:
return data
raise exceptions.ValidationError(errors)
class ShapeSerializer(serializers.Serializer):
type = serializers.ChoiceField(choices=models.ShapeType.choices())
occluded = serializers.BooleanField(default=False)
outside = serializers.BooleanField(default=False, required=False)
z_order = serializers.IntegerField(default=0)
rotation = serializers.FloatField(default=0, min_value=0, max_value=360)
points = OptimizedFloatListField(allow_empty=True, required=False)
def validate(self, attrs):
attrs = super().validate(attrs)
shape_type = attrs["type"]
num_points = len(attrs.get("points", ()))
def bad_num_points_unless(condition: bool) -> None:
if not condition:
raise serializers.ValidationError(
{"points": f"invalid length for shape type '{shape_type}'"}
)
if shape_type in {models.ShapeType.RECTANGLE, models.ShapeType.ELLIPSE}:
bad_num_points_unless(num_points == 4)
elif shape_type == models.ShapeType.POLYGON:
bad_num_points_unless(num_points >= 6 and num_points % 2 == 0)
elif shape_type == models.ShapeType.POLYLINE:
bad_num_points_unless(num_points >= 4 and num_points % 2 == 0)
elif shape_type == models.ShapeType.POINTS:
bad_num_points_unless(num_points >= 2 and num_points % 2 == 0)
elif shape_type == models.ShapeType.CUBOID:
bad_num_points_unless(num_points == 16)
elif shape_type == models.ShapeType.MASK:
bad_num_points_unless(num_points >= 5)
elif shape_type == models.ShapeType.SKELETON:
bad_num_points_unless(num_points == 0)
else:
assert False, f"Unknown shape type '{shape_type}'"
return attrs
class SubLabeledShapeSerializer(
ShapeSerializer,
AnnotationSerializer,
FrameAnnotationSerializer,
AttributedAnnotationSerializer,
ScoredAnnotationSerializer,
):
pass
class LabeledShapeSerializer(SubLabeledShapeSerializer):
elements = SubLabeledShapeSerializer(many=True, required=False)
def validate(self, attrs):
attrs = super().validate(attrs)
num_elements = len(attrs.get("elements", ()))
if attrs["type"] == models.ShapeType.SKELETON:
if num_elements == 0:
raise serializers.ValidationError(
{"elements": "at least one required for skeleton shape"}
)
else:
if num_elements != 0:
raise serializers.ValidationError(
{"elements": "not allowed for non-skeleton shape"}
)
return attrs
def _convert_annotation(obj, keys):
d = OrderedDict([(key, obj[key]) for key in keys])
if "group" in d:
# backward compatibility; TODO: disallow null on the DB level
d["group"] = d["group"] or 0
return d
def _convert_attributes(attr_set):
attr_keys = ["spec_id", "value"]
return [OrderedDict([(key, attr[key]) for key in attr_keys]) for attr in attr_set]
class LabeledImageSerializerFromDB(serializers.BaseSerializer):
# Use this serializer to export data from the database
# Because default DRF serializer is too slow on huge collections
def to_representation(self, instance):
def convert_tag(tag):
result = _convert_annotation(tag, ["id", "label_id", "frame", "group", "source"])
result["attributes"] = _convert_attributes(tag["attributes"])
return result
return convert_tag(instance)
class LabeledShapeSerializerFromDB(serializers.BaseSerializer):
# Use this serializer to export data from the database
# Because default DRF serializer is too slow on huge collections
def to_representation(self, instance):
def convert_shape(shape):
result = _convert_annotation(
shape,
[
"id",
"label_id",
"type",
"frame",
"group",
"source",
"score",
"occluded",
"outside",
"z_order",
"rotation",
"points",
],
)
result["attributes"] = _convert_attributes(shape["attributes"])
if shape.get("elements", None) is not None and shape["parent"] is None:
result["elements"] = [convert_shape(element) for element in shape["elements"]]
return result
return convert_shape(instance)
class LabeledTrackSerializerFromDB(serializers.BaseSerializer):
# Use this serializer to export data from the database
# Because default DRF serializer is too slow on huge collections
def to_representation(self, instance):
def convert_track(track):
shape_keys = [
"id",
"type",
"frame",
"occluded",
"outside",
"z_order",
"rotation",
"points",
"attributes",
]
result = _convert_annotation(track, ["id", "label_id", "frame", "group", "source"])
result["shapes"] = [_convert_annotation(shape, shape_keys) for shape in track["shapes"]]
result["attributes"] = _convert_attributes(track["attributes"])
for shape in result["shapes"]:
shape["attributes"] = _convert_attributes(shape["attributes"])
if track.get("elements", None) is not None and track["parent"] is None:
result["elements"] = [convert_track(element) for element in track["elements"]]
return result
return convert_track(instance)
class LabeledIntervalSerializerFromDB(serializers.BaseSerializer):
# Use this serializer to export data from the database
# Because default DRF serializer is too slow on huge collections
def to_representation(self, instance):
def convert_interval(interval):
result = _convert_annotation(
interval,
[
"id",
"label_id",
"start",
"stop",
"group",
"source",
"score",
],
)
result["attributes"] = _convert_attributes(interval["attributes"])
return result
return convert_interval(instance)
class TrackedShapeSerializer(ShapeSerializer, AttributedAnnotationSerializer):
id = serializers.IntegerField(default=None, allow_null=True)
frame = serializers.IntegerField(min_value=0)
class SubLabeledTrackSerializer(
AnnotationSerializer, FrameAnnotationSerializer, AttributedAnnotationSerializer
):
shapes = TrackedShapeSerializer(many=True, allow_empty=True)
class LabeledTrackSerializer(SubLabeledTrackSerializer):
elements = SubLabeledTrackSerializer(many=True, required=False)
class LabeledIntervalSerializer(
AnnotationSerializer,
AttributedAnnotationSerializer,
ScoredAnnotationSerializer,
):
start = serializers.IntegerField(min_value=0)
stop = serializers.IntegerField(min_value=0, allow_null=True)
class LabeledDataSerializer(serializers.Serializer):
version = serializers.IntegerField(default=0) # TODO: remove
tags = LabeledImageSerializer(many=True, default=[])
shapes = LabeledShapeSerializer(many=True, default=[])
tracks = LabeledTrackSerializer(many=True, default=[])
intervals = LabeledIntervalSerializer(many=True, default=[])
class FileInfoSerializer(serializers.Serializer):
name = serializers.CharField(max_length=MAX_FILENAME_LENGTH)
type = serializers.ChoiceField(choices=["REG", "DIR"])
mime_type = serializers.CharField(max_length=255)
class AnnotationFileSerializer(serializers.Serializer):
annotation_file = serializers.FileField()
class DatasetFileSerializer(serializers.Serializer):
dataset_file = serializers.FileField()
@staticmethod
def validate_dataset_file(value):
if os.path.splitext(value.name)[1] != ".zip":
raise serializers.ValidationError("Dataset file should be zip archive")
return value
class TaskFileSerializer(serializers.Serializer):
task_file = serializers.FileField()
class ProjectFileSerializer(serializers.Serializer):
project_file = serializers.FileField()
class CommentReadSerializer(serializers.ModelSerializer):
owner = BasicUserSerializer(allow_null=True, required=False)
class Meta:
model = models.Comment
fields = ("id", "issue", "owner", "message", "created_date", "updated_date")
read_only_fields = fields
class CommentWriteSerializer(WriteOnceMixin, serializers.ModelSerializer):
def to_representation(self, instance):
serializer = CommentReadSerializer(instance, context=self.context)
return serializer.data
class Meta:
model = models.Comment
fields = ("issue", "message")
write_once_fields = ("issue",)
class IssueReadSerializer(serializers.ModelSerializer):
owner = BasicUserSerializer(allow_null=True, required=False)
assignee = BasicUserSerializer(allow_null=True, required=False)
position = serializers.ListField(child=serializers.FloatField(), allow_empty=False)
comments = CommentsSummarySerializer(models.Comment, url_filter_key="issue_id")
class Meta:
model = models.Issue
fields = (
"id",
"frame",
"position",
"job",
"owner",
"assignee",
"created_date",
"updated_date",
"resolved",
"comments",
)
read_only_fields = fields
extra_kwargs = {
"created_date": {"allow_null": True},
"updated_date": {"allow_null": True},
}
class IssueWriteSerializer(WriteOnceMixin, serializers.ModelSerializer):
position = serializers.ListField(
child=serializers.FloatField(),
allow_empty=False,
)
message = serializers.CharField(style={"base_template": "textarea.html"})
def to_representation(self, instance):
serializer = IssueReadSerializer(instance, context=self.context)
return serializer.data
def create(self, validated_data):
message = validated_data.pop("message")
db_issue = super().create(validated_data)
models.Comment.objects.create(issue=db_issue, message=message, owner=db_issue.owner)
return db_issue
class Meta:
model = models.Issue
fields = ("frame", "position", "job", "assignee", "message", "resolved")
write_once_fields = ("frame", "job", "message")
class ManifestSerializer(serializers.ModelSerializer):
class Meta:
model = models.Manifest
fields = ("filename",)
extra_kwargs = {"filename": {"validators": [CanonicalRelativePathValidator()]}}
def to_internal_value(self, data):
return super().to_internal_value({"filename": data})
# pylint: disable=no-self-use
def to_representation(self, instance):
return instance.filename if instance else instance
class CloudStorageReadSerializer(serializers.ModelSerializer):
owner = BasicUserSerializer(required=False, allow_null=True)
manifests = ManifestSerializer(many=True, default=[])
class Meta:
model = models.CloudStorage
exclude = ["credentials"]
read_only_fields = ("created_date", "updated_date", "owner", "organization")
extra_kwargs = {"organization": {"allow_null": True}}
@extend_schema_serializer(
examples=[
OpenApiExample(
"Create Amazon S3 cloud storage with credentials",
description="",
value={
"provider_type": models.CloudProviderChoice.AMAZON_S3,
"resource": "somebucket",
"display_name": "Bucket",
"credentials_type": models.CredentialsTypeChoice.KEY_SECRET_KEY_PAIR,
"key": "XXX",
"secret_key": "XXX",
"specific_attributes": "region=eu-central-1",
"description": "Some description",
"manifests": ["manifest.jsonl"],
},
request_only=True,
),
OpenApiExample(
"Create Amazon S3 cloud storage without credentials",
value={
"provider_type": models.CloudProviderChoice.AMAZON_S3,
"resource": "somebucket",
"display_name": "Bucket",
"credentials_type": models.CredentialsTypeChoice.ANONYMOUS_ACCESS,
"manifests": ["manifest.jsonl"],
},
request_only=True,
),
OpenApiExample(
"Create Azure cloud storage",
value={
"provider_type": models.CloudProviderChoice.AZURE_BLOB_STORAGE,
"resource": "sonecontainer",
"display_name": "Container",
"credentials_type": models.CredentialsTypeChoice.ACCOUNT_NAME_TOKEN_PAIR,
"account_name": "someaccount",
"session_token": "xxx",
"manifests": ["manifest.jsonl"],
},
request_only=True,
),
OpenApiExample(
"Create GCS",
value={
"provider_type": models.CloudProviderChoice.GOOGLE_CLOUD_STORAGE,
"resource": "somebucket",
"display_name": "Bucket",
"credentials_type": models.CredentialsTypeChoice.KEY_FILE_PATH,
"key_file": "file",
"manifests": ["manifest.jsonl"],
},
request_only=True,
),
]
)
class CloudStorageWriteSerializer(serializers.ModelSerializer):
owner = BasicUserSerializer(required=False)
session_token = serializers.CharField(max_length=440, allow_blank=True, required=False)
key = serializers.CharField(max_length=40, allow_blank=True, required=False)
secret_key = serializers.CharField(max_length=64, allow_blank=True, required=False)
key_file = serializers.FileField(required=False)
account_name = serializers.CharField(max_length=24, allow_blank=True, required=False)
manifests = ManifestSerializer(many=True, default=[])
connection_string = serializers.CharField(max_length=1024, allow_blank=True, required=False)
class Meta:
model = models.CloudStorage
fields = (
"provider_type",
"resource",
"display_name",
"owner",
"credentials_type",
"created_date",
"updated_date",
"session_token",
"account_name",
"key",
"secret_key",
"connection_string",
"key_file",
"specific_attributes",
"description",
"id",
"manifests",
"organization",
)
read_only_fields = ("created_date", "updated_date", "owner", "organization")
extra_kwargs = {"organization": {"allow_null": True}}
# pylint: disable=no-self-use
def validate_specific_attributes(self, value):
if value:
attributes = value.split("&")
for attribute in attributes:
if not len(attribute.split("=")) == 2:
raise serializers.ValidationError("Invalid specific attributes")
return value
def validate(self, attrs):
provider_type = attrs.get("provider_type")
if provider_type == models.CloudProviderChoice.AZURE_BLOB_STORAGE:
if not attrs.get("account_name", "") and not attrs.get("connection_string", ""):
raise serializers.ValidationError(
"Account name or connection string for Azure container was not specified"
)
# Amazon S3: https://docs.aws.amazon.com/AmazonS3/latest/userguide/bucketnamingrules.html?icmpid=docs_amazons3_console
# ABS: https://learn.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names
# GCS: https://cloud.google.com/storage/docs/buckets#naming
ALLOWED_RESOURCE_NAME_SYMBOLS = string.ascii_lowercase + string.digits + "-"
if provider_type == models.CloudProviderChoice.GOOGLE_CLOUD_STORAGE:
ALLOWED_RESOURCE_NAME_SYMBOLS += "_."
elif provider_type == models.CloudProviderChoice.AMAZON_S3:
ALLOWED_RESOURCE_NAME_SYMBOLS += "."
# We need to check only basic naming rule
if (resource := attrs.get("resource")) and (
diff := (set(resource) - set(ALLOWED_RESOURCE_NAME_SYMBOLS))
):
raise serializers.ValidationError(
{
"resource": f"Invalid characters ({','.join(diff)}) were found.",
}
)
return attrs
def _validate_prefix(self, value: str) -> None:
if value.startswith("/"):
raise serializers.ValidationError('Prefix cannot start with forward slash ("/").')
if "" in value.strip("/").split("/"):
raise serializers.ValidationError("Prefix cannot contain multiple slashes in a row.")
@staticmethod
def _manifests_validation(storage, manifests):
# check manifest files availability
for manifest in manifests:
file_status = storage.get_file_status(manifest)
if file_status == Status.NOT_FOUND:
raise serializers.ValidationError(
{
"manifests": "The '{}' file does not exist on '{}' cloud storage".format(
manifest, storage.name
)
}
)
elif file_status == Status.FORBIDDEN:
raise serializers.ValidationError(
{
"manifests": "The '{}' file does not available on '{}' cloud storage. Access denied".format(
manifest, storage.name
)
}
)
def create(self, validated_data):
provider_type = validated_data.get("provider_type")
key_file = validated_data.pop("key_file", None)
# we need to save it to temporary file to check the granted permissions
temporary_file = None
if key_file:
with NamedTemporaryFile(mode="wb", prefix="cvat", delete=False) as temp_key:
temp_key.write(key_file.read())
temporary_file = temp_key.name
key_file.close()
del key_file
credentials = Credentials(
account_name=validated_data.pop("account_name", ""),
key=validated_data.pop("key", ""),
secret_key=validated_data.pop("secret_key", ""),
session_token=validated_data.pop("session_token", ""),
key_file_path=temporary_file,
credentials_type=validated_data.get("credentials_type"),
connection_string=validated_data.pop("connection_string", ""),
)
details = {
"resource": validated_data.get("resource"),
"credentials": credentials,
"specific_attributes": parse_specific_attributes(
validated_data.get("specific_attributes", "")
),
}
if prefix := details["specific_attributes"].get("prefix"):
self._validate_prefix(prefix)
storage = get_cloud_storage_instance(cloud_provider=provider_type, **details)
storage_status = storage.get_status()
if storage_status == Status.AVAILABLE:
manifests = [m.get("filename") for m in validated_data.pop("manifests")]
self._manifests_validation(storage, manifests)
db_storage = models.CloudStorage.objects.create(
credentials=credentials.convert_to_db(), **validated_data
)
db_storage.save()
manifest_file_instances = [
models.Manifest(filename=manifest, cloud_storage=db_storage)
for manifest in manifests
]
db_utils.bulk_create(models.Manifest, manifest_file_instances)
cloud_storage_path = db_storage.get_storage_dirname()
if os.path.isdir(cloud_storage_path):
shutil.rmtree(cloud_storage_path)
os.makedirs(cloud_storage_path)
if temporary_file:
# so, gcs key file is valid and we need to set correct path to the file
real_path_to_key_file = db_storage.get_key_file_path()
shutil.copyfile(temporary_file, real_path_to_key_file)
os.remove(temporary_file)
credentials.key_file_path = real_path_to_key_file
db_storage.credentials = credentials.convert_to_db()
db_storage.save()
return db_storage
elif storage_status == Status.FORBIDDEN:
field = "credentials"
message = (
"Cannot create resource {} with specified credentials. Access forbidden.".format(
storage.name
)
)
else:
field = "resource"
message = "The resource {} not found. It may have been deleted.".format(storage.name)
if temporary_file:
os.remove(temporary_file)
slogger.glob.error(message)
raise serializers.ValidationError({field: message})
@transaction.atomic
def update(self, instance, validated_data):
credentials = Credentials()
credentials.convert_from_db(
{
"type": instance.credentials_type,
"value": instance.credentials,
}
)
credentials_dict = {
k: v
for k, v in validated_data.items()
if k
in {
"key",
"secret_key",
"account_name",
"session_token",
"key_file_path",
"credentials_type",
"connection_string",
}
}
key_file = validated_data.pop("key_file", None)
temporary_file = None
if key_file:
with NamedTemporaryFile(mode="wb", prefix="cvat", delete=False) as temp_key:
temp_key.write(key_file.read())
temporary_file = temp_key.name
credentials_dict["key_file_path"] = temporary_file
key_file.close()
del key_file
if prefix := parse_specific_attributes(validated_data.get("specific_attributes", "")).get(
"prefix"
):
self._validate_prefix(prefix)
credentials.mapping_with_new_values(credentials_dict)
instance.credentials = credentials.convert_to_db()
for field in (
"credentials_type",
"resource",
"display_name",
"description",
"specific_attributes",
):
if field in validated_data:
setattr(instance, field, validated_data[field])
# check cloud storage existing
details = {
"resource": instance.resource,
"credentials": credentials,
"specific_attributes": parse_specific_attributes(instance.specific_attributes),
}
storage = get_cloud_storage_instance(cloud_provider=instance.provider_type, **details)
storage_status = storage.get_status()
if storage_status == Status.AVAILABLE:
new_manifest_names = set(i.get("filename") for i in validated_data.get("manifests", []))
previous_manifest_names = set(i.filename for i in instance.manifests.all())
delta_to_delete = tuple(previous_manifest_names - new_manifest_names)
delta_to_create = tuple(new_manifest_names - previous_manifest_names)
if delta_to_delete:
instance.manifests.filter(filename__in=delta_to_delete).delete()
if delta_to_create:
# check manifest files existing
self._manifests_validation(storage, delta_to_create)
manifest_instances = [
models.Manifest(filename=f, cloud_storage=instance) for f in delta_to_create
]
db_utils.bulk_create(models.Manifest, manifest_instances)
if temporary_file:
# so, gcs key file is valid and we need to set correct path to the file
real_path_to_key_file = instance.get_key_file_path()
shutil.copyfile(temporary_file, real_path_to_key_file)
os.remove(temporary_file)
instance.credentials = real_path_to_key_file
instance.save()
return instance
elif storage_status == Status.FORBIDDEN:
field = "credentials"
message = (
"Cannot update resource {} with specified credentials. Access forbidden.".format(
storage.name
)
)
else:
field = "resource"
message = "The resource {} not found. It may have been deleted.".format(storage.name)
if temporary_file:
os.remove(temporary_file)
slogger.glob.error(message)
raise serializers.ValidationError({field: message})
class CloudStorageContentSerializer(serializers.Serializer):
next = serializers.CharField(
required=False,
allow_null=True,
allow_blank=True,
help_text="This token is used to continue listing files in the bucket.",
)
content = FileInfoSerializer(many=True)
def _update_related_storages(
instance: models.Project | models.Task,
*,
validated_data: dict[str, Any],
workspace_transferring: bool = False,
dst_organization_id: int | None = None,
) -> None:
for storage_type in ("source_storage", "target_storage"):
# storage_instance maybe None
storage_instance: models.Storage | None = getattr(instance, storage_type, None)
new_conf = validated_data.pop(storage_type, {})
if not new_conf:
if (
not workspace_transferring
or workspace_transferring
and (
not storage_instance
or storage_instance
and not storage_instance.cloud_storage_id
)
):
continue
try:
original_cs = models.CloudStorage.objects.annotate(Count("manifests")).get(
pk=storage_instance.cloud_storage_id
)
except models.CloudStorage.DoesNotExist:
# generally should not occur
new_conf = {
"location": models.Location.LOCAL,
"cloud_storage_id": None,
}
else:
similar_ones: list[models.CloudStorage] = []
msg_no_similar_cs = "Could not find a similar cloud storage in the new workspace"
qs_with_similar_ones = (
models.CloudStorage.objects.filter(
provider_type=original_cs.provider_type,
resource=original_cs.resource,
organization_id=dst_organization_id,
)
.annotate(Count("manifests"))
.filter(manifests__count=original_cs.manifests__count)
)
for cs in qs_with_similar_ones:
if (
cs.get_specific_attributes() != original_cs.get_specific_attributes()
# we do not need to compare manifest contents since the content
# will be synchronized with the original CS file on the next content request
or list(
cs.manifests.values_list("filename", flat=True).order_by("filename")
)
!= list(
original_cs.manifests.values_list("filename", flat=True).order_by(
"filename"
)
)
):
continue
similar_ones.append(cs)
if len(similar_ones) != 1:
raise serializers.ValidationError(msg_no_similar_cs)
new_conf = {
"location": models.Location.CLOUD_STORAGE,
"cloud_storage_id": similar_ones[0].pk,
}
storage_serializer = StorageSerializer(storage_instance, data=new_conf)
storage_serializer.is_valid(raise_exception=True)
storage_instance = storage_serializer.save()
setattr(instance, storage_type, storage_instance)
def _configure_related_storages(validated_data: dict[str, Any]) -> dict[str, models.Storage | None]:
storages = {
"source_storage": None,
"target_storage": None,
}
for i in storages:
if storage_conf := validated_data.get(i):
if (
cloud_storage_id := storage_conf.get("cloud_storage_id")
) and not models.CloudStorage.objects.filter(id=cloud_storage_id).exists():
raise serializers.ValidationError(
f"The specified cloud storage {cloud_storage_id} does not exist."
)
storage_instance = models.Storage(**storage_conf)
storage_instance.save()
storages[i] = storage_instance
return storages
class AssetReadSerializer(WriteOnceMixin, serializers.ModelSerializer):
filename = serializers.CharField(required=True, max_length=MAX_FILENAME_LENGTH)
owner = BasicUserSerializer(required=False)
class Meta:
model = models.Asset
fields = (
"uuid",
"filename",
"created_date",
"owner",
"guide_id",
)
read_only_fields = fields
class AssetWriteSerializer(WriteOnceMixin, serializers.ModelSerializer):
file = serializers.FileField(
required=True, write_only=True, allow_empty_file=False, max_length=MAX_FILENAME_LENGTH
)
guide_id = serializers.IntegerField(required=True)
def validate_file(self, value):
if not isinstance(value, UploadedFile):
raise serializers.ValidationError(
"Invalid asset_file type. Expected an UploadedFile instance."
)
if value.size / (1024 * 1024) > settings.ASSET_MAX_SIZE_MB:
raise serializers.ValidationError(
f"Maximum size of asset is {settings.ASSET_MAX_SIZE_MB} MB"
)
if value.content_type not in settings.ASSET_SUPPORTED_TYPES:
raise serializers.ValidationError(
f"File is not supported as an asset. Supported are {settings.ASSET_SUPPORTED_TYPES}"
)
guessed_type, guessed_encoding = mimetypes.guess_type(value.name)
if guessed_type != value.content_type or guessed_encoding is not None:
raise serializers.ValidationError(
"Provided Content-Type does not match the file extension."
)
return value
def create(self, validated_data):
asset_file = validated_data.pop("file")
asset_uuid = str(uuid.uuid4())
dirname = os.path.join(settings.ASSETS_ROOT, asset_uuid)
basename = asset_file.name
filename = os.path.join(dirname, basename)
os.makedirs(dirname)
try:
if asset_file.content_type in ("image/jpeg", "image/png"):
image = Image.open(asset_file)
if any(x > settings.ASSET_MAX_IMAGE_SIZE for x in image.size):
scale_factor = settings.ASSET_MAX_IMAGE_SIZE / max(image.size)
image = image.resize(int(x * scale_factor) for x in image.size)
image.save(filename)
else:
with open(filename, "wb") as destination:
for chunk in asset_file.chunks():
destination.write(chunk)
av_scan_paths(dirname)
return models.Asset.objects.create(
**validated_data,
uuid=asset_uuid,
filename=basename,
content_size=get_path_size(dirname),
)
except Exception:
if os.path.exists(filename):
os.remove(filename)
os.rmdir(dirname)
raise
class Meta:
model = models.Asset
fields = (
"guide_id",
"file",
)
write_once_fields = ("guide_id",)
class AnnotationGuideReadSerializer(WriteOnceMixin, serializers.ModelSerializer):
class Meta:
model = models.AnnotationGuide
fields = (
"id",
"task_id",
"project_id",
"created_date",
"updated_date",
"markdown",
)
read_only_fields = fields
class AnnotationGuideWriteSerializer(WriteOnceMixin, serializers.ModelSerializer):
project_id = serializers.IntegerField(required=False, allow_null=True)
task_id = serializers.IntegerField(required=False, allow_null=True)
def validate_markdown(self, markdown: str) -> str:
if (
len(models.AnnotationGuide.get_asset_ids_from_markdown(markdown))
> settings.ASSET_MAX_COUNT_PER_GUIDE
):
raise serializers.ValidationError("Maximum number of assets per guide reached")
return markdown
@transaction.atomic
def create(self, validated_data):
project_id = validated_data.get("project_id", None)
task_id = validated_data.get("task_id", None)
if project_id is None and task_id is None:
raise serializers.ValidationError("One of project_id or task_id must be specified")
if project_id is not None and task_id is not None:
raise serializers.ValidationError("Both project_id and task_id must not be specified")
project = None
task = None
if project_id is not None:
try:
project = models.Project.objects.get(id=project_id)
except models.Project.DoesNotExist:
raise serializers.ValidationError(
f"The specified project #{project_id} does not exist."
)
if task_id is not None:
try:
task = models.Task.objects.get(id=task_id)
except models.Task.DoesNotExist:
raise serializers.ValidationError(f"The specified task #{task_id} does not exist.")
db_data = models.AnnotationGuide.objects.create(
**validated_data, project=project, task=task
)
return db_data
class Meta:
model = models.AnnotationGuide
fields = (
"id",
"task_id",
"project_id",
"markdown",
)