Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Component "Image processing" () 

* Init new component for image processing.

* Add demo platform

* address comments

* add unittest v1 for demo

* Add unittest for alpr

* Add openalpr local test

* Add openalpr cloud platform

* Add unittest openalpr cloud platform

* Update stale docstring

* Address paulus comments

* Update stale docstring

* Add coro to function

* Add coro to cloud
pull/5318/head
Pascal Vizeli 2017-01-14 08:18:03 +01:00 committed by GitHub
parent 5bba9a63a5
commit c2492d1493
15 changed files with 1439 additions and 0 deletions

40
homeassistant/components/camera/__init__.py
View File

@ -10,8 +10,13 @@ from datetime import timedelta
import logging
import hashlib
import aiohttp
from aiohttp import web
import async_timeout
from homeassistant.const import ATTR_ENTITY_PICTURE
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.helpers.config_validation import PLATFORM_SCHEMA # noqa
@ -29,6 +34,41 @@ STATE_IDLE = 'idle'
ENTITY_IMAGE_URL = '/api/camera_proxy/{0}?token={1}'
@asyncio.coroutine
def async_get_image(hass, entity_id, timeout=10):
"""Fetch a image from a camera entity."""
websession = async_get_clientsession(hass)
state = hass.states.get(entity_id)
if state is None:
raise HomeAssistantError(
"No entity '{0}' for grab a image".format(entity_id))
url = "{0}{1}".format(
hass.config.api.base_url,
state.attributes.get(ATTR_ENTITY_PICTURE)
)
response = None
try:
with async_timeout.timeout(timeout, loop=hass.loop):
response = yield from websession.get(url)
if response.status != 200:
raise HomeAssistantError("Error {0} on {1}".format(
response.status, url))
image = yield from response.read()
return image
except (asyncio.TimeoutError, aiohttp.errors.ClientError):
raise HomeAssistantError("Can't connect to {0}".format(url))
finally:
if response is not None:
yield from response.release()
@asyncio.coroutine
def async_setup(hass, config):
"""Setup the camera component."""

2
homeassistant/components/demo.py
View File

@ -23,12 +23,14 @@ COMPONENTS_WITH_DEMO_PLATFORM = [
'cover',
'device_tracker',
'fan',
'image_processing',
'light',
'lock',
'media_player',
'notify',
'sensor',
'switch',
'tts',
]

127
homeassistant/components/image_processing/__init__.py Normal file
View File

