Merge pull request #5 from robmarkcole/development

Development

Author: robmarkcole

.devcontainer/Dockerfile

@@ -115,3 +115,6 @@ dmypy.json
 # Pyre type checker
 .pyre/
 .DS_Store
+
+venv
+.vscode

.devcontainer/devcontainer.json

@@ -1,99 +1,117 @@
 """
 Deepstack core.
 """
-import imghdr
 import requests
 from PIL import Image
+from typing import Union, List, Set, Dict
 
 ## Const
-CLASSIFIER = "deepstack"
 HTTP_OK = 200
 HTTP_BAD_REQUEST = 400
 HTTP_UNAUTHORIZED = 401
-TIMEOUT = 30  # seconds
+DEFAULT_TIMEOUT = 10  # seconds
 
 
-def get_matched_faces(predictions: dict):
+def format_confidence(confidence: Union[str, float]) -> float:
+    """Takes a confidence from the API like 
+       0.55623 and returne 55.6 (%).
     """
-    Get the predicted faces and their confidence.
+    return round(float(confidence) * 100, 1)
+
+
+def get_confidences_above_threshold(
+    confidences: List[float], confidence_threshold: float
+) -> List[float]:
+    """Takes a list of confidences and returns those above a confidence_threshold."""
+    return [val for val in confidences if val >= confidence_threshold]
+
+
+def get_object_labels(predictions: List[Dict]) -> Set[str]:
     """
-    try:
-        matched_faces = {
-            face["userid"]: round(face["confidence"] * 100, 1)
-            for face in predictions
-            if not face["userid"] == "unknown"
-        }
-        return matched_faces
-    except:
-        return {}
+    Get a list of the unique object labels predicted.
+    """
+    labels = [pred["label"] for pred in predictions]
+    return set(labels)
 
 
-def is_valid_image(file_path: str):
+def get_label_confidences(predictions: List[Dict], target_label: str):
     """
-    Check file_path is valid image, using PIL then imghdr.
+    Return the list of confidences of instances of target label.
     """
-    try:
-        with Image.open(file_path):
-            pass
+    confidences = [
+        pred["confidence"] for pred in predictions if pred["label"] == target_label
+    ]
+    return confidences
 
-        image_extension = imghdr.what(file_path)
-        if image_extension in ["jpeg", ".jpg", ".png"]:
-            return True
-        return False
-    except Exception as exc:
-        print(exc)
-        return False
 
+def get_objects_summary(predictions: List[Dict]):
+    """
+    Get a summary of the objects detected.
+    """
+    labels = get_object_labels(predictions)
+    return {
+        label: len(get_label_confidences(predictions, target_label=label))
+        for label in labels
+    }
 
-def post_image(url: str, image: bytes):
-    """Post an image to the classifier."""
+
+def post_image(url: str, image: bytes, api_key: str, timeout: int):
+    """Post an image to Deepstack."""
     try:
-        response = requests.post(url, files={"image": image}, timeout=TIMEOUT)
+        response = requests.post(
+            url, files={"image": image}, data={"api_key": api_key}, timeout=timeout
+        )
         return response
-    except requests.exceptions.ConnectionError:
-        print("ConnectionError: Is %s running?", CLASSIFIER)
-        return None
     except requests.exceptions.Timeout:
-        print("Timeout error from %s", CLASSIFIER)
-        return None
+        raise DeepstackException(
+            f"Timeout connecting to Deepstack, current timeout is {timeout} seconds"
+        )
+
 
+class DeepstackException(Exception):
+    pass
 
-class DeepstackFace:
-    """Work with faces."""
 
-    def __init__(self, ip_address: str, port: str):
+class DeepstackObject:
+    """The object detection API locates and classifies 80 
+    different kinds of objects in a single image.."""
 
-        self._url_check = "http://{}:{}/v1/vision/face/recognize".format(
+    def __init__(
+        self,
+        ip_address: str,
+        port: str,
+        api_key: str = "",
+        timeout: int = DEFAULT_TIMEOUT,
+    ):
+
+        self._url_object_detection = "http://{}:{}/v1/vision/detection".format(
             ip_address, port
         )
-
-        self._faces = None
-        self._matched = {}
+        self._api_key = api_key
+        self._timeout = timeout
+        self._predictions = []
 
     def process_file(self, file_path: str):
         """Process an image file."""
-        if is_valid_image(file_path):
-            with open(file_path, "rb") as image_bytes:
-                self.process_image_bytes(image_bytes)
+        with open(file_path, "rb") as image_bytes:
+            self.process_image_bytes(image_bytes)
 
     def process_image_bytes(self, image_bytes: bytes):
         """Process an image."""
-        response = post_image(self._url_check, image_bytes)
-        if response:
-            if response.status_code == HTTP_OK:
-                predictions_json = response.json()["predictions"]
-                self._faces = len(predictions_json)
-                self._matched = get_matched_faces(predictions_json)
+        self._predictions = []
+
+        response = post_image(
+            self._url_object_detection, image_bytes, self._api_key, self._timeout
+        )
 
-        else:
-            self._faces = None
-            self._matched = {}
+        if response.status_code == HTTP_OK:
+            if response.json()["success"]:
+                self._predictions = response.json()["predictions"]
+            else:
+                error = response.json()["error"]
+                raise DeepstackException(f"Error from Deepstack: {error}")
 
     @property
-    def attributes(self):
+    def predictions(self):
         """Return the classifier attributes."""
-        return {
-            "faces": self._faces,
-            "matched_faces": self._matched,
-            "total_matched_faces": len(self._matched),
-        }
+        return self._predictions