evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add unittests for FFmpeg and spliting binary sensor (#5659)

Browse source 
* Spliting ffmpeg binary sensor and move service to component.

* unittests for component

* add unittest for binary_sensor

* exclude camera for tests
This commit is contained in:
Pascal Vizeli 2017-01-31 16:48:03 +01:00 • committed by Paulus Schoutsen
parent 9ae574c7d9
commit 88d9d787a6
9 changed files with 722 additions and 308 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.coveragerc
View file

@ -116,9 +116,6 @@ omit =
homeassistant/components/knx.py
homeassistant/components/*/knx.py
homeassistant/components/ffmpeg.py
homeassistant/components/*/ffmpeg.py
homeassistant/components/zoneminder.py
homeassistant/components/*/zoneminder.py
@ -141,6 +138,7 @@ omit =
homeassistant/components/browser.py
homeassistant/components/camera/amcrest.py
homeassistant/components/camera/bloomsky.py
homeassistant/components/camera/ffmpeg.py
homeassistant/components/camera/foscam.py
homeassistant/components/camera/mjpeg.py
homeassistant/components/camera/rpi_camera.py

279
homeassistant/components/binary_sensor/ffmpeg.py
View file

@ -1,279 +0,0 @@
"""
Provides a binary sensor which is a collection of ffmpeg tools.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.ffmpeg/
"""
import asyncio
import logging
import os
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.components.binary_sensor import (
BinarySensorDevice, PLATFORM_SCHEMA, DOMAIN)
from homeassistant.components.ffmpeg import (
DATA_FFMPEG, CONF_INPUT, CONF_OUTPUT, CONF_EXTRA_ARGUMENTS)
from homeassistant.config import load_yaml_config_file
from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP, EVENT_HOMEASSISTANT_START, CONF_NAME,
ATTR_ENTITY_ID)
DEPENDENCIES = ['ffmpeg']
_LOGGER = logging.getLogger(__name__)
SERVICE_START = 'ffmpeg_start'
SERVICE_STOP = 'ffmpeg_stop'
SERVICE_RESTART = 'ffmpeg_restart'
DATA_FFMPEG_DEVICE = 'ffmpeg_binary_sensor'
FFMPEG_SENSOR_NOISE = 'noise'
FFMPEG_SENSOR_MOTION = 'motion'
MAP_FFMPEG_BIN = [
FFMPEG_SENSOR_NOISE,
FFMPEG_SENSOR_MOTION
]
CONF_INITIAL_STATE = 'initial_state'
CONF_TOOL = 'tool'
CONF_PEAK = 'peak'
CONF_DURATION = 'duration'
CONF_RESET = 'reset'
CONF_CHANGES = 'changes'
CONF_REPEAT = 'repeat'
CONF_REPEAT_TIME = 'repeat_time'
DEFAULT_NAME = 'FFmpeg'
DEFAULT_INIT_STATE = True
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_TOOL): vol.In(MAP_FFMPEG_BIN),
vol.Required(CONF_INPUT): cv.string,
vol.Optional(CONF_INITIAL_STATE, default=DEFAULT_INIT_STATE): cv.boolean,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_EXTRA_ARGUMENTS): cv.string,
vol.Optional(CONF_OUTPUT): cv.string,
vol.Optional(CONF_PEAK, default=-30): vol.Coerce(int),
vol.Optional(CONF_DURATION, default=1):
vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(CONF_RESET, default=10):
vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(CONF_CHANGES, default=10):
vol.All(vol.Coerce(float), vol.Range(min=0, max=99)),
vol.Optional(CONF_REPEAT, default=0):
vol.All(vol.Coerce(int), vol.Range(min=0)),
vol.Optional(CONF_REPEAT_TIME, default=0):
vol.All(vol.Coerce(int), vol.Range(min=0)),
})
SERVICE_FFMPEG_SCHEMA = vol.Schema({
vol.Optional(ATTR_ENTITY_ID): cv.entity_ids,
})
def restart(hass, entity_id=None):
"""Restart a ffmpeg process on entity."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_RESTART, data)
@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
"""Create the binary sensor."""
from haffmpeg import SensorNoise, SensorMotion
# check source
if not hass.data[DATA_FFMPEG].async_run_test(config.get(CONF_INPUT)):
return
# generate sensor object
if config.get(CONF_TOOL) == FFMPEG_SENSOR_NOISE:
entity = FFmpegNoise(hass, SensorNoise, config)
else:
entity = FFmpegMotion(hass, SensorMotion, config)
@asyncio.coroutine
def async_shutdown(event):
"""Stop ffmpeg."""
yield from entity.async_shutdown_ffmpeg()
hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, async_shutdown)
# start on startup
if config.get(CONF_INITIAL_STATE):
@asyncio.coroutine
def async_start(event):
"""Start ffmpeg."""
yield from entity.async_start_ffmpeg()
yield from entity.async_update_ha_state()
hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_START, async_start)
# add to system
yield from async_add_devices([entity])
# exists service?
if hass.services.has_service(DOMAIN, SERVICE_RESTART):
hass.data[DATA_FFMPEG_DEVICE].append(entity)
return
hass.data[DATA_FFMPEG_DEVICE] = [entity]
descriptions = yield from hass.loop.run_in_executor(
None, load_yaml_config_file,
os.path.join(os.path.dirname(__file__), 'services.yaml'))
# register service
@asyncio.coroutine
def async_service_handle(service):
"""Handle service binary_sensor.ffmpeg_restart."""
entity_ids = service.data.get('entity_id')
if entity_ids:
_devices = [device for device in hass.data[DATA_FFMPEG_DEVICE]
if device.entity_id in entity_ids]
else:
_devices = hass.data[DATA_FFMPEG_DEVICE]
tasks = []
for device in _devices:
if service.service == SERVICE_START:
tasks.append(device.async_start_ffmpeg())
elif service.service == SERVICE_STOP:
tasks.append(device.async_shutdown_ffmpeg())
else:
tasks.append(device.async_restart_ffmpeg())
if tasks:
yield from asyncio.wait(tasks, loop=hass.loop)
hass.services.async_register(
DOMAIN, SERVICE_START, async_service_handle,
descriptions.get(SERVICE_START), schema=SERVICE_FFMPEG_SCHEMA)
hass.services.async_register(
DOMAIN, SERVICE_STOP, async_service_handle,
descriptions.get(SERVICE_STOP), schema=SERVICE_FFMPEG_SCHEMA)
hass.services.async_register(
DOMAIN, SERVICE_RESTART, async_service_handle,
descriptions.get(SERVICE_RESTART), schema=SERVICE_FFMPEG_SCHEMA)
class FFmpegBinarySensor(BinarySensorDevice):
"""A binary sensor which use ffmpeg for noise detection."""
def __init__(self, hass, ffobj, config):
"""Constructor for binary sensor noise detection."""
self._manager = hass.data[DATA_FFMPEG]
self._state = False
self._config = config
self._name = config.get(CONF_NAME)
self._ffmpeg = ffobj(
self._manager.binary, hass.loop, self._async_callback)
def _async_callback(self, state):
"""HA-FFmpeg callback for noise detection."""
self._state = state
self.hass.async_add_job(self.async_update_ha_state())
def async_start_ffmpeg(self):
"""Start a FFmpeg instance.
This method must be run in the event loop and returns a coroutine.
"""
raise NotImplementedError()
def async_shutdown_ffmpeg(self):
"""For STOP event to shutdown ffmpeg.
This method must be run in the event loop and returns a coroutine.
"""
return self._ffmpeg.close()
@asyncio.coroutine
def async_restart_ffmpeg(self):
"""Restart processing."""
yield from self.async_shutdown_ffmpeg()
yield from self.async_start_ffmpeg()
@property
def is_on(self):
"""True if the binary sensor is on."""
return self._state
@property
def should_poll(self):
"""Return True if entity has to be polled for state."""
return False
@property
def name(self):
"""Return the name of the entity."""
return self._name
@property
def available(self):
"""Return True if entity is available."""
return self._ffmpeg.is_running
class FFmpegNoise(FFmpegBinarySensor):
"""A binary sensor which use ffmpeg for noise detection."""
def async_start_ffmpeg(self):
"""Start a FFmpeg instance.
This method must be run in the event loop and returns a coroutine.
"""
# init config
self._ffmpeg.set_options(
time_duration=self._config.get(CONF_DURATION),
time_reset=self._config.get(CONF_RESET),
peak=self._config.get(CONF_PEAK),
)
# run
return self._ffmpeg.open_sensor(
input_source=self._config.get(CONF_INPUT),
output_dest=self._config.get(CONF_OUTPUT),
extra_cmd=self._config.get(CONF_EXTRA_ARGUMENTS),
)
@property
def sensor_class(self):
"""Return the class of this sensor, from SENSOR_CLASSES."""
return "sound"
class FFmpegMotion(FFmpegBinarySensor):
"""A binary sensor which use ffmpeg for noise detection."""
def async_start_ffmpeg(self):
"""Start a FFmpeg instance.
This method must be run in the event loop and returns a coroutine.
"""
# init config
self._ffmpeg.set_options(
time_reset=self._config.get(CONF_RESET),
time_repeat=self._config.get(CONF_REPEAT_TIME),
repeat=self._config.get(CONF_REPEAT),
changes=self._config.get(CONF_CHANGES),
)
# run
return self._ffmpeg.open_sensor(
input_source=self._config.get(CONF_INPUT),
extra_cmd=self._config.get(CONF_EXTRA_ARGUMENTS),
)
@property
def sensor_class(self):
"""Return the class of this sensor, from SENSOR_CLASSES."""
return "motion"

127
homeassistant/components/binary_sensor/ffmpeg_motion.py Normal file
View file

@ -0,0 +1,127 @@
"""
Provides a binary sensor which is a collection of ffmpeg tools.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.ffmpeg_motion/
"""
import asyncio
import logging
import voluptuous as vol
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from homeassistant.components.binary_sensor import (
BinarySensorDevice, PLATFORM_SCHEMA)
from homeassistant.components.ffmpeg import (
FFmpegBase, DATA_FFMPEG, CONF_INPUT, CONF_EXTRA_ARGUMENTS,
CONF_INITIAL_STATE)
from homeassistant.const import CONF_NAME
DEPENDENCIES = ['ffmpeg']
_LOGGER = logging.getLogger(__name__)
CONF_RESET = 'reset'
CONF_CHANGES = 'changes'
CONF_REPEAT = 'repeat'
CONF_REPEAT_TIME = 'repeat_time'
DEFAULT_NAME = 'FFmpeg Motion'
DEFAULT_INIT_STATE = True
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_INPUT): cv.string,
vol.Optional(CONF_INITIAL_STATE, default=DEFAULT_INIT_STATE): cv.boolean,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_EXTRA_ARGUMENTS): cv.string,
vol.Optional(CONF_RESET, default=10):
vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(CONF_CHANGES, default=10):
vol.All(vol.Coerce(float), vol.Range(min=0, max=99)),
vol.Inclusive(CONF_REPEAT, 'repeat'):
vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Inclusive(CONF_REPEAT_TIME, 'repeat'):
vol.All(vol.Coerce(int), vol.Range(min=1)),
})
@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
"""Create the binary sensor."""
manager = hass.data[DATA_FFMPEG]
# check source
if not manager.async_run_test(config.get(CONF_INPUT)):
return
# generate sensor object
entity = FFmpegMotion(hass, manager, config)
# add to system
manager.async_register_device(entity)
yield from async_add_devices([entity])
class FFmpegBinarySensor(FFmpegBase, BinarySensorDevice):
"""A binary sensor which use ffmpeg for noise detection."""
def __init__(self, hass, config):
"""Constructor for binary sensor noise detection."""
super().__init__(config.get(CONF_INITIAL_STATE))
self._state = False
self._config = config
self._name = config.get(CONF_NAME)
@callback
def _async_callback(self, state):
"""HA-FFmpeg callback for noise detection."""
self._state = state
self.hass.async_add_job(self.async_update_ha_state())
@property
def is_on(self):
"""True if the binary sensor is on."""
return self._state
@property
def name(self):
"""Return the name of the entity."""
return self._name
class FFmpegMotion(FFmpegBinarySensor):
"""A binary sensor which use ffmpeg for noise detection."""
def __init__(self, hass, manager, config):
"""Initialize ffmpeg motion binary sensor."""
from haffmpeg import SensorMotion
super().__init__(hass, config)
self.ffmpeg = SensorMotion(
manager.binary, hass.loop, self._async_callback)
def async_start_ffmpeg(self):
"""Start a FFmpeg instance.
This method must be run in the event loop and returns a coroutine.
"""
# init config
self.ffmpeg.set_options(
time_reset=self._config.get(CONF_RESET),
time_repeat=self._config.get(CONF_REPEAT_TIME, 0),
repeat=self._config.get(CONF_REPEAT, 0),
changes=self._config.get(CONF_CHANGES),
)
# run
return self.ffmpeg.open_sensor(
input_source=self._config.get(CONF_INPUT),
extra_cmd=self._config.get(CONF_EXTRA_ARGUMENTS),
)
@property
def sensor_class(self):
"""Return the class of this sensor, from SENSOR_CLASSES."""
return "motion"

96
homeassistant/components/binary_sensor/ffmpeg_noise.py Normal file
View file

@ -0,0 +1,96 @@
"""
Provides a binary sensor which is a collection of ffmpeg tools.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/binary_sensor.ffmpeg_noise/
"""
import asyncio
import logging
import voluptuous as vol
import homeassistant.helpers.config_validation as cv
from homeassistant.components.binary_sensor import PLATFORM_SCHEMA
from homeassistant.components.binary_sensor.ffmpeg_motion import (
FFmpegBinarySensor)
from homeassistant.components.ffmpeg import (
DATA_FFMPEG, CONF_INPUT, CONF_OUTPUT, CONF_EXTRA_ARGUMENTS,
CONF_INITIAL_STATE)
from homeassistant.const import CONF_NAME
DEPENDENCIES = ['ffmpeg']
_LOGGER = logging.getLogger(__name__)
CONF_PEAK = 'peak'
CONF_DURATION = 'duration'
CONF_RESET = 'reset'
DEFAULT_NAME = 'FFmpeg Noise'
DEFAULT_INIT_STATE = True
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Required(CONF_INPUT): cv.string,
vol.Optional(CONF_INITIAL_STATE, default=DEFAULT_INIT_STATE): cv.boolean,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_EXTRA_ARGUMENTS): cv.string,
vol.Optional(CONF_OUTPUT): cv.string,
vol.Optional(CONF_PEAK, default=-30): vol.Coerce(int),
vol.Optional(CONF_DURATION, default=1):
vol.All(vol.Coerce(int), vol.Range(min=1)),
vol.Optional(CONF_RESET, default=10):
vol.All(vol.Coerce(int), vol.Range(min=1)),
})
@asyncio.coroutine
def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
"""Create the binary sensor."""
manager = hass.data[DATA_FFMPEG]
# check source
if not manager.async_run_test(config.get(CONF_INPUT)):
return
# generate sensor object
entity = FFmpegNoise(hass, manager, config)
# add to system
manager.async_register_device(entity)
yield from async_add_devices([entity])
class FFmpegNoise(FFmpegBinarySensor):
"""A binary sensor which use ffmpeg for noise detection."""
def __init__(self, hass, manager, config):
"""Initialize ffmpeg noise binary sensor."""
from haffmpeg import SensorNoise
super().__init__(hass, config)
self.ffmpeg = SensorNoise(
manager.binary, hass.loop, self._async_callback)
def async_start_ffmpeg(self):
"""Start a FFmpeg instance.
This method must be run in the event loop and returns a coroutine.
"""
# init config
self.ffmpeg.set_options(
time_duration=self._config.get(CONF_DURATION),
time_reset=self._config.get(CONF_RESET),
peak=self._config.get(CONF_PEAK),
)
# run
return self.ffmpeg.open_sensor(
input_source=self._config.get(CONF_INPUT),
output_dest=self._config.get(CONF_OUTPUT),
extra_cmd=self._config.get(CONF_EXTRA_ARGUMENTS),
)
@property
def sensor_class(self):
"""Return the class of this sensor, from SENSOR_CLASSES."""
return "sound"

25
homeassistant/components/binary_sensor/services.yaml
View file

@ -1,25 +0,0 @@
# Describes the format for available binary_sensor services
ffmpeg_start:
description: Send a start command to a ffmpeg based sensor.
fields:
entity_id:
description: Name(s) of entites that will start. Platform dependent.
example: 'binary_sensor.ffmpeg_noise'
ffmpeg_stop:
description: Send a stop command to a ffmpeg based sensor.
fields:
entity_id:
description: Name(s) of entites that will stop. Platform dependent.
example: 'binary_sensor.ffmpeg_noise'
ffmpeg_restart:
description: Send a restart command to a ffmpeg based sensor.
fields:
entity_id:
description: Name(s) of entites that will restart. Platform dependent.
example: 'binary_sensor.ffmpeg_noise'

146
homeassistant/components/ffmpeg.py
View file

@ -6,18 +6,29 @@ https://home-assistant.io/components/ffmpeg/
"""
import asyncio
import logging
import os
import voluptuous as vol
from homeassistant.core import callback
from homeassistant.const import (
ATTR_ENTITY_ID, EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP)
from homeassistant.config import load_yaml_config_file
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity import Entity
DOMAIN = 'ffmpeg'
REQUIREMENTS = ["ha-ffmpeg==1.2"]
_LOGGER = logging.getLogger(__name__)
SERVICE_START = 'start'
SERVICE_STOP = 'stop'
SERVICE_RESTART = 'restart'
DATA_FFMPEG = 'ffmpeg'
CONF_INITIAL_STATE = 'initial_state'
CONF_INPUT = 'input'
CONF_FFMPEG_BIN = 'ffmpeg_bin'
CONF_EXTRA_ARGUMENTS = 'extra_arguments'
@ -34,18 +45,82 @@ CONFIG_SCHEMA = vol.Schema({
}),
}, extra=vol.ALLOW_EXTRA)
SERVICE_FFMPEG_SCHEMA = vol.Schema({
vol.Optional(ATTR_ENTITY_ID): cv.entity_ids,
})
def start(hass, entity_id=None):
"""Start a ffmpeg process on entity."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_START, data)
def stop(hass, entity_id=None):
"""Stop a ffmpeg process on entity."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_STOP, data)
def restart(hass, entity_id=None):
"""Restart a ffmpeg process on entity."""
data = {ATTR_ENTITY_ID: entity_id} if entity_id else {}
hass.services.call(DOMAIN, SERVICE_RESTART, data)
@asyncio.coroutine
def async_setup(hass, config):
"""Setup the FFmpeg component."""
conf = config.get(DOMAIN, {})
hass.data[DATA_FFMPEG] = FFmpegManager(
manager = FFmpegManager(
hass,
conf.get(CONF_FFMPEG_BIN, DEFAULT_BINARY),
conf.get(CONF_RUN_TEST, DEFAULT_RUN_TEST)
)
descriptions = yield from hass.loop.run_in_executor(
None, load_yaml_config_file,
os.path.join(os.path.dirname(__file__), 'services.yaml'))
# register service
@asyncio.coroutine
def async_service_handle(service):
"""Handle service ffmpeg process."""
entity_ids = service.data.get(ATTR_ENTITY_ID)
if entity_ids:
devices = [device for device in manager.entities
if device.entity_id in entity_ids]
else:
devices = manager.entities
tasks = []
for device in devices:
if service.service == SERVICE_START:
tasks.append(device.async_start_ffmpeg())
elif service.service == SERVICE_STOP:
tasks.append(device.async_stop_ffmpeg())
else:
tasks.append(device.async_restart_ffmpeg())
if tasks:
yield from asyncio.wait(tasks, loop=hass.loop)
hass.services.async_register(
DOMAIN, SERVICE_START, async_service_handle,
descriptions[DOMAIN].get(SERVICE_START), schema=SERVICE_FFMPEG_SCHEMA)
hass.services.async_register(
DOMAIN, SERVICE_STOP, async_service_handle,
descriptions[DOMAIN].get(SERVICE_STOP), schema=SERVICE_FFMPEG_SCHEMA)
hass.services.async_register(
DOMAIN, SERVICE_RESTART, async_service_handle,
descriptions[DOMAIN].get(SERVICE_RESTART),
schema=SERVICE_FFMPEG_SCHEMA)
hass.data[DATA_FFMPEG] = manager
return True
@ -58,12 +133,42 @@ class FFmpegManager(object):
self._cache = {}
self._bin = ffmpeg_bin
self._run_test = run_test
self._entities = []
@property
def binary(self):
"""Return ffmpeg binary from config."""
return self._bin
@property
def entities(self):
"""Return ffmpeg entities for services."""
return self._entities
@callback
def async_register_device(self, device):
"""Register a ffmpeg process/device."""
self._entities.append(device)
@asyncio.coroutine
def async_shutdown(event):
"""Stop ffmpeg process."""
yield from device.async_stop_ffmpeg()
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_STOP, async_shutdown)
# start on startup
if device.initial_state:
@asyncio.coroutine
def async_start(event):
"""Start ffmpeg process."""
yield from device.async_start_ffmpeg()
yield from device.async_update_ha_state()
self.hass.bus.async_listen_once(
EVENT_HOMEASSISTANT_START, async_start)
@asyncio.coroutine
def async_run_test(self, input_source):
"""Run test on this input. TRUE is deactivate or run correct.
@ -86,3 +191,42 @@ class FFmpegManager(object):
return False
self._cache[input_source] = True
return True
class FFmpegBase(Entity):
"""Interface object for ffmpeg."""
def __init__(self, initial_state=True):
"""Initialize ffmpeg base object."""
self.ffmpeg = None
self.initial_state = initial_state
@property
def available(self):
"""Return True if entity is available."""
return self.ffmpeg.is_running
@property
def should_poll(self):
"""Return True if entity has to be polled for state."""
return False
def async_start_ffmpeg(self):
"""Start a ffmpeg process.
This method must be run in the event loop and returns a coroutine.
"""
raise NotImplementedError()
def async_stop_ffmpeg(self):
"""Stop a ffmpeg process.
This method must be run in the event loop and returns a coroutine.
"""
return self.ffmpeg.close()
@asyncio.coroutine
def async_restart_ffmpeg(self):
"""Stop a ffmpeg process."""
yield from self.async_stop_ffmpeg()
yield from self.async_start_ffmpeg()

22
homeassistant/components/services.yaml
View file

@ -261,3 +261,25 @@ hdmi_cec:
standby:
description: Standby all devices which supports it.
ffmpeg:
start:
description: Send a start command to a ffmpeg based sensor.
fields:
entity_id:
description: Name(s) of entites that will start. Platform dependent.
example: 'binary_sensor.ffmpeg_noise'
stop:
description: Send a stop command to a ffmpeg based sensor.
fields:
entity_id:
description: Name(s) of entites that will stop. Platform dependent.
example: 'binary_sensor.ffmpeg_noise'
restart:
description: Send a restart command to a ffmpeg based sensor.
fields:
entity_id:
description: Name(s) of entites that will restart. Platform dependent.
example: 'binary_sensor.ffmpeg_noise'

104
tests/components/binary_sensor/test_ffmpeg.py Normal file
View file

@ -0,0 +1,104 @@
"""The tests for Home Assistant ffmpeg binary sensor."""
from unittest.mock import patch
from homeassistant.bootstrap import setup_component
from homeassistant.util.async import run_callback_threadsafe
from tests.common import (
get_test_home_assistant, assert_setup_component, mock_coro)
class TestFFmpegNoiseSetup(object):
"""Test class for ffmpeg."""
def setup_method(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.config = {
'ffmpeg': {
'run_test': False,
},
'binary_sensor': {
'platform': 'ffmpeg_noise',
'input': 'testinputvideo',
},
}
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_component(self):
"""Setup ffmpeg component."""
with assert_setup_component(1, 'binary_sensor'):
setup_component(self.hass, 'binary_sensor', self.config)
assert self.hass.data['ffmpeg'].binary == 'ffmpeg'
assert len(self.hass.data['ffmpeg'].entities) == 1
@patch('haffmpeg.SensorNoise.open_sensor', return_value=mock_coro()())
def test_setup_component_start(self, mock_start):
"""Setup ffmpeg component."""
with assert_setup_component(1, 'binary_sensor'):
setup_component(self.hass, 'binary_sensor', self.config)
assert self.hass.data['ffmpeg'].binary == 'ffmpeg'
assert len(self.hass.data['ffmpeg'].entities) == 1
entity = self.hass.data['ffmpeg'].entities[0]
self.hass.start()
assert mock_start.called
assert entity.state == 'off'
run_callback_threadsafe(
self.hass.loop, entity._async_callback, True).result()
assert entity.state == 'on'
class TestFFmpegMotionSetup(object):
"""Test class for ffmpeg."""
def setup_method(self):
"""Setup things to be run when tests are started."""
self.hass = get_test_home_assistant()
self.config = {
'ffmpeg': {
'run_test': False,
},
'binary_sensor': {
'platform': 'ffmpeg_motion',
'input': 'testinputvideo',
},
}
def teardown_method(self):
"""Stop everything that was started."""
self.hass.stop()
def test_setup_component(self):
"""Setup ffmpeg component."""
with assert_setup_component(1, 'binary_sensor'):
setup_component(self.hass, 'binary_sensor', self.config)
assert self.hass.data['ffmpeg'].binary == 'ffmpeg'
assert len(self.hass.data['ffmpeg'].entities) == 1
@patch('haffmpeg.SensorMotion.open_sensor', return_value=mock_coro()())
def test_setup_component_start(self, mock_start):
"""Setup ffmpeg component."""
with assert_setup_component(1, 'binary_sensor'):
setup_component(self.hass, 'binary_sensor', self.config)
assert self.hass.data['ffmpeg'].binary == 'ffmpeg'
assert len(self.hass.data['ffmpeg'].entities) == 1
entity = self.hass.data['ffmpeg'].entities[0]
self.hass.start()
assert mock_start.called
assert entity.state == 'off'
run_callback_threadsafe(
self.hass.loop, entity._async_callback, True).result()
assert entity.state == 'on'

227
tests/components/test_ffmpeg.py Normal file
View file

@ -0,0 +1,227 @@
"""The tests for Home Assistant ffmpeg."""
import asyncio
from unittest.mock import patch, MagicMock
import homeassistant.components.ffmpeg as ffmpeg
from homeassistant.bootstrap import setup_component
from homeassistant.util.async import (
run_callback_threadsafe, run_coroutine_threadsafe)
from tests.common import (
get_test_home_assistant, assert_setup_component, mock_coro)
class MockFFmpegDev(ffmpeg.FFmpegBase):
"""FFmpeg device mock."""
def __init__(self, initial_state=True, entity_id='test.ffmpeg_device'):
"""Initialize mock."""
super().__init__(initial_state)
self.entity_id = entity_id
self.ffmpeg = MagicMock
self.called_stop = False
self.called_start = False
self.called_restart = False
@asyncio.coroutine
def async_start_ffmpeg(self):
"""Mock start."""
self.called_start = True
@asyncio.coroutine
def async_stop_ffmpeg(self):
"""Mock stop."""
self.called_stop = True
@asyncio.coroutine
def async_restart_ffmpeg(self):
"""Mock restart."""
self.called_restart = True
class TestFFmpegSetup(object):
"""Test class for ffmpeg."""
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 ffmpeg component."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
assert self.hass.data[ffmpeg.DATA_FFMPEG].binary == 'ffmpeg'
def test_setup_component_test_service(self):
"""Setup ffmpeg component test services."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
assert self.hass.services.has_service(ffmpeg.DOMAIN, 'start')
assert self.hass.services.has_service(ffmpeg.DOMAIN, 'stop')
assert self.hass.services.has_service(ffmpeg.DOMAIN, 'restart')
def test_setup_component_test_register(self):
"""Setup ffmpeg component test register."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
self.hass.bus.async_listen_once = MagicMock()
ffmpeg_dev = MockFFmpegDev()
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
run_callback_threadsafe(
self.hass.loop, manager.async_register_device, ffmpeg_dev).result()
assert self.hass.bus.async_listen_once.called
assert self.hass.bus.async_listen_once.call_count == 2
assert len(manager.entities) == 1
assert manager.entities[0] == ffmpeg_dev
def test_setup_component_test_register_no_startup(self):
"""Setup ffmpeg component test register without startup."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
self.hass.bus.async_listen_once = MagicMock()
ffmpeg_dev = MockFFmpegDev(False)
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
run_callback_threadsafe(
self.hass.loop, manager.async_register_device, ffmpeg_dev).result()
assert self.hass.bus.async_listen_once.called
assert self.hass.bus.async_listen_once.call_count == 1
assert len(manager.entities) == 1
assert manager.entities[0] == ffmpeg_dev
def test_setup_component_test_servcie_start(self):
"""Setup ffmpeg component test service start."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
ffmpeg_dev = MockFFmpegDev(False)
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
run_callback_threadsafe(
self.hass.loop, manager.async_register_device, ffmpeg_dev).result()
ffmpeg.start(self.hass)
self.hass.block_till_done()
assert ffmpeg_dev.called_start
def test_setup_component_test_servcie_stop(self):
"""Setup ffmpeg component test service stop."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
ffmpeg_dev = MockFFmpegDev(False)
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
run_callback_threadsafe(
self.hass.loop, manager.async_register_device, ffmpeg_dev).result()
ffmpeg.stop(self.hass)
self.hass.block_till_done()
assert ffmpeg_dev.called_stop
def test_setup_component_test_servcie_restart(self):
"""Setup ffmpeg component test service restart."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
ffmpeg_dev = MockFFmpegDev(False)
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
run_callback_threadsafe(
self.hass.loop, manager.async_register_device, ffmpeg_dev).result()
ffmpeg.restart(self.hass)
self.hass.block_till_done()
assert ffmpeg_dev.called_restart
def test_setup_component_test_servcie_start_with_entity(self):
"""Setup ffmpeg component test service start."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
ffmpeg_dev = MockFFmpegDev(False)
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
run_callback_threadsafe(
self.hass.loop, manager.async_register_device, ffmpeg_dev).result()
ffmpeg.start(self.hass, 'test.ffmpeg_device')
self.hass.block_till_done()
assert ffmpeg_dev.called_start
def test_setup_component_test_run_test_false(self):
"""Setup ffmpeg component test run_test false."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {
'run_test': False,
}})
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
assert run_coroutine_threadsafe(
manager.async_run_test("blabalblabla"), self.hass.loop).result()
assert len(manager._cache) == 0
@patch('haffmpeg.Test.run_test',
return_value=mock_coro(return_value=True)())
def test_setup_component_test_run_test(self, mock_test):
"""Setup ffmpeg component test run_test."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
assert run_coroutine_threadsafe(
manager.async_run_test("blabalblabla"), self.hass.loop).result()
assert mock_test.called
assert mock_test.call_count == 1
assert len(manager._cache) == 1
assert manager._cache['blabalblabla']
assert run_coroutine_threadsafe(
manager.async_run_test("blabalblabla"), self.hass.loop).result()
assert mock_test.called
assert mock_test.call_count == 1
assert len(manager._cache) == 1
assert manager._cache['blabalblabla']
@patch('haffmpeg.Test.run_test',
return_value=mock_coro(return_value=False)())
def test_setup_component_test_run_test_test_fail(self, mock_test):
"""Setup ffmpeg component test run_test."""
with assert_setup_component(2):
setup_component(self.hass, ffmpeg.DOMAIN, {ffmpeg.DOMAIN: {}})
manager = self.hass.data[ffmpeg.DATA_FFMPEG]
assert not run_coroutine_threadsafe(
manager.async_run_test("blabalblabla"), self.hass.loop).result()
assert mock_test.called
assert mock_test.call_count == 1
assert len(manager._cache) == 1
assert not manager._cache['blabalblabla']
assert not run_coroutine_threadsafe(
manager.async_run_test("blabalblabla"), self.hass.loop).result()
assert mock_test.called
assert mock_test.call_count == 1
assert len(manager._cache) == 1
assert not manager._cache['blabalblabla']