camera.py

Go to the documentation of this file.

"""Support ezviz camera devices."""
 
from __future__ import annotations
 
import logging
 
from pyezviz.exceptions import HTTPError, InvalidHost, PyEzvizError
 
from homeassistant.components import ffmpeg
from homeassistant.components.camera import Camera, CameraEntityFeature
from homeassistant.components.ffmpeg import get_ffmpeg_manager
from homeassistant.components.stream import CONF_USE_WALLCLOCK_AS_TIMESTAMPS
from homeassistant.config_entries import (
    SOURCE_IGNORE,
    SOURCE_INTEGRATION_DISCOVERY,
    ConfigEntry,
)
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.entity_platform import (
    AddEntitiesCallback,
    async_get_current_platform,
)
 
from .const import (
    ATTR_SERIAL,
    CONF_FFMPEG_ARGUMENTS,
    DATA_COORDINATOR,
    DEFAULT_CAMERA_USERNAME,
    DEFAULT_FFMPEG_ARGUMENTS,
    DOMAIN,
    SERVICE_WAKE_DEVICE,
)
from .coordinator import EzvizDataUpdateCoordinator
from .entity import EzvizEntity
 
_LOGGER = logging.getLogger(__name__)
 
 
async def async_setup_entry(
    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
    """Set up EZVIZ cameras based on a config entry."""
 
    coordinator: EzvizDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
        DATA_COORDINATOR
    ]
 
    camera_entities = []
 
    for camera, value in coordinator.data.items():
        camera_rtsp_entry = [
            item
            for item in hass.config_entries.async_entries(DOMAIN)
            if item.unique_id == camera and item.source != SOURCE_IGNORE
        ]
 
        if camera_rtsp_entry:
            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']}:{value['local_rtsp_port']}{ffmpeg_arguments}"
            _LOGGER.debug(
                "Configuring Camera %s with ip: %s rtsp port: %s ffmpeg arguments: %s",
                camera,
                value["local_ip"],
                value["local_rtsp_port"],
                ffmpeg_arguments,
            )
 
        else:
            discovery_flow.async_create_flow(
                hass,
                DOMAIN,
                context={"source": SOURCE_INTEGRATION_DISCOVERY},
                data={
                    ATTR_SERIAL: camera,
                    CONF_IP_ADDRESS: value["local_ip"],
                },
            )
 
            _LOGGER.warning(
                (
                    "Found camera with serial %s without configuration. Please go to"
                    " integration to complete setup"
                ),
                camera,
            )
 
            ffmpeg_arguments = DEFAULT_FFMPEG_ARGUMENTS
            camera_username = DEFAULT_CAMERA_USERNAME
            camera_password = None
            camera_rtsp_stream = ""
 
        camera_entities.append(
            EzvizCamera(
                hass,
                coordinator,
                camera,
                camera_username,
                camera_password,
                camera_rtsp_stream,
                value["local_rtsp_port"],
                ffmpeg_arguments,
            )
        )
 
    async_add_entities(camera_entities)
 
    platform = async_get_current_platform()
 
    platform.async_register_entity_service(
        SERVICE_WAKE_DEVICE, None, "perform_wake_device"
    )
 
 
class EzvizCamera(EzvizEntity, Camera):
    """An implementation of a EZVIZ security camera."""
 
    _attr_name = None
 
    def __init__(
        self,
        hass: HomeAssistant,
        coordinator: EzvizDataUpdateCoordinator,
        serial: str,
        camera_username: str,
        camera_password: str | None,
        camera_rtsp_stream: str | None,
        local_rtsp_port: int,
        ffmpeg_arguments: str | None,
    ) -> None:
        """Initialize a EZVIZ security camera."""
        super().__init__(coordinator, serial)
        Camera.__init__(self)
        self.stream_options[CONF_USE_WALLCLOCK_AS_TIMESTAMPS] = True
        self._username_username = camera_username
        self._password_password = camera_password
        self._rtsp_stream_rtsp_stream = camera_rtsp_stream
        self._local_rtsp_port_local_rtsp_port = local_rtsp_port
        self._ffmpeg_arguments_ffmpeg_arguments = ffmpeg_arguments
        self._ffmpeg_ffmpeg = get_ffmpeg_manager(hass)
        self._attr_unique_id_attr_unique_id = serial
        if camera_password:
            self._attr_supported_features_attr_supported_features = CameraEntityFeature.STREAM
 
    @property
    def available(self) -> bool:
        """Return True if entity is available."""
        return self.datadatadata["status"] != 2
 
    @property
    def is_on(self) -> bool:
        """Return true if on."""
        return bool(self.datadatadata["status"])
 
    @property
    def is_recording(self) -> bool:
        """Return true if the device is recording."""
        return self.datadatadata["alarm_notify"]
 
    @property
    def motion_detection_enabled(self) -> bool:
        """Camera Motion Detection Status."""
        return self.datadatadata["alarm_notify"]
 
    def enable_motion_detection(self) -> None:
        """Enable motion detection in camera."""
        try:
            self.coordinator.ezviz_client.set_camera_defence(self._serial_serial, 1)
 
        except InvalidHost as err:
            raise InvalidHost("Error enabling motion detection") from err
 
    def disable_motion_detection(self) -> None:
        """Disable motion detection."""
        try:
            self.coordinator.ezviz_client.set_camera_defence(self._serial_serial, 0)
 
        except InvalidHost as err:
            raise InvalidHost("Error disabling motion detection") from err
 
    async def async_camera_image(
        self, width: int | None = None, height: int | None = None
    ) -> bytes | None:
        """Return a frame from the camera stream."""
        if self._rtsp_stream_rtsp_stream is None:
            return None
        return await ffmpeg.async_get_image(
            self.hasshasshass, self._rtsp_stream_rtsp_stream, width=width, height=height
        )
 
    async def stream_source(self) -> str | None:
        """Return the stream source."""
        if self._password_password is None:
            return None
        local_ip = self.datadatadata["local_ip"]
        self._rtsp_stream_rtsp_stream = (
            f"rtsp://{self._username}:{self._password}@"
            f"{local_ip}:{self._local_rtsp_port}{self._ffmpeg_arguments}"
        )
        _LOGGER.debug(
            "Configuring Camera %s with ip: %s rtsp port: %s ffmpeg arguments: %s",
            self._serial_serial,
            local_ip,
            self._local_rtsp_port_local_rtsp_port,
            self._ffmpeg_arguments_ffmpeg_arguments,
        )
 
        return self._rtsp_stream_rtsp_stream
 
    def perform_wake_device(self) -> None:
        """Basically wakes the camera by querying the device."""
        try:
            self.coordinator.ezviz_client.get_detection_sensibility(self._serial_serial)
        except (HTTPError, PyEzvizError) as err:
            raise PyEzvizError("Cannot wake device") from err