@ -0,0 +1,127 @@
"""
Provides functionality to interact with image processing services.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/image_processing/
"""
import asyncio
from datetime import timedelta
import logging
import os
import voluptuous as vol
from homeassistant.config import load_yaml_config_file
from homeassistant.const import (
ATTR_ENTITY_ID, CONF_NAME, CONF_ENTITY_ID)
from homeassistant.exceptions import HomeAssistantError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_component import EntityComponent
from homeassistant.loader import get_component
DOMAIN = 'image_processing'
DEPENDENCIES = ['camera']
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=10)
SERVICE_SCAN = 'scan'
ATTR_CONFIDENCE = 'confidence'
CONF_SOURCE = 'source'
CONF_CONFIDENCE = 'confidence'
DEFAULT_TIMEOUT = 10
SOURCE_SCHEMA = vol.Schema({
vol.Required(CONF_ENTITY_ID): cv.entity_id,
vol.Optional(CONF_NAME): cv.string,
})
PLATFORM_SCHEMA = cv.PLATFORM_SCHEMA.extend({
vol.Optional(CONF_SOURCE): vol.All(cv.ensure_list, [SOURCE_SCHEMA]),
})
SERVICE_SCAN_SCHEMA = vol.Schema({
vol.Optional(ATTR_ENTITY_ID): cv.entity_ids,
})
def scan(hass, entity_id=None):
"""Force process a image."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else None
hass.services.call(DOMAIN, SERVICE_SCAN, data)
@asyncio.coroutine
def async_setup(hass, config):
"""Setup image processing."""
component = EntityComponent(_LOGGER, DOMAIN, hass, SCAN_INTERVAL)
yield from component.async_setup(config)
descriptions = yield from hass.loop.run_in_executor(
None, load_yaml_config_file,
os.path.join(os.path.dirname(__file__), 'services.yaml'))
@asyncio.coroutine
def async_scan_service(service):
"""Service handler for scan."""
image_entities = component.async_extract_from_service(service)
update_task = [entity.async_update_ha_state(True) for
entity in image_entities]
if update_task:
yield from asyncio.wait(update_task, loop=hass.loop)
hass.services.async_register(
DOMAIN, SERVICE_SCAN, async_scan_service,
descriptions.get(SERVICE_SCAN), schema=SERVICE_SCAN_SCHEMA)
return True
class ImageProcessingEntity(Entity):
"""Base entity class for image processing."""
timeout = DEFAULT_TIMEOUT
@property
def camera_entity(self):
"""Return camera entity id from process pictures."""
return None
def process_image(self, image):
"""Process image."""
raise NotImplementedError()
def async_process_image(self, image):
"""Process image.
This method must be run in the event loop and returns a coroutine.
"""
return self.hass.loop.run_in_executor(None, self.process_image, image)
@asyncio.coroutine
def async_update(self):
"""Update image and process it.
This method is a coroutine.
"""
camera = get_component('camera')
image = None
try:
image = yield from camera.async_get_image(
self.hass, self.camera_entity, timeout=self.timeout)
except HomeAssistantError as err:
_LOGGER.error("Error on receive image from entity: %s", err)
return
# process image data
yield from self.async_process_image(image)

84
homeassistant/components/image_processing/demo.py Normal file
View File

@ -0,0 +1,84 @@
"""
Support for the demo image processing.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/demo/
"""
from homeassistant.components.image_processing import ImageProcessingEntity
from homeassistant.components.image_processing.openalpr_local import (
ImageProcessingAlprEntity)
def setup_platform(hass, config, add_devices, discovery_info=None):
"""Setup the demo image_processing platform."""
add_devices([
DemoImageProcessing('camera.demo_camera', "Demo"),
DemoImageProcessingAlpr('camera.demo_camera', "Demo Alpr")
])
class DemoImageProcessing(ImageProcessingEntity):
"""Demo alpr image processing entity."""
def __init__(self, camera_entity, name):
"""Initialize demo alpr."""
self._name = name
self._camera = camera_entity
self._count = 0
@property
def camera_entity(self):
"""Return camera entity id from process pictures."""
return self._camera
@property
def name(self):
"""Return the name of the entity."""
return self._name
@property
def state(self):
"""Return the state of the entity."""
return self._count
def process_image(self, image):
"""Process image."""
self._count += 1
class DemoImageProcessingAlpr(ImageProcessingAlprEntity):
"""Demo alpr image processing entity."""
def __init__(self, camera_entity, name):
"""Initialize demo alpr."""
super().__init__()
self._name = name
self._camera = camera_entity
@property
def camera_entity(self):
"""Return camera entity id from process pictures."""
return self._camera
@property
def confidence(self):
"""Return minimum confidence for send events."""
return 80
@property
def name(self):
"""Return the name of the entity."""
return self._name
def process_image(self, image):
"""Process image."""
demo_data = {
'AC3829': 98.3,
'BE392034': 95.5,
'CD02394': 93.4,
'DF923043': 90.8
}
self.process_plates(demo_data, 1)

151
homeassistant/components/image_processing/openalpr_cloud.py Normal file
View File

@ -0,0 +1,151 @@
"""
Component that will help set the openalpr cloud for alpr processing.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/image_processing.openalpr_cloud/
"""
import asyncio
from base64 import b64encode
import logging
import aiohttp
import async_timeout
import voluptuous as vol
from homeassistant.core import split_entity_id
from homeassistant.const import CONF_API_KEY
from homeassistant.components.image_processing import (
PLATFORM_SCHEMA, CONF_CONFIDENCE, CONF_SOURCE, CONF_ENTITY_ID, CONF_NAME)
from homeassistant.components.image_processing.openalpr_local import (
ImageProcessingAlprEntity)
from homeassistant.helpers.aiohttp_client import async_get_clientsession
import homeassistant.helpers.config_validation as cv
_LOGGER = logging.getLogger(__name__)
OPENALPR_API_URL = "https://api.openalpr.com/v1/recognize"
OPENALPR_REGIONS = [
'us',
'eu',
'au',
'auwide',
'gb',
'kr',
'mx',
'sg',
]
CONF_REGION = 'region'
DEFAULT_CONFIDENCE = 80
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_API_KEY): cv.string,
vol.Required(CONF_REGION):
vol.All(vol.Lower, vol.In(OPENALPR_REGIONS)),
vol.Optional(CONF_CONFIDENCE, default=DEFAULT_CONFIDENCE):
vol.All(vol.Coerce(float), vol.Range(min=0, max=100))
})
@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
"""Set up the openalpr cloud api platform."""
confidence = config[CONF_CONFIDENCE]
params = {
'secret_key': config[CONF_API_KEY],
'tasks': "plate",
'return_image': 0,
'country': config[CONF_REGION],
}
entities = []
for camera in config[CONF_SOURCE]:
entities.append(OpenAlprCloudEntity(
camera[CONF_ENTITY_ID], params, confidence, camera.get(CONF_NAME)
))
yield from async_add_devices(entities)
class OpenAlprCloudEntity(ImageProcessingAlprEntity):
"""OpenAlpr cloud entity."""
def __init__(self, camera_entity, params, confidence, name=None):
"""Initialize openalpr local api."""
super().__init__()
self._params = params
self._camera = camera_entity
self._confidence = confidence
if name:
self._name = name
else:
self._name = "OpenAlpr {0}".format(
split_entity_id(camera_entity)[1])
@property
def confidence(self):
"""Return minimum confidence for send events."""
return self._confidence
@property
def camera_entity(self):
"""Return camera entity id from process pictures."""
return self._camera
@property
def name(self):
"""Return the name of the entity."""
return self._name
@asyncio.coroutine
def async_process_image(self, image):
"""Process image.
This method is a coroutine.
"""
websession = async_get_clientsession(self.hass)
params = self._params.copy()
params['image_bytes'] = str(b64encode(image), 'utf-8')
data = None
request = None
try:
with async_timeout.timeout(self.timeout, loop=self.hass.loop):
request = yield from websession.post(
OPENALPR_API_URL, params=params
)
data = yield from request.json()
if request.status != 200:
_LOGGER.error("Error %d -> %s.",
request.status, data.get('error'))
return
except (asyncio.TimeoutError, aiohttp.errors.ClientError):
_LOGGER.error("Timeout for openalpr api.")
return
finally:
if request is not None:
yield from request.release()
# processing api data
vehicles = 0
result = {}
for row in data['plate']['results']:
vehicles += 1
for p_data in row['candidates']:
try:
result.update(
{p_data['plate']: float(p_data['confidence'])})
except ValueError:
continue
self.async_process_plates(result, vehicles)

218
homeassistant/components/image_processing/openalpr_local.py Normal file
View File

@ -0,0 +1,218 @@
"""
Component that will help set the openalpr local for alpr processing.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/image_processing.openalpr_local/
"""
import asyncio
import logging
import io
import re
import voluptuous as vol
from homeassistant.core import split_entity_id, callback
from homeassistant.const import STATE_UNKNOWN
import homeassistant.helpers.config_validation as cv
from homeassistant.components.image_processing import (
PLATFORM_SCHEMA, ImageProcessingEntity, CONF_CONFIDENCE, CONF_SOURCE,
CONF_ENTITY_ID, CONF_NAME, ATTR_ENTITY_ID, ATTR_CONFIDENCE)
from homeassistant.util.async import run_callback_threadsafe
_LOGGER = logging.getLogger(__name__)
RE_ALPR_PLATE = re.compile(r"^plate\d*:")
RE_ALPR_RESULT = re.compile(r"- (\w*)\s*confidence: (\d*.\d*)")
EVENT_FOUND_PLATE = 'found_plate'
ATTR_PLATE = 'plate'
ATTR_PLATES = 'plates'
ATTR_VEHICLES = 'vehicles'
OPENALPR_REGIONS = [
'us',
'eu',
'au',
'auwide',
'gb',
'kr',
'mx',
'sg',
]
CONF_REGION = 'region'
CONF_ALPR_BIN = 'alp_bin'
DEFAULT_BINARY = 'alpr'
DEFAULT_CONFIDENCE = 80
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_REGION):
vol.All(vol.Lower, vol.In(OPENALPR_REGIONS)),
vol.Optional(CONF_ALPR_BIN, default=DEFAULT_BINARY): cv.string,
vol.Optional(CONF_CONFIDENCE, default=DEFAULT_CONFIDENCE):
vol.All(vol.Coerce(float), vol.Range(min=0, max=100))
})
@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
"""Set up the openalpr local platform."""
command = [config[CONF_ALPR_BIN], '-c', config[CONF_REGION], '-']
confidence = config[CONF_CONFIDENCE]
entities = []
for camera in config[CONF_SOURCE]:
entities.append(OpenAlprLocalEntity(
camera[CONF_ENTITY_ID], command, confidence, camera.get(CONF_NAME)
))
yield from async_add_devices(entities)
class ImageProcessingAlprEntity(ImageProcessingEntity):
"""Base entity class for alpr image processing."""
def __init__(self):
"""Initialize base alpr entity."""
self.plates = {} # last scan data
self.vehicles = 0 # vehicles count
@property
def confidence(self):
"""Return minimum confidence for send events."""
return None
@property
def state(self):
"""Return the state of the entity."""
confidence = 0
plate = STATE_UNKNOWN
# search high plate
for i_pl, i_co in self.plates.items():
if i_co > confidence:
confidence = i_co
plate = i_pl
return plate
@property
def state_attributes(self):
"""Return device specific state attributes."""
attr = {
ATTR_PLATES: self.plates,
ATTR_VEHICLES: self.vehicles
}
return attr
def process_plates(self, plates, vehicles):
"""Send event with new plates and store data."""
run_callback_threadsafe(
self.hass.loop, self.async_process_plates, plates, vehicles
).result()
@callback
def async_process_plates(self, plates, vehicles):
"""Send event with new plates and store data.
plates are a dict in follow format:
{ 'plate': confidence }
This method must be run in the event loop.
"""
plates = {plate: confidence for plate, confidence in plates.items()
if confidence >= self.confidence}
new_plates = set(plates) - set(self.plates)
# send events
for i_plate in new_plates:
self.hass.async_add_job(
self.hass.bus.async_fire, EVENT_FOUND_PLATE, {
ATTR_PLATE: i_plate,
ATTR_ENTITY_ID: self.entity_id,
ATTR_CONFIDENCE: plates.get(i_plate),
}
)
# update entity store
self.plates = plates
self.vehicles = vehicles
class OpenAlprLocalEntity(ImageProcessingAlprEntity):
"""OpenAlpr local api entity."""
def __init__(self, camera_entity, command, confidence, name=None):
"""Initialize openalpr local api."""
super().__init__()
self._cmd = command
self._camera = camera_entity
self._confidence = confidence
if name:
self._name = name
else:
self._name = "OpenAlpr {0}".format(
split_entity_id(camera_entity)[1])
@property
def confidence(self):
"""Return minimum confidence for send events."""
return self._confidence
@property
def camera_entity(self):
"""Return camera entity id from process pictures."""
return self._camera
@property
def name(self):
"""Return the name of the entity."""
return self._name
@asyncio.coroutine
def async_process_image(self, image):
"""Process image.
This method is a coroutine.
"""
result = {}
vehicles = 0
alpr = yield from asyncio.create_subprocess_exec(
*self._cmd,
loop=self.hass.loop,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL
)
# send image
stdout, _ = yield from alpr.communicate(input=image)
stdout = io.StringIO(str(stdout, 'utf-8'))
while True:
line = stdout.readline()
if not line:
break
new_plates = RE_ALPR_PLATE.search(line)
new_result = RE_ALPR_RESULT.search(line)
# found new vehicle
if new_plates:
vehicles += 1
continue
# found plate result
if new_result:
try:
result.update(
{new_result.group(1): float(new_result.group(2))})
except ValueError:
continue
self.async_process_plates(result, vehicles)

9
homeassistant/components/image_processing/services.yaml Normal file
View File

@ -0,0 +1,9 @@
# Describes the format for available image_processing services
scan:
description: Process an image immediately
fields:
entity_id:
description: Name(s) of entities to scan immediately
example: 'image_processing.alpr_garage'

101
tests/components/camera/test_init.py Normal file
View File

@ -0,0 +1,101 @@
"""The tests for the camera component."""
import asyncio
from unittest.mock import patch
import pytest
from homeassistant.bootstrap import setup_component
from homeassistant.const import ATTR_ENTITY_PICTURE
import homeassistant.components.camera as camera
from homeassistant.exceptions import HomeAssistantError
from homeassistant.util.async import run_coroutine_threadsafe
from tests.common import get_test_home_assistant, assert_setup_component
class TestSetupCamera(object):
"""Test class for setup camera."""
def setup_method(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_component(self):
"""Setup demo platfrom on camera component."""
config = {
camera.DOMAIN: {
'platform': 'demo'
}
}
with assert_setup_component(1, camera.DOMAIN):
setup_component(self.hass, camera.DOMAIN, config)
class TestGetImage(object):
"""Test class for camera."""
def setup_method(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
config = {
camera.DOMAIN: {
'platform': 'demo'
}
}
setup_component(self.hass, camera.DOMAIN, config)
state = self.hass.states.get('camera.demo_camera')
self.url = "{0}{1}".format(
self.hass.config.api.base_url,
state.attributes.get(ATTR_ENTITY_PICTURE))
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
@patch('homeassistant.components.camera.demo.DemoCamera.camera_image',
autospec=True, return_value=b'Test')
def test_get_image_from_camera(self, mock_camera):
"""Grab a image from camera entity."""
self.hass.start()
image = run_coroutine_threadsafe(camera.async_get_image(
self.hass, 'camera.demo_camera'), self.hass.loop).result()
assert mock_camera.called
assert image == b'Test'
def test_get_image_without_exists_camera(self):
"""Try to get image without exists camera."""
self.hass.states.remove('camera.demo_camera')
with pytest.raises(HomeAssistantError):
run_coroutine_threadsafe(camera.async_get_image(
self.hass, 'camera.demo_camera'), self.hass.loop).result()
def test_get_image_with_timeout(self, aioclient_mock):
"""Try to get image with timeout."""
aioclient_mock.get(self.url, exc=asyncio.TimeoutError())
with pytest.raises(HomeAssistantError):
run_coroutine_threadsafe(camera.async_get_image(
self.hass, 'camera.demo_camera'), self.hass.loop).result()
assert len(aioclient_mock.mock_calls) == 1
def test_get_image_with_bad_http_state(self, aioclient_mock):
"""Try to get image with bad http status."""
aioclient_mock.get(self.url, status=400)
with pytest.raises(HomeAssistantError):
run_coroutine_threadsafe(camera.async_get_image(
self.hass, 'camera.demo_camera'), self.hass.loop).result()
assert len(aioclient_mock.mock_calls) == 1

1
tests/components/image_processing/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Test 'image_processing' component plaforms."""

209
tests/components/image_processing/test_init.py Normal file
View File

@ -0,0 +1,209 @@
"""The tests for the image_processing component."""
from unittest.mock import patch, PropertyMock
from homeassistant.core import callback
from homeassistant.const import ATTR_ENTITY_PICTURE
from homeassistant.bootstrap import setup_component
from homeassistant.exceptions import HomeAssistantError
import homeassistant.components.image_processing as ip
from tests.common import get_test_home_assistant, assert_setup_component
class TestSetupImageProcessing(object):
"""Test class for setup image processing."""
def setup_method(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_component(self):
"""Setup demo platfrom on image_process component."""
config = {
ip.DOMAIN: {
'platform': 'demo'
}
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
def test_setup_component_with_service(self):
"""Setup demo platfrom on image_process component test service."""
config = {
ip.DOMAIN: {
'platform': 'demo'
}
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
assert self.hass.services.has_service(ip.DOMAIN, 'scan')
class TestImageProcessing(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()
config = {
ip.DOMAIN: {
'platform': 'demo'
},
'camera': {
'platform': 'demo'
},
}
with patch('homeassistant.components.image_processing.demo.'
'DemoImageProcessing.should_poll',
new_callable=PropertyMock(return_value=False)):
setup_component(self.hass, ip.DOMAIN, config)
state = self.hass.states.get('camera.demo_camera')
self.url = "{0}{1}".format(
self.hass.config.api.base_url,
state.attributes.get(ATTR_ENTITY_PICTURE))
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
@patch('homeassistant.components.camera.demo.DemoCamera.camera_image',
autospec=True, return_value=b'Test')
@patch('homeassistant.components.image_processing.demo.'
'DemoImageProcessing.process_image', autospec=True)
def test_get_image_from_camera(self, mock_process, mock_camera):
"""Grab a image from camera entity."""
self.hass.start()
ip.scan(self.hass, entity_id='image_processing.demo')
self.hass.block_till_done()
assert mock_camera.called
assert mock_process.called
assert mock_process.call_args[0][1] == b'Test'
@patch('homeassistant.components.camera.async_get_image',
side_effect=HomeAssistantError())
@patch('homeassistant.components.image_processing.demo.'
'DemoImageProcessing.process_image', autospec=True)
def test_get_image_without_exists_camera(self, mock_process, mock_image):
"""Try to get image without exists camera."""
self.hass.states.remove('camera.demo_camera')
ip.scan(self.hass, entity_id='image_processing.demo')
self.hass.block_till_done()
assert mock_image.called
assert not mock_process.called
class TestImageProcessingAlpr(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()
config = {
ip.DOMAIN: {
'platform': 'demo'
},
'camera': {
'platform': 'demo'
},
}
with patch('homeassistant.components.image_processing.demo.'
'DemoImageProcessingAlpr.should_poll',
new_callable=PropertyMock(return_value=False)):
setup_component(self.hass, ip.DOMAIN, config)
state = self.hass.states.get('camera.demo_camera')
self.url = "{0}{1}".format(
self.hass.config.api.base_url,
state.attributes.get(ATTR_ENTITY_PICTURE))
self.alpr_events = []
@callback
def mock_alpr_event(event):
"""Mock event."""
self.alpr_events.append(event)
self.hass.bus.listen('found_plate', mock_alpr_event)
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_alpr_event_single_call(self, aioclient_mock):
"""Setup and scan a picture and test plates from event."""
aioclient_mock.get(self.url, content=b'image')
ip.scan(self.hass, entity_id='image_processing.demo_alpr')
self.hass.block_till_done()
state = self.hass.states.get('image_processing.demo_alpr')
assert len(self.alpr_events) == 4
assert state.state == 'AC3829'
event_data = [event.data for event in self.alpr_events if
event.data.get('plate') == 'AC3829']
assert len(event_data) == 1
assert event_data[0]['plate'] == 'AC3829'
assert event_data[0]['confidence'] == 98.3
assert event_data[0]['entity_id'] == 'image_processing.demo_alpr'
def test_alpr_event_double_call(self, aioclient_mock):
"""Setup and scan a picture and test plates from event."""
aioclient_mock.get(self.url, content=b'image')
ip.scan(self.hass, entity_id='image_processing.demo_alpr')
ip.scan(self.hass, entity_id='image_processing.demo_alpr')
self.hass.block_till_done()
state = self.hass.states.get('image_processing.demo_alpr')
assert len(self.alpr_events) == 4
assert state.state == 'AC3829'
event_data = [event.data for event in self.alpr_events if
event.data.get('plate') == 'AC3829']
assert len(event_data) == 1
assert event_data[0]['plate'] == 'AC3829'
assert event_data[0]['confidence'] == 98.3
assert event_data[0]['entity_id'] == 'image_processing.demo_alpr'
@patch('homeassistant.components.image_processing.demo.'
'DemoImageProcessingAlpr.confidence',
new_callable=PropertyMock(return_value=95))
def test_alpr_event_single_call_confidence(self, confidence_mock,
aioclient_mock):
"""Setup and scan a picture and test plates from event."""
aioclient_mock.get(self.url, content=b'image')
ip.scan(self.hass, entity_id='image_processing.demo_alpr')
self.hass.block_till_done()
state = self.hass.states.get('image_processing.demo_alpr')
assert len(self.alpr_events) == 2
assert state.state == 'AC3829'
event_data = [event.data for event in self.alpr_events if
event.data.get('plate') == 'AC3829']
assert len(event_data) == 1
assert event_data[0]['plate'] == 'AC3829'
assert event_data[0]['confidence'] == 98.3
assert event_data[0]['entity_id'] == 'image_processing.demo_alpr'

212
tests/components/image_processing/test_openalpr_cloud.py Normal file
View File

@ -0,0 +1,212 @@
"""The tests for the openalpr clooud platform."""
import asyncio
from unittest.mock import patch, PropertyMock
from homeassistant.core import callback
from homeassistant.const import ATTR_ENTITY_PICTURE
from homeassistant.bootstrap import setup_component
import homeassistant.components.image_processing as ip
from homeassistant.components.image_processing.openalpr_cloud import (
OPENALPR_API_URL)
from tests.common import (
get_test_home_assistant, assert_setup_component, load_fixture)
class TestOpenAlprCloudlSetup(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 teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_platform(self):
"""Setup platform with one entity."""
config = {
ip.DOMAIN: {
'platform': 'openalpr_cloud',
'source': {
'entity_id': 'camera.demo_camera'
},
'region': 'eu',
'api_key': 'sk_abcxyz123456',
},
'camera': {
'platform': 'demo'
},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
assert self.hass.states.get('image_processing.openalpr_demo_camera')
def test_setup_platform_name(self):
"""Setup platform with one entity and set name."""
config = {
ip.DOMAIN: {
'platform': 'openalpr_cloud',
'source': {
'entity_id': 'camera.demo_camera',
'name': 'test local'
},
'region': 'eu',
'api_key': 'sk_abcxyz123456',
},
'camera': {
'platform': 'demo'
},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
assert self.hass.states.get('image_processing.test_local')
def test_setup_platform_without_api_key(self):
"""Setup platform with one entity without api_key."""
config = {
ip.DOMAIN: {
'platform': 'openalpr_cloud',
'source': {
'entity_id': 'camera.demo_camera'
},
'region': 'eu',
},
'camera': {
'platform': 'demo'
},
}
with assert_setup_component(0, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
def test_setup_platform_without_region(self):
"""Setup platform with one entity without region."""
config = {
ip.DOMAIN: {
'platform': 'openalpr_cloud',
'source': {
'entity_id': 'camera.demo_camera'
},
'api_key': 'sk_abcxyz123456',
},
'camera': {
'platform': 'demo'
},
}
with assert_setup_component(0, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
class TestOpenAlprCloud(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()
config = {
ip.DOMAIN: {
'platform': 'openalpr_cloud',
'source': {
'entity_id': 'camera.demo_camera',
'name': 'test local'
},
'region': 'eu',
'api_key': 'sk_abcxyz123456',
},
'camera': {
'platform': 'demo'
},
}
with patch('homeassistant.components.image_processing.openalpr_cloud.'
'OpenAlprCloudEntity.should_poll',
new_callable=PropertyMock(return_value=False)):
setup_component(self.hass, ip.DOMAIN, config)
state = self.hass.states.get('camera.demo_camera')
self.url = "{0}{1}".format(
self.hass.config.api.base_url,
state.attributes.get(ATTR_ENTITY_PICTURE))
self.alpr_events = []
@callback
def mock_alpr_event(event):
"""Mock event."""
self.alpr_events.append(event)
self.hass.bus.listen('found_plate', mock_alpr_event)
self.params = {
'secret_key': "sk_abcxyz123456",
'tasks': "plate",
'return_image': 0,
'country': 'eu',
'image_bytes': "aW1hZ2U="
}
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_openalpr_process_image(self, aioclient_mock):
"""Setup and scan a picture and test plates from event."""
aioclient_mock.get(self.url, content=b'image')
aioclient_mock.post(
OPENALPR_API_URL, params=self.params,
text=load_fixture('alpr_cloud.json'), status=200
)
ip.scan(self.hass, entity_id='image_processing.test_local')
self.hass.block_till_done()
state = self.hass.states.get('image_processing.test_local')
assert len(aioclient_mock.mock_calls) == 2
assert len(self.alpr_events) == 5
assert state.attributes.get('vehicles') == 1
assert state.state == 'H786P0J'
event_data = [event.data for event in self.alpr_events if
event.data.get('plate') == 'H786P0J']
assert len(event_data) == 1
assert event_data[0]['plate'] == 'H786P0J'
assert event_data[0]['confidence'] == float(90.436699)
assert event_data[0]['entity_id'] == \
'image_processing.test_local'
def test_openalpr_process_image_api_error(self, aioclient_mock):
"""Setup and scan a picture and test api error."""
aioclient_mock.get(self.url, content=b'image')
aioclient_mock.post(
OPENALPR_API_URL, params=self.params,
text="{'error': 'error message'}", status=400
)
ip.scan(self.hass, entity_id='image_processing.test_local')
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
assert len(self.alpr_events) == 0
def test_openalpr_process_image_api_timeout(self, aioclient_mock):
"""Setup and scan a picture and test api error."""
aioclient_mock.get(self.url, content=b'image')
aioclient_mock.post(
OPENALPR_API_URL, params=self.params,
exc=asyncio.TimeoutError()
)
ip.scan(self.hass, entity_id='image_processing.test_local')
self.hass.block_till_done()
assert len(aioclient_mock.mock_calls) == 2
assert len(self.alpr_events) == 0

165
tests/components/image_processing/test_openalpr_local.py Normal file
View File

@ -0,0 +1,165 @@
"""The tests for the openalpr local platform."""
import asyncio
from unittest.mock import patch, PropertyMock, MagicMock
from homeassistant.core import callback
from homeassistant.const import ATTR_ENTITY_PICTURE
from homeassistant.bootstrap import setup_component
import homeassistant.components.image_processing as ip
from tests.common import (
get_test_home_assistant, assert_setup_component, load_fixture)
@asyncio.coroutine
def mock_async_subprocess():
"""Get a Popen mock back."""
async_popen = MagicMock()
@asyncio.coroutine
def communicate(input=None):
"""Communicate mock."""
fixture = bytes(load_fixture('alpr_stdout.txt'), 'utf-8')
return (fixture, None)
async_popen.communicate = communicate
return async_popen
class TestOpenAlprLocalSetup(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 teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_platform(self):
"""Setup platform with one entity."""
config = {
ip.DOMAIN: {
'platform': 'openalpr_local',
'source': {
'entity_id': 'camera.demo_camera'
},
'region': 'eu',
},
'camera': {
'platform': 'demo'
},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
assert self.hass.states.get('image_processing.openalpr_demo_camera')
def test_setup_platform_name(self):
"""Setup platform with one entity and set name."""
config = {
ip.DOMAIN: {
'platform': 'openalpr_local',
'source': {
'entity_id': 'camera.demo_camera',
'name': 'test local'
},
'region': 'eu',
},
'camera': {
'platform': 'demo'
},
}
with assert_setup_component(1, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
assert self.hass.states.get('image_processing.test_local')
def test_setup_platform_without_region(self):
"""Setup platform with one entity without region."""
config = {
ip.DOMAIN: {
'platform': 'openalpr_local',
'source': {
'entity_id': 'camera.demo_camera'
},
},
'camera': {
'platform': 'demo'
},
}
with assert_setup_component(0, ip.DOMAIN):
setup_component(self.hass, ip.DOMAIN, config)
class TestOpenAlprLocal(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()
config = {
ip.DOMAIN: {
'platform': 'openalpr_local',
'source': {
'entity_id': 'camera.demo_camera',
'name': 'test local'
},
'region': 'eu',
},
'camera': {
'platform': 'demo'
},
}
with patch('homeassistant.components.image_processing.openalpr_local.'
'OpenAlprLocalEntity.should_poll',
new_callable=PropertyMock(return_value=False)):
setup_component(self.hass, ip.DOMAIN, config)
state = self.hass.states.get('camera.demo_camera')
self.url = "{0}{1}".format(
self.hass.config.api.base_url,
state.attributes.get(ATTR_ENTITY_PICTURE))
self.alpr_events = []
@callback
def mock_alpr_event(event):
"""Mock event."""
self.alpr_events.append(event)
self.hass.bus.listen('found_plate', mock_alpr_event)
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
@patch('asyncio.create_subprocess_exec',
return_value=mock_async_subprocess())
def test_openalpr_process_image(self, popen_mock, aioclient_mock):
"""Setup and scan a picture and test plates from event."""
aioclient_mock.get(self.url, content=b'image')
ip.scan(self.hass, entity_id='image_processing.test_local')
self.hass.block_till_done()
state = self.hass.states.get('image_processing.test_local')
assert popen_mock.called
assert len(self.alpr_events) == 5
assert state.attributes.get('vehicles') == 1
assert state.state == 'PE3R2X'
event_data = [event.data for event in self.alpr_events if
event.data.get('plate') == 'PE3R2X']
assert len(event_data) == 1
assert event_data[0]['plate'] == 'PE3R2X'
assert event_data[0]['confidence'] == float(98.9371)
assert event_data[0]['entity_id'] == \
'image_processing.test_local'

103
tests/fixtures/alpr_cloud.json vendored Normal file
View File

@ -0,0 +1,103 @@
{
"plate":{
"data_type":"alpr_results",
"epoch_time":1483953071942,
"img_height":640,
"img_width":480,
"results":[
{
"plate":"H786P0J",
"confidence":90.436699,
"region_confidence":0,
"region":"",
"plate_index":0,
"processing_time_ms":16.495636,
"candidates":[
{
"matches_template":0,
"plate":"H786P0J",
"confidence":90.436699
},
{
"matches_template":0,
"plate":"H786POJ",
"confidence":88.046814
},
{
"matches_template":0,
"plate":"H786PDJ",
"confidence":85.58432
},
{
"matches_template":0,
"plate":"H786PQJ",
"confidence":85.472939
},
{
"matches_template":0,
"plate":"HS786P0J",
"confidence":75.455666
},
{
"matches_template":0,
"plate":"H2786P0J",
"confidence":75.256081
},
{
"matches_template":0,
"plate":"H3786P0J",
"confidence":65.228058
},
{
"matches_template":0,
"plate":"H786PGJ",
"confidence":63.303329
},
{
"matches_template":0,
"plate":"HS786POJ",
"confidence":83.065773
},
{
"matches_template":0,
"plate":"H2786POJ",
"confidence":52.866196
}
],
"coordinates":[
{
"y":384,
"x":156
},
{
"y":384,
"x":289
},
{
"y":409,
"x":289
},
{
"y":409,
"x":156
}
],
"matches_template":0,
"requested_topn":10
}
],
"version":2,
"processing_time_ms":115.687286,
"regions_of_interest":[
]
},
"image_bytes":"",
"img_width":480,
"credits_monthly_used":5791,
"img_height":640,
"total_processing_time":120.71599999762839,
"credits_monthly_total":10000000000,
"image_bytes_prefix":"data:image/jpeg;base64,",
"credit_cost":1
}

12
tests/fixtures/alpr_stdout.txt vendored Normal file
View File

@ -0,0 +1,12 @@
plate0: top 10 results -- Processing Time = 58.1879ms.
- PE3R2X confidence: 98.9371
- PE32X confidence: 98.1385
- PE3R2 confidence: 97.5444
- PE3R2Y confidence: 86.1448
- P63R2X confidence: 82.9016
- FE3R2X confidence: 72.1147
- PE32 confidence: 66.7458
- PE32Y confidence: 65.3462
- P632X confidence: 62.1031
- P63R2 confidence: 61.5089

5
tests/test_util/aiohttp.py
View File

@ -150,6 +150,11 @@ class AiohttpClientMockResponse:
"""Return mock response as a string."""
return self.response.decode(encoding)
@asyncio.coroutine
def json(self, encoding='utf-8'):
"""Return mock response as a json."""
return _json.loads(self.response.decode(encoding))
@asyncio.coroutine
def release(self):
"""Mock release."""