jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-23 16:57:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Config flow for ONVIF (#34520)

Browse Source
This commit is contained in:
Jason Hunter 2020-05-01 02:15:40 -04:00 committed by GitHub
parent e7157f2164
commit 850b5cb02b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 1292 additions and 263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -507,6 +507,7 @@ omit =
homeassistant/components/ombi/*
homeassistant/components/onewire/sensor.py
homeassistant/components/onkyo/media_player.py
homeassistant/components/onvif/__init__.py
homeassistant/components/onvif/camera.py
homeassistant/components/opencv/*
homeassistant/components/openevse/sensor.py

1
CODEOWNERS
View File

@ -277,6 +277,7 @@ homeassistant/components/ohmconnect/* @robbiet480
homeassistant/components/ombi/* @larssont
homeassistant/components/onboarding/* @home-assistant/core
homeassistant/components/onewire/* @garbled1
homeassistant/components/onvif/* @hunterjm
homeassistant/components/openerz/* @misialq
homeassistant/components/opentherm_gw/* @mvn23
homeassistant/components/openuv/* @bachya

87
homeassistant/components/onvif/__init__.py
View File

@ -1 +1,86 @@
"""The onvif component."""
"""The ONVIF integration."""
import asyncio
import voluptuous as vol
from homeassistant.components.ffmpeg import CONF_EXTRA_ARGUMENTS
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_per_platform
from .const import (
CONF_PROFILE,
CONF_RTSP_TRANSPORT,
DEFAULT_ARGUMENTS,
DEFAULT_PROFILE,
DOMAIN,
RTSP_TRANS_PROTOCOLS,
)
CONFIG_SCHEMA = vol.Schema({DOMAIN: vol.Schema({})}, extra=vol.ALLOW_EXTRA)
PLATFORMS = ["camera"]
async def async_setup(hass: HomeAssistant, config: dict):
"""Set up the ONVIF component."""
# Import from yaml
configs = {}
for p_type, p_config in config_per_platform(config, "camera"):
if p_type != DOMAIN:
continue
config = p_config.copy()
profile = config.get(CONF_PROFILE, DEFAULT_PROFILE)
if config[CONF_HOST] not in configs.keys():
configs[config[CONF_HOST]] = config
configs[config[CONF_HOST]][CONF_PROFILE] = [profile]
else:
configs[config[CONF_HOST]][CONF_PROFILE].append(profile)
for conf in configs.values():
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=conf
)
)
return True
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Set up ONVIF from a config entry."""
if not entry.options:
await async_populate_options(hass, entry)
for component in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setup(entry, component)
)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
"""Unload a config entry."""
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
return unload_ok
async def async_populate_options(hass, entry):
"""Populate default options for device."""
options = {
CONF_EXTRA_ARGUMENTS: DEFAULT_ARGUMENTS,
CONF_RTSP_TRANSPORT: RTSP_TRANS_PROTOCOLS[0],
}
hass.config_entries.async_update_entry(entry, options=options)

467
homeassistant/components/onvif/camera.py
View File

@ -1,7 +1,6 @@
"""Support for ONVIF Cameras with FFmpeg as decoder."""
import asyncio
import datetime as dt
import logging
import os
from typing import Optional
@ -16,10 +15,9 @@ import voluptuous as vol
from zeep.asyncio import AsyncTransport
from zeep.exceptions import Fault
from homeassistant.components.camera import PLATFORM_SCHEMA, SUPPORT_STREAM, Camera
from homeassistant.components.camera import SUPPORT_STREAM, Camera
from homeassistant.components.ffmpeg import CONF_EXTRA_ARGUMENTS, DATA_FFMPEG
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
@ -27,131 +25,85 @@ from homeassistant.const import (
CONF_USERNAME,
)
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers import config_validation as cv, entity_platform
from homeassistant.helpers.aiohttp_client import (
async_aiohttp_proxy_stream,
async_get_clientsession,
)
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.service import async_extract_entity_ids
import homeassistant.util.dt as dt_util
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "ONVIF Camera"
DEFAULT_PORT = 5000
DEFAULT_USERNAME = "admin"
DEFAULT_PASSWORD = "888888"
DEFAULT_ARGUMENTS = "-pred 1"
DEFAULT_PROFILE = 0
CONF_PROFILE = "profile"
CONF_RTSP_TRANSPORT = "rtsp_transport"
ATTR_PAN = "pan"
ATTR_TILT = "tilt"
ATTR_ZOOM = "zoom"
ATTR_DISTANCE = "distance"
ATTR_SPEED = "speed"
ATTR_MOVE_MODE = "move_mode"
ATTR_CONTINUOUS_DURATION = "continuous_duration"
ATTR_PRESET = "preset"
DIR_UP = "UP"
DIR_DOWN = "DOWN"
DIR_LEFT = "LEFT"
DIR_RIGHT = "RIGHT"
ZOOM_OUT = "ZOOM_OUT"
ZOOM_IN = "ZOOM_IN"
PAN_FACTOR = {DIR_RIGHT: 1, DIR_LEFT: -1}
TILT_FACTOR = {DIR_UP: 1, DIR_DOWN: -1}
ZOOM_FACTOR = {ZOOM_IN: 1, ZOOM_OUT: -1}
CONTINUOUS_MOVE = "ContinuousMove"
RELATIVE_MOVE = "RelativeMove"
ABSOLUTE_MOVE = "AbsoluteMove"
GOTOPRESET_MOVE = "GotoPreset"
SERVICE_PTZ = "ptz"
DOMAIN = "onvif"
ONVIF_DATA = "onvif"
ENTITIES = "entities"
RTSP_TRANS_PROTOCOLS = ["tcp", "udp", "udp_multicast", "http"]
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_HOST): cv.string,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PASSWORD, default=DEFAULT_PASSWORD): cv.string,
vol.Optional(CONF_USERNAME, default=DEFAULT_USERNAME): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_EXTRA_ARGUMENTS, default=DEFAULT_ARGUMENTS): cv.string,
vol.Optional(CONF_RTSP_TRANSPORT, default=RTSP_TRANS_PROTOCOLS[0]): vol.In(
RTSP_TRANS_PROTOCOLS
),
vol.Optional(CONF_PROFILE, default=DEFAULT_PROFILE): vol.All(
vol.Coerce(int), vol.Range(min=0)
),
}
)
SERVICE_PTZ_SCHEMA = vol.Schema(
{
ATTR_ENTITY_ID: cv.entity_ids,
vol.Optional(ATTR_PAN): vol.In([DIR_LEFT, DIR_RIGHT]),
vol.Optional(ATTR_TILT): vol.In([DIR_UP, DIR_DOWN]),
vol.Optional(ATTR_ZOOM): vol.In([ZOOM_OUT, ZOOM_IN]),
ATTR_MOVE_MODE: vol.In(
[CONTINUOUS_MOVE, RELATIVE_MOVE, ABSOLUTE_MOVE, GOTOPRESET_MOVE]
),
vol.Optional(ATTR_CONTINUOUS_DURATION, default=0.5): cv.small_float,
vol.Optional(ATTR_DISTANCE, default=0.1): cv.small_float,
vol.Optional(ATTR_SPEED, default=0.5): cv.small_float,
vol.Optional(ATTR_PRESET, default="0"): cv.string,
}
from .const import (
ABSOLUTE_MOVE,
ATTR_CONTINUOUS_DURATION,
ATTR_DISTANCE,
ATTR_MOVE_MODE,
ATTR_PAN,
ATTR_PRESET,
ATTR_SPEED,
ATTR_TILT,
ATTR_ZOOM,
CONF_PROFILE,
CONF_RTSP_TRANSPORT,
CONTINUOUS_MOVE,
DIR_DOWN,
DIR_LEFT,
DIR_RIGHT,
DIR_UP,
DOMAIN,
ENTITIES,
GOTOPRESET_MOVE,
LOGGER,
PAN_FACTOR,
RELATIVE_MOVE,
SERVICE_PTZ,
TILT_FACTOR,
ZOOM_FACTOR,
ZOOM_IN,
ZOOM_OUT,
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up a ONVIF camera."""
_LOGGER.debug("Setting up the ONVIF camera platform")
async def async_setup_entry(hass, config_entry, async_add_entities):
"""Set up the ONVIF camera video stream."""
platform = entity_platform.current_platform.get()
async def async_handle_ptz(service):
"""Handle PTZ service call."""
pan = service.data.get(ATTR_PAN)
tilt = service.data.get(ATTR_TILT)
zoom = service.data.get(ATTR_ZOOM)
distance = service.data[ATTR_DISTANCE]
speed = service.data[ATTR_SPEED]
move_mode = service.data.get(ATTR_MOVE_MODE)
continuous_duration = service.data[ATTR_CONTINUOUS_DURATION]
preset = service.data[ATTR_PRESET]
all_cameras = hass.data[ONVIF_DATA][ENTITIES]
entity_ids = await async_extract_entity_ids(hass, service)
target_cameras = []
if not entity_ids:
target_cameras = all_cameras
else:
target_cameras = [
camera for camera in all_cameras if camera.entity_id in entity_ids
]
for camera in target_cameras:
await camera.async_perform_ptz(
pan, tilt, zoom, distance, speed, move_mode, continuous_duration, preset
)
hass.services.async_register(
DOMAIN, SERVICE_PTZ, async_handle_ptz, schema=SERVICE_PTZ_SCHEMA
# Create PTZ service
platform.async_register_entity_service(
SERVICE_PTZ,
{
vol.Optional(ATTR_PAN): vol.In([DIR_LEFT, DIR_RIGHT]),
vol.Optional(ATTR_TILT): vol.In([DIR_UP, DIR_DOWN]),
vol.Optional(ATTR_ZOOM): vol.In([ZOOM_OUT, ZOOM_IN]),
vol.Optional(ATTR_DISTANCE, default=0.1): cv.small_float,
vol.Optional(ATTR_SPEED, default=0.5): cv.small_float,
vol.Optional(ATTR_MOVE_MODE, default=RELATIVE_MOVE): vol.In(
[CONTINUOUS_MOVE, RELATIVE_MOVE, ABSOLUTE_MOVE, GOTOPRESET_MOVE]
),
vol.Optional(ATTR_CONTINUOUS_DURATION, default=0.5): cv.small_float,
vol.Optional(ATTR_PRESET, default="0"): cv.string,
},
"async_perform_ptz",
)
_LOGGER.debug("Constructing the ONVIFHassCamera")
base_config = {
CONF_NAME: config_entry.data[CONF_NAME],
CONF_HOST: config_entry.data[CONF_HOST],
CONF_PORT: config_entry.data[CONF_PORT],
CONF_USERNAME: config_entry.data[CONF_USERNAME],
CONF_PASSWORD: config_entry.data[CONF_PASSWORD],
CONF_EXTRA_ARGUMENTS: config_entry.options[CONF_EXTRA_ARGUMENTS],
CONF_RTSP_TRANSPORT: config_entry.options[CONF_RTSP_TRANSPORT],
}
hass_camera = ONVIFHassCamera(hass, config)
entities = []
for profile in config_entry.data[CONF_PROFILE]:
config = {**base_config, CONF_PROFILE: profile}
camera = ONVIFHassCamera(hass, config)
await camera.async_initialize()
entities.append(camera)
await hass_camera.async_initialize()
async_add_entities([hass_camera])
return
async_add_entities(entities)
return True
class ONVIFHassCamera(Camera):
@ -161,9 +113,9 @@ class ONVIFHassCamera(Camera):
"""Initialize an ONVIF camera."""
super().__init__()
_LOGGER.debug("Importing dependencies")
LOGGER.debug("Importing dependencies")
_LOGGER.debug("Setting up the ONVIF camera component")
LOGGER.debug("Setting up the ONVIF camera component")
self._username = config.get(CONF_USERNAME)
self._password = config.get(CONF_PASSWORD)
@ -172,13 +124,18 @@ class ONVIFHassCamera(Camera):
self._name = config.get(CONF_NAME)
self._ffmpeg_arguments = config.get(CONF_EXTRA_ARGUMENTS)
self._profile_index = config.get(CONF_PROFILE)
self._profile_token = None
self._profile_name = None
self._ptz_service = None
self._input = None
self._snapshot = None
self.stream_options[CONF_RTSP_TRANSPORT] = config.get(CONF_RTSP_TRANSPORT)
self._manufacturer = None
self._model = None
self._firmware_version = None
self._mac = None
_LOGGER.debug(
LOGGER.debug(
"Setting up the ONVIF camera device @ '%s:%s'", self._host, self._port
)
@ -201,29 +158,39 @@ class ONVIFHassCamera(Camera):
the camera. Also retrieves the ONVIF profiles.
"""
try:
_LOGGER.debug("Updating service addresses")
LOGGER.debug("Updating service addresses")
await self._camera.update_xaddrs()
await self.async_obtain_device_info()
await self.async_obtain_mac_address()
await self.async_check_date_and_time()
await self.async_obtain_profile_token()
await self.async_obtain_input_uri()
await self.async_obtain_snapshot_uri()
self.setup_ptz()
except ClientConnectionError as err:
_LOGGER.warning(
LOGGER.warning(
"Couldn't connect to camera '%s', but will retry later. Error: %s",
self._name,
err,
)
raise PlatformNotReady
except Fault as err:
_LOGGER.error(
LOGGER.error(
"Couldn't connect to camera '%s', please verify "
"that the credentials are correct. Error: %s",
self._name,
err,
)
async def async_obtain_device_info(self):
"""Obtain the MAC address of the camera to use as the unique ID."""
devicemgmt = self._camera.create_devicemgmt_service()
device_info = await devicemgmt.GetDeviceInformation()
self._manufacturer = device_info.Manufacturer
self._model = device_info.Model
self._firmware_version = device_info.FirmwareVersion
async def async_obtain_mac_address(self):
"""Obtain the MAC address of the camera to use as the unique ID."""
devicemgmt = self._camera.create_devicemgmt_service()
@ -234,15 +201,15 @@ class ONVIFHassCamera(Camera):
async def async_check_date_and_time(self):
"""Warns if camera and system date not synced."""
_LOGGER.debug("Setting up the ONVIF device management service")
LOGGER.debug("Setting up the ONVIF device management service")
devicemgmt = self._camera.create_devicemgmt_service()
_LOGGER.debug("Retrieving current camera date/time")
LOGGER.debug("Retrieving current camera date/time")
try:
system_date = dt_util.utcnow()
device_time = await devicemgmt.GetSystemDateAndTime()
if not device_time:
_LOGGER.debug(
LOGGER.debug(
"""Couldn't get camera '%s' date/time.
GetSystemDateAndTime() return null/empty""",
self._name,
@ -260,7 +227,7 @@ class ONVIFHassCamera(Camera):
cdate = device_time.LocalDateTime
if cdate is None:
_LOGGER.warning("Could not retrieve date/time on this camera")
LOGGER.warning("Could not retrieve date/time on this camera")
else:
cam_date = dt.datetime(
cdate.Date.Year,
@ -275,19 +242,19 @@ class ONVIFHassCamera(Camera):
cam_date_utc = cam_date.astimezone(dt_util.UTC)
_LOGGER.debug("TimeZone for date/time: %s", tzone)
LOGGER.debug("TimeZone for date/time: %s", tzone)
_LOGGER.debug("Camera date/time: %s", cam_date)
LOGGER.debug("Camera date/time: %s", cam_date)
_LOGGER.debug("Camera date/time in UTC: %s", cam_date_utc)
LOGGER.debug("Camera date/time in UTC: %s", cam_date_utc)
_LOGGER.debug("System date/time: %s", system_date)
LOGGER.debug("System date/time: %s", system_date)
dt_diff = cam_date - system_date
dt_diff_seconds = dt_diff.total_seconds()
if dt_diff_seconds > 5:
_LOGGER.warning(
LOGGER.warning(
"The date/time on the camera (UTC) is '%s', "
"which is different from the system '%s', "
"this could lead to authentication issues",
@ -295,7 +262,7 @@ class ONVIFHassCamera(Camera):
system_date,
)
except ServerDisconnectedError as err:
_LOGGER.warning(
LOGGER.warning(
"Couldn't get camera '%s' date/time. Error: %s", self._name, err
)
@ -306,10 +273,10 @@ class ONVIFHassCamera(Camera):
profiles = await media_service.GetProfiles()
_LOGGER.debug("Retrieved '%d' profiles", len(profiles))
LOGGER.debug("Retrieved '%d' profiles", len(profiles))
if self._profile_index >= len(profiles):
_LOGGER.warning(
LOGGER.warning(
"ONVIF Camera '%s' doesn't provide profile %d."
" Using the last profile.",
self._name,
@ -317,51 +284,32 @@ class ONVIFHassCamera(Camera):
)
self._profile_index = -1
_LOGGER.debug("Using profile index '%d'", self._profile_index)
LOGGER.debug("Using profile index '%d'", self._profile_index)
return profiles[self._profile_index].token
self._profile_token = profiles[self._profile_index].token
self._profile_name = profiles[self._profile_index].Name
except exceptions.ONVIFError as err:
_LOGGER.error(
LOGGER.error(
"Couldn't retrieve profile token of camera '%s'. Error: %s",
self._name,
err,
)
return None
async def async_obtain_input_uri(self):
"""Set the input uri for the camera."""
_LOGGER.debug(
LOGGER.debug(
"Connecting with ONVIF Camera: %s on port %s", self._host, self._port
)
try:
_LOGGER.debug("Retrieving profiles")
media_service = self._camera.create_media_service()
profiles = await media_service.GetProfiles()
_LOGGER.debug("Retrieved '%d' profiles", len(profiles))
if self._profile_index >= len(profiles):
_LOGGER.warning(
"ONVIF Camera '%s' doesn't provide profile %d."
" Using the last profile.",
self._name,
self._profile_index,
)
self._profile_index = -1
_LOGGER.debug("Using profile index '%d'", self._profile_index)
_LOGGER.debug("Retrieving stream uri")
LOGGER.debug("Retrieving stream uri")
# Fix Onvif setup error on Goke GK7102 based IP camera
# where we need to recreate media_service #26781
media_service = self._camera.create_media_service()
req = media_service.create_type("GetStreamUri")
req.ProfileToken = profiles[self._profile_index].token
req.ProfileToken = self._profile_token
req.StreamSetup = {
"Stream": "RTP-Unicast",
"Transport": {"Protocol": "RTSP"},
@ -374,155 +322,143 @@ class ONVIFHassCamera(Camera):
"rtsp://", f"rtsp://{self._username}:{self._password}@", 1
)
_LOGGER.debug(
LOGGER.debug(
"ONVIF Camera Using the following URL for %s: %s",
self._name,
uri_for_log,
)
except exceptions.ONVIFError as err:
_LOGGER.error("Couldn't setup camera '%s'. Error: %s", self._name, err)
LOGGER.error("Couldn't setup camera '%s'. Error: %s", self._name, err)
async def async_obtain_snapshot_uri(self):
"""Set the snapshot uri for the camera."""
_LOGGER.debug(
LOGGER.debug(
"Connecting with ONVIF Camera: %s on port %s", self._host, self._port
)
try:
_LOGGER.debug("Retrieving profiles")
media_service = self._camera.create_media_service()
profiles = await media_service.GetProfiles()
_LOGGER.debug("Retrieved '%d' profiles", len(profiles))
if self._profile_index >= len(profiles):
_LOGGER.warning(
"ONVIF Camera '%s' doesn't provide profile %d."
" Using the last profile.",
self._name,
self._profile_index,
)
self._profile_index = -1
_LOGGER.debug("Using profile index '%d'", self._profile_index)
_LOGGER.debug("Retrieving snapshot uri")
LOGGER.debug("Retrieving snapshot uri")
# Fix Onvif setup error on Goke GK7102 based IP camera
# where we need to recreate media_service #26781
media_service = self._camera.create_media_service()
req = media_service.create_type("GetSnapshotUri")
req.ProfileToken = profiles[self._profile_index].token
req.ProfileToken = self._profile_token
try:
snapshot_uri = await media_service.GetSnapshotUri(req)
self._snapshot = snapshot_uri.Uri
except ServerDisconnectedError as err:
_LOGGER.debug("Camera does not support GetSnapshotUri: %s", err)
LOGGER.debug("Camera does not support GetSnapshotUri: %s", err)
_LOGGER.debug(
LOGGER.debug(
"ONVIF Camera Using the following URL for %s snapshot: %s",
self._name,
self._snapshot,
)
except exceptions.ONVIFError as err:
_LOGGER.error("Couldn't setup camera '%s'. Error: %s", self._name, err)
LOGGER.error("Couldn't setup camera '%s'. Error: %s", self._name, err)
def setup_ptz(self):
"""Set up PTZ if available."""
_LOGGER.debug("Setting up the ONVIF PTZ service")
LOGGER.debug("Setting up the ONVIF PTZ service")
if self._camera.get_service("ptz", create=False) is None:
_LOGGER.debug("PTZ is not available")
LOGGER.debug("PTZ is not available")
else:
self._ptz_service = self._camera.create_ptz_service()
_LOGGER.debug("Completed set up of the ONVIF camera component")
LOGGER.debug("Completed set up of the ONVIF camera component")
async def async_perform_ptz(
self, pan, tilt, zoom, distance, speed, move_mode, continuous_duration, preset
self,
distance,
speed,
move_mode,
continuous_duration,
preset,
pan=None,
tilt=None,
zoom=None,
):
"""Perform a PTZ action on the camera."""
if self._ptz_service is None:
_LOGGER.warning("PTZ actions are not supported on camera '%s'", self._name)
LOGGER.warning("PTZ actions are not supported on camera '%s'", self._name)
return
if self._ptz_service:
pan_val = distance * PAN_FACTOR.get(pan, 0)
tilt_val = distance * TILT_FACTOR.get(tilt, 0)
zoom_val = distance * ZOOM_FACTOR.get(zoom, 0)
speed_val = speed
preset_val = preset
_LOGGER.debug(
"Calling %s PTZ | Pan = %4.2f | Tilt = %4.2f | Zoom = %4.2f | Speed = %4.2f | Preset = %s",
move_mode,
pan_val,
tilt_val,
zoom_val,
speed_val,
preset_val,
)
try:
req = self._ptz_service.create_type(move_mode)
req.ProfileToken = await self.async_obtain_profile_token()
if move_mode == CONTINUOUS_MOVE:
req.Velocity = {
"PanTilt": {"x": pan_val, "y": tilt_val},
"Zoom": {"x": zoom_val},
}
pan_val = distance * PAN_FACTOR.get(pan, 0)
tilt_val = distance * TILT_FACTOR.get(tilt, 0)
zoom_val = distance * ZOOM_FACTOR.get(zoom, 0)
speed_val = speed
preset_val = preset
LOGGER.debug(
"Calling %s PTZ | Pan = %4.2f | Tilt = %4.2f | Zoom = %4.2f | Speed = %4.2f | Preset = %s",
move_mode,
pan_val,
tilt_val,
zoom_val,
speed_val,
preset_val,
)
try:
req = self._ptz_service.create_type(move_mode)
req.ProfileToken = self._profile_token
if move_mode == CONTINUOUS_MOVE:
req.Velocity = {
"PanTilt": {"x": pan_val, "y": tilt_val},
"Zoom": {"x": zoom_val},
}
await self._ptz_service.ContinuousMove(req)
await asyncio.sleep(continuous_duration)
req = self._ptz_service.create_type("Stop")
req.ProfileToken = await self.async_obtain_profile_token()
await self._ptz_service.Stop({"ProfileToken": req.ProfileToken})
elif move_mode == RELATIVE_MOVE:
req.Translation = {
"PanTilt": {"x": pan_val, "y": tilt_val},
"Zoom": {"x": zoom_val},
}
req.Speed = {
"PanTilt": {"x": speed_val, "y": speed_val},
"Zoom": {"x": speed_val},
}
await self._ptz_service.RelativeMove(req)
elif move_mode == ABSOLUTE_MOVE:
req.Position = {
"PanTilt": {"x": pan_val, "y": tilt_val},
"Zoom": {"x": zoom_val},
}
req.Speed = {
"PanTilt": {"x": speed_val, "y": speed_val},
"Zoom": {"x": speed_val},
}
await self._ptz_service.AbsoluteMove(req)
elif move_mode == GOTOPRESET_MOVE:
req.PresetToken = preset_val
req.Speed = {
"PanTilt": {"x": speed_val, "y": speed_val},
"Zoom": {"x": speed_val},
}
await self._ptz_service.GotoPreset(req)
except exceptions.ONVIFError as err:
if "Bad Request" in err.reason:
self._ptz_service = None
_LOGGER.debug("Camera '%s' doesn't support PTZ.", self._name)
else:
_LOGGER.debug("Camera '%s' doesn't support PTZ.", self._name)
await self._ptz_service.ContinuousMove(req)
await asyncio.sleep(continuous_duration)
req = self._ptz_service.create_type("Stop")
req.ProfileToken = self._profile_token
await self._ptz_service.Stop({"ProfileToken": req.ProfileToken})
elif move_mode == RELATIVE_MOVE:
req.Translation = {
"PanTilt": {"x": pan_val, "y": tilt_val},
"Zoom": {"x": zoom_val},
}
req.Speed = {
"PanTilt": {"x": speed_val, "y": speed_val},
"Zoom": {"x": speed_val},
}
await self._ptz_service.RelativeMove(req)
elif move_mode == ABSOLUTE_MOVE:
req.Position = {
"PanTilt": {"x": pan_val, "y": tilt_val},
"Zoom": {"x": zoom_val},
}
req.Speed = {
"PanTilt": {"x": speed_val, "y": speed_val},
"Zoom": {"x": speed_val},
}
await self._ptz_service.AbsoluteMove(req)
elif move_mode == GOTOPRESET_MOVE:
req.PresetToken = preset_val
req.Speed = {
"PanTilt": {"x": speed_val, "y": speed_val},
"Zoom": {"x": speed_val},
}
await self._ptz_service.GotoPreset(req)
except exceptions.ONVIFError as err:
if "Bad Request" in err.reason:
self._ptz_service = None
LOGGER.debug("Camera '%s' doesn't support PTZ.", self._name)
else:
LOGGER.error("Error trying to perform PTZ action: %s", err)
async def async_added_to_hass(self):
"""Handle entity addition to hass."""
_LOGGER.debug("Camera '%s' added to hass", self._name)
LOGGER.debug("Camera '%s' added to hass", self._name)
if ONVIF_DATA not in self.hass.data:
self.hass.data[ONVIF_DATA] = {}
self.hass.data[ONVIF_DATA][ENTITIES] = []
self.hass.data[ONVIF_DATA][ENTITIES].append(self)
if DOMAIN not in self.hass.data:
self.hass.data[DOMAIN] = {}
self.hass.data[DOMAIN][ENTITIES] = []
self.hass.data[DOMAIN][ENTITIES].append(self)
async def async_camera_image(self):
"""Return a still image response from the camera."""
_LOGGER.debug("Retrieving image from camera '%s'", self._name)
LOGGER.debug("Retrieving image from camera '%s'", self._name)
image = None
if self._snapshot is not None:
@ -537,7 +473,7 @@ class ONVIFHassCamera(Camera):
if response.status_code < 300:
return response.content
except requests.exceptions.RequestException as error:
_LOGGER.error(
LOGGER.error(
"Fetch snapshot image failed from %s, falling back to FFmpeg; %s",
self._name,
error,
@ -564,7 +500,7 @@ class ONVIFHassCamera(Camera):
async def handle_async_mjpeg_stream(self, request):
"""Generate an HTTP MJPEG stream from the camera."""
_LOGGER.debug("Handling mjpeg stream from camera '%s'", self._name)
LOGGER.debug("Handling mjpeg stream from camera '%s'", self._name)
ffmpeg_manager = self.hass.data[DATA_FFMPEG]
stream = CameraMjpeg(ffmpeg_manager.binary, loop=self.hass.loop)
@ -596,7 +532,7 @@ class ONVIFHassCamera(Camera):
@property
def name(self):
"""Return the name of this camera."""
return self._name
return f"{self._name} - {self._profile_name}"
@property
def unique_id(self) -> Optional[str]:
@ -604,3 +540,14 @@ class ONVIFHassCamera(Camera):
if self._profile_index:
return f"{self._mac}_{self._profile_index}"
return self._mac
@property
def device_info(self):
"""Return a device description for device registry."""
return {
"identifiers": {(DOMAIN, self._mac)},
"name": self._name,
"manufacturer": self._manufacturer,
"model": self._model,
"sw_version": self._firmware_version,
}

311
homeassistant/components/onvif/config_flow.py Normal file
View File

@ -0,0 +1,311 @@
"""Config flow for ONVIF."""
import os
from pprint import pformat
from typing import List
from urllib.parse import urlparse
import onvif
from onvif import ONVIFCamera, exceptions
import voluptuous as vol
from wsdiscovery.discovery import ThreadedWSDiscovery as WSDiscovery
from wsdiscovery.scope import Scope
from wsdiscovery.service import Service
from zeep.asyncio import AsyncTransport
from zeep.exceptions import Fault
from homeassistant import config_entries
from homeassistant.components.ffmpeg import CONF_EXTRA_ARGUMENTS
from homeassistant.const import (
CONF_HOST,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
)
from homeassistant.core import callback
from homeassistant.helpers.aiohttp_client import async_get_clientsession
# pylint: disable=unused-import
from .const import (
CONF_DEVICE_ID,
CONF_PROFILE,
CONF_RTSP_TRANSPORT,
DEFAULT_ARGUMENTS,
DEFAULT_PORT,
DOMAIN,
LOGGER,
RTSP_TRANS_PROTOCOLS,
)
CONF_MANUAL_INPUT = "Manually configure ONVIF device"
def wsdiscovery() -> List[Service]:
"""Get ONVIF Profile S devices from network."""
discovery = WSDiscovery(ttl=4)
discovery.start()
services = discovery.searchServices(
scopes=[Scope("onvif://www.onvif.org/Profile/Streaming")]
)
discovery.stop()
return services
async def async_discovery(hass) -> bool:
"""Return if there are devices that can be discovered."""
LOGGER.debug("Starting ONVIF discovery...")
services = await hass.async_add_executor_job(wsdiscovery)
devices = []
for service in services:
url = urlparse(service.getXAddrs()[0])
device = {
CONF_DEVICE_ID: None,
CONF_NAME: service.getEPR(),
CONF_HOST: url.hostname,
CONF_PORT: url.port or 80,
}
for scope in service.getScopes():
scope_str = scope.getValue()
if scope_str.lower().startswith("onvif://www.onvif.org/name"):
device[CONF_NAME] = scope_str.split("/")[-1]
if scope_str.lower().startswith("onvif://www.onvif.org/mac"):
device[CONF_DEVICE_ID] = scope_str.split("/")[-1]
devices.append(device)
return devices
class OnvifFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a ONVIF config flow."""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Get the options flow for this handler."""
return OnvifOptionsFlowHandler(config_entry)
def __init__(self):
"""Initialize the ONVIF config flow."""
self.device_id = None
self.devices = []
self.onvif_config = {}
async def async_step_user(self, user_input=None):
"""Handle user flow."""
if user_input is not None:
return await self.async_step_device()
return self.async_show_form(step_id="user")
async def async_step_device(self, user_input=None):
"""Handle WS-Discovery.
Let user choose between discovered devices and manual configuration.
If no device is found allow user to manually input configuration.
"""
if user_input:
if CONF_MANUAL_INPUT == user_input[CONF_HOST]:
return await self.async_step_manual_input()
for device in self.devices:
name = f"{device[CONF_NAME]} ({device[CONF_HOST]})"
if name == user_input[CONF_HOST]:
self.device_id = device[CONF_DEVICE_ID]
self.onvif_config = {
CONF_NAME: device[CONF_NAME],
CONF_HOST: device[CONF_HOST],
CONF_PORT: device[CONF_PORT],
}
return await self.async_step_auth()
discovery = await async_discovery(self.hass)
for device in discovery:
configured = False
for entry in self._async_current_entries():
if entry.unique_id == device[CONF_DEVICE_ID]:
configured = True
break
if not configured:
self.devices.append(device)
LOGGER.debug("Discovered ONVIF devices %s", pformat(self.devices))
if self.devices:
names = []
for device in self.devices:
names.append(f"{device[CONF_NAME]} ({device[CONF_HOST]})")
names.append(CONF_MANUAL_INPUT)
return self.async_show_form(
step_id="device",
data_schema=vol.Schema({vol.Optional(CONF_HOST): vol.In(names)}),
)
return await self.async_step_manual_input()
async def async_step_manual_input(self, user_input=None):
"""Manual configuration."""
if user_input:
self.onvif_config = user_input
return await self.async_step_auth()
return self.async_show_form(
step_id="manual_input",
data_schema=vol.Schema(
{
vol.Required(CONF_NAME): str,
vol.Required(CONF_HOST): str,
vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
}
),
)
async def async_step_auth(self, user_input=None):
"""Username and Password configuration for ONVIF device."""
if user_input:
self.onvif_config[CONF_USERNAME] = user_input[CONF_USERNAME]
self.onvif_config[CONF_PASSWORD] = user_input[CONF_PASSWORD]
return await self.async_step_profiles()
return self.async_show_form(
step_id="auth",
data_schema=vol.Schema(
{vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
),
)
async def async_step_profiles(self, user_input=None):
"""Fetch ONVIF device profiles."""
errors = {}
LOGGER.debug(
"Fetching profiles from ONVIF device %s", pformat(self.onvif_config)
)
device = get_device(
self.hass,
self.onvif_config[CONF_HOST],
self.onvif_config[CONF_PORT],
self.onvif_config[CONF_USERNAME],
self.onvif_config[CONF_PASSWORD],
)
await device.update_xaddrs()
try:
# Get the MAC address to use as the unique ID for the config flow
if not self.device_id:
devicemgmt = device.create_devicemgmt_service()
network_interfaces = await devicemgmt.GetNetworkInterfaces()
for interface in network_interfaces:
if interface.Enabled:
self.device_id = interface.Info.HwAddress
if self.device_id is None:
return self.async_abort(reason="no_mac")
await self.async_set_unique_id(self.device_id, raise_on_progress=False)
self._abort_if_unique_id_configured(
updates={
CONF_HOST: self.onvif_config[CONF_HOST],
CONF_PORT: self.onvif_config[CONF_PORT],
CONF_NAME: self.onvif_config[CONF_NAME],
}
)
if not self.onvif_config.get(CONF_PROFILE):
self.onvif_config[CONF_PROFILE] = []
media_service = device.create_media_service()
profiles = await media_service.GetProfiles()
LOGGER.debug("Media Profiles %s", pformat(profiles))
for key, profile in enumerate(profiles):
if profile.VideoEncoderConfiguration.Encoding != "H264":
continue
self.onvif_config[CONF_PROFILE].append(key)
if not self.onvif_config[CONF_PROFILE]:
return self.async_abort(reason="no_h264")
title = f"{self.onvif_config[CONF_NAME]} - {self.device_id}"
return self.async_create_entry(title=title, data=self.onvif_config)
except exceptions.ONVIFError as err:
LOGGER.error(
"Couldn't setup ONVIF device '%s'. Error: %s",
self.onvif_config[CONF_NAME],
err,
)
return self.async_abort(reason="onvif_error")
except Fault:
errors["base"] = "connection_failed"
return self.async_show_form(step_id="auth", errors=errors)
async def async_step_import(self, user_input):
"""Handle import."""
self.onvif_config = user_input
return await self.async_step_profiles()
class OnvifOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle ONVIF options."""
def __init__(self, config_entry):
"""Initialize ONVIF options flow."""
self.config_entry = config_entry
self.options = dict(config_entry.options)
async def async_step_init(self, user_input=None):
"""Manage the ONVIF options."""
return await self.async_step_onvif_devices()
async def async_step_onvif_devices(self, user_input=None):
"""Manage the ONVIF devices options."""
if user_input is not None:
self.options[CONF_EXTRA_ARGUMENTS] = user_input[CONF_EXTRA_ARGUMENTS]
self.options[CONF_RTSP_TRANSPORT] = user_input[CONF_RTSP_TRANSPORT]
return self.async_create_entry(title="", data=self.options)
return self.async_show_form(
step_id="onvif_devices",
data_schema=vol.Schema(
{
vol.Optional(
CONF_EXTRA_ARGUMENTS,
default=self.config_entry.options.get(
CONF_EXTRA_ARGUMENTS, DEFAULT_ARGUMENTS
),
): str,
vol.Optional(
CONF_RTSP_TRANSPORT,
default=self.config_entry.options.get(
CONF_RTSP_TRANSPORT, RTSP_TRANS_PROTOCOLS[0]
),
): vol.In(RTSP_TRANS_PROTOCOLS),
}
),
)
def get_device(hass, host, port, username, password) -> ONVIFCamera:
"""Get ONVIFCamera instance."""
session = async_get_clientsession(hass)
transport = AsyncTransport(None, session=session)
device = ONVIFCamera(
host,
port,
username,
password,
f"{os.path.dirname(onvif.__file__)}/wsdl/",
transport=transport,
)
return device

47
homeassistant/components/onvif/const.py Normal file
View File

@ -0,0 +1,47 @@
"""Constants for the onvif component."""
import logging
LOGGER = logging.getLogger(__package__)
DOMAIN = "onvif"
ONVIF_DATA = "onvif"
ENTITIES = "entities"
DEFAULT_NAME = "ONVIF Camera"
DEFAULT_PORT = 5000
DEFAULT_USERNAME = "admin"
DEFAULT_PASSWORD = "888888"
DEFAULT_ARGUMENTS = "-pred 1"
DEFAULT_PROFILE = 0
CONF_DEVICE_ID = "deviceid"
CONF_PROFILE = "profile"
CONF_RTSP_TRANSPORT = "rtsp_transport"
RTSP_TRANS_PROTOCOLS = ["tcp", "udp", "udp_multicast", "http"]
ATTR_PAN = "pan"
ATTR_TILT = "tilt"
ATTR_ZOOM = "zoom"
ATTR_DISTANCE = "distance"
ATTR_SPEED = "speed"
ATTR_MOVE_MODE = "move_mode"
ATTR_CONTINUOUS_DURATION = "continuous_duration"
ATTR_PRESET = "preset"
DIR_UP = "UP"
DIR_DOWN = "DOWN"
DIR_LEFT = "LEFT"
DIR_RIGHT = "RIGHT"
ZOOM_OUT = "ZOOM_OUT"
ZOOM_IN = "ZOOM_IN"
PAN_FACTOR = {DIR_RIGHT: 1, DIR_LEFT: -1}
TILT_FACTOR = {DIR_UP: 1, DIR_DOWN: -1}
ZOOM_FACTOR = {ZOOM_IN: 1, ZOOM_OUT: -1}
CONTINUOUS_MOVE = "ContinuousMove"
RELATIVE_MOVE = "RelativeMove"
ABSOLUTE_MOVE = "AbsoluteMove"
GOTOPRESET_MOVE = "GotoPreset"
SERVICE_PTZ = "ptz"
ENTITIES = "entities"

5
homeassistant/components/onvif/manifest.json
View File

@ -2,7 +2,8 @@
"domain": "onvif",
"name": "ONVIF",
"documentation": "https://www.home-assistant.io/integrations/onvif",
"requirements": ["onvif-zeep-async==0.2.0"],
"requirements": ["onvif-zeep-async==0.2.0", "WSDiscovery==2.0.0"],
"dependencies": ["ffmpeg"],
"codeowners": []
"codeowners": ["@hunterjm"],
"config_flow": true
}

58
homeassistant/components/onvif/strings.json Normal file
View File

@ -0,0 +1,58 @@
{
"config": {
"abort": {
"already_configured": "ONVIF device is already configured.",
"already_in_progress": "Config flow for ONVIF device is already in progress.",
"onvif_error": "Error setting up ONVIF device. Check logs for more information.",
"no_h264": "There were no H264 streams available. Check the profile configuration on your device.",
"no_mac": "Could not configure unique ID for ONVIF device."
},
"error": {
"connection_failed": "Could not connect to ONVIF service with provided credentials."
},
"step": {
"user": {
"title": "ONVIF device setup",
"description": "By clicking submit, we will search your network for ONVIF devices that support Profile S.\n\nSome manufacturers have started to disable ONVIF by default. Please ensure ONVIF is enabled in your camera's configuration."
},
"device": {
"data": {
"host": "Select discovered ONVIF device"
},
"title": "Select ONVIF device"
},
"manual_input": {
"data": {
"host": "Host",
"port": "Port"
},
"title": "Configure ONVIF device"
},
"auth": {
"title": "Configure authentication",
"data": {
"username": "Username",
"password": "Password"
}
},
"configure_profile": {
"description": "Create camera entity for {profile} at {resolution} resolution?",
"title": "Configure Profiles",
"data": {
"include": "Create camera entity"
}
}
}
},
"options": {
"step": {
"onvif_devices": {
"data": {
"extra_arguments": "Extra FFMPEG arguments",
"rtsp_transport": "RTSP transport mechanism"
},
"title": "ONVIF Device Options"
}
}
}
}

59
homeassistant/components/onvif/translations/en.json Normal file
View File

@ -0,0 +1,59 @@
{
"config": {
"abort": {
"already_configured": "ONVIF device is already configured.",
"already_in_progress": "Config flow for ONVIF device is already in progress.",
"no_h264": "There were no H264 streams available. Check the profile configuration on your device.",
"no_mac": "Could not configure unique ID for ONVIF device.",
"onvif_error": "Error setting up ONVIF device. Check logs for more information."
},
"error": {
"connection_failed": "Could not connect to ONVIF service with provided credentials."
},
"step": {
"auth": {
"data": {
"password": "Password",
"username": "Username"
},
"title": "Configure authentication"
},
"configure_profile": {
"data": {
"include": "Create camera entity"
},
"description": "Create camera entity for {profile} at {resolution} resolution?",
"title": "Configure Profiles"
},
"device": {
"data": {
"host": "Select discovered ONVIF device"
},
"title": "Select ONVIF device"
},
"manual_input": {
"data": {
"host": "Host",
"port": "Port"
},
"title": "Configure ONVIF device"
},
"user": {
"description": "By clicking submit, we will search your network for ONVIF devices that support Profile S.\n\nSome manufacturers have started to disable ONVIF by default. Please ensure ONVIF is enabled in your camera's configuration.",
"title": "ONVIF device setup"
}
}
},
"options": {
"step": {
"onvif_devices": {
"data": {
"extra_arguments": "Extra FFMPEG arguments",
"rtsp_transport": "RTSP transport mechanism"
},
"title": "ONVIF Device Options"
}
}
},
"title": "ONVIF"
}

1
homeassistant/generated/config_flows.py
View File

@ -91,6 +91,7 @@ FLOWS = [
"nuheat",
"nut",
"nws",
"onvif",
"opentherm_gw",
"openuv",
"owntracks",

3
requirements_all.txt
View File

@ -101,6 +101,9 @@ TwitterAPI==2.5.11
# homeassistant.components.tof
# VL53L1X2==0.1.5
# homeassistant.components.onvif
WSDiscovery==2.0.0
# homeassistant.components.waze_travel_time
WazeRouteCalculator==0.12

6
requirements_test_all.txt
View File

@ -23,6 +23,9 @@ PyTransportNSW==0.1.1
# homeassistant.components.remember_the_milk
RtmAPI==0.7.2
# homeassistant.components.onvif
WSDiscovery==2.0.0
# homeassistant.components.yessssms
YesssSMS==0.4.1
@ -389,6 +392,9 @@ numpy==1.18.2
# homeassistant.components.google
oauth2client==4.0.0
# homeassistant.components.onvif
onvif-zeep-async==0.2.0
# homeassistant.components.openerz
openerz-api==0.1.0

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

@ -0,0 +1 @@
"""Tests for the ONVIF integration."""

508
tests/components/onvif/test_config_flow.py Normal file
View File

@ -0,0 +1,508 @@
"""Test ONVIF config flow."""
from asyncio import Future
from asynctest import MagicMock, patch
from onvif.exceptions import ONVIFError
from zeep.exceptions import Fault
from homeassistant import config_entries, data_entry_flow
from homeassistant.components.onvif import config_flow
from tests.common import MockConfigEntry
URN = "urn:uuid:123456789"
NAME = "TestCamera"
HOST = "1.2.3.4"
PORT = 80
USERNAME = "admin"
PASSWORD = "12345"
MAC = "aa:bb:cc:dd:ee"
DISCOVERY = [
{
"EPR": URN,
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
"MAC": MAC,
},
{
"EPR": "urn:uuid:987654321",
config_flow.CONF_NAME: "TestCamera2",
config_flow.CONF_HOST: "5.6.7.8",
config_flow.CONF_PORT: PORT,
"MAC": "ee:dd:cc:bb:aa",
},
]
def setup_mock_onvif_device(
mock_device, with_h264=True, two_profiles=False, with_interfaces=True
):
"""Prepare mock ONVIF device."""
devicemgmt = MagicMock()
interface = MagicMock()
interface.Enabled = True
interface.Info.HwAddress = MAC
devicemgmt.GetNetworkInterfaces.return_value = Future()
devicemgmt.GetNetworkInterfaces.return_value.set_result(
[interface] if with_interfaces else []
)
media_service = MagicMock()
profile1 = MagicMock()
profile1.VideoEncoderConfiguration.Encoding = "H264" if with_h264 else "MJPEG"
profile2 = MagicMock()
profile2.VideoEncoderConfiguration.Encoding = "H264" if two_profiles else "MJPEG"
media_service.GetProfiles.return_value = Future()
media_service.GetProfiles.return_value.set_result([profile1, profile2])
mock_device.update_xaddrs.return_value = Future()
mock_device.update_xaddrs.return_value.set_result(True)
mock_device.create_devicemgmt_service = MagicMock(return_value=devicemgmt)
mock_device.create_media_service = MagicMock(return_value=media_service)
def mock_constructor(
host,
port,
user,
passwd,
wsdl_dir,
encrypt=True,
no_cache=False,
adjust_time=False,
transport=None,
):
"""Fake the controller constructor."""
return mock_device
mock_device.side_effect = mock_constructor
def setup_mock_discovery(
mock_discovery, with_name=False, with_mac=False, two_devices=False
):
"""Prepare mock discovery result."""
services = []
for item in DISCOVERY:
service = MagicMock()
service.getXAddrs = MagicMock(
return_value=[
f"http://{item[config_flow.CONF_HOST]}:{item[config_flow.CONF_PORT]}/onvif/device_service"
]
)
service.getEPR = MagicMock(return_value=item["EPR"])
scopes = []
if with_name:
scope = MagicMock()
scope.getValue = MagicMock(
return_value=f"onvif://www.onvif.org/name/{item[config_flow.CONF_NAME]}"
)
scopes.append(scope)
if with_mac:
scope = MagicMock()
scope.getValue = MagicMock(
return_value=f"onvif://www.onvif.org/mac/{item['MAC']}"
)
scopes.append(scope)
service.getScopes = MagicMock(return_value=scopes)
services.append(service)
mock_discovery.return_value = services
def setup_mock_camera(mock_camera):
"""Prepare mock HASS camera."""
mock_camera.async_initialize.return_value = Future()
mock_camera.async_initialize.return_value.set_result(True)
def mock_constructor(hass, config):
"""Fake the controller constructor."""
return mock_camera
mock_camera.side_effect = mock_constructor
async def setup_onvif_integration(
hass, config=None, options=None, unique_id=MAC, entry_id="1", source="user",
):
"""Create an ONVIF config entry."""
if not config:
config = {
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
config_flow.CONF_PROFILE: [0],
}
config_entry = MockConfigEntry(
domain=config_flow.DOMAIN,
source=source,
data={**config},
connection_class=config_entries.CONN_CLASS_LOCAL_PUSH,
options=options or {},
entry_id=entry_id,
unique_id=unique_id,
)
config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_device, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.camera.ONVIFHassCamera"
) as mock_camera:
setup_mock_onvif_device(mock_device, two_profiles=True)
# no discovery
mock_discovery.return_value = []
setup_mock_camera(mock_camera)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
return config_entry
async def test_flow_discovered_devices(hass):
"""Test that config flow works for discovered devices."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_device, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.camera.ONVIFHassCamera"
) as mock_camera:
setup_mock_onvif_device(mock_device)
setup_mock_discovery(mock_discovery)
setup_mock_camera(mock_camera)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "device"
assert len(result["data_schema"].schema[config_flow.CONF_HOST].container) == 3
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={config_flow.CONF_HOST: f"{URN} ({HOST})"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == f"{URN} - {MAC}"
assert result["data"] == {
config_flow.CONF_NAME: URN,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
config_flow.CONF_PROFILE: [0],
}
async def test_flow_discovered_devices_ignore_configured_manual_input(hass):
"""Test that config flow discovery ignores configured devices."""
await setup_onvif_integration(hass)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_device, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.camera.ONVIFHassCamera"
) as mock_camera:
setup_mock_onvif_device(mock_device)
setup_mock_discovery(mock_discovery, with_mac=True)
setup_mock_camera(mock_camera)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "device"
assert len(result["data_schema"].schema[config_flow.CONF_HOST].container) == 2
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={config_flow.CONF_HOST: config_flow.CONF_MANUAL_INPUT},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "manual_input"
async def test_flow_discovery_ignore_existing_and_abort(hass):
"""Test that config flow discovery ignores setup devices."""
await setup_onvif_integration(hass)
await setup_onvif_integration(
hass,
config={
config_flow.CONF_NAME: DISCOVERY[1]["EPR"],
config_flow.CONF_HOST: DISCOVERY[1][config_flow.CONF_HOST],
config_flow.CONF_PORT: DISCOVERY[1][config_flow.CONF_PORT],
config_flow.CONF_USERNAME: "",
config_flow.CONF_PASSWORD: "",
},
unique_id=DISCOVERY[1]["MAC"],
entry_id="2",
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_device, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.camera.ONVIFHassCamera"
) as mock_camera:
setup_mock_onvif_device(mock_device)
setup_mock_discovery(mock_discovery, with_name=True, with_mac=True)
setup_mock_camera(mock_camera)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
# It should skip to manual entry if the only devices are already configured
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "manual_input"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
# It should abort if already configured and entered manually
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
async def test_flow_manual_entry(hass):
"""Test that config flow works for discovered devices."""
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN, context={"source": "user"}
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.onvif.config_flow.get_device"
) as mock_device, patch(
"homeassistant.components.onvif.config_flow.wsdiscovery"
) as mock_discovery, patch(
"homeassistant.components.onvif.camera.ONVIFHassCamera"
) as mock_camera:
setup_mock_onvif_device(mock_device, two_profiles=True)
# no discovery
mock_discovery.return_value = []
setup_mock_camera(mock_camera)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "manual_input"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == f"{NAME} - {MAC}"
assert result["data"] == {
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
config_flow.CONF_PROFILE: [0, 1],
}
async def test_flow_import_no_mac(hass):
"""Test that config flow fails when no MAC available."""
with patch("homeassistant.components.onvif.config_flow.get_device") as mock_device:
setup_mock_onvif_device(mock_device, with_interfaces=False)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
config_flow.CONF_PROFILE: [0],
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_mac"
async def test_flow_import_no_h264(hass):
"""Test that config flow fails when no MAC available."""
with patch("homeassistant.components.onvif.config_flow.get_device") as mock_device:
setup_mock_onvif_device(mock_device, with_h264=False)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_h264"
async def test_flow_import_onvif_api_error(hass):
"""Test that config flow fails when ONVIF API fails."""
with patch("homeassistant.components.onvif.config_flow.get_device") as mock_device:
setup_mock_onvif_device(mock_device)
mock_device.create_devicemgmt_service = MagicMock(
side_effect=ONVIFError("Could not get device mgmt service")
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "onvif_error"
async def test_flow_import_onvif_auth_error(hass):
"""Test that config flow fails when ONVIF API fails."""
with patch("homeassistant.components.onvif.config_flow.get_device") as mock_device:
setup_mock_onvif_device(mock_device)
mock_device.create_devicemgmt_service = MagicMock(
side_effect=Fault("Auth Error")
)
result = await hass.config_entries.flow.async_init(
config_flow.DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
config_flow.CONF_NAME: NAME,
config_flow.CONF_HOST: HOST,
config_flow.CONF_PORT: PORT,
config_flow.CONF_USERNAME: USERNAME,
config_flow.CONF_PASSWORD: PASSWORD,
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "auth"
assert result["errors"]["base"] == "connection_failed"
async def test_option_flow(hass):
"""Test config flow options."""
entry = await setup_onvif_integration(hass)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "onvif_devices"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
config_flow.CONF_EXTRA_ARGUMENTS: "",
config_flow.CONF_RTSP_TRANSPORT: config_flow.RTSP_TRANS_PROTOCOLS[1],
},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {
config_flow.CONF_EXTRA_ARGUMENTS: "",
config_flow.CONF_RTSP_TRANSPORT: config_flow.RTSP_TRANS_PROTOCOLS[1],
}