jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-27 10:47:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update Ezviz Component (#45722)

Browse Source 
* Update Ezviz Component

* Update Ezviz for pylint test

* Update Ezviz component pylint tests

* Update Ezviz component tests

* Update Ezviz Component tests

* Update Ezviz component pylint error

* Fix ezviz component config flow tests

* Update ezviz component

* Update Ezviz component

* Add sensor platforms

* issue with requirements file

* Update binary_sensor to include switches

* Updates to Ezviz sensors

* Removed enum private method.

* Fix switch args

* Update homeassistant/components/ezviz/switch.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* config flow checks login info

* Config_flow now imports ezviz from camera platform

* Update test

* Updated config_flow with unique_id and remove period from logging

* Added two camera services and clarified service descryptions in services.yaml

* Fixed variable name mistake with new service

* Added french integration translation

* Config_flow add camera rtsp credentials as seperate entities, with user step and import step

* rerun hassfest after rebase

* Removed region from legacy config schema, removed logging in camera platform setup that could contain credentials, removed unused constant.

* Regenerate requirements

* Fix tests and add config_flow import config test

* Added addition test to config_flow to test successfull camera entity create.

* Add to tests method to end in create entry, config_flow cleanup, use entry instead of entry.data

* Removed all services, sorted platforms in init file.

* Changed RTSP logging to debug from warning. (Forgot to change this before commit)

* Cleanup typing, change platform order, bump pyezviz version

* Added types to entries, allow creation of main entry if deleted by validating existance of type

* Config_flow doesn't store serial under entry data, camera rtsp read from entry and not stored in hass, removed duplicate abort if unique id from config flow

* Fix test of config_flow

* Update tests/components/ezviz/test_config_flow.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update tests/components/ezviz/test_config_flow.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update tests/components/ezviz/test_config_flow.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Bumped pyezviz api version, added api pyezvizerror exception raised in api (on HTTPError), cleanup unused imports.

* rebase

* cleanup coordinator, bump pyezviz api version, move async_setup_entry to add entry options to camera entries. (order change)

* Added discovery step in config_flow if cameras detected without rtsp config entry

* Reload main integration after addition or completion of camera rtsp config entry

* Add tests for discovery config_flow, added a few other output asserts

* Camera platform call discover flow with hass.async_create_task. Fixes to config_flow for discovery step

* Fix config_flow discovery, add check to legacy yaml camera platform import, move camera private method to camera import step

* Remove not needed check from config_flow import step.

* Cleanup config_flow

* Added config_flow description for discovered camera

* Reordered description in config_flow confim step.

* Added serial to flow_step description for discovered camera, readded camera attributes for rtsp stream url (allows user to check RTSP cred), added local ip and firmware upgade available.

* Bumped pyezviz version and changed region code to region url. (Russia uses a completly different url). PyEzviz adds a Local IP sensor, removed camera entity attributes.

* Add RSTP describe auth check from API to config_flow

* url as vol.in options in Config_flow

* Config_flow changes to discovery step, added exceptions, fixed tests, added rtsp config validate module mock to test disovery confirm step

* Add test for config_flow step user_camera

* Added tests for abort flow

* Extend tests on custom url flow step

* Fix exceptions in config_flow, fix test for discovery import exception test

* Bump pyezviz api version

* Bump api version, added config_flow function to wake hybernating camera before testing credentials, removed "user camera" step from config flow not needed as cameras are discovered.

* Create pyezviz Api instance for config_flow wake hybernating camera, fixed tests and added fixture to mock method

* Added alarm_control_panel with support to arm/disarm all cameras, fixed camera is available attribute (returns 2 if unavailable, 1 if available)

* Skip ignored entities when setup up camera RTSP stream

* Remove alarm_control_panel, add additional config_flow tests

* Cleanup tests, add tests for discovery_step.

* Add test for config_flow rtsp test step1 exceptions

* Removed redundant except from second step in test RTSP method

* All tests to CREATE or ABORT, added step exception for general HTTP error so user can retry in case of trasient network condition

* Ammended tests with output checks for step_id, error, data, create entry method calls.

* bumped ezviz api now rases library exceptions. Config_flow, coordiantor and init raises library exceptions. Updated test sideeffect for library exceptions

* Bump api version, Create mock ezviz cloud account on discovery tests first to allow more complete testing of step.

* Add abort to rtsp verification method if cloud account was deleted and add tests

* Update tests/components/ezviz/__init__.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/const.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update tests/components/ezviz/__init__.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Undo config import change to password key for yaml, move hass.data.setdefault to async_setup_entry and remove async_setup

* Fixed tests by removing _patch_async_setup as this was removed from init.

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Update homeassistant/components/ezviz/camera.py

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Changed L67 on camera config to complete suggestion for cleanup

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
RenierM26 2021-04-09 12:39:19 +02:00 committed by GitHub
parent e30cf88459
commit 155322584d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 1843 additions and 154 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.coveragerc
View File

@ -274,7 +274,13 @@ omit =
homeassistant/components/eufy/*
homeassistant/components/everlights/light.py
homeassistant/components/evohome/*
homeassistant/components/ezviz/*
homeassistant/components/ezviz/__init__.py
homeassistant/components/ezviz/camera.py
homeassistant/components/ezviz/coordinator.py
homeassistant/components/ezviz/const.py
homeassistant/components/ezviz/binary_sensor.py
homeassistant/components/ezviz/sensor.py
homeassistant/components/ezviz/switch.py
homeassistant/components/familyhub/camera.py
homeassistant/components/faa_delays/__init__.py
homeassistant/components/faa_delays/binary_sensor.py

2
CODEOWNERS
View File

@ -148,7 +148,7 @@ homeassistant/components/eq3btsmart/* @rytilahti
homeassistant/components/esphome/* @OttoWinter
homeassistant/components/essent/* @TheLastProject
homeassistant/components/evohome/* @zxdavb
homeassistant/components/ezviz/* @baqs
homeassistant/components/ezviz/* @RenierM26 @baqs
homeassistant/components/faa_delays/* @ntilley905
homeassistant/components/fastdotcom/* @rohankapoorcom
homeassistant/components/file/* @fabaff

130
homeassistant/components/ezviz/__init__.py
View File

@ -1 +1,129 @@
"""Support for Ezviz devices via Ezviz Cloud API."""
"""Support for Ezviz camera."""
import asyncio
from datetime import timedelta
import logging
from pyezviz.client import EzvizClient, HTTPError, InvalidURL, PyEzvizError
from homeassistant.const import (
CONF_PASSWORD,
CONF_TIMEOUT,
CONF_TYPE,
CONF_URL,
CONF_USERNAME,
)
from homeassistant.exceptions import ConfigEntryNotReady
from .const import (
ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD,
CONF_FFMPEG_ARGUMENTS,
DATA_COORDINATOR,
DATA_UNDO_UPDATE_LISTENER,
DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT,
DOMAIN,
)
from .coordinator import EzvizDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=30)
PLATFORMS = [
"binary_sensor",
"camera",
"sensor",
"switch",
]
async def async_setup_entry(hass, entry):
"""Set up Ezviz from a config entry."""
hass.data.setdefault(DOMAIN, {})
if not entry.options:
options = {
CONF_FFMPEG_ARGUMENTS: entry.data.get(
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
),
CONF_TIMEOUT: entry.data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
}
hass.config_entries.async_update_entry(entry, options=options)
if entry.data.get(CONF_TYPE) == ATTR_TYPE_CAMERA:
if hass.data.get(DOMAIN):
# Should only execute on addition of new camera entry.
# Fetch Entry id of main account and reload it.
for item in hass.config_entries.async_entries():
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
_LOGGER.info("Reload Ezviz integration with new camera rtsp entry")
await hass.config_entries.async_reload(item.entry_id)
return True
try:
ezviz_client = await hass.async_add_executor_job(
_get_ezviz_client_instance, entry
)
except (InvalidURL, HTTPError, PyEzvizError) as error:
_LOGGER.error("Unable to connect to Ezviz service: %s", str(error))
raise ConfigEntryNotReady from error
coordinator = EzvizDataUpdateCoordinator(hass, api=ezviz_client)
await coordinator.async_refresh()
if not coordinator.last_update_success:
raise ConfigEntryNotReady
undo_listener = entry.add_update_listener(_async_update_listener)
hass.data[DOMAIN][entry.entry_id] = {
DATA_COORDINATOR: coordinator,
DATA_UNDO_UPDATE_LISTENER: undo_listener,
}
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, entry):
"""Unload a config entry."""
if entry.data.get(CONF_TYPE) == ATTR_TYPE_CAMERA:
return True
unload_ok = all(
await asyncio.gather(
*[
hass.config_entries.async_forward_entry_unload(entry, component)
for component in PLATFORMS
]
)
)
if unload_ok:
hass.data[DOMAIN][entry.entry_id][DATA_UNDO_UPDATE_LISTENER]()
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
async def _async_update_listener(hass, entry):
"""Handle options update."""
await hass.config_entries.async_reload(entry.entry_id)
def _get_ezviz_client_instance(entry):
"""Initialize a new instance of EzvizClientApi."""
ezviz_client = EzvizClient(
entry.data[CONF_USERNAME],
entry.data[CONF_PASSWORD],
entry.data[CONF_URL],
entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
)
ezviz_client.login()
return ezviz_client

77
homeassistant/components/ezviz/binary_sensor.py Normal file
View File

@ -0,0 +1,77 @@
"""Support for Ezviz binary sensors."""
import logging
from pyezviz.constants import BinarySensorType
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DATA_COORDINATOR, DOMAIN, MANUFACTURER
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Ezviz sensors based on a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR]
sensors = []
sensor_type_name = "None"
for idx, camera in enumerate(coordinator.data):
for name in camera:
# Only add sensor with value.
if camera.get(name) is None:
continue
if name in BinarySensorType.__members__:
sensor_type_name = getattr(BinarySensorType, name).value
sensors.append(
EzvizBinarySensor(coordinator, idx, name, sensor_type_name)
)
async_add_entities(sensors)
class EzvizBinarySensor(CoordinatorEntity, BinarySensorEntity):
"""Representation of a Ezviz sensor."""
def __init__(self, coordinator, idx, name, sensor_type_name):
"""Initialize the sensor."""
super().__init__(coordinator)
self._idx = idx
self._camera_name = self.coordinator.data[self._idx]["name"]
self._name = name
self._sensor_name = f"{self._camera_name}.{self._name}"
self.sensor_type_name = sensor_type_name
self._serial = self.coordinator.data[self._idx]["serial"]
@property
def name(self):
"""Return the name of the Ezviz sensor."""
return self._sensor_name
@property
def is_on(self):
"""Return the state of the sensor."""
return self.coordinator.data[self._idx][self._name]
@property
def unique_id(self):
"""Return the unique ID of this sensor."""
return f"{self._serial}_{self._sensor_name}"
@property
def device_info(self):
"""Return the device_info of the device."""
return {
"identifiers": {(DOMAIN, self._serial)},
"name": self.coordinator.data[self._idx]["name"],
"model": self.coordinator.data[self._idx]["device_sub_category"],
"manufacturer": MANUFACTURER,
"sw_version": self.coordinator.data[self._idx]["version"],
}
@property
def device_class(self):
"""Device class for the sensor."""
return self.sensor_type_name

330
homeassistant/components/ezviz/camera.py
View File

@ -1,28 +1,30 @@
"""This component provides basic support for Ezviz IP cameras."""
"""Support ezviz camera devices."""
import asyncio
from datetime import timedelta
import logging
# pylint: disable=import-error
from haffmpeg.tools import IMAGE_JPEG, ImageFrame
from pyezviz.camera import EzvizCamera
from pyezviz.client import EzvizClient, PyEzvizError
import voluptuous as vol
from homeassistant.components.camera import PLATFORM_SCHEMA, SUPPORT_STREAM, Camera
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.components.ffmpeg import DATA_FFMPEG
from homeassistant.config_entries import SOURCE_DISCOVERY, SOURCE_IGNORE, SOURCE_IMPORT
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
_LOGGER = logging.getLogger(__name__)
CONF_CAMERAS = "cameras"
DEFAULT_CAMERA_USERNAME = "admin"
DEFAULT_RTSP_PORT = "554"
DATA_FFMPEG = "ffmpeg"
EZVIZ_DATA = "ezviz"
ENTITIES = "entities"
from .const import (
ATTR_SERIAL,
CONF_CAMERAS,
CONF_FFMPEG_ARGUMENTS,
DATA_COORDINATOR,
DEFAULT_CAMERA_USERNAME,
DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_RTSP_PORT,
DOMAIN,
MANUFACTURER,
)
CAMERA_SCHEMA = vol.Schema(
{vol.Required(CONF_USERNAME): cv.string, vol.Required(CONF_PASSWORD): cv.string}
@ -36,162 +38,162 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
}
)
_LOGGER = logging.getLogger(__name__)
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Ezviz IP Cameras."""
MIN_TIME_BETWEEN_SESSION_RENEW = timedelta(seconds=90)
conf_cameras = config[CONF_CAMERAS]
account = config[CONF_USERNAME]
password = config[CONF_PASSWORD]
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up a Ezviz IP Camera from platform config."""
_LOGGER.warning(
"Loading ezviz via platform config is deprecated, it will be automatically imported. Please remove it afterwards"
)
try:
ezviz_client = EzvizClient(account, password)
ezviz_client.login()
cameras = ezviz_client.load_cameras()
except PyEzvizError as exp:
_LOGGER.error(exp)
# Check if entry config exists and skips import if it does.
if hass.config_entries.async_entries(DOMAIN):
return
# now, let's build the HASS devices
# Check if importing camera account.
if CONF_CAMERAS in config:
cameras_conf = config[CONF_CAMERAS]
for serial, camera in cameras_conf.items():
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data={
ATTR_SERIAL: serial,
CONF_USERNAME: camera[CONF_USERNAME],
CONF_PASSWORD: camera[CONF_PASSWORD],
},
)
)
# Check if importing main ezviz cloud account.
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
data=config,
)
)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Ezviz cameras based on a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR]
camera_config_entries = hass.config_entries.async_entries(DOMAIN)
camera_entities = []
# Add the cameras as devices in HASS
for camera in cameras:
camera_username = DEFAULT_CAMERA_USERNAME
camera_password = ""
camera_rtsp_stream = ""
camera_serial = camera["serial"]
for idx, camera in enumerate(coordinator.data):
# There seem to be a bug related to localRtspPort in Ezviz API...
local_rtsp_port = DEFAULT_RTSP_PORT
if camera["local_rtsp_port"] and camera["local_rtsp_port"] != 0:
camera_rtsp_entry = [
item
for item in camera_config_entries
if item.unique_id == camera[ATTR_SERIAL]
]
if camera["local_rtsp_port"] != 0:
local_rtsp_port = camera["local_rtsp_port"]
if camera_serial in conf_cameras:
camera_username = conf_cameras[camera_serial][CONF_USERNAME]
camera_password = conf_cameras[camera_serial][CONF_PASSWORD]
camera_rtsp_stream = f"rtsp://{camera_username}:{camera_password}@{camera['local_ip']}:{local_rtsp_port}"
if camera_rtsp_entry:
conf_cameras = camera_rtsp_entry[0]
# Skip ignored entities.
if conf_cameras.source == SOURCE_IGNORE:
continue
ffmpeg_arguments = conf_cameras.options.get(
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
)
camera_username = conf_cameras.data[CONF_USERNAME]
camera_password = conf_cameras.data[CONF_PASSWORD]
camera_rtsp_stream = f"rtsp://{camera_username}:{camera_password}@{camera['local_ip']}:{local_rtsp_port}{ffmpeg_arguments}"
_LOGGER.debug(
"Camera %s source stream: %s", camera["serial"], camera_rtsp_stream
"Camera %s source stream: %s", camera[ATTR_SERIAL], camera_rtsp_stream
)
else:
_LOGGER.info(
"Found camera with serial %s without configuration. Add it to configuration.yaml to see the camera stream",
camera_serial,
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_DISCOVERY},
data={
ATTR_SERIAL: camera[ATTR_SERIAL],
CONF_IP_ADDRESS: camera["local_ip"],
},
)
)
camera["username"] = camera_username
camera["password"] = camera_password
camera["rtsp_stream"] = camera_rtsp_stream
camera_username = DEFAULT_CAMERA_USERNAME
camera_password = ""
camera_rtsp_stream = ""
ffmpeg_arguments = DEFAULT_FFMPEG_ARGUMENTS
_LOGGER.warning(
"Found camera with serial %s without configuration. Please go to integration to complete setup",
camera[ATTR_SERIAL],
)
camera["ezviz_camera"] = EzvizCamera(ezviz_client, camera_serial)
camera_entities.append(
EzvizCamera(
hass,
coordinator,
idx,
camera_username,
camera_password,
camera_rtsp_stream,
local_rtsp_port,
ffmpeg_arguments,
)
)
camera_entities.append(HassEzvizCamera(**camera))
add_entities(camera_entities)
async_add_entities(camera_entities)
class HassEzvizCamera(Camera):
"""An implementation of a Foscam IP camera."""
class EzvizCamera(CoordinatorEntity, Camera, RestoreEntity):
"""An implementation of a Ezviz security camera."""
def __init__(self, **data):
"""Initialize an Ezviz camera."""
super().__init__()
def __init__(
self,
hass,
coordinator,
idx,
camera_username,
camera_password,
camera_rtsp_stream,
local_rtsp_port,
ffmpeg_arguments,
):
"""Initialize a Ezviz security camera."""
super().__init__(coordinator)
Camera.__init__(self)
self._username = camera_username
self._password = camera_password
self._rtsp_stream = camera_rtsp_stream
self._idx = idx
self._ffmpeg = hass.data[DATA_FFMPEG]
self._local_rtsp_port = local_rtsp_port
self._ffmpeg_arguments = ffmpeg_arguments
self._username = data["username"]
self._password = data["password"]
self._rtsp_stream = data["rtsp_stream"]
self._ezviz_camera = data["ezviz_camera"]
self._serial = data["serial"]
self._name = data["name"]
self._status = data["status"]
self._privacy = data["privacy"]
self._audio = data["audio"]
self._ir_led = data["ir_led"]
self._state_led = data["state_led"]
self._follow_move = data["follow_move"]
self._alarm_notify = data["alarm_notify"]
self._alarm_sound_mod = data["alarm_sound_mod"]
self._encrypted = data["encrypted"]
self._local_ip = data["local_ip"]
self._detection_sensibility = data["detection_sensibility"]
self._device_sub_category = data["device_sub_category"]
self._local_rtsp_port = data["local_rtsp_port"]
self._ffmpeg = None
def update(self):
"""Update the camera states."""
data = self._ezviz_camera.status()
self._name = data["name"]
self._status = data["status"]
self._privacy = data["privacy"]
self._audio = data["audio"]
self._ir_led = data["ir_led"]
self._state_led = data["state_led"]
self._follow_move = data["follow_move"]
self._alarm_notify = data["alarm_notify"]
self._alarm_sound_mod = data["alarm_sound_mod"]
self._encrypted = data["encrypted"]
self._local_ip = data["local_ip"]
self._detection_sensibility = data["detection_sensibility"]
self._device_sub_category = data["device_sub_category"]
self._local_rtsp_port = data["local_rtsp_port"]
async def async_added_to_hass(self):
"""Subscribe to ffmpeg and add camera to list."""
self._ffmpeg = self.hass.data[DATA_FFMPEG]
@property
def should_poll(self) -> bool:
"""Return True if entity has to be polled for state.
False if entity pushes its state to HA.
"""
return True
@property
def extra_state_attributes(self):
"""Return the Ezviz-specific camera state attributes."""
return {
# if privacy == true, the device closed the lid or did a 180° tilt
"privacy": self._privacy,
# is the camera listening ?
"audio": self._audio,
# infrared led on ?
"ir_led": self._ir_led,
# state led on ?
"state_led": self._state_led,
# if true, the camera will move automatically to follow movements
"follow_move": self._follow_move,
# if true, if some movement is detected, the app is notified
"alarm_notify": self._alarm_notify,
# if true, if some movement is detected, the camera makes some sound
"alarm_sound_mod": self._alarm_sound_mod,
# are the camera's stored videos/images encrypted?
"encrypted": self._encrypted,
# camera's local ip on local network
"local_ip": self._local_ip,
# from 1 to 9, the higher is the sensibility, the more it will detect small movements
"detection_sensibility": self._detection_sensibility,
}
self._serial = self.coordinator.data[self._idx]["serial"]
self._name = self.coordinator.data[self._idx]["name"]
self._local_ip = self.coordinator.data[self._idx]["local_ip"]
@property
def available(self):
"""Return True if entity is available."""
return self._status
if self.coordinator.data[self._idx]["status"] == 2:
return False
@property
def brand(self):
"""Return the camera brand."""
return "Ezviz"
return True
@property
def supported_features(self):
@ -200,20 +202,40 @@ class HassEzvizCamera(Camera):
return SUPPORT_STREAM
return 0
@property
def name(self):
"""Return the name of this device."""
return self._name
@property
def model(self):
"""Return the camera model."""
return self._device_sub_category
"""Return the model of this device."""
return self.coordinator.data[self._idx]["device_sub_category"]
@property
def brand(self):
"""Return the manufacturer of this device."""
return MANUFACTURER
@property
def is_on(self):
"""Return true if on."""
return self._status
return bool(self.coordinator.data[self._idx]["status"])
@property
def name(self):
def is_recording(self):
"""Return true if the device is recording."""
return self.coordinator.data[self._idx]["alarm_notify"]
@property
def motion_detection_enabled(self):
"""Camera Motion Detection Status."""
return self.coordinator.data[self._idx]["alarm_notify"]
@property
def unique_id(self):
"""Return the name of this camera."""
return self._name
return self._serial
async def async_camera_image(self):
"""Return a frame from the camera stream."""
@ -224,12 +246,24 @@ class HassEzvizCamera(Camera):
)
return image
@property
def device_info(self):
"""Return the device_info of the device."""
return {
"identifiers": {(DOMAIN, self._serial)},
"name": self.coordinator.data[self._idx]["name"],
"model": self.coordinator.data[self._idx]["device_sub_category"],
"manufacturer": MANUFACTURER,
"sw_version": self.coordinator.data[self._idx]["version"],
}
async def stream_source(self):
"""Return the stream source."""
local_ip = self.coordinator.data[self._idx]["local_ip"]
if self._local_rtsp_port:
rtsp_stream_source = (
f"rtsp://{self._username}:{self._password}@"
f"{self._local_ip}:{self._local_rtsp_port}"
f"{local_ip}:{self._local_rtsp_port}{self._ffmpeg_arguments}"
)
_LOGGER.debug(
"Camera %s source stream: %s", self._serial, rtsp_stream_source

374
homeassistant/components/ezviz/config_flow.py Normal file
View File

@ -0,0 +1,374 @@
"""Config flow for ezviz."""
import logging
from pyezviz.client import EzvizClient, HTTPError, InvalidURL, PyEzvizError
from pyezviz.test_cam_rtsp import AuthTestResultFailed, InvalidHost, TestRTSPAuth
import voluptuous as vol
from homeassistant.config_entries import CONN_CLASS_CLOUD_POLL, ConfigFlow, OptionsFlow
from homeassistant.const import (
CONF_CUSTOMIZE,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_TIMEOUT,
CONF_TYPE,
CONF_URL,
CONF_USERNAME,
)
from homeassistant.core import callback
from .const import ( # pylint: disable=unused-import
ATTR_SERIAL,
ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD,
CONF_FFMPEG_ARGUMENTS,
DEFAULT_CAMERA_USERNAME,
DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT,
DOMAIN,
EU_URL,
RUSSIA_URL,
)
_LOGGER = logging.getLogger(__name__)
def _get_ezviz_client_instance(data):
"""Initialize a new instance of EzvizClientApi."""
ezviz_client = EzvizClient(
data[CONF_USERNAME],
data[CONF_PASSWORD],
data.get(CONF_URL, EU_URL),
data.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
)
ezviz_client.login()
return ezviz_client
def _test_camera_rtsp_creds(data):
"""Try DESCRIBE on RTSP camera with credentials."""
test_rtsp = TestRTSPAuth(
data[CONF_IP_ADDRESS], data[CONF_USERNAME], data[CONF_PASSWORD]
)
test_rtsp.main()
class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Ezviz."""
VERSION = 1
CONNECTION_CLASS = CONN_CLASS_CLOUD_POLL
async def _validate_and_create_auth(self, data):
"""Try to login to ezviz cloud account and create entry if successful."""
await self.async_set_unique_id(data[CONF_USERNAME])
self._abort_if_unique_id_configured()
# Verify cloud credentials by attempting a login request.
try:
await self.hass.async_add_executor_job(_get_ezviz_client_instance, data)
except InvalidURL as err:
raise InvalidURL from err
except HTTPError as err:
raise InvalidHost from err
except PyEzvizError as err:
raise PyEzvizError from err
auth_data = {
CONF_USERNAME: data[CONF_USERNAME],
CONF_PASSWORD: data[CONF_PASSWORD],
CONF_URL: data.get(CONF_URL, EU_URL),
CONF_TYPE: ATTR_TYPE_CLOUD,
}
return self.async_create_entry(title=data[CONF_USERNAME], data=auth_data)
async def _validate_and_create_camera_rtsp(self, data):
"""Try DESCRIBE on RTSP camera with credentials."""
# Get Ezviz cloud credentials from config entry
ezviz_client_creds = {
CONF_USERNAME: None,
CONF_PASSWORD: None,
CONF_URL: None,
}
for item in self._async_current_entries():
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
ezviz_client_creds = {
CONF_USERNAME: item.data.get(CONF_USERNAME),
CONF_PASSWORD: item.data.get(CONF_PASSWORD),
CONF_URL: item.data.get(CONF_URL),
}
# Abort flow if user removed cloud account before adding camera.
if ezviz_client_creds[CONF_USERNAME] is None:
return self.async_abort(reason="ezviz_cloud_account_missing")
# We need to wake hibernating cameras.
# First create EZVIZ API instance.
try:
ezviz_client = await self.hass.async_add_executor_job(
_get_ezviz_client_instance, ezviz_client_creds
)
except InvalidURL as err:
raise InvalidURL from err
except HTTPError as err:
raise InvalidHost from err
except PyEzvizError as err:
raise PyEzvizError from err
# Secondly try to wake hybernating camera.
try:
await self.hass.async_add_executor_job(
ezviz_client.get_detection_sensibility, data[ATTR_SERIAL]
)
except HTTPError as err:
raise InvalidHost from err
# Thirdly attempts an authenticated RTSP DESCRIBE request.
try:
await self.hass.async_add_executor_job(_test_camera_rtsp_creds, data)
except InvalidHost as err:
raise InvalidHost from err
except AuthTestResultFailed as err:
raise AuthTestResultFailed from err
return self.async_create_entry(
title=data[ATTR_SERIAL],
data={
CONF_USERNAME: data[CONF_USERNAME],
CONF_PASSWORD: data[CONF_PASSWORD],
CONF_TYPE: ATTR_TYPE_CAMERA,
},
)
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""Get the options flow for this handler."""
return EzvizOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None):
"""Handle a flow initiated by the user."""
# Check if ezviz cloud account is present in entry config,
# abort if already configured.
for item in self._async_current_entries():
if item.data.get(CONF_TYPE) == ATTR_TYPE_CLOUD:
return self.async_abort(reason="already_configured_account")
errors = {}
if user_input is not None:
if user_input[CONF_URL] == CONF_CUSTOMIZE:
self.context["data"] = {
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
}
return await self.async_step_user_custom_url()
if CONF_TIMEOUT not in user_input:
user_input[CONF_TIMEOUT] = DEFAULT_TIMEOUT
try:
return await self._validate_and_create_auth(user_input)
except InvalidURL:
errors["base"] = "invalid_host"
except InvalidHost:
errors["base"] = "cannot_connect"
except PyEzvizError:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
data_schema = vol.Schema(
{
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Required(CONF_URL, default=EU_URL): vol.In(
[EU_URL, RUSSIA_URL, CONF_CUSTOMIZE]
),
}
)
return self.async_show_form(
step_id="user", data_schema=data_schema, errors=errors
)
async def async_step_user_custom_url(self, user_input=None):
"""Handle a flow initiated by the user for custom region url."""
errors = {}
if user_input is not None:
user_input[CONF_USERNAME] = self.context["data"][CONF_USERNAME]
user_input[CONF_PASSWORD] = self.context["data"][CONF_PASSWORD]
if CONF_TIMEOUT not in user_input:
user_input[CONF_TIMEOUT] = DEFAULT_TIMEOUT
try:
return await self._validate_and_create_auth(user_input)
except InvalidURL:
errors["base"] = "invalid_host"
except InvalidHost:
errors["base"] = "cannot_connect"
except PyEzvizError:
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
data_schema_custom_url = vol.Schema(
{
vol.Required(CONF_URL, default=EU_URL): str,
}
)
return self.async_show_form(
step_id="user_custom_url", data_schema=data_schema_custom_url, errors=errors
)
async def async_step_discovery(self, discovery_info):
"""Handle a flow for discovered camera without rtsp config entry."""
await self.async_set_unique_id(discovery_info[ATTR_SERIAL])
self._abort_if_unique_id_configured()
self.context["title_placeholders"] = {"serial": self.unique_id}
self.context["data"] = {CONF_IP_ADDRESS: discovery_info[CONF_IP_ADDRESS]}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None):
"""Confirm and create entry from discovery step."""
errors = {}
if user_input is not None:
user_input[ATTR_SERIAL] = self.unique_id
user_input[CONF_IP_ADDRESS] = self.context["data"][CONF_IP_ADDRESS]
try:
return await self._validate_and_create_camera_rtsp(user_input)
except (InvalidHost, InvalidURL):
errors["base"] = "invalid_host"
except (PyEzvizError, AuthTestResultFailed):
errors["base"] = "invalid_auth"
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Unexpected exception")
return self.async_abort(reason="unknown")
discovered_camera_schema = vol.Schema(
{
vol.Required(CONF_USERNAME, default=DEFAULT_CAMERA_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
}
)
return self.async_show_form(
step_id="confirm",
data_schema=discovered_camera_schema,
errors=errors,
description_placeholders={
"serial": self.unique_id,
CONF_IP_ADDRESS: self.context["data"][CONF_IP_ADDRESS],
},
)
async def async_step_import(self, import_config):
"""Handle config import from yaml."""
_LOGGER.debug("import config: %s", import_config)
# Check importing camera.
if ATTR_SERIAL in import_config:
return await self.async_step_import_camera(import_config)
# Validate and setup of main ezviz cloud account.
try:
return await self._validate_and_create_auth(import_config)
except InvalidURL:
_LOGGER.error("Error importing Ezviz platform config: invalid host")
return self.async_abort(reason="invalid_host")
except InvalidHost:
_LOGGER.error("Error importing Ezviz platform config: cannot connect")
return self.async_abort(reason="cannot_connect")
except (AuthTestResultFailed, PyEzvizError):
_LOGGER.error("Error importing Ezviz platform config: invalid auth")
return self.async_abort(reason="invalid_auth")
except Exception: # pylint: disable=broad-except
_LOGGER.exception(
"Error importing ezviz platform config: unexpected exception"
)
return self.async_abort(reason="unknown")
async def async_step_import_camera(self, data):
"""Create RTSP auth entry per camera in config."""
await self.async_set_unique_id(data[ATTR_SERIAL])
self._abort_if_unique_id_configured()
_LOGGER.debug("Create camera with: %s", data)
cam_serial = data.pop(ATTR_SERIAL)
data[CONF_TYPE] = ATTR_TYPE_CAMERA
return self.async_create_entry(title=cam_serial, data=data)
class EzvizOptionsFlowHandler(OptionsFlow):
"""Handle Ezviz client options."""
def __init__(self, config_entry):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input=None):
"""Manage Ezviz options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
options = {
vol.Optional(
CONF_TIMEOUT,
default=self.config_entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
): int,
vol.Optional(
CONF_FFMPEG_ARGUMENTS,
default=self.config_entry.options.get(
CONF_FFMPEG_ARGUMENTS, DEFAULT_FFMPEG_ARGUMENTS
),
): str,
}
return self.async_show_form(step_id="init", data_schema=vol.Schema(options))

42
homeassistant/components/ezviz/const.py Normal file
View File

@ -0,0 +1,42 @@
"""Constants for the ezviz integration."""
DOMAIN = "ezviz"
MANUFACTURER = "Ezviz"
# Configuration
ATTR_SERIAL = "serial"
CONF_CAMERAS = "cameras"
ATTR_SWITCH = "switch"
ATTR_ENABLE = "enable"
ATTR_DIRECTION = "direction"
ATTR_SPEED = "speed"
ATTR_LEVEL = "level"
ATTR_TYPE = "type_value"
DIR_UP = "up"
DIR_DOWN = "down"
DIR_LEFT = "left"
DIR_RIGHT = "right"
ATTR_LIGHT = "LIGHT"
ATTR_SOUND = "SOUND"
ATTR_INFRARED_LIGHT = "INFRARED_LIGHT"
ATTR_PRIVACY = "PRIVACY"
ATTR_SLEEP = "SLEEP"
ATTR_MOBILE_TRACKING = "MOBILE_TRACKING"
ATTR_TRACKING = "TRACKING"
CONF_FFMPEG_ARGUMENTS = "ffmpeg_arguments"
ATTR_HOME = "HOME_MODE"
ATTR_AWAY = "AWAY_MODE"
ATTR_TYPE_CLOUD = "EZVIZ_CLOUD_ACCOUNT"
ATTR_TYPE_CAMERA = "CAMERA_ACCOUNT"
# Defaults
EU_URL = "apiieu.ezvizlife.com"
RUSSIA_URL = "apirus.ezvizru.com"
DEFAULT_CAMERA_USERNAME = "admin"
DEFAULT_RTSP_PORT = "554"
DEFAULT_TIMEOUT = 25
DEFAULT_FFMPEG_ARGUMENTS = ""
# Data
DATA_COORDINATOR = "coordinator"
DATA_UNDO_UPDATE_LISTENER = "undo_update_listener"

38
homeassistant/components/ezviz/coordinator.py Normal file
View File

@ -0,0 +1,38 @@
"""Provides the ezviz DataUpdateCoordinator."""
from datetime import timedelta
import logging
from async_timeout import timeout
from pyezviz.client import HTTPError, InvalidURL, PyEzvizError
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
class EzvizDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching Ezviz data."""
def __init__(self, hass, *, api):
"""Initialize global Ezviz data updater."""
self.ezviz_client = api
update_interval = timedelta(seconds=30)
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
def _update_data(self):
"""Fetch data from Ezviz via camera load function."""
cameras = self.ezviz_client.load_cameras()
return cameras
async def _async_update_data(self):
"""Fetch data from Ezviz."""
try:
async with timeout(35):
return await self.hass.async_add_executor_job(self._update_data)
except (InvalidURL, HTTPError, PyEzvizError) as error:
raise UpdateFailed(f"Invalid response from API: {error}") from error

7
homeassistant/components/ezviz/manifest.json
View File

@ -1,8 +1,9 @@
{
"disabled": "Dependency contains code that breaks Home Assistant.",
"domain": "ezviz",
"name": "Ezviz",
"documentation": "https://www.home-assistant.io/integrations/ezviz",
"codeowners": ["@baqs"],
"requirements": ["pyezviz==0.1.5"]
"dependencies": ["ffmpeg"],
"codeowners": ["@RenierM26", "@baqs"],
"requirements": ["pyezviz==0.1.8.7"],
"config_flow": true
}

75
homeassistant/components/ezviz/sensor.py Normal file
View File

@ -0,0 +1,75 @@
"""Support for Ezviz sensors."""
import logging
from pyezviz.constants import SensorType
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DATA_COORDINATOR, DOMAIN, MANUFACTURER
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Ezviz sensors based on a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR]
sensors = []
sensor_type_name = "None"
for idx, camera in enumerate(coordinator.data):
for name in camera:
# Only add sensor with value.
if camera.get(name) is None:
continue
if name in SensorType.__members__:
sensor_type_name = getattr(SensorType, name).value
sensors.append(EzvizSensor(coordinator, idx, name, sensor_type_name))
async_add_entities(sensors)
class EzvizSensor(CoordinatorEntity, Entity):
"""Representation of a Ezviz sensor."""
def __init__(self, coordinator, idx, name, sensor_type_name):
"""Initialize the sensor."""
super().__init__(coordinator)
self._idx = idx
self._camera_name = self.coordinator.data[self._idx]["name"]
self._name = name
self._sensor_name = f"{self._camera_name}.{self._name}"
self.sensor_type_name = sensor_type_name
self._serial = self.coordinator.data[self._idx]["serial"]
@property
def name(self):
"""Return the name of the Ezviz sensor."""
return self._sensor_name
@property
def state(self):
"""Return the state of the sensor."""
return self.coordinator.data[self._idx][self._name]
@property
def unique_id(self):
"""Return the unique ID of this sensor."""
return f"{self._serial}_{self._sensor_name}"
@property
def device_info(self):
"""Return the device_info of the device."""
return {
"identifiers": {(DOMAIN, self._serial)},
"name": self.coordinator.data[self._idx]["name"],
"model": self.coordinator.data[self._idx]["device_sub_category"],
"manufacturer": MANUFACTURER,
"sw_version": self.coordinator.data[self._idx]["version"],
}
@property
def device_class(self):
"""Device class for the sensor."""
return self.sensor_type_name

52
homeassistant/components/ezviz/strings.json Normal file
View File

@ -0,0 +1,52 @@
{
"config": {
"flow_title": "{serial}",
"step": {
"user": {
"title": "Connect to Ezviz Cloud",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"url": "[%key:common::config_flow::data::url%]"
}
},
"user_custom_url": {
"title": "Connect to custom Ezviz URL",
"description": "Manually specify your region URL",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"url": "[%key:common::config_flow::data::url%]"
}
},
"confirm": {
"title": "Discovered Ezviz Camera",
"description": "Enter RTSP credentials for Ezviz camera {serial} with IP {ip_address}",
"data": {
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_host": "[%key:common::config_flow::error::invalid_host%]"
},
"abort": {
"already_configured_account": "[%key:common::config_flow::abort::already_configured_account%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"ezviz_cloud_account_missing": "Ezviz cloud account missing. Please reconfigure Ezviz cloud account"
}
},
"options": {
"step": {
"init": {
"data": {
"timeout": "Request Timeout (seconds)",
"ffmpeg_arguments": "Arguments passed to ffmpeg for cameras"
}
}
}
}
}

90
homeassistant/components/ezviz/switch.py Normal file
View File

@ -0,0 +1,90 @@
"""Support for Ezviz Switch sensors."""
import logging
from pyezviz.constants import DeviceSwitchType
from homeassistant.components.switch import DEVICE_CLASS_SWITCH, SwitchEntity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DATA_COORDINATOR, DOMAIN, MANUFACTURER
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass, entry, async_add_entities):
"""Set up Ezviz switch based on a config entry."""
coordinator = hass.data[DOMAIN][entry.entry_id][DATA_COORDINATOR]
switch_entities = []
supported_switches = []
for switches in DeviceSwitchType:
supported_switches.append(switches.value)
supported_switches = set(supported_switches)
for idx, camera in enumerate(coordinator.data):
if not camera.get("switches"):
continue
for switch in camera["switches"]:
if switch not in supported_switches:
continue
switch_entities.append(EzvizSwitch(coordinator, idx, switch))
async_add_entities(switch_entities)
class EzvizSwitch(CoordinatorEntity, SwitchEntity):
"""Representation of a Ezviz sensor."""
def __init__(self, coordinator, idx, switch):
"""Initialize the switch."""
super().__init__(coordinator)
self._idx = idx
self._camera_name = self.coordinator.data[self._idx]["name"]
self._name = switch
self._sensor_name = f"{self._camera_name}.{DeviceSwitchType(self._name).name}"
self._serial = self.coordinator.data[self._idx]["serial"]
self._device_class = DEVICE_CLASS_SWITCH
@property
def name(self):
"""Return the name of the Ezviz switch."""
return f"{self._camera_name}.{DeviceSwitchType(self._name).name}"
@property
def is_on(self):
"""Return the state of the switch."""
return self.coordinator.data[self._idx]["switches"][self._name]
@property
def unique_id(self):
"""Return the unique ID of this switch."""
return f"{self._serial}_{self._sensor_name}"
def turn_on(self, **kwargs):
"""Change a device switch on the camera."""
_LOGGER.debug("Set EZVIZ Switch '%s' to on", self._name)
self.coordinator.ezviz_client.switch_status(self._serial, self._name, 1)
def turn_off(self, **kwargs):
"""Change a device switch on the camera."""
_LOGGER.debug("Set EZVIZ Switch '%s' to off", self._name)
self.coordinator.ezviz_client.switch_status(self._serial, self._name, 0)
@property
def device_info(self):
"""Return the device_info of the device."""
return {
"identifiers": {(DOMAIN, self._serial)},
"name": self.coordinator.data[self._idx]["name"],
"model": self.coordinator.data[self._idx]["device_sub_category"],
"manufacturer": MANUFACTURER,
"sw_version": self.coordinator.data[self._idx]["version"],
}
@property
def device_class(self):
"""Device class for the sensor."""
return self._device_class

52
homeassistant/components/ezviz/translations/en.json Normal file
View File

@ -0,0 +1,52 @@
{
"config": {
"abort": {
"already_configured_account": "Account is already configured.",
"unknown": "Unexpected error",
"ezviz_cloud_account_missing": "Ezviz cloud account missing. Please reconfigure Ezviz cloud account"
},
"error": {
"cannot_connect": "Failed to connect",
"invalid_auth": "Invalid authentication",
"invalid_host": "Invalid IP or URL"
},
"flow_title": "{serial}",
"step": {
"user": {
"data": {
"username": "Username",
"password": "Password",
"url": "URL"
},
"title": "Connect to Ezviz Cloud"
},
"user_custom_url": {
"data": {
"username": "Username",
"password": "Password",
"url": "URL"
},
"title": "Connect to custom Ezviz URL",
"description": "Manually specify your region URL"
},
"confirm": {
"data": {
"username": "Username",
"password": "Password"
},
"title": "Discovered Ezviz Camera",
"description": "Enter RTSP credentials for Ezviz camera {serial} with IP as {ip_address}"
}
}
},
"options": {
"step": {
"init": {
"data": {
"timeout": "Request Timeout (seconds)",
"ffmpeg_arguments": "Arguments passed to ffmpeg for cameras"
}
}
}
}
}

1
homeassistant/generated/config_flows.py
View File

@ -66,6 +66,7 @@ FLOWS = [
"enphase_envoy",
"epson",
"esphome",
"ezviz",
"faa_delays",
"fireservicerota",
"flick_electric",

3
requirements_all.txt
View File

@ -1384,6 +1384,9 @@ pyephember==0.3.1
# homeassistant.components.everlights
pyeverlights==0.1.0
# homeassistant.components.ezviz
pyezviz==0.1.8.7
# homeassistant.components.fido
pyfido==2.1.1

3
requirements_test_all.txt
View File

@ -737,6 +737,9 @@ pyeconet==0.1.13
# homeassistant.components.everlights
pyeverlights==0.1.0
# homeassistant.components.ezviz
pyezviz==0.1.8.7
# homeassistant.components.fido
pyfido==2.1.1

118
tests/components/ezviz/__init__.py Normal file
View File

@ -0,0 +1,118 @@
"""Tests for the Ezviz integration."""
from unittest.mock import patch
from homeassistant.components.ezviz.const import (
ATTR_SERIAL,
ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD,
CONF_CAMERAS,
CONF_FFMPEG_ARGUMENTS,
DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT,
DOMAIN,
)
from homeassistant.const import (
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_TIMEOUT,
CONF_TYPE,
CONF_URL,
CONF_USERNAME,
)
from homeassistant.helpers.typing import HomeAssistantType
from tests.common import MockConfigEntry
ENTRY_CONFIG = {
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_URL: "apiieu.ezvizlife.com",
CONF_TYPE: ATTR_TYPE_CLOUD,
}
ENTRY_OPTIONS = {
CONF_FFMPEG_ARGUMENTS: DEFAULT_FFMPEG_ARGUMENTS,
CONF_TIMEOUT: DEFAULT_TIMEOUT,
}
USER_INPUT_VALIDATE = {
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_URL: "apiieu.ezvizlife.com",
}
USER_INPUT = {
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_URL: "apiieu.ezvizlife.com",
CONF_TYPE: ATTR_TYPE_CLOUD,
}
USER_INPUT_CAMERA_VALIDATE = {
ATTR_SERIAL: "C666666",
CONF_PASSWORD: "test-password",
CONF_USERNAME: "test-username",
}
USER_INPUT_CAMERA = {
CONF_PASSWORD: "test-password",
CONF_USERNAME: "test-username",
CONF_TYPE: ATTR_TYPE_CAMERA,
}
YAML_CONFIG = {
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
CONF_URL: "apiieu.ezvizlife.com",
CONF_CAMERAS: {
"C666666": {CONF_USERNAME: "test-username", CONF_PASSWORD: "test-password"}
},
}
YAML_INVALID = {
"C666666": {CONF_USERNAME: "test-username", CONF_PASSWORD: "test-password"}
}
YAML_CONFIG_CAMERA = {
ATTR_SERIAL: "C666666",
CONF_USERNAME: "test-username",
CONF_PASSWORD: "test-password",
}
DISCOVERY_INFO = {
ATTR_SERIAL: "C666666",
CONF_USERNAME: None,
CONF_PASSWORD: None,
CONF_IP_ADDRESS: "127.0.0.1",
}
TEST = {
CONF_USERNAME: None,
CONF_PASSWORD: None,
CONF_IP_ADDRESS: "127.0.0.1",
}
def _patch_async_setup_entry(return_value=True):
return patch(
"homeassistant.components.ezviz.async_setup_entry",
return_value=return_value,
)
async def init_integration(
hass: HomeAssistantType,
*,
data: dict = ENTRY_CONFIG,
options: dict = ENTRY_OPTIONS,
skip_entry_setup: bool = False,
) -> MockConfigEntry:
"""Set up the Ezviz integration in Home Assistant."""
entry = MockConfigEntry(domain=DOMAIN, data=data, options=options)
entry.add_to_hass(hass)
if not skip_entry_setup:
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

48
tests/components/ezviz/conftest.py Normal file
View File

@ -0,0 +1,48 @@
"""Define fixtures available for all tests."""
from unittest.mock import MagicMock, patch
from pyezviz import EzvizClient
from pyezviz.test_cam_rtsp import TestRTSPAuth
from pytest import fixture
@fixture(autouse=True)
def mock_ffmpeg(hass):
"""Mock ffmpeg is loaded."""
hass.config.components.add("ffmpeg")
@fixture
def ezviz_test_rtsp_config_flow(hass):
"""Mock the EzvizApi for easier testing."""
with patch.object(TestRTSPAuth, "main", return_value=True), patch(
"homeassistant.components.ezviz.config_flow.TestRTSPAuth"
) as mock_ezviz_test_rtsp:
instance = mock_ezviz_test_rtsp.return_value = TestRTSPAuth(
"test-ip",
"test-username",
"test-password",
)
instance.main = MagicMock(return_value=True)
yield mock_ezviz_test_rtsp
@fixture
def ezviz_config_flow(hass):
"""Mock the EzvizAPI for easier config flow testing."""
with patch.object(EzvizClient, "login", return_value=True), patch(
"homeassistant.components.ezviz.config_flow.EzvizClient"
) as mock_ezviz:
instance = mock_ezviz.return_value = EzvizClient(
"test-username",
"test-password",
"local.host",
"1",
)
instance.login = MagicMock(return_value=True)
instance.get_detection_sensibility = MagicMock(return_value=True)
yield mock_ezviz

547
tests/components/ezviz/test_config_flow.py Normal file
View File

@ -0,0 +1,547 @@
"""Test the Ezviz config flow."""
from unittest.mock import patch
from pyezviz.client import HTTPError, InvalidURL, PyEzvizError
from pyezviz.test_cam_rtsp import AuthTestResultFailed, InvalidHost
from homeassistant.components.ezviz.const import (
ATTR_SERIAL,
ATTR_TYPE_CAMERA,
ATTR_TYPE_CLOUD,
CONF_FFMPEG_ARGUMENTS,
DEFAULT_FFMPEG_ARGUMENTS,
DEFAULT_TIMEOUT,
DOMAIN,
)
from homeassistant.config_entries import SOURCE_DISCOVERY, SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import (
CONF_CUSTOMIZE,
CONF_IP_ADDRESS,
CONF_PASSWORD,
CONF_TIMEOUT,
CONF_TYPE,
CONF_URL,
CONF_USERNAME,
)
from homeassistant.data_entry_flow import (
RESULT_TYPE_ABORT,
RESULT_TYPE_CREATE_ENTRY,
RESULT_TYPE_FORM,
)
from homeassistant.setup import async_setup_component
from . import (
DISCOVERY_INFO,
USER_INPUT,
USER_INPUT_CAMERA,
USER_INPUT_CAMERA_VALIDATE,
USER_INPUT_VALIDATE,
YAML_CONFIG,
YAML_CONFIG_CAMERA,
YAML_INVALID,
_patch_async_setup_entry,
init_integration,
)
async def test_user_form(hass, ezviz_config_flow):
"""Test the user initiated form."""
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {}
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "test-username"
assert result["data"] == {**USER_INPUT}
assert len(mock_setup_entry.mock_calls) == 1
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "already_configured_account"
async def test_user_custom_url(hass, ezviz_config_flow):
"""Test custom url step."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_USERNAME: "test-user", CONF_PASSWORD: "test-pass", CONF_URL: "customize"},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user_custom_url"
assert result["errors"] == {}
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_URL: "test-user"},
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {
CONF_PASSWORD: "test-pass",
CONF_TYPE: ATTR_TYPE_CLOUD,
CONF_URL: "test-user",
CONF_USERNAME: "test-user",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_async_step_import(hass, ezviz_config_flow):
"""Test the config import flow."""
await async_setup_component(hass, "persistent_notification", {})
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == USER_INPUT
assert len(mock_setup_entry.mock_calls) == 1
async def test_async_step_import_camera(hass, ezviz_config_flow):
"""Test the config import camera flow."""
await async_setup_component(hass, "persistent_notification", {})
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG_CAMERA
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == USER_INPUT_CAMERA
assert len(mock_setup_entry.mock_calls) == 1
async def test_async_step_import_2nd_form_returns_camera(hass, ezviz_config_flow):
"""Test we get the user initiated form."""
await async_setup_component(hass, "persistent_notification", {})
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG
)
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == USER_INPUT
assert len(mock_setup_entry.mock_calls) == 1
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=USER_INPUT_CAMERA_VALIDATE
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == USER_INPUT_CAMERA
assert len(mock_setup_entry.mock_calls) == 1
async def test_async_step_import_abort(hass, ezviz_config_flow):
"""Test the config import flow with invalid data."""
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_INVALID
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_step_discovery_abort_if_cloud_account_missing(hass):
"""Test discovery and confirm step, abort if cloud account was removed."""
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DISCOVERY}, data=DISCOVERY_INFO
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "ezviz_cloud_account_missing"
async def test_async_step_discovery(
hass, ezviz_config_flow, ezviz_test_rtsp_config_flow
):
"""Test discovery and confirm step."""
with patch("homeassistant.components.ezviz.PLATFORMS", []):
await init_integration(hass)
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_DISCOVERY}, data=DISCOVERY_INFO
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {}
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"] == {
CONF_PASSWORD: "test-pass",
CONF_TYPE: ATTR_TYPE_CAMERA,
CONF_USERNAME: "test-user",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_options_flow(hass):
"""Test updating options."""
with patch("homeassistant.components.ezviz.PLATFORMS", []):
entry = await init_integration(hass)
assert entry.options[CONF_FFMPEG_ARGUMENTS] == DEFAULT_FFMPEG_ARGUMENTS
assert entry.options[CONF_TIMEOUT] == DEFAULT_TIMEOUT
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "init"
assert result["errors"] is None
with _patch_async_setup_entry() as mock_setup_entry:
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_FFMPEG_ARGUMENTS: "/H.264", CONF_TIMEOUT: 25},
)
await hass.async_block_till_done()
assert result["type"] == RESULT_TYPE_CREATE_ENTRY
assert result["data"][CONF_FFMPEG_ARGUMENTS] == "/H.264"
assert result["data"][CONF_TIMEOUT] == 25
assert len(mock_setup_entry.mock_calls) == 0
async def test_user_form_exception(hass, ezviz_config_flow):
"""Test we handle exception on user form."""
ezviz_config_flow.side_effect = PyEzvizError
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "invalid_auth"}
ezviz_config_flow.side_effect = InvalidURL
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "invalid_host"}
ezviz_config_flow.side_effect = HTTPError
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"] == {"base": "cannot_connect"}
ezviz_config_flow.side_effect = Exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
USER_INPUT_VALIDATE,
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_import_exception(hass, ezviz_config_flow):
"""Test we handle unexpected exception on import."""
ezviz_config_flow.side_effect = PyEzvizError
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "invalid_auth"
ezviz_config_flow.side_effect = InvalidURL
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "invalid_host"
ezviz_config_flow.side_effect = HTTPError
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"
ezviz_config_flow.side_effect = Exception
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=YAML_CONFIG
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_discover_exception_step1(
hass,
ezviz_config_flow,
):
"""Test we handle unexpected exception on discovery."""
with patch("homeassistant.components.ezviz.PLATFORMS", []):
await init_integration(hass)
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_DISCOVERY},
data={ATTR_SERIAL: "C66666", CONF_IP_ADDRESS: "test-ip"},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {}
# Test Step 1
ezviz_config_flow.side_effect = PyEzvizError
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {"base": "invalid_auth"}
ezviz_config_flow.side_effect = InvalidURL
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {"base": "invalid_host"}
ezviz_config_flow.side_effect = HTTPError
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {"base": "invalid_host"}
ezviz_config_flow.side_effect = Exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_discover_exception_step3(
hass,
ezviz_config_flow,
ezviz_test_rtsp_config_flow,
):
"""Test we handle unexpected exception on discovery."""
with patch("homeassistant.components.ezviz.PLATFORMS", []):
await init_integration(hass)
await async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_DISCOVERY},
data={ATTR_SERIAL: "C66666", CONF_IP_ADDRESS: "test-ip"},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {}
# Test Step 3
ezviz_test_rtsp_config_flow.side_effect = AuthTestResultFailed
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {"base": "invalid_auth"}
ezviz_test_rtsp_config_flow.side_effect = InvalidHost
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "confirm"
assert result["errors"] == {"base": "invalid_host"}
ezviz_test_rtsp_config_flow.side_effect = Exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"
async def test_user_custom_url_exception(hass, ezviz_config_flow):
"""Test we handle unexpected exception."""
ezviz_config_flow.side_effect = PyEzvizError()
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_USERNAME: "test-user",
CONF_PASSWORD: "test-pass",
CONF_URL: CONF_CUSTOMIZE,
},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user_custom_url"
assert result["errors"] == {}
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_URL: "test-user"},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user_custom_url"
assert result["errors"] == {"base": "invalid_auth"}
ezviz_config_flow.side_effect = InvalidURL
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_URL: "test-user"},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user_custom_url"
assert result["errors"] == {"base": "invalid_host"}
ezviz_config_flow.side_effect = HTTPError
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_URL: "test-user"},
)
assert result["type"] == RESULT_TYPE_FORM
assert result["step_id"] == "user_custom_url"
assert result["errors"] == {"base": "cannot_connect"}
ezviz_config_flow.side_effect = Exception
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{CONF_URL: "test-user"},
)
assert result["type"] == RESULT_TYPE_ABORT
assert result["reason"] == "unknown"