-
Notifications
You must be signed in to change notification settings - Fork 123
/
Copy pathbenchmark_snapshot.py
172 lines (149 loc) · 5.08 KB
/
benchmark_snapshot.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
import argparse
import math
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from timeit import default_timer
import psutil
import requests
from PIL import Image
from psutil import cpu_percent, process_iter
from plate_recognition import recognition_api
def parse_arguments():
parser = argparse.ArgumentParser(description="Benchmark SDK.")
parser.add_argument(
"--sdk-url",
help="Url to self hosted sdk For example, http://localhost:8080",
default="http://localhost:8080",
)
parser.add_argument(
"--threads", help="Use thread to parallelize API calls", default=4, type=int
)
parser.add_argument("--image", default="assets/car-4k.jpg")
parser.add_argument("--mmc", action="store_true")
parser.add_argument("--iterations", default=50, type=int)
parser.add_argument("--blur", action="store_true")
return parser.parse_args()
def print_table(results):
if not results:
return
print("| Mode | Resolution | Speed | l_min | l_max |")
print("| -------- | ---------- | ------- | ------ | ------ |")
for result in results:
print(
"| {mode:8s} | {resolution:10s} | {avg:7.1f} | {min:6.1f} | {max:6.1f} |".format(
**result
)
)
def blur_api(url, fp):
"""
Upload an Image to url for burring
"""
response = requests.post(url, files={"upload": fp})
if response.status_code < 200 or response.status_code > 300:
if response.status_code == 400:
msg = response.json().get("error")
else:
msg = response.text
raise Exception(f"Error performing blur: {msg}")
blur_data = response.json().get("blur")
if blur_data is None:
raise Exception(
"Error - ensure blurring on server is enabled - "
"https://guides.platerecognizer.com/docs/blur/api-reference#post-parameters"
)
return blur_data
def call_duration(path, sdk_url, config, mmc, blur):
now = default_timer()
with open(path, "rb") as fp:
if blur:
blur_api(sdk_url, fp)
else:
recognition_api(
fp,
sdk_url=sdk_url,
config=config,
mmc="true" if mmc else "false",
)
return (default_timer() - now) * 1000
def benchmark(args, executor):
image = Image.open(args.image)
for resolution in [(800, 600), (1280, 720), (1920, 1080), (2560, 1440)]:
image.resize(resolution).save("/tmp/platerec-benchmark.jpg")
if args.blur:
configs = [{}]
else:
configs = [{}, dict(mode="fast")]
for config in configs:
now = default_timer()
stats = list(
executor.map(
partial(
call_duration,
sdk_url=args.sdk_url,
config=config,
mmc=args.mmc,
blur=args.blur,
),
["/tmp/platerec-benchmark.jpg"] * args.iterations,
)
)
duration = (default_timer() - now) * 1000
yield dict(
resolution="%sx%s" % resolution,
mode=config.get("mode", "regular"),
min=min(stats),
max=max(stats),
avg=duration / args.iterations,
)
def mem_usage():
usage = {}
for process in process_iter():
try:
if "main.py" in process.cmdline() or "start.sh" in process.cmdline():
usage[process.pid] = process.memory_info()
except psutil.ZombieProcess:
pass
return usage
def convert_size(size_bytes):
if size_bytes == 0:
return "0B"
sign = ""
if size_bytes < 0:
size_bytes *= -1
sign = "-"
size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")
i = int(math.floor(math.log(size_bytes, 1024)))
p = math.pow(1024, i)
s = round(size_bytes / p, 2)
return f"{sign}{s} {size_name[i]}"
def main():
args = parse_arguments()
initial_mem = mem_usage()
cpu_percent() # first time this is called it will return a meaningless 0.0
with ThreadPoolExecutor(max_workers=args.threads) as executor:
# Warmup
list(
executor.map(
partial(
call_duration,
sdk_url=args.sdk_url,
config={},
mmc=args.mmc,
blur=args.blur,
),
[args.image] * 2,
)
)
# Benchmark
results = list(benchmark(args, executor))
# Memory Usage
print(f"CPU: {cpu_percent()}%")
for pid, mem in mem_usage().items():
print(
f"PID: {pid:5}, "
f"RES {convert_size(mem.rss):10} ({convert_size(mem.rss - initial_mem[pid].rss):10}), "
f"SHR {convert_size(mem.shared):10} ({convert_size(mem.shared - initial_mem[pid].shared):10})"
)
print_table(results)
if __name__ == "__main__":
main()