13
13
URL_CUSTOM = "http://{ip}:{port}/v1/vision/custom/{custom_model}"
14
14
URL_OBJECT_DETECTION = "http://{ip}:{port}/v1/vision/detection"
15
15
URL_FACE_DETECTION = "http://{ip}:{port}/v1/vision/face"
16
- URL_FACE_REGISTRATION = "http://{ip}:{port}/v1/vision/face/register"
17
- URL_FACE_RECOGNITION = "http://{ip}:{port}/v1/vision/face/recognize"
16
+ URL_FACE_REGISTER = "http://{ip}:{port}/v1/vision/face/register"
17
+ URL_FACE_RECOGNIZE = "http://{ip}:{port}/v1/vision/face/recognize"
18
18
URL_SCENE_DETECTION = "http://{ip}:{port}/v1/vision/scene"
19
19
20
20
@@ -25,16 +25,14 @@ def format_confidence(confidence: Union[str, float]) -> float:
25
25
return round (float (confidence ) * 100 , 1 )
26
26
27
27
28
- def get_confidences_above_threshold (
29
- confidences : List [float ], confidence_threshold : float
30
- ) -> List [float ]:
28
+ def get_confidences_above_threshold (confidences : List [float ], confidence_threshold : float ) -> List [float ]:
31
29
"""Takes a list of confidences and returns those above a confidence_threshold."""
32
30
return [val for val in confidences if val >= confidence_threshold ]
33
31
34
32
35
- def get_recognised_faces (predictions : List [Dict ]) -> List [Dict ]:
33
+ def get_recognized_faces (predictions : List [Dict ]) -> List [Dict ]:
36
34
"""
37
- Get the recognised faces.
35
+ Get the recognized faces.
38
36
"""
39
37
try :
40
38
matched_faces = {
@@ -59,9 +57,7 @@ def get_object_confidences(predictions: List[Dict], target_object: str):
59
57
"""
60
58
Return the list of confidences of instances of target label.
61
59
"""
62
- confidences = [
63
- pred ["confidence" ] for pred in predictions if pred ["label" ] == target_object
64
- ]
60
+ confidences = [pred ["confidence" ] for pred in predictions if pred ["label" ] == target_object ]
65
61
return confidences
66
62
67
63
@@ -70,26 +66,17 @@ def get_objects_summary(predictions: List[Dict]):
70
66
Get a summary of the objects detected.
71
67
"""
72
68
objects = get_objects (predictions )
73
- return {
74
- target_object : len (get_object_confidences (predictions , target_object ))
75
- for target_object in objects
76
- }
69
+ return {target_object : len (get_object_confidences (predictions , target_object )) for target_object in objects }
77
70
78
71
79
- def post_image (
80
- url : str , image_bytes : bytes , api_key : str , timeout : int , data : dict = {}
81
- ):
72
+ def post_image (url : str , image_bytes : bytes , api_key : str , timeout : int , data : dict = {}):
82
73
"""Post an image to Deepstack."""
83
74
try :
84
- data ["api_key" ] = api_key
85
- response = requests .post (
86
- url , files = {"image" : image_bytes }, data = data , timeout = timeout
87
- )
75
+ data ["api_key" ] = api_key # Insert the api_key
76
+ response = requests .post (url , files = {"image" : image_bytes }, data = data , timeout = timeout )
88
77
return response
89
78
except requests .exceptions .Timeout :
90
- raise DeepstackException (
91
- f"Timeout connecting to Deepstack, current timeout is { timeout } seconds"
92
- )
79
+ raise DeepstackException (f"Timeout connecting to Deepstack, current timeout is { timeout } seconds" )
93
80
except requests .exceptions .ConnectionError as exc :
94
81
raise DeepstackException (f"Connection error: { exc } " )
95
82
@@ -106,35 +93,39 @@ def __init__(
106
93
api_key : str = "" ,
107
94
timeout : int = DEFAULT_TIMEOUT ,
108
95
url_detect : str = None ,
109
- url_recognise : str = None ,
96
+ url_recognize : str = None ,
110
97
url_register : str = None ,
111
98
):
112
99
113
100
self ._url_detect = url_detect
114
- self ._url_recognise = url_recognise
101
+ self ._url_recognize = url_recognize
115
102
self ._url_register = url_register
116
103
self ._api_key = api_key
117
104
self ._timeout = timeout
118
105
self ._response = None
119
106
120
107
def detect (self , image_bytes : bytes ):
121
- """Process image_bytes, performing detection ."""
108
+ """Process image_bytes and detect ."""
122
109
self ._response = None
123
- response = post_image (
124
- self ._url_detect , image_bytes , self ._api_key , self ._timeout
125
- )
110
+ response = post_image (self ._url_detect , image_bytes , self ._api_key , self ._timeout )
126
111
127
112
if not response .status_code == HTTP_OK :
128
- raise DeepstackException (
129
- f"Error from request, status code: { response .status_code } "
130
- )
113
+ raise DeepstackException (f"Error from request, status code: { response .status_code } " )
131
114
return
132
115
133
116
self ._response = response .json ()
134
117
if not self ._response ["success" ]:
135
118
error = self ._response ["error" ]
136
119
raise DeepstackException (f"Error from Deepstack: { error } " )
137
120
121
+ def recognize (self ):
122
+ """Process image_bytes and recognize."""
123
+ raise NotImplementedError
124
+
125
+ def register (self ):
126
+ """Perform a registration."""
127
+ raise NotImplementedError
128
+
138
129
@property
139
130
def predictions (self ):
140
131
"""Return the predictions."""
@@ -145,26 +136,15 @@ class DeepstackObject(Deepstack):
145
136
"""Work with objects"""
146
137
147
138
def __init__ (
148
- self ,
149
- ip : str ,
150
- port : str ,
151
- api_key : str = "" ,
152
- timeout : int = DEFAULT_TIMEOUT ,
153
- custom_model : str = None ,
139
+ self , ip : str , port : str , api_key : str = "" , timeout : int = DEFAULT_TIMEOUT , custom_model : str = None ,
154
140
):
155
141
if not custom_model :
156
142
super ().__init__ (
157
- api_key ,
158
- timeout ,
159
- url_detect = URL_OBJECT_DETECTION .format (ip = ip , port = port ),
143
+ api_key , timeout , url_detect = URL_OBJECT_DETECTION .format (ip = ip , port = port ),
160
144
)
161
145
else :
162
146
super ().__init__ (
163
- api_key ,
164
- timeout ,
165
- url_detect = URL_CUSTOM .format (
166
- ip = ip , port = port , custom_model = custom_model
167
- ),
147
+ api_key , timeout , url_detect = URL_CUSTOM .format (ip = ip , port = port , custom_model = custom_model ),
168
148
)
169
149
170
150
@property
@@ -177,16 +157,10 @@ class DeepstackScene(Deepstack):
177
157
"""Work with scenes"""
178
158
179
159
def __init__ (
180
- self ,
181
- ip : str ,
182
- port : str ,
183
- api_key : str = "" ,
184
- timeout : int = DEFAULT_TIMEOUT ,
160
+ self , ip : str , port : str , api_key : str = "" , timeout : int = DEFAULT_TIMEOUT ,
185
161
):
186
162
super ().__init__ (
187
- api_key ,
188
- timeout ,
189
- url_detect = URL_SCENE_DETECTION .format (ip = self ._ip , port = self ._port ),
163
+ api_key , timeout , url_detect = URL_SCENE_DETECTION .format (ip = self ._ip , port = self ._port ),
190
164
)
191
165
192
166
@property
@@ -199,26 +173,22 @@ class DeepstackFace(Deepstack):
199
173
"""Work with objects"""
200
174
201
175
def __init__ (
202
- self ,
203
- ip : str ,
204
- port : str ,
205
- api_key : str = "" ,
206
- timeout : int = DEFAULT_TIMEOUT ,
176
+ self , ip : str , port : str , api_key : str = "" , timeout : int = DEFAULT_TIMEOUT ,
207
177
):
208
178
super ().__init__ (
209
179
api_key ,
210
180
timeout ,
211
181
url_detect = URL_FACE_DETECTION .format (ip = self ._ip , port = self ._port ),
212
- url_register = URL_FACE_REGISTRATION .format (ip = self ._ip , port = self ._port ),
213
- url_recognise = URL_FACE_RECOGNITION .format (ip = self ._ip , port = self ._port ),
182
+ url_register = URL_FACE_REGISTER .format (ip = self ._ip , port = self ._port ),
183
+ url_recognize = URL_FACE_RECOGNIZE .format (ip = self ._ip , port = self ._port ),
214
184
)
215
185
216
186
@property
217
187
def predictions (self ):
218
188
"""Return the classifier attributes."""
219
189
return self ._response ["predictions" ]
220
190
221
- def register_face (self , name : str , image_bytes : bytes ):
191
+ def register (self , name : str , image_bytes : bytes ):
222
192
"""
223
193
Register a face name to a file.
224
194
"""
@@ -231,24 +201,20 @@ def register_face(self, name: str, image_bytes: bytes):
231
201
data = {"userid" : name },
232
202
)
233
203
234
- if response .status_code == 200 and response .json ()["success" ] == True :
204
+ if response .status_code == HTTP_OK and response .json ()["success" ] == True :
235
205
return
236
206
237
- elif response .status_code == 200 and response .json ()["success" ] == False :
207
+ elif response .status_code == HTTP_OK and response .json ()["success" ] == False :
238
208
error = response .json ()["error" ]
239
209
raise DeepstackException (f"Error from Deepstack: { error } " )
240
210
241
- def recognise (self , image_bytes : bytes ):
211
+ def recognize (self , image_bytes : bytes ):
242
212
"""Process image_bytes, performing recognition."""
243
213
244
- response = post_image (
245
- self ._url_recognise , image_bytes , self ._api_key , self ._timeout
246
- )
214
+ response = post_image (self ._url_recognize , image_bytes , self ._api_key , self ._timeout )
247
215
248
216
if not response .status_code == HTTP_OK :
249
- raise DeepstackException (
250
- f"Error from request, status code: { response .status_code } "
251
- )
217
+ raise DeepstackException (f"Error from request, status code: { response .status_code } " )
252
218
return
253
219
254
220
self ._response = response .json ()
0 commit comments