mirror of
https://github.com/home-assistant/core.git
synced 2025-07-16 09:47:13 +00:00
Use entity description for Ezviz sensors (#56634)
This commit is contained in:
parent
b155d2bbe5
commit
48c2cfa6f8
@ -301,6 +301,7 @@ omit =
|
|||||||
homeassistant/components/ezviz/camera.py
|
homeassistant/components/ezviz/camera.py
|
||||||
homeassistant/components/ezviz/coordinator.py
|
homeassistant/components/ezviz/coordinator.py
|
||||||
homeassistant/components/ezviz/const.py
|
homeassistant/components/ezviz/const.py
|
||||||
|
homeassistant/components/ezviz/entity.py
|
||||||
homeassistant/components/ezviz/binary_sensor.py
|
homeassistant/components/ezviz/binary_sensor.py
|
||||||
homeassistant/components/ezviz/sensor.py
|
homeassistant/components/ezviz/sensor.py
|
||||||
homeassistant/components/ezviz/switch.py
|
homeassistant/components/ezviz/switch.py
|
||||||
|
@ -1,19 +1,36 @@
|
|||||||
"""Support for Ezviz binary sensors."""
|
"""Support for Ezviz binary sensors."""
|
||||||
import logging
|
from __future__ import annotations
|
||||||
|
|
||||||
from pyezviz.constants import BinarySensorType
|
from homeassistant.components.binary_sensor import (
|
||||||
|
DEVICE_CLASS_MOTION,
|
||||||
from homeassistant.components.binary_sensor import BinarySensorEntity
|
DEVICE_CLASS_UPDATE,
|
||||||
|
BinarySensorEntity,
|
||||||
|
BinarySensorEntityDescription,
|
||||||
|
)
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.entity import DeviceInfo
|
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
||||||
|
|
||||||
from .const import DATA_COORDINATOR, DOMAIN, MANUFACTURER
|
from .const import DATA_COORDINATOR, DOMAIN
|
||||||
from .coordinator import EzvizDataUpdateCoordinator
|
from .coordinator import EzvizDataUpdateCoordinator
|
||||||
|
from .entity import EzvizEntity
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
PARALLEL_UPDATES = 1
|
||||||
|
|
||||||
|
BINARY_SENSOR_TYPES: dict[str, BinarySensorEntityDescription] = {
|
||||||
|
"Motion_Trigger": BinarySensorEntityDescription(
|
||||||
|
key="Motion_Trigger",
|
||||||
|
device_class=DEVICE_CLASS_MOTION,
|
||||||
|
),
|
||||||
|
"alarm_schedules_enabled": BinarySensorEntityDescription(
|
||||||
|
key="alarm_schedules_enabled"
|
||||||
|
),
|
||||||
|
"encrypted": BinarySensorEntityDescription(key="encrypted"),
|
||||||
|
"upgrade_available": BinarySensorEntityDescription(
|
||||||
|
key="upgrade_available",
|
||||||
|
device_class=DEVICE_CLASS_UPDATE,
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(
|
async def async_setup_entry(
|
||||||
@ -23,24 +40,19 @@ async def async_setup_entry(
|
|||||||
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
||||||
DATA_COORDINATOR
|
DATA_COORDINATOR
|
||||||
]
|
]
|
||||||
sensors = []
|
|
||||||
|
|
||||||
for idx, camera in enumerate(coordinator.data):
|
async_add_entities(
|
||||||
for name in camera:
|
[
|
||||||
# Only add sensor with value.
|
EzvizBinarySensor(coordinator, camera, binary_sensor)
|
||||||
if camera.get(name) is None:
|
for camera in coordinator.data
|
||||||
continue
|
for binary_sensor, value in coordinator.data[camera].items()
|
||||||
|
if binary_sensor in BINARY_SENSOR_TYPES
|
||||||
if name in BinarySensorType.__members__:
|
if value is not None
|
||||||
sensor_type_name = getattr(BinarySensorType, name).value
|
]
|
||||||
sensors.append(
|
|
||||||
EzvizBinarySensor(coordinator, idx, name, sensor_type_name)
|
|
||||||
)
|
)
|
||||||
|
|
||||||
async_add_entities(sensors)
|
|
||||||
|
|
||||||
|
class EzvizBinarySensor(EzvizEntity, BinarySensorEntity):
|
||||||
class EzvizBinarySensor(CoordinatorEntity, BinarySensorEntity):
|
|
||||||
"""Representation of a Ezviz sensor."""
|
"""Representation of a Ezviz sensor."""
|
||||||
|
|
||||||
coordinator: EzvizDataUpdateCoordinator
|
coordinator: EzvizDataUpdateCoordinator
|
||||||
@ -48,46 +60,17 @@ class EzvizBinarySensor(CoordinatorEntity, BinarySensorEntity):
|
|||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
coordinator: EzvizDataUpdateCoordinator,
|
coordinator: EzvizDataUpdateCoordinator,
|
||||||
idx: int,
|
serial: str,
|
||||||
name: str,
|
binary_sensor: str,
|
||||||
sensor_type_name: str,
|
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the sensor."""
|
"""Initialize the sensor."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator, serial)
|
||||||
self._idx = idx
|
self._sensor_name = binary_sensor
|
||||||
self._camera_name = self.coordinator.data[self._idx]["name"]
|
self._attr_name = f"{self._camera_name} {binary_sensor.title()}"
|
||||||
self._name = name
|
self._attr_unique_id = f"{serial}_{self._camera_name}.{binary_sensor}"
|
||||||
self._sensor_name = f"{self._camera_name}.{self._name}"
|
self.entity_description = BINARY_SENSOR_TYPES[binary_sensor]
|
||||||
self.sensor_type_name = sensor_type_name
|
|
||||||
self._serial = self.coordinator.data[self._idx]["serial"]
|
|
||||||
|
|
||||||
@property
|
|
||||||
def name(self) -> str:
|
|
||||||
"""Return the name of the Ezviz sensor."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_on(self) -> bool:
|
def is_on(self) -> bool:
|
||||||
"""Return the state of the sensor."""
|
"""Return the state of the sensor."""
|
||||||
return self.coordinator.data[self._idx][self._name]
|
return self.data[self._sensor_name]
|
||||||
|
|
||||||
@property
|
|
||||||
def unique_id(self) -> str:
|
|
||||||
"""Return the unique ID of this sensor."""
|
|
||||||
return f"{self._serial}_{self._sensor_name}"
|
|
||||||
|
|
||||||
@property
|
|
||||||
def device_info(self) -> DeviceInfo:
|
|
||||||
"""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) -> str:
|
|
||||||
"""Device class for the sensor."""
|
|
||||||
return self.sensor_type_name
|
|
||||||
|
@ -18,9 +18,7 @@ from homeassistant.config_entries import (
|
|||||||
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
|
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers import config_validation as cv, entity_platform
|
from homeassistant.helpers import config_validation as cv, entity_platform
|
||||||
from homeassistant.helpers.entity import DeviceInfo
|
|
||||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
||||||
|
|
||||||
from .const import (
|
from .const import (
|
||||||
ATTR_DIRECTION,
|
ATTR_DIRECTION,
|
||||||
@ -40,7 +38,6 @@ from .const import (
|
|||||||
DIR_RIGHT,
|
DIR_RIGHT,
|
||||||
DIR_UP,
|
DIR_UP,
|
||||||
DOMAIN,
|
DOMAIN,
|
||||||
MANUFACTURER,
|
|
||||||
SERVICE_ALARM_SOUND,
|
SERVICE_ALARM_SOUND,
|
||||||
SERVICE_ALARM_TRIGER,
|
SERVICE_ALARM_TRIGER,
|
||||||
SERVICE_DETECTION_SENSITIVITY,
|
SERVICE_DETECTION_SENSITIVITY,
|
||||||
@ -48,6 +45,7 @@ from .const import (
|
|||||||
SERVICE_WAKE_DEVICE,
|
SERVICE_WAKE_DEVICE,
|
||||||
)
|
)
|
||||||
from .coordinator import EzvizDataUpdateCoordinator
|
from .coordinator import EzvizDataUpdateCoordinator
|
||||||
|
from .entity import EzvizEntity
|
||||||
|
|
||||||
CAMERA_SCHEMA = vol.Schema(
|
CAMERA_SCHEMA = vol.Schema(
|
||||||
{vol.Required(CONF_USERNAME): cv.string, vol.Required(CONF_PASSWORD): cv.string}
|
{vol.Required(CONF_USERNAME): cv.string, vol.Required(CONF_PASSWORD): cv.string}
|
||||||
@ -115,41 +113,37 @@ async def async_setup_entry(
|
|||||||
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
||||||
DATA_COORDINATOR
|
DATA_COORDINATOR
|
||||||
]
|
]
|
||||||
camera_config_entries = hass.config_entries.async_entries(DOMAIN)
|
|
||||||
|
|
||||||
camera_entities = []
|
camera_entities = []
|
||||||
|
|
||||||
for idx, camera in enumerate(coordinator.data):
|
for camera, value in coordinator.data.items():
|
||||||
|
|
||||||
# There seem to be a bug related to localRtspPort in Ezviz API...
|
|
||||||
local_rtsp_port = DEFAULT_RTSP_PORT
|
|
||||||
|
|
||||||
camera_rtsp_entry = [
|
camera_rtsp_entry = [
|
||||||
item
|
item
|
||||||
for item in camera_config_entries
|
for item in hass.config_entries.async_entries(DOMAIN)
|
||||||
if item.unique_id == camera[ATTR_SERIAL]
|
if item.unique_id == camera and item.source != SOURCE_IGNORE
|
||||||
]
|
]
|
||||||
|
|
||||||
if camera["local_rtsp_port"] != 0:
|
# There seem to be a bug related to localRtspPort in Ezviz API.
|
||||||
local_rtsp_port = camera["local_rtsp_port"]
|
local_rtsp_port = (
|
||||||
|
value["local_rtsp_port"]
|
||||||
if camera_rtsp_entry:
|
if value["local_rtsp_port"] != 0
|
||||||
conf_cameras = camera_rtsp_entry[0]
|
else DEFAULT_RTSP_PORT
|
||||||
|
|
||||||
# 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]
|
if camera_rtsp_entry:
|
||||||
camera_password = conf_cameras.data[CONF_PASSWORD]
|
|
||||||
|
|
||||||
camera_rtsp_stream = f"rtsp://{camera_username}:{camera_password}@{camera['local_ip']}:{local_rtsp_port}{ffmpeg_arguments}"
|
ffmpeg_arguments = camera_rtsp_entry[0].options[CONF_FFMPEG_ARGUMENTS]
|
||||||
|
camera_username = camera_rtsp_entry[0].data[CONF_USERNAME]
|
||||||
|
camera_password = camera_rtsp_entry[0].data[CONF_PASSWORD]
|
||||||
|
|
||||||
|
camera_rtsp_stream = f"rtsp://{camera_username}:{camera_password}@{value['local_ip']}:{local_rtsp_port}{ffmpeg_arguments}"
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"Camera %s source stream: %s", camera[ATTR_SERIAL], camera_rtsp_stream
|
"Configuring Camera %s with ip: %s rtsp port: %s ffmpeg arguments: %s",
|
||||||
|
camera,
|
||||||
|
value["local_ip"],
|
||||||
|
local_rtsp_port,
|
||||||
|
ffmpeg_arguments,
|
||||||
)
|
)
|
||||||
|
|
||||||
else:
|
else:
|
||||||
@ -159,26 +153,27 @@ async def async_setup_entry(
|
|||||||
DOMAIN,
|
DOMAIN,
|
||||||
context={"source": SOURCE_DISCOVERY},
|
context={"source": SOURCE_DISCOVERY},
|
||||||
data={
|
data={
|
||||||
ATTR_SERIAL: camera[ATTR_SERIAL],
|
ATTR_SERIAL: camera,
|
||||||
CONF_IP_ADDRESS: camera["local_ip"],
|
CONF_IP_ADDRESS: value["local_ip"],
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
camera_username = DEFAULT_CAMERA_USERNAME
|
|
||||||
camera_password = ""
|
|
||||||
camera_rtsp_stream = ""
|
|
||||||
ffmpeg_arguments = DEFAULT_FFMPEG_ARGUMENTS
|
|
||||||
_LOGGER.warning(
|
_LOGGER.warning(
|
||||||
"Found camera with serial %s without configuration. Please go to integration to complete setup",
|
"Found camera with serial %s without configuration. Please go to integration to complete setup",
|
||||||
camera[ATTR_SERIAL],
|
camera,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
ffmpeg_arguments = DEFAULT_FFMPEG_ARGUMENTS
|
||||||
|
camera_username = DEFAULT_CAMERA_USERNAME
|
||||||
|
camera_password = None
|
||||||
|
camera_rtsp_stream = ""
|
||||||
|
|
||||||
camera_entities.append(
|
camera_entities.append(
|
||||||
EzvizCamera(
|
EzvizCamera(
|
||||||
hass,
|
hass,
|
||||||
coordinator,
|
coordinator,
|
||||||
idx,
|
camera,
|
||||||
camera_username,
|
camera_username,
|
||||||
camera_password,
|
camera_password,
|
||||||
camera_rtsp_stream,
|
camera_rtsp_stream,
|
||||||
@ -230,7 +225,7 @@ async def async_setup_entry(
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class EzvizCamera(CoordinatorEntity, Camera):
|
class EzvizCamera(EzvizEntity, Camera):
|
||||||
"""An implementation of a Ezviz security camera."""
|
"""An implementation of a Ezviz security camera."""
|
||||||
|
|
||||||
coordinator: EzvizDataUpdateCoordinator
|
coordinator: EzvizDataUpdateCoordinator
|
||||||
@ -239,69 +234,51 @@ class EzvizCamera(CoordinatorEntity, Camera):
|
|||||||
self,
|
self,
|
||||||
hass: HomeAssistant,
|
hass: HomeAssistant,
|
||||||
coordinator: EzvizDataUpdateCoordinator,
|
coordinator: EzvizDataUpdateCoordinator,
|
||||||
idx: int,
|
serial: str,
|
||||||
camera_username: str,
|
camera_username: str,
|
||||||
camera_password: str,
|
camera_password: str | None,
|
||||||
camera_rtsp_stream: str | None,
|
camera_rtsp_stream: str | None,
|
||||||
local_rtsp_port: int | None,
|
local_rtsp_port: int,
|
||||||
ffmpeg_arguments: str | None,
|
ffmpeg_arguments: str | None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize a Ezviz security camera."""
|
"""Initialize a Ezviz security camera."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator, serial)
|
||||||
Camera.__init__(self)
|
Camera.__init__(self)
|
||||||
self._username = camera_username
|
self._username = camera_username
|
||||||
self._password = camera_password
|
self._password = camera_password
|
||||||
self._rtsp_stream = camera_rtsp_stream
|
self._rtsp_stream = camera_rtsp_stream
|
||||||
self._idx = idx
|
|
||||||
self._ffmpeg = hass.data[DATA_FFMPEG]
|
|
||||||
self._local_rtsp_port = local_rtsp_port
|
self._local_rtsp_port = local_rtsp_port
|
||||||
self._ffmpeg_arguments = ffmpeg_arguments
|
self._ffmpeg_arguments = ffmpeg_arguments
|
||||||
|
self._ffmpeg = hass.data[DATA_FFMPEG]
|
||||||
self._serial = self.coordinator.data[self._idx]["serial"]
|
self._attr_unique_id = serial
|
||||||
self._name = self.coordinator.data[self._idx]["name"]
|
self._attr_name = self.data["name"]
|
||||||
self._local_ip = self.coordinator.data[self._idx]["local_ip"]
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def available(self) -> bool:
|
def available(self) -> bool:
|
||||||
"""Return True if entity is available."""
|
"""Return True if entity is available."""
|
||||||
return self.coordinator.data[self._idx]["status"] != 2
|
return self.data["status"] != 2
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def supported_features(self) -> int:
|
def supported_features(self) -> int:
|
||||||
"""Return supported features."""
|
"""Return supported features."""
|
||||||
if self._rtsp_stream:
|
if self._password:
|
||||||
return SUPPORT_STREAM
|
return SUPPORT_STREAM
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
@property
|
|
||||||
def name(self) -> str:
|
|
||||||
"""Return the name of this device."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
|
||||||
def model(self) -> str:
|
|
||||||
"""Return the model of this device."""
|
|
||||||
return self.coordinator.data[self._idx]["device_sub_category"]
|
|
||||||
|
|
||||||
@property
|
|
||||||
def brand(self) -> str:
|
|
||||||
"""Return the manufacturer of this device."""
|
|
||||||
return MANUFACTURER
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_on(self) -> bool:
|
def is_on(self) -> bool:
|
||||||
"""Return true if on."""
|
"""Return true if on."""
|
||||||
return bool(self.coordinator.data[self._idx]["status"])
|
return bool(self.data["status"])
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_recording(self) -> bool:
|
def is_recording(self) -> bool:
|
||||||
"""Return true if the device is recording."""
|
"""Return true if the device is recording."""
|
||||||
return self.coordinator.data[self._idx]["alarm_notify"]
|
return self.data["alarm_notify"]
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def motion_detection_enabled(self) -> bool:
|
def motion_detection_enabled(self) -> bool:
|
||||||
"""Camera Motion Detection Status."""
|
"""Camera Motion Detection Status."""
|
||||||
return self.coordinator.data[self._idx]["alarm_notify"]
|
return self.data["alarm_notify"]
|
||||||
|
|
||||||
def enable_motion_detection(self) -> None:
|
def enable_motion_detection(self) -> None:
|
||||||
"""Enable motion detection in camera."""
|
"""Enable motion detection in camera."""
|
||||||
@ -319,11 +296,6 @@ class EzvizCamera(CoordinatorEntity, Camera):
|
|||||||
except InvalidHost as err:
|
except InvalidHost as err:
|
||||||
raise InvalidHost("Error disabling motion detection") from err
|
raise InvalidHost("Error disabling motion detection") from err
|
||||||
|
|
||||||
@property
|
|
||||||
def unique_id(self) -> str:
|
|
||||||
"""Return the name of this camera."""
|
|
||||||
return self._serial
|
|
||||||
|
|
||||||
async def async_camera_image(
|
async def async_camera_image(
|
||||||
self, width: int | None = None, height: int | None = None
|
self, width: int | None = None, height: int | None = None
|
||||||
) -> bytes | None:
|
) -> bytes | None:
|
||||||
@ -334,31 +306,24 @@ class EzvizCamera(CoordinatorEntity, Camera):
|
|||||||
self.hass, self._rtsp_stream, width=width, height=height
|
self.hass, self._rtsp_stream, width=width, height=height
|
||||||
)
|
)
|
||||||
|
|
||||||
@property
|
|
||||||
def device_info(self) -> DeviceInfo:
|
|
||||||
"""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) -> str | None:
|
async def stream_source(self) -> str | None:
|
||||||
"""Return the stream source."""
|
"""Return the stream source."""
|
||||||
local_ip = self.coordinator.data[self._idx]["local_ip"]
|
if self._password is None:
|
||||||
if self._local_rtsp_port:
|
return None
|
||||||
rtsp_stream_source = (
|
local_ip = self.data["local_ip"]
|
||||||
|
self._rtsp_stream = (
|
||||||
f"rtsp://{self._username}:{self._password}@"
|
f"rtsp://{self._username}:{self._password}@"
|
||||||
f"{local_ip}:{self._local_rtsp_port}{self._ffmpeg_arguments}"
|
f"{local_ip}:{self._local_rtsp_port}{self._ffmpeg_arguments}"
|
||||||
)
|
)
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"Camera %s source stream: %s", self._serial, rtsp_stream_source
|
"Configuring Camera %s with ip: %s rtsp port: %s ffmpeg arguments: %s",
|
||||||
|
self._serial,
|
||||||
|
local_ip,
|
||||||
|
self._local_rtsp_port,
|
||||||
|
self._ffmpeg_arguments,
|
||||||
)
|
)
|
||||||
self._rtsp_stream = rtsp_stream_source
|
|
||||||
return rtsp_stream_source
|
return self._rtsp_stream
|
||||||
return None
|
|
||||||
|
|
||||||
def perform_ptz(self, direction: str, speed: int) -> None:
|
def perform_ptz(self, direction: str, speed: int) -> None:
|
||||||
"""Perform a PTZ action on the camera."""
|
"""Perform a PTZ action on the camera."""
|
||||||
|
36
homeassistant/components/ezviz/entity.py
Normal file
36
homeassistant/components/ezviz/entity.py
Normal file
@ -0,0 +1,36 @@
|
|||||||
|
"""An abstract class common to all Ezviz entities."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from homeassistant.helpers.entity import DeviceInfo, Entity
|
||||||
|
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||||
|
|
||||||
|
from .const import DOMAIN, MANUFACTURER
|
||||||
|
from .coordinator import EzvizDataUpdateCoordinator
|
||||||
|
|
||||||
|
|
||||||
|
class EzvizEntity(CoordinatorEntity, Entity):
|
||||||
|
"""Generic entity encapsulating common features of Ezviz device."""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
coordinator: EzvizDataUpdateCoordinator,
|
||||||
|
serial: str,
|
||||||
|
) -> None:
|
||||||
|
"""Initialize the entity."""
|
||||||
|
super().__init__(coordinator)
|
||||||
|
self._serial = serial
|
||||||
|
self._camera_name = self.data["name"]
|
||||||
|
self._attr_device_info: DeviceInfo = {
|
||||||
|
"identifiers": {(DOMAIN, serial)},
|
||||||
|
"name": self.data["name"],
|
||||||
|
"model": self.data["device_sub_category"],
|
||||||
|
"manufacturer": MANUFACTURER,
|
||||||
|
"sw_version": self.data["version"],
|
||||||
|
}
|
||||||
|
|
||||||
|
@property
|
||||||
|
def data(self) -> dict[str, Any]:
|
||||||
|
"""Return coordinator data for this entity."""
|
||||||
|
return self.coordinator.data[self._serial]
|
@ -4,7 +4,7 @@
|
|||||||
"documentation": "https://www.home-assistant.io/integrations/ezviz",
|
"documentation": "https://www.home-assistant.io/integrations/ezviz",
|
||||||
"dependencies": ["ffmpeg"],
|
"dependencies": ["ffmpeg"],
|
||||||
"codeowners": ["@RenierM26", "@baqs"],
|
"codeowners": ["@RenierM26", "@baqs"],
|
||||||
"requirements": ["pyezviz==0.1.8.9"],
|
"requirements": ["pyezviz==0.1.9.4"],
|
||||||
"config_flow": true,
|
"config_flow": true,
|
||||||
"iot_class": "cloud_polling"
|
"iot_class": "cloud_polling"
|
||||||
}
|
}
|
||||||
|
@ -1,21 +1,42 @@
|
|||||||
"""Support for Ezviz sensors."""
|
"""Support for Ezviz sensors."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
from homeassistant.components.binary_sensor import DEVICE_CLASS_MOTION
|
||||||
|
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
|
||||||
from pyezviz.constants import SensorType
|
|
||||||
|
|
||||||
from homeassistant.components.sensor import SensorEntity
|
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
|
from homeassistant.const import DEVICE_CLASS_BATTERY, PERCENTAGE
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.entity import DeviceInfo
|
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
||||||
|
|
||||||
from .const import DATA_COORDINATOR, DOMAIN, MANUFACTURER
|
from .const import DATA_COORDINATOR, DOMAIN
|
||||||
from .coordinator import EzvizDataUpdateCoordinator
|
from .coordinator import EzvizDataUpdateCoordinator
|
||||||
|
from .entity import EzvizEntity
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
PARALLEL_UPDATES = 1
|
||||||
|
|
||||||
|
SENSOR_TYPES: dict[str, SensorEntityDescription] = {
|
||||||
|
"sw_version": SensorEntityDescription(key="sw_version"),
|
||||||
|
"battery_level": SensorEntityDescription(
|
||||||
|
key="battery_level",
|
||||||
|
native_unit_of_measurement=PERCENTAGE,
|
||||||
|
device_class=DEVICE_CLASS_BATTERY,
|
||||||
|
),
|
||||||
|
"alarm_sound_mod": SensorEntityDescription(key="alarm_sound_mod"),
|
||||||
|
"detection_sensibility": SensorEntityDescription(key="detection_sensibility"),
|
||||||
|
"last_alarm_time": SensorEntityDescription(key="last_alarm_time"),
|
||||||
|
"Seconds_Last_Trigger": SensorEntityDescription(
|
||||||
|
key="Seconds_Last_Trigger",
|
||||||
|
entity_registry_enabled_default=False,
|
||||||
|
),
|
||||||
|
"last_alarm_pic": SensorEntityDescription(key="last_alarm_pic"),
|
||||||
|
"supported_channels": SensorEntityDescription(key="supported_channels"),
|
||||||
|
"local_ip": SensorEntityDescription(key="local_ip"),
|
||||||
|
"wan_ip": SensorEntityDescription(key="wan_ip"),
|
||||||
|
"PIR_Status": SensorEntityDescription(
|
||||||
|
key="PIR_Status",
|
||||||
|
device_class=DEVICE_CLASS_MOTION,
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(
|
async def async_setup_entry(
|
||||||
@ -25,69 +46,34 @@ async def async_setup_entry(
|
|||||||
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
||||||
DATA_COORDINATOR
|
DATA_COORDINATOR
|
||||||
]
|
]
|
||||||
sensors = []
|
|
||||||
|
|
||||||
for idx, camera in enumerate(coordinator.data):
|
async_add_entities(
|
||||||
for name in camera:
|
[
|
||||||
# Only add sensor with value.
|
EzvizSensor(coordinator, camera, sensor)
|
||||||
if camera.get(name) is None:
|
for camera in coordinator.data
|
||||||
continue
|
for sensor, value in coordinator.data[camera].items()
|
||||||
|
if sensor in SENSOR_TYPES
|
||||||
if name in SensorType.__members__:
|
if value is not None
|
||||||
sensor_type_name = getattr(SensorType, name).value
|
]
|
||||||
sensors.append(EzvizSensor(coordinator, idx, name, sensor_type_name))
|
)
|
||||||
|
|
||||||
async_add_entities(sensors)
|
|
||||||
|
|
||||||
|
|
||||||
class EzvizSensor(CoordinatorEntity, SensorEntity):
|
class EzvizSensor(EzvizEntity, SensorEntity):
|
||||||
"""Representation of a Ezviz sensor."""
|
"""Representation of a Ezviz sensor."""
|
||||||
|
|
||||||
coordinator: EzvizDataUpdateCoordinator
|
coordinator: EzvizDataUpdateCoordinator
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self, coordinator: EzvizDataUpdateCoordinator, serial: str, sensor: str
|
||||||
coordinator: EzvizDataUpdateCoordinator,
|
|
||||||
idx: int,
|
|
||||||
name: str,
|
|
||||||
sensor_type_name: str,
|
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the sensor."""
|
"""Initialize the sensor."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator, serial)
|
||||||
self._idx = idx
|
self._sensor_name = sensor
|
||||||
self._camera_name = self.coordinator.data[self._idx]["name"]
|
self._attr_name = f"{self._camera_name} {sensor.title()}"
|
||||||
self._name = name
|
self._attr_unique_id = f"{serial}_{self._camera_name}.{sensor}"
|
||||||
self._sensor_name = f"{self._camera_name}.{self._name}"
|
self.entity_description = SENSOR_TYPES[sensor]
|
||||||
self.sensor_type_name = sensor_type_name
|
|
||||||
self._serial = self.coordinator.data[self._idx]["serial"]
|
|
||||||
|
|
||||||
@property
|
|
||||||
def name(self) -> str:
|
|
||||||
"""Return the name of the Ezviz sensor."""
|
|
||||||
return self._name
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def native_value(self) -> int | str:
|
def native_value(self) -> int | str:
|
||||||
"""Return the state of the sensor."""
|
"""Return the state of the sensor."""
|
||||||
return self.coordinator.data[self._idx][self._name]
|
return self.data[self._sensor_name]
|
||||||
|
|
||||||
@property
|
|
||||||
def unique_id(self) -> str:
|
|
||||||
"""Return the unique ID of this sensor."""
|
|
||||||
return f"{self._serial}_{self._sensor_name}"
|
|
||||||
|
|
||||||
@property
|
|
||||||
def device_info(self) -> DeviceInfo:
|
|
||||||
"""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) -> str:
|
|
||||||
"""Device class for the sensor."""
|
|
||||||
return self.sensor_type_name
|
|
||||||
|
@ -1,7 +1,6 @@
|
|||||||
"""Support for Ezviz Switch sensors."""
|
"""Support for Ezviz Switch sensors."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from pyezviz.constants import DeviceSwitchType
|
from pyezviz.constants import DeviceSwitchType
|
||||||
@ -10,14 +9,11 @@ from pyezviz.exceptions import HTTPError, PyEzvizError
|
|||||||
from homeassistant.components.switch import DEVICE_CLASS_SWITCH, SwitchEntity
|
from homeassistant.components.switch import DEVICE_CLASS_SWITCH, SwitchEntity
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
from homeassistant.core import HomeAssistant
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.entity import DeviceInfo
|
|
||||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
|
||||||
|
|
||||||
from .const import DATA_COORDINATOR, DOMAIN, MANUFACTURER
|
from .const import DATA_COORDINATOR, DOMAIN
|
||||||
from .coordinator import EzvizDataUpdateCoordinator
|
from .coordinator import EzvizDataUpdateCoordinator
|
||||||
|
from .entity import EzvizEntity
|
||||||
_LOGGER = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
|
|
||||||
async def async_setup_entry(
|
async def async_setup_entry(
|
||||||
@ -27,51 +23,40 @@ async def async_setup_entry(
|
|||||||
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
|
||||||
DATA_COORDINATOR
|
DATA_COORDINATOR
|
||||||
]
|
]
|
||||||
switch_entities = []
|
|
||||||
supported_switches = {switches.value for switches in DeviceSwitchType}
|
supported_switches = {switches.value for switches in DeviceSwitchType}
|
||||||
|
|
||||||
for idx, camera in enumerate(coordinator.data):
|
async_add_entities(
|
||||||
if not camera.get("switches"):
|
[
|
||||||
continue
|
EzvizSwitch(coordinator, camera, switch)
|
||||||
for switch in camera["switches"]:
|
for camera in coordinator.data
|
||||||
if switch not in supported_switches:
|
for switch in coordinator.data[camera].get("switches")
|
||||||
continue
|
if switch in supported_switches
|
||||||
switch_entities.append(EzvizSwitch(coordinator, idx, switch))
|
]
|
||||||
|
)
|
||||||
async_add_entities(switch_entities)
|
|
||||||
|
|
||||||
|
|
||||||
class EzvizSwitch(CoordinatorEntity, SwitchEntity):
|
class EzvizSwitch(EzvizEntity, SwitchEntity):
|
||||||
"""Representation of a Ezviz sensor."""
|
"""Representation of a Ezviz sensor."""
|
||||||
|
|
||||||
coordinator: EzvizDataUpdateCoordinator
|
coordinator: EzvizDataUpdateCoordinator
|
||||||
|
ATTR_DEVICE_CLASS = DEVICE_CLASS_SWITCH
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self, coordinator: EzvizDataUpdateCoordinator, idx: int, switch: str
|
self, coordinator: EzvizDataUpdateCoordinator, serial: str, switch: str
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Initialize the switch."""
|
"""Initialize the switch."""
|
||||||
super().__init__(coordinator)
|
super().__init__(coordinator, serial)
|
||||||
self._idx = idx
|
|
||||||
self._camera_name = self.coordinator.data[self._idx]["name"]
|
|
||||||
self._name = switch
|
self._name = switch
|
||||||
self._sensor_name = f"{self._camera_name}.{DeviceSwitchType(self._name).name}"
|
self._attr_name = f"{self._camera_name} {DeviceSwitchType(switch).name.title()}"
|
||||||
self._serial = self.coordinator.data[self._idx]["serial"]
|
self._attr_unique_id = (
|
||||||
self._device_class = DEVICE_CLASS_SWITCH
|
f"{serial}_{self._camera_name}.{DeviceSwitchType(switch).name}"
|
||||||
|
)
|
||||||
@property
|
|
||||||
def name(self) -> str:
|
|
||||||
"""Return the name of the Ezviz switch."""
|
|
||||||
return f"{DeviceSwitchType(self._name).name}"
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def is_on(self) -> bool:
|
def is_on(self) -> bool:
|
||||||
"""Return the state of the switch."""
|
"""Return the state of the switch."""
|
||||||
return self.coordinator.data[self._idx]["switches"][self._name]
|
return self.data["switches"][self._name]
|
||||||
|
|
||||||
@property
|
|
||||||
def unique_id(self) -> str:
|
|
||||||
"""Return the unique ID of this switch."""
|
|
||||||
return f"{self._serial}_{self._sensor_name}"
|
|
||||||
|
|
||||||
async def async_turn_on(self, **kwargs: Any) -> None:
|
async def async_turn_on(self, **kwargs: Any) -> None:
|
||||||
"""Change a device switch on the camera."""
|
"""Change a device switch on the camera."""
|
||||||
@ -98,19 +83,3 @@ class EzvizSwitch(CoordinatorEntity, SwitchEntity):
|
|||||||
|
|
||||||
if update_ok:
|
if update_ok:
|
||||||
await self.coordinator.async_request_refresh()
|
await self.coordinator.async_request_refresh()
|
||||||
|
|
||||||
@property
|
|
||||||
def device_info(self) -> DeviceInfo:
|
|
||||||
"""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) -> str:
|
|
||||||
"""Device class for the sensor."""
|
|
||||||
return self._device_class
|
|
||||||
|
@ -1465,7 +1465,7 @@ pyephember==0.3.1
|
|||||||
pyeverlights==0.1.0
|
pyeverlights==0.1.0
|
||||||
|
|
||||||
# homeassistant.components.ezviz
|
# homeassistant.components.ezviz
|
||||||
pyezviz==0.1.8.9
|
pyezviz==0.1.9.4
|
||||||
|
|
||||||
# homeassistant.components.fido
|
# homeassistant.components.fido
|
||||||
pyfido==2.1.1
|
pyfido==2.1.1
|
||||||
|
@ -853,7 +853,7 @@ pyefergy==0.1.2
|
|||||||
pyeverlights==0.1.0
|
pyeverlights==0.1.0
|
||||||
|
|
||||||
# homeassistant.components.ezviz
|
# homeassistant.components.ezviz
|
||||||
pyezviz==0.1.8.9
|
pyezviz==0.1.9.4
|
||||||
|
|
||||||
# homeassistant.components.fido
|
# homeassistant.components.fido
|
||||||
pyfido==2.1.1
|
pyfido==2.1.1
|
||||||
|
Loading…
x
Reference in New Issue
Block a user