annotations upload updated

Author: VaghinakDev

pytest.ini

@@ -3,4 +3,4 @@ minversion = 3.7
 log_cli=true
 python_files = test_*.py
 ;pytest_plugins = ['pytest_profiling']
-addopts = -n auto --dist=loadscope
+;addopts = -n auto --dist=loadscope

requirements.txt

@@ -20,3 +20,5 @@ email-validator>=1.0.3
 nest-asyncio==1.5.4
 jsonschema==3.2.0
 pandas>=1.1.4
+aiofiles==0.8.0
+

src/superannotate/lib/app/analytics/aggregators.py

@@ -257,7 +257,7 @@ def aggregate_video_annotations_as_df(self, annotation_paths: List[str]):
                         )
                     if not timestamps:
                         raws.append(parameter_raw)
-                if not parameters and instance_type != 'tag':
+                if not parameters and instance_type != "tag":
                     raws.append(instance_raw)
             if not instances:
                 raws.append(raw_data)

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

@@ -20,6 +20,7 @@
 from typing import Set
 from typing import Tuple
 
+import aiofiles
 import boto3
 import jsonschema.validators
 import lib.core as constants
@@ -188,28 +189,28 @@ def prepare_annotation(self, annotation: dict, size) -> dict:
         )
         return annotation
 
-    def get_annotation(
+    async def get_annotation(
         self, path: str
     ) -> (Optional[Tuple[io.StringIO]], Optional[io.BytesIO]):
         mask = None
-
         if self._client_s3_bucket:
-            file = self.get_annotation_from_s3(self._client_s3_bucket, path)
-
+            content = self.get_annotation_from_s3(self._client_s3_bucket, path).read()
         else:
-            file = open(path)
+            async with aiofiles.open(path) as file:
+                content = await file.read()
             if self._project.type == constants.ProjectType.PIXEL.value:
-                mask = open(
+                async with aiofiles.open(
                     path.replace(
                         constants.PIXEL_ANNOTATION_POSTFIX,
                         constants.ANNOTATION_MASK_POSTFIX,
                     ),
                     "rb",
-                )
-        _tmp = file.read()
-        if not isinstance(_tmp, bytes):
-            _tmp = _tmp.encode("utf8")
-        file = io.BytesIO(_tmp)
+                ) as mask:
+                    mask = await mask.read()
+
+        if not isinstance(content, bytes):
+            content = content.encode("utf8")
+        file = io.BytesIO(content)
         file.seek(0)
         size = file.getbuffer().nbytes
         annotation = json.load(file)
@@ -393,7 +394,7 @@ async def upload_big_annotations(self):
             if item:
                 await self._upload_big_annotation(item)
             else:
-                await self._big_files_queue.put(None)
+                await self._big_files_queue.put_nowait(None)
                 break
 
     async def distribute_queues(self, items_to_upload: list):
@@ -403,28 +404,32 @@ async def distribute_queues(self, items_to_upload: list):
             for idx, (item, processed) in enumerate(data):
                 if not processed:
                     try:
-                        annotation, mask, size = self.get_annotation(item.path)
+                        annotation, mask, size = await self.get_annotation(item.path)
                         t_item = copy.copy(item)
                         annotation_file = io.StringIO()
                         json.dump(annotation, annotation_file)
                         annotation_file.seek(0)
-                        del annotation
                         t_item.data = annotation_file
                         t_item.mask = mask
-                        if size > BIG_FILE_THRESHOLD:
-                            if self._big_files_queue.qsize() > 4:
-                                continue
-                            else:
+                        while True:
+                            if size > BIG_FILE_THRESHOLD:
+                                if self._big_files_queue.qsize() > 32:
+                                    await asyncio.sleep(3)
+                                    continue
                                 self._big_files_queue.put_nowait(t_item)
-                        else:
-                            self._small_files_queue.put_nowait(t_item)
+                                break
+                            else:
+                                self._small_files_queue.put_nowait(t_item)
+                                break
                     except Exception as e:
                         self.reporter.log_debug(str(e))
                         self._report.failed_annotations.append(item.name)
-                    finally:
                         self.reporter.update_progress()
                         data[idx][1] = True
                         processed_count += 1
+                    self.reporter.update_progress()
+                    data[idx][1] = True
+                    processed_count += 1
         self._big_files_queue.put_nowait(None)
         self._small_files_queue.put_nowait(None)
 
@@ -434,12 +439,12 @@ async def run_workers(self, items_to_upload):
             self._small_files_queue = asyncio.Queue()
             await asyncio.gather(
                 self.distribute_queues(items_to_upload),
-                self.upload_small_annotations(),
                 self.upload_big_annotations(),
                 self.upload_big_annotations(),
                 self.upload_big_annotations(),
                 return_exceptions=True,
             )
+            await asyncio.gather(self.upload_small_annotations())
         except Exception as e:
             self.reporter.log_error(f"Error {str(e)}")
 

src/superannotate/lib/infrastructure/services.py

@@ -83,6 +83,7 @@ def __init__(
     def assets_provider_url(self):
         if self.api_url != constance.BACKEND_URL:
             return "https://assets-provider.devsuperannotate.com/api/v1.01/"
+            # return "https://e53a-178-160-196-42.ngrok.io/api/v1.01/"
         return "https://assets-provider.superannotate.com/api/v1/"
 
     @lru_cache(maxsize=32)
@@ -1418,10 +1419,8 @@ async def upload_big_annotation(
         data: io.StringIO,
         chunk_size: int,
     ) -> bool:
-        headers = self.default_headers
-        headers["Content-Type"] = "application/x-www-form-urlencoded"
         async with aiohttp.ClientSession(
-            connector=aiohttp.TCPConnector(ssl=False),
+            connector=aiohttp.TCPConnector(ssl=False), headers=self.default_headers
         ) as session:
             params = {
                 "team_id": team_id,
@@ -1438,12 +1437,12 @@ async def upload_big_annotation(
                     "project_id": project_id,
                     "folder_id": folder_id,
                 },
-                headers=headers,
             )
             if not start_response.ok:
                 raise AppException(str(await start_response.text()))
             process_info = await start_response.json()
             params["path"] = process_info["path"]
+            headers = copy.copy(self.default_headers)
             headers["upload_id"] = process_info["upload_id"]
             chunk_id = 1
             data_sent = False
@@ -1459,7 +1458,7 @@ async def upload_big_annotation(
                         ),
                         params=params,
                         headers=headers,
-                        data={"data_chunk": chunk},
+                        data=json.dumps({"data_chunk": chunk}),
                     )
                     if not response.ok:
                         raise AppException(str(await response.text()))

tests/data_set/sample_big_json_vector/aearth_mov_004.jpg.json

@@ -1 +1,14 @@
-{"metadata":{"name":"aearth_mov_008.jpg","projectId":1290,"isPredicted":false,"status":"InProgress","pinned":false,"annotatorEmail":null,"qaEmail":null},"instances":[],"tags":[],"comments":[]}
+{
+  "metadata": {
+    "name": "aearth_mov_008.jpg",
+    "projectId": 1290,
+    "isPredicted": false,
+    "status": "InProgress",
+    "pinned": false,
+    "annotatorEmail": null,
+    "qaEmail": null
+  },
+  "instances": [],
+  "tags": [],
+  "comments": []
+}

tests/integration/annotations/test_upload_annotations_from_folder_to_project.py

@@ -58,7 +58,7 @@ def test_4_annotation_folder_upload_download(self):
     def test_upload_big_annotations(self):
         sa.attach_items(
             self.PROJECT_NAME,
-            [{"name": f"aearth_mov_00{i}.jpg", "url": f"url_{i}"} for i in range(1, 2)]  # noqa
+            [{"name": f"aearth_mov_00{i}.jpg", "url": f"url_{i}"} for i in range(1, 6)]  # noqa
         )
         sa.create_annotation_classes_from_classes_json(
             self.PROJECT_NAME, f"{self.big_annotations_folder_path}/classes/classes.json"
@@ -68,4 +68,4 @@ def test_upload_big_annotations(self):
         )
         assert len(uploaded) == 5
         annotations = sa.get_annotations(self.PROJECT_NAME)
-        assert all([len(annotation["instances"]) > 1 for annotation in annotations])
+        assert [len(annotation["instances"]) > 1 for annotation in annotations].count(True) == 4

tests/integration/annotations/validations/test_vector_annotation_validation.py

@@ -69,6 +69,7 @@ def test_should_be_valid(self):
                 "instances": [
                     {
                         "type": "bbox",
+                        "createdAt": '2021-11-18T13.36.53.293Z',
                         "createdBy": {'email': 'arturn@superannotate.com', 'role': 'dmin'},
                         "points": {
                             "x1": 437.16,