change upload annotations assets data

Author: Narek Mkhitaryan

src/superannotate/lib/core/serviceproviders.py

@@ -367,7 +367,7 @@ async def upload_small_annotations(
         self,
         project: entities.ProjectEntity,
         folder: entities.FolderEntity,
-        items_name_file_map: Dict[str, io.StringIO],
+        items_name_data_map: Dict[str, dict],
     ) -> UploadAnnotationsResponse:
         raise NotImplementedError
 

src/superannotate/lib/core/usecases/annotations.py

@@ -47,6 +47,7 @@
 from lib.core.types import PriorityScoreEntity
 from lib.core.usecases.base import BaseReportableUseCase
 from lib.core.video_convertor import VideoFrameGenerator
+from lib.infrastructure.utils import divide_to_chunks
 from pydantic import BaseModel
 
 logger = logging.getLogger("sa")
@@ -119,11 +120,6 @@ def get_or_raise(response: ServiceResponse):
         raise AppException(response.error)
 
 
-def divide_to_chunks(it, size):
-    it = iter(it)
-    return iter(lambda: tuple(islice(it, size)), ())
-
-
 def log_report(
     report: Report,
 ):
@@ -148,7 +144,6 @@ class ItemToUpload(BaseModel):
     item: BaseItemEntity
     annotation_json: Optional[dict]
     path: Optional[str]
-    file: Optional[io.StringIO]
     file_size: Optional[int]
     mask: Optional[io.BytesIO]
 
@@ -186,7 +181,7 @@ async def upload_small_annotations(
     report: Report,
     callback: Callable = None,
 ):
-    async def upload(_chunk):
+    async def upload(_chunk: List[ItemToUpload]):
         failed_annotations, missing_classes, missing_attr_groups, missing_attrs = (
             [],
             [],
@@ -197,7 +192,7 @@ async def upload(_chunk):
             response = await service_provider.annotations.upload_small_annotations(
                 project=project,
                 folder=folder,
-                items_name_file_map={i.item.name: i.file for i in chunk},
+                items_name_data_map={i.item.name: i.annotation_json for i in chunk},
             )
             if response.ok:
                 if response.data.failed_items:  # noqa
@@ -221,9 +216,9 @@ async def upload(_chunk):
             reporter.update_progress(len(chunk))
 
     _size = 0
-    chunk = []
+    chunk: List[ItemToUpload] = []
     while True:
-        item_data = await queue.get()
+        item_data: ItemToUpload = await queue.get()
         queue.task_done()
         if not item_data:
             queue.put_nowait(None)
@@ -253,11 +248,14 @@ async def upload_big_annotations(
 ):
     async def _upload_big_annotation(item_data: ItemToUpload) -> Tuple[str, bool]:
         try:
+            buff = io.StringIO()
+            json.dump(item_data.annotation_json, buff, allow_nan=False)
+            buff.seek(0)
             is_uploaded = await service_provider.annotations.upload_big_annotation(
                 project=project,
                 folder=folder,
                 item_id=item_data.item.id,
-                data=item_data.file,
+                data=buff,
                 chunk_size=5 * 1024 * 1024,
             )
             if is_uploaded and callback:
@@ -335,15 +333,14 @@ async def distribute_queues(self, items_to_upload: List[ItemToUpload]):
             for idx, (item_to_upload, processed) in enumerate(data):
                 if not processed:
                     try:
-                        item_to_upload.file = io.StringIO()
+                        file = io.StringIO()
                         json.dump(
                             item_to_upload.annotation_json,
-                            item_to_upload.file,
+                            file,
                             allow_nan=False,
                         )
-                        item_to_upload.file.seek(0, os.SEEK_END)
-                        item_to_upload.file_size = item_to_upload.file.tell()
-                        item_to_upload.file.seek(0)
+                        file.seek(0, os.SEEK_END)
+                        item_to_upload.file_size = file.tell()
                         while True:
                             if item_to_upload.file_size > BIG_FILE_THRESHOLD:
                                 if self._big_files_queue.qsize() > 32:
@@ -723,13 +720,10 @@ async def distribute_queues(self, items_to_upload: List[ItemToUpload]):
                 if not processed:
                     try:
                         (
-                            annotation,
+                            item_to_upload.annotation_json,
                             item_to_upload.mask,
                             item_to_upload.file_size,
                         ) = await self.get_annotation(item_to_upload.path)
-                        item_to_upload.file = io.StringIO()
-                        json.dump(annotation, item_to_upload.file, allow_nan=False)
-                        item_to_upload.file.seek(0)
                         while True:
                             if item_to_upload.file_size > BIG_FILE_THRESHOLD:
                                 if self._big_files_queue.qsize() > 32:
@@ -1018,7 +1012,7 @@ def execute(self):
                         self._service_provider.annotations.upload_small_annotations(
                             project=self._project,
                             folder=self._folder,
-                            items_name_file_map={self._image.name: annotation_file},
+                            items_name_data_map={self._image.name: annotation_json},
                         )
                     )
                     if response.ok:

src/superannotate/lib/infrastructure/services/annotation.py

@@ -162,8 +162,8 @@ def get_upload_chunks(
         response = self.client.request(
             url=urljoin(self.assets_provider_url, self.URL_CLASSIFY_ITEM_SIZE),
             method="POST",
-            params={"project_id": project.id, "limit": len(item_ids)},
-            data={"item_ids": item_ids},
+            params={"limit": len(item_ids)},
+            data={"project_id": project.id, "item_ids": item_ids},
         )
         if not response.ok:
             raise AppException(response.error)
@@ -246,12 +246,12 @@ async def upload_small_annotations(
         self,
         project: entities.ProjectEntity,
         folder: entities.FolderEntity,
-        items_name_file_map: Dict[str, io.StringIO],
+        items_name_data_map: Dict[str, dict],
     ) -> UploadAnnotationsResponse:
         url = urljoin(
             self.assets_provider_url,
             (
-                f"{self.URL_UPLOAD_ANNOTATIONS}?{'&'.join(f'image_names[]={item_name}' for item_name in items_name_file_map.keys())}"
+                f"{self.URL_UPLOAD_ANNOTATIONS}?{'&'.join(f'image_names[]={item_name}' for item_name in items_name_data_map.keys())}"
             ),
         )
 
@@ -261,12 +261,19 @@ async def upload_small_annotations(
             headers=headers,
             connector=aiohttp.TCPConnector(ssl=False),
         ) as session:
-            data = aiohttp.FormData(quote_fields=False)
-            for key, file in items_name_file_map.items():
-                file.seek(0)
-                data.add_field(
+            form_data = aiohttp.FormData(
+                quote_fields=False,
+            )
+            tmp = {}
+            for name, data in items_name_data_map.items():
+                tmp[name] = io.StringIO()
+                json.dump({"data": data}, tmp[name], allow_nan=False)
+                tmp[name].seek(0)
+
+            for key, data in tmp.items():
+                form_data.add_field(
                     key,
-                    bytes(file.read(), "ascii"),
+                    data,
                     filename=key,
                     content_type="application/json",
                 )
@@ -276,7 +283,9 @@ async def upload_small_annotations(
                 "project_id": project.id,
                 "folder_id": folder.id,
             }
-            _response = await session.request("post", url, params=params, data=data)
+            _response = await session.request(
+                "post", url, params=params, data=form_data
+            )
             if not _response.ok:
                 logger.debug(f"Status code {str(_response.status)}")
                 logger.debug(await _response.text())
@@ -324,7 +333,8 @@ async def upload_big_annotation(
                 params["chunk_id"] = chunk_id
                 if chunk:
                     data_sent = True
-                    response = await session.post(
+                    response = await session.request(
+                        "post",
                         urljoin(
                             self.assets_provider_url,
                             self.URL_START_FILE_SEND_PART.format(item_id=item_id),
@@ -341,7 +351,8 @@ async def upload_big_annotation(
                 if len(chunk) < chunk_size:
                     break
             del params["chunk_id"]
-            response = await session.post(
+            response = await session.request(
+                "post",
                 urljoin(
                     self.assets_provider_url,
                     self.URL_START_FILE_SEND_FINISH.format(item_id=item_id),
@@ -352,7 +363,8 @@ async def upload_big_annotation(
             if not response.ok:
                 raise AppException(str(await response.text()))
             del params["path"]
-            response = await session.post(
+            response = await session.request(
+                "post",
                 urljoin(
                     self.assets_provider_url,
                     self.URL_START_FILE_SYNC.format(item_id=item_id),
@@ -363,7 +375,8 @@ async def upload_big_annotation(
             if not response.ok:
                 raise AppException(str(await response.text()))
             while True:
-                response = await session.get(
+                response = await session.request(
+                    "get",
                     urljoin(
                         self.assets_provider_url,
                         self.URL_START_FILE_SYNC_STATUS.format(item_id=item_id),

src/superannotate/lib/infrastructure/utils.py

@@ -1,3 +1,4 @@
+from itertools import islice
 from pathlib import Path
 from typing import Optional
 from typing import Tuple
@@ -6,6 +7,11 @@
 from superannotate.lib.app.exceptions import PathError
 
 
+def divide_to_chunks(it, size):
+    it = iter(it)
+    return iter(lambda: tuple(islice(it, size)), ())
+
+
 def split_project_path(project_path: str) -> Tuple[str, Optional[str]]:
     path = Path(project_path)
     if len(path.parts) > 3: