in inference-units/tps.py [0:0]
def __init__(self, *args, **kwargs):
for (dirpath, dirnames, filenames) in walk(image_base_path):
self.images = [Path(image_base_path) / filename for filename in filenames]
break
super(WebserviceUser, self).__init__(*args, **kwargs)
config = Config(
retries={
'max_attempts': 1,
'mode': 'standard'
}
)
self.client = boto3.client(
'lookoutvision',
aws_region,
config=config
)
def detection_tests(image_path, arg):
try:
start_time = time.time()
with open(image_path, 'rb') as image:
r = self.client.detect_anomalies(ProjectName=project_name, ContentType='image/jpeg',Body=image.read(), ModelVersion=model_version)
except Exception as exception:
total_time = int((time.time() - start_time) * 1000)
self.environment.events.request_failure.fire(request_type="GET",
name=inspect.stack()[0][3], response_time=total_time,
response_length=0, exception=exception)
else:
total_time = int((time.time() - start_time) * 1000)
self.environment.events.request_success.fire(request_type="GET", name=inspect.stack()[0][3],
response_time=total_time, response_length=0)
self.tasks = [functools.partial(detection_tests, image) for image in self.images]