Skip to content

Commit 2db7c76

Browse files
VaghinakDevVaghinakDev
committed
Fix get anntoations
1 parent 9f9d7f5 commit 2db7c76

File tree

8 files changed

+116
-111
lines changed

8 files changed

+116
-111
lines changed

‎pytest.ini‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,4 @@ minversion = 3.7
33
log_cli=true
44
python_files = test_*.py
55
;pytest_plugins = ['pytest_profiling']
6-
;addopts = -n auto --dist=loadscope
6+
addopts = -n auto --dist=loadscope

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -176,9 +176,7 @@ def __add_attributes_to_raws(self, raws, attributes, element_raw):
176176
for attribute_id, attribute in enumerate(attributes):
177177
attribute_raw = copy.copy(element_raw)
178178
attribute_raw.attributeId = attribute_id
179-
attribute_raw.attributeGroupName = attribute.get(
180-
"groupName"
181-
)
179+
attribute_raw.attributeGroupName = attribute.get("groupName")
182180
attribute_raw.attributeName = attribute.get("name")
183181
raws.append(attribute_raw)
184182
if not attributes:
@@ -240,8 +238,8 @@ def aggregate_video_annotations_as_df(self, annotation_paths: List[str]):
240238
)
241239
instance_raw.pointLabels = instance["meta"].get("pointLabels")
242240
parameters = instance.get("parameters", [])
243-
if instance_raw.type == 'tag':
244-
attributes = instance["meta"].get("attributes",[])
241+
if instance_raw.type == "tag":
242+
attributes = instance["meta"].get("attributes", [])
245243
raws = self.__add_attributes_to_raws(raws, attributes, instance_raw)
246244
for parameter_id, parameter in enumerate(parameters):
247245
parameter_raw = copy.copy(instance_raw)
@@ -254,7 +252,9 @@ def aggregate_video_annotations_as_df(self, annotation_paths: List[str]):
254252
timestamp_raw.timestampId = timestamp_id
255253
timestamp_raw.meta = self.MAPPERS[instance_type](timestamp)
256254
attributes = timestamp.get("attributes", [])
257-
raws = self.__add_attributes_to_raws(raws, attributes, timestamp_raw)
255+
raws = self.__add_attributes_to_raws(
256+
raws, attributes, timestamp_raw
257+
)
258258
if not timestamps:
259259
raws.append(parameter_raw)
260260
if not parameters:

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

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -949,6 +949,7 @@ def __init__(
949949
items: BaseReadOnlyRepository,
950950
folders: BaseReadOnlyRepository,
951951
classes: BaseReadOnlyRepository,
952+
images: BaseManageableRepository,
952953
callback: Callable = None,
953954
):
954955
super().__init__(reporter)
@@ -962,6 +963,7 @@ def __init__(
962963
self._folders = folders
963964
self._classes = classes
964965
self._callback = callback
966+
self._images = images
965967

966968
def validate_item_names(self):
967969
if self._item_names:
@@ -1050,12 +1052,21 @@ def execute(self):
10501052

10511053
if not folders:
10521054
loop = asyncio.new_event_loop()
1055+
if not self._item_names:
1056+
condition = (
1057+
Condition("team_id", self._project.team_id, EQ)
1058+
& Condition("project_id", self._project.id, EQ)
1059+
& Condition("folder_id", self._folder.uuid, EQ)
1060+
)
1061+
item_names = [item.name for item in self._images.get_all(condition)]
1062+
else:
1063+
item_names = self._item_names
10531064
count = loop.run_until_complete(
10541065
self._backend_client.download_annotations(
10551066
team_id=self._project.team_id,
10561067
project_id=self._project.id,
10571068
folder_id=self._folder.uuid,
1058-
items=self._item_names,
1069+
items=item_names,
10591070
reporter=self.reporter,
10601071
download_path=f"{export_path}{'/' + self._folder.name if not self._folder.is_root else ''}",
10611072
postfix=postfix,
@@ -1066,12 +1077,21 @@ def execute(self):
10661077
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
10671078
coroutines = []
10681079
for folder in folders:
1080+
if not self._item_names:
1081+
condition = (
1082+
Condition("team_id", self._project.team_id, EQ)
1083+
& Condition("project_id", self._project.id, EQ)
1084+
& Condition("folder_id", folder.uuid, EQ)
1085+
)
1086+
item_names = [item.name for item in self._images.get_all(condition)]
1087+
else:
1088+
item_names = self._item_names
10691089
coroutines.append(
10701090
self._backend_client.download_annotations(
10711091
team_id=self._project.team_id,
10721092
project_id=self._project.id,
10731093
folder_id=folder.uuid,
1074-
items=self._item_names,
1094+
items=item_names,
10751095
reporter=self.reporter,
10761096
download_path=f"{export_path}{'/' + folder.name if not folder.is_root else ''}", # noqa
10771097
postfix=postfix,

‎src/superannotate/lib/core/video_convertor.py‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,6 @@ def pairwise(data: list):
9999
next(b, None)
100100
return zip(a, b)
101101

102-
103102
def get_median(self, annotations: List[dict]) -> dict:
104103
if len(annotations) >= 1:
105104
return annotations[0]

‎src/superannotate/lib/infrastructure/controller.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1633,6 +1633,7 @@ def download_annotations(
16331633
recursive=recursive,
16341634
item_names=item_names,
16351635
items=self.items,
1636+
images=self.images,
16361637
folders=self.folders,
16371638
classes=AnnotationClassRepository(
16381639
service=self._backend_client, project=project

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

Lines changed: 18 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1182,32 +1182,26 @@ async def download_annotations(
11821182
callback: Callable = None,
11831183
) -> int:
11841184

1185-
async with aiohttp.ClientSession(
1186-
raise_for_status=True,
1185+
query_params = {
1186+
"team_id": team_id,
1187+
"project_id": project_id,
1188+
"folder_id": folder_id,
1189+
}
1190+
handler = StreamedAnnotations(
11871191
headers=self.default_headers,
1188-
connector=aiohttp.TCPConnector(ssl=False),
1189-
) as session:
1190-
query_params = {
1191-
"team_id": team_id,
1192-
"project_id": project_id,
1193-
"folder_id": folder_id,
1194-
}
1195-
handler = StreamedAnnotations(
1196-
self.default_headers,
1197-
reporter,
1198-
map_function=lambda x: {"image_names": x},
1199-
callback=callback,
1200-
)
1192+
reporter=reporter,
1193+
map_function=lambda x: {"image_names": x},
1194+
callback=callback,
1195+
)
12011196

1202-
return await handler.download_data(
1203-
url=urljoin(self.assets_provider_url, self.URL_GET_ANNOTATIONS),
1204-
data=items,
1205-
params=query_params,
1206-
chunk_size=self.DEFAULT_CHUNK_SIZE,
1207-
download_path=download_path,
1208-
postfix=postfix,
1209-
session=session,
1210-
)
1197+
return await handler.download_data(
1198+
url=urljoin(self.assets_provider_url, self.URL_GET_ANNOTATIONS),
1199+
data=items,
1200+
params=query_params,
1201+
chunk_size=self.DEFAULT_CHUNK_SIZE,
1202+
download_path=download_path,
1203+
postfix=postfix,
1204+
)
12111205

12121206
def get_integrations(self, team_id: int) -> List[dict]:
12131207
get_integrations_url = urljoin(

‎src/superannotate/lib/infrastructure/stream_data_handler.py‎

Lines changed: 68 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,9 @@
88

99
from lib.core.reporter import Reporter
1010

11+
_seconds = 2 ** 10
12+
TIMEOUT = aiohttp.ClientTimeout(total=_seconds, sock_connect=_seconds, sock_read=_seconds)
13+
1114

1215
class StreamedAnnotations:
1316
DELIMITER = b"\\n;)\\n"
@@ -34,15 +37,12 @@ async def fetch(
3437
data: dict = None,
3538
params: dict = None,
3639
):
37-
kwargs = {
38-
"params": params,
39-
"json": {
40-
"folder_id": params.pop("folder_id")
41-
}
42-
}
40+
kwargs = {"params": params, "json": {"folder_id": params.pop("folder_id")}}
4341
if data:
4442
kwargs["json"].update(data)
45-
response = await session._request(method, url, **kwargs, raise_for_status=True)
43+
response = await session._request(
44+
method, url, **kwargs, raise_for_status=True, timeout=TIMEOUT
45+
)
4646
buffer = b""
4747
async for line in response.content.iter_any():
4848
slices = line.split(self.DELIMITER)
@@ -63,49 +63,59 @@ async def fetch(
6363
async def process_chunk(
6464
self,
6565
method: str,
66-
session: aiohttp.ClientSession,
6766
url: str,
6867
data: dict = None,
6968
params: dict = None,
69+
verify_ssl=False,
7070
):
71-
async for annotation in self.fetch(
72-
method,
73-
session,
74-
url,
75-
self._process_data(data),
76-
params=params,
77-
):
78-
self._annotations.append(
79-
self._callback(annotation) if self._callback else annotation
80-
)
71+
async with aiohttp.ClientSession(
72+
headers=self._headers,
73+
timeout=TIMEOUT,
74+
connector=aiohttp.TCPConnector(ssl=verify_ssl, keepalive_timeout=2 ** 32),
75+
) as session:
76+
async for annotation in self.fetch(
77+
method,
78+
session,
79+
url,
80+
self._process_data(data),
81+
params=params,
82+
):
83+
self._annotations.append(
84+
self._callback(annotation) if self._callback else annotation
85+
)
8186

8287
async def store_chunk(
8388
self,
8489
method: str,
85-
session: aiohttp.ClientSession,
8690
url: str,
8791
download_path,
8892
postfix,
8993
data: dict = None,
9094
params: dict = None,
9195
):
92-
async for annotation in self.fetch(
93-
method,
94-
session,
95-
url,
96-
self._process_data(data),
97-
params=params,
98-
):
99-
self._annotations.append(
100-
self._callback(annotation) if self._callback else annotation
101-
)
102-
self._store_annotation(
103-
download_path,
104-
postfix,
105-
annotation,
106-
self._callback,
107-
)
108-
self._items_downloaded += 1
96+
async with aiohttp.ClientSession(
97+
raise_for_status=True,
98+
headers=self._headers,
99+
timeout=TIMEOUT,
100+
connector=aiohttp.TCPConnector(ssl=False, keepalive_timeout=2 ** 32),
101+
) as session:
102+
async for annotation in self.fetch(
103+
method,
104+
session,
105+
url,
106+
self._process_data(data),
107+
params=params,
108+
):
109+
self._annotations.append(
110+
self._callback(annotation) if self._callback else annotation
111+
)
112+
self._store_annotation(
113+
download_path,
114+
postfix,
115+
annotation,
116+
self._callback,
117+
)
118+
self._items_downloaded += 1
109119

110120
async def get_data(
111121
self,
@@ -116,19 +126,19 @@ async def get_data(
116126
chunk_size: int = 5000,
117127
verify_ssl: bool = False,
118128
):
119-
async with aiohttp.ClientSession(
120-
headers=self._headers,
121-
connector=aiohttp.TCPConnector(ssl=verify_ssl),
122-
) as session:
123-
params["limit"] = chunk_size
124-
await asyncio.gather(
125-
*[
126-
self.process_chunk(
127-
method=method, session=session, url=url, data=data[i: i + chunk_size],
128-
params=copy.copy(params)
129-
) for i in
130-
range(0, len(data), chunk_size)]
131-
)
129+
params["limit"] = chunk_size
130+
await asyncio.gather(
131+
*[
132+
self.process_chunk(
133+
method=method,
134+
url=url,
135+
data=data[i: i + chunk_size],
136+
params=copy.copy(params),
137+
verify_ssl=verify_ssl,
138+
)
139+
for i in range(0, len(data), chunk_size)
140+
]
141+
)
132142

133143
return self._annotations
134144

@@ -150,7 +160,6 @@ async def download_data(
150160
data: list,
151161
download_path: str,
152162
postfix: str,
153-
session,
154163
method: str = "post",
155164
params=None,
156165
chunk_size: int = 5000,
@@ -162,9 +171,14 @@ async def download_data(
162171
await asyncio.gather(
163172
*[
164173
self.store_chunk(
165-
method=method, session=session, url=url, data=data[i: i + chunk_size], params=copy.copy(params),
166-
download_path=download_path, postfix=postfix
167-
) for i in
168-
range(0, len(data), chunk_size)]
174+
method=method,
175+
url=url,
176+
data=data[i: i + chunk_size], # noqa
177+
params=copy.copy(params),
178+
download_path=download_path,
179+
postfix=postfix,
180+
)
181+
for i in range(0, len(data), chunk_size)
182+
]
169183
)
170184
return self._items_downloaded

‎tests/integration/annotations/test_.py‎

Lines changed: 0 additions & 23 deletions
This file was deleted.

0 commit comments

Comments
 (0)