[CVAT] Oracle fixes (#3224)

* Fix invalid project data urls in ExO db

* Fix gt bbox shift in boxes_from_points tasks

* Fix gt annotation replacement in the annotation results

Author: zhiltsov-max

packages/examples/cvat/exchange-oracle/src/handlers/job_creation.py

@@ -30,13 +30,13 @@
 import src.services.cvat as db_service
 from src.chain.escrow import get_escrow_manifest
 from src.core.config import Config
-from src.core.storage import compose_data_bucket_filename
+from src.core.storage import compose_data_bucket_filename, compose_data_bucket_prefix
 from src.core.types import CvatLabelTypes, TaskStatuses, TaskTypes
 from src.db import SessionLocal
 from src.log import ROOT_LOGGER_NAME
 from src.models.cvat import Project
 from src.services.cloud import CloudProviders, StorageClient
-from src.services.cloud.utils import BucketAccessInfo, compose_bucket_url
+from src.services.cloud.utils import BucketAccessInfo
 from src.utils.annotations import InstanceSegmentsToBbox, ProjectLabels, is_point_in_bbox
 from src.utils.assignments import parse_manifest
 from src.utils.logging import NullLogger, format_sequence, get_function_logger
@@ -396,11 +396,7 @@ def build(self):
                 manifest.annotation.type,
                 escrow_address,
                 chain_id,
-                compose_bucket_url(
-                    data_bucket.bucket_name,
-                    bucket_host=data_bucket.host_url,
-                    provider=data_bucket.provider,
-                ),
+                data_bucket.to_url(),
                 cvat_webhook_id=cvat_webhook.id,
             )
 
@@ -1469,6 +1465,10 @@ def _prepare_gt_roi_dataset(self):
             categories=self._gt_dataset.categories(), media_type=dm.Image
         )
 
+        roi_info_by_point_id: dict[int, skeletons_from_boxes_task.RoiInfo] = {
+            roi_info.point_id: roi_info for roi_info in self._rois
+        }
+
         for sample in self._gt_dataset:
             for gt_bbox in sample.annotations:
                 assert isinstance(gt_bbox, dm.Bbox)
@@ -1478,10 +1478,15 @@ def _prepare_gt_roi_dataset(self):
                     self.escrow_address, self.chain_id, self._roi_filenames[point_id]
                 )
 
+                # update gt bbox coordinates to match RoI shift
+                roi_info = roi_info_by_point_id[point_id]
+                new_x = gt_bbox.points[0] - roi_info.roi_x
+                new_y = gt_bbox.points[1] - roi_info.roi_y
+
                 self._gt_roi_dataset.put(
                     sample.wrap(
                         id=os.path.splitext(gt_roi_filename)[0],
-                        annotations=[gt_bbox],
+                        annotations=[gt_bbox.wrap(x=new_x, y=new_y)],
                         media=dm.Image(path=gt_roi_filename, size=sample.media_as(dm.Image).size),
                         attributes=filter_dict(sample.attributes, exclude_keys=["id"]),
                     )
@@ -1495,7 +1500,6 @@ def _create_on_cvat(self):
         assert self._label_configuration is not _unset
         assert self._gt_roi_dataset is not _unset
 
-        input_data_bucket = BucketAccessInfo.parse_obj(self.manifest.data.data_url)
         oracle_bucket = self.oracle_data_bucket
 
         # Register cloud storage on CVAT to pass user dataset
@@ -1535,11 +1539,9 @@ def _create_on_cvat(self):
                 self.manifest.annotation.type,
                 self.escrow_address,
                 self.chain_id,
-                compose_bucket_url(
-                    input_data_bucket.bucket_name,
-                    bucket_host=input_data_bucket.host_url,
-                    provider=input_data_bucket.provider,
-                ),
+                oracle_bucket.to_url().rstrip("/")
+                + "/"
+                + compose_data_bucket_prefix(self.escrow_address, self.chain_id),
                 cvat_webhook_id=cvat_webhook.id,
             )
             db_service.get_project_by_id(session, project_id, for_update=True)  # lock the row
@@ -2635,7 +2637,6 @@ def _task_params_label_key(ts):
             for skeleton_label_id, skeleton_label in enumerate(self.manifest.annotation.labels)
         }
 
-        input_data_bucket = BucketAccessInfo.parse_obj(self.manifest.data.data_url)
         oracle_bucket = self.oracle_data_bucket
 
         # Register cloud storage on CVAT to pass user dataset
@@ -2714,11 +2715,9 @@ def _task_params_label_key(ts):
                         self.manifest.annotation.type,
                         self.escrow_address,
                         self.chain_id,
-                        compose_bucket_url(
-                            input_data_bucket.bucket_name,
-                            bucket_host=input_data_bucket.host_url,
-                            provider=input_data_bucket.provider,
-                        ),
+                        oracle_bucket.to_url().rstrip("/")
+                        + "/"
+                        + compose_data_bucket_prefix(self.escrow_address, self.chain_id),
                         cvat_webhook_id=cvat_webhook.id,
                     )
                     created_projects.append(project_id)

packages/examples/cvat/exchange-oracle/src/services/cloud/types.py

@@ -8,6 +8,7 @@
 from urllib.parse import urlparse
 
 import pydantic
+from httpx import URL
 
 from src.core import manifest
 from src.core.config import Config, StorageConfig
@@ -113,7 +114,7 @@ def from_url(cls, url: str) -> BucketAccessInfo:
             )
         elif Config.features.enable_custom_cloud_host:
             # Check if netloc is an ip address
-            # or localhost with port (or its /etc/hosts aliast, e.g. minio:9000)
+            # or localhost with port (or its /etc/hosts alias, e.g. minio:9000)
             if is_ipv4(parsed_url.netloc) or re.fullmatch(r"\w+:\d{4}", parsed_url.netloc):
                 host = parsed_url.netloc
                 bucket_name, path = parsed_url.path.lstrip("/").split("/", maxsplit=1)
@@ -190,3 +191,18 @@ def parse_obj(
             return cls.from_storage_config(data)
 
         raise TypeError(f"Unsupported data type ({type(data)}) was provided")
+
+    def to_url(self) -> str:
+        url = URL(self.host_url)
+
+        if Config.features.enable_custom_cloud_host and (
+            not url.host.endswith(DEFAULT_S3_HOST) and not url.host.endswith(DEFAULT_GCS_HOST)
+        ):
+            url = url.copy_with(path="/".join(["", self.bucket_name, url.path.lstrip("/")]))
+        else:
+            url = url.copy_with(host=f"{self.bucket_name}.{url.host}")
+
+        if self.path:
+            url = url.copy_with(path="/".join(["", url.path.lstrip("/"), self.path.lstrip("/")]))
+
+        return str(url)

packages/examples/cvat/exchange-oracle/src/services/cloud/utils.py

@@ -1,19 +1,9 @@
 from src.services.cloud.client import StorageClient
-from src.services.cloud.gcs import DEFAULT_GCS_HOST, GcsClient
-from src.services.cloud.s3 import DEFAULT_S3_HOST, S3Client
+from src.services.cloud.gcs import GcsClient
+from src.services.cloud.s3 import S3Client
 from src.services.cloud.types import BucketAccessInfo, CloudProviders
 
 
-def compose_bucket_url(
-    bucket_name: str, provider: CloudProviders, *, bucket_host: str | None = None
-) -> str:
-    match provider:
-        case CloudProviders.aws:
-            return f"https://{bucket_name}.{bucket_host or DEFAULT_S3_HOST}/"
-        case CloudProviders.gcs:
-            return f"https://{bucket_name}.{bucket_host or DEFAULT_GCS_HOST}/"
-
-
 def make_client(
     bucket_info: BucketAccessInfo,
 ) -> StorageClient:

packages/examples/cvat/recording-oracle/src/handlers/process_intermediate_results.py

@@ -266,7 +266,7 @@ def transform_item(self, item: dm.DatasetItem) -> dm.DatasetItem:
                     item = item.wrap(id=item.id[len(self._prefix) :])
                 return item
 
-        prefix = BucketAccessInfo.parse_obj(self.manifest.data.data_url).path.lstrip("/\\") + "/"
+        prefix = BucketAccessInfo.parse_obj(self.manifest.data.data_url).path.strip("/\\") + "/"
 
         # Remove prefixes if it can be done safely
         sample_ids = {sample.id for sample in merged_dataset}