WIP: Adds facebox_face_detect

homeassistant/components/image_processing/facebox_face_detect.py

@@ -0,0 +1,107 @@
+"""
+Component that will perform facial detection via a local facebox instance.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/image_processing.facebox_face_detect
+"""
+import base64
+import requests
+import logging
+import time
+import voluptuous as vol
+
+from homeassistant.core import split_entity_id
+import homeassistant.helpers.config_validation as cv
+from homeassistant.components.image_processing import (
+    PLATFORM_SCHEMA, CONF_SOURCE, CONF_ENTITY_ID,
+    CONF_NAME)
+from homeassistant.const import (CONF_IP_ADDRESS, CONF_PORT)
+from homeassistant.components.image_processing.microsoft_face_identify import (
+    ImageProcessingFaceEntity)
+
+_LOGGER = logging.getLogger(__name__)
+
+ROUNDING_DECIMALS = 2
+
+PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
+    vol.Required(CONF_IP_ADDRESS): cv.string,
+    vol.Required(CONF_PORT): cv.string,
+})
+
+
+def setup_platform(hass, config, add_devices, discovery_info=None):
+    """Set up the classifier."""
+    entities = []
+    for camera in config[CONF_SOURCE]:
+        entities.append(FaceboxFaceDetectEntity(
+            config[CONF_IP_ADDRESS],
+            config[CONF_PORT],
+            camera[CONF_ENTITY_ID],
+            camera.get(CONF_NAME)
+        ))
+    add_devices(entities)
+
+
+class FaceboxFaceDetectEntity(ImageProcessingFaceEntity):
+    """Perform a classification via a Facebox."""
+
+    def __init__(self, ip, port, camera_entity, name=None):
+        """Init with the API key and model id"""
+        super().__init__()
+        self._url = "http://{}:{}/facebox/check".format(ip, port)
+        self._camera = camera_entity
+        if name:
+            self._name = name
+        else:
+            self._name = "Facebox {0}".format(
+                split_entity_id(camera_entity)[1])
+        self._response_time = None
+        self.total_faces = 0
+        self.faces = []
+
+    def process_image(self, image):
+        """Process an image."""
+        timer_start = time.perf_counter()
+        response = {}
+        try:
+            response = requests.post(
+                self._url,
+                json=self.encode_image(image),
+                timeout=30
+                ).json()
+        except requests.exceptions.ConnectionError:
+            _LOGGER.error("ConnectionError: Is Facebox running?")
+            response['success'] = False
+
+        if response['success']:
+            elapsed_time = time.perf_counter() - timer_start
+            self._response_time = "{} seconds".format(
+                    str(round(elapsed_time, 1)))
+            self.total_faces = response['facesCount']
+            self.faces = response['faces']
+
+        else:
+            self.total_faces = "Request_failed"
+            self.faces = []
+
+    def encode_image(self, image):
+        """base64 encode an image stream."""
+        base64_img = base64.b64encode(image).decode('ascii')
+        return {"base64": base64_img}
+
+    @property
+    def camera_entity(self):
+        """Return camera entity id from process pictures."""
+        return self._camera
+
+    @property
+    def name(self):
+        """Return the name of the sensor."""
+        return self._name
+
+    @property
+    def device_state_attributes(self):
+        """Return the other state attributes."""
+        return {
+            'response_time': self._response_time,
+        }

tests/components/image_processing/test_facebox_face_detect.py

@@ -0,0 +1,68 @@
+"""The tests for the facebox component."""
+import requests_mock
+
+from homeassistant.const import (CONF_IP_ADDRESS, CONF_PORT)
+from homeassistant.setup import setup_component
+import homeassistant.components.image_processing as ip
+
+from tests.common import (
+    get_test_home_assistant, assert_setup_component)
+
+MOCK_IP = '192.168.0.1'
+MOCK_PORT = '8080'
+
+MOCK_RESPONSE = """
+{"facesCount": 1,
+"success": True,
+"faces":['face_data']}
+"""
+
+VALID_ENTITY_ID = 'image_processing.facebox_demo_camera'
+VALID_CONFIG = {
+    ip.DOMAIN: {
+        'platform': 'facebox_face_detect',
+        CONF_IP_ADDRESS: MOCK_IP,
+        CONF_PORT: MOCK_PORT,
+        ip.CONF_SOURCE: {
+            ip.CONF_ENTITY_ID: 'camera.demo_camera'}
+        },
+    'camera': {
+        'platform': 'demo'
+        }
+    }
+
+
+class TestFaceboxSetup(object):
+    """Test class for image processing."""
+
+    def setup_method(self):
+        """Setup things to be run when tests are started."""
+        self.hass = get_test_home_assistant()
+
+    def test_setup_platform(self):
+        """Setup platform with one entity."""
+
+        with assert_setup_component(1, ip.DOMAIN):
+            setup_component(self.hass, ip.DOMAIN, VALID_CONFIG)
+
+        assert self.hass.states.get(VALID_ENTITY_ID)
+
+    def test_process_image(self):
+        """Test processing of an image."""
+
+        with assert_setup_component(1, ip.DOMAIN):
+            setup_component(self.hass, ip.DOMAIN, VALID_CONFIG)
+        assert self.hass.states.get(VALID_ENTITY_ID)
+
+        with requests_mock.Mocker() as mock_req:
+            url = "http://{}:{}/facebox/check".format(MOCK_IP, MOCK_PORT)
+            mock_req.get(url, text=MOCK_RESPONSE)
+            ip.scan(self.hass, entity_id=VALID_ENTITY_ID)
+            self.hass.block_till_done()
+
+        state = self.hass.states.get(VALID_ENTITY_ID)
+        assert state.state == '1'
+
+    def teardown_method(self):
+        """Stop everything that was started."""
+        self.hass.stop()