Refactor quality reports

cvat/apps/dataset_manager/bindings.py

@@ -17,6 +17,7 @@
 from typing import Any, Callable, Literal, NamedTuple, Optional, Union
 
 import datumaro as dm
+import datumaro.util
 import defusedxml.ElementTree as ET
 import rq
 from attr import attrib, attrs
@@ -2128,9 +2129,9 @@ def match_dm_item(
     if frame_number is None:
         frame_number = instance_data.match_frame(item.id, root_hint, path_has_ext=False)
     if frame_number is None:
-        frame_number = dm.util.cast(item.attributes.get('frame', item.id), int)
+        frame_number = datumaro.util.cast(item.attributes.get('frame', item.id), int)
     if frame_number is None and is_video:
-        frame_number = dm.util.cast(osp.basename(item.id)[len('frame_'):], int)
+        frame_number = datumaro.util.cast(osp.basename(item.id)[len('frame_'):], int)
 
     if not frame_number in instance_data.frame_info:
         raise CvatImportError("Could not match item id: "
@@ -2254,9 +2255,15 @@ def import_dm_annotations(dm_dataset: dm.Dataset, instance_data: Union[ProjectDa
                     # because in some formats return type can be different
                     # from bool / None
                     # https://github.com/openvinotoolkit/datumaro/issues/719
-                    occluded = dm.util.cast(ann.attributes.pop('occluded', None), to_bool) is True
-                    keyframe = dm.util.cast(ann.attributes.get('keyframe', None), to_bool) is True
-                    outside = dm.util.cast(ann.attributes.pop('outside', None), to_bool) is True
+                    occluded = datumaro.util.cast(
+                        ann.attributes.pop('occluded', None), to_bool
+                    ) is True
+                    keyframe = datumaro.util.cast(
+                        ann.attributes.get('keyframe', None), to_bool
+                    ) is True
+                    outside = datumaro.util.cast(
+                        ann.attributes.pop('outside', None), to_bool
+                    ) is True
 
                     track_id = ann.attributes.pop('track_id', None)
                     source = ann.attributes.pop('source').lower() \

cvat/apps/engine/models.py

@@ -11,7 +11,7 @@
 import shutil
 import uuid
 from abc import ABCMeta, abstractmethod
-from collections.abc import Collection, Sequence
+from collections.abc import Collection, Iterable, Sequence
 from enum import Enum
 from functools import cached_property
 from typing import Any, ClassVar, Optional
@@ -453,7 +453,7 @@ def get_tmp_dirname(self) -> str:
 
 
 @transaction.atomic(savepoint=False)
-def clear_annotations_in_jobs(job_ids):
+def clear_annotations_in_jobs(job_ids: Iterable[int]):
     for job_ids_chunk in take_by(job_ids, chunk_size=1000):
         TrackedShapeAttributeVal.objects.filter(shape__track__job_id__in=job_ids_chunk).delete()
         TrackedShape.objects.filter(track__job_id__in=job_ids_chunk).delete()

cvat/apps/engine/utils.py

@@ -15,7 +15,7 @@
 import traceback
 import urllib.parse
 from collections import namedtuple
-from collections.abc import Generator, Iterable, Iterator, Mapping, Sequence
+from collections.abc import Generator, Iterable, Mapping, Sequence
 from contextlib import nullcontext, suppress
 from itertools import islice
 from multiprocessing import cpu_count
@@ -135,7 +135,7 @@ def parse_specific_attributes(specific_attributes):
     } if parsed_specific_attributes else dict()
 
 
-def parse_exception_message(msg):
+def parse_exception_message(msg: str) -> str:
     parsed_msg = msg
     try:
         if 'ErrorDetail' in msg:
@@ -463,9 +463,7 @@ def format_list(
 _V = TypeVar("_V")
 
 
-def grouped(
-    items: Iterator[_V] | Iterable[_V], *, key: Callable[[_V], _K]
-) -> Mapping[_K, Sequence[_V]]:
+def grouped(items: Iterable[_V], *, key: Callable[[_V], _K]) -> Mapping[_K, Sequence[_V]]:
     """
     Returns a mapping with input iterable elements grouped by key, for example:
 

cvat/apps/quality_control/permissions.py

@@ -17,7 +17,7 @@
 
 class QualityReportPermission(OpenPolicyAgentPermission):
     obj: Optional[QualityReport]
-    job_owner_id: Optional[int]
+    rq_job_owner_id: Optional[int]
     task_id: Optional[int]
 
     class Scopes(StrEnum):
@@ -27,10 +27,10 @@ class Scopes(StrEnum):
         VIEW_STATUS = "view:status"
 
     @classmethod
-    def create_scope_check_status(cls, request, job_owner_id: int, iam_context=None):
+    def create_scope_check_status(cls, request, rq_job_owner_id: int, iam_context=None):
         if not iam_context and request:
             iam_context = get_iam_context(request, None)
-        return cls(**iam_context, scope=cls.Scopes.VIEW_STATUS, job_owner_id=job_owner_id)
+        return cls(**iam_context, scope=cls.Scopes.VIEW_STATUS, rq_job_owner_id=rq_job_owner_id)
 
     @classmethod
     def create_scope_view(cls, request, report: Union[int, QualityReport], iam_context=None):
@@ -102,8 +102,8 @@ def create(cls, request, view, obj, iam_context):
         return permissions
 
     def __init__(self, **kwargs):
-        if "job_owner_id" in kwargs:
-            self.job_owner_id = int(kwargs.pop("job_owner_id"))
+        if "rq_job_owner_id" in kwargs:
+            self.rq_job_owner_id = int(kwargs.pop("rq_job_owner_id"))
 
         super().__init__(**kwargs)
         self.url = settings.IAM_OPA_DATA_URL + "/quality_reports/allow"
@@ -164,7 +164,7 @@ def get_resource(self):
                 ),
             }
         elif self.scope == self.Scopes.VIEW_STATUS:
-            data = {"owner": {"id": self.job_owner_id}}
+            data = {"owner": {"id": self.rq_job_owner_id}}
 
         return data