camera.py

Go to the documentation of this file.

"""Support for Amcrest IP cameras."""
 
from __future__ import annotations
 
import asyncio
from collections.abc import Callable
from datetime import timedelta
import logging
from typing import TYPE_CHECKING, Any
 
import aiohttp
from aiohttp import web
from amcrest import AmcrestError
from haffmpeg.camera import CameraMjpeg
import voluptuous as vol
 
from homeassistant.components.camera import Camera, CameraEntityFeature
from homeassistant.components.ffmpeg import FFmpegManager, get_ffmpeg_manager
from homeassistant.const import ATTR_ENTITY_ID, CONF_NAME, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import (
    async_aiohttp_proxy_stream,
    async_aiohttp_proxy_web,
    async_get_clientsession,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
from .const import (
    CAMERA_WEB_SESSION_TIMEOUT,
    CAMERAS,
    COMM_TIMEOUT,
    DATA_AMCREST,
    DEVICES,
    RESOLUTION_TO_STREAM,
    SERVICE_UPDATE,
    SNAPSHOT_TIMEOUT,
)
from .helpers import log_update_error, service_signal
 
if TYPE_CHECKING:
    from . import AmcrestDevice
 
_LOGGER = logging.getLogger(__name__)
 
SCAN_INTERVAL = timedelta(seconds=15)
 
STREAM_SOURCE_LIST = ["snapshot", "mjpeg", "rtsp"]
 
_SRV_EN_REC = "enable_recording"
_SRV_DS_REC = "disable_recording"
_SRV_EN_AUD = "enable_audio"
_SRV_DS_AUD = "disable_audio"
_SRV_EN_MOT_REC = "enable_motion_recording"
_SRV_DS_MOT_REC = "disable_motion_recording"
_SRV_GOTO = "goto_preset"
_SRV_CBW = "set_color_bw"
_SRV_TOUR_ON = "start_tour"
_SRV_TOUR_OFF = "stop_tour"
 
_SRV_PTZ_CTRL = "ptz_control"
_ATTR_PTZ_TT = "travel_time"
_ATTR_PTZ_MOV = "movement"
_MOV = [
    "zoom_out",
    "zoom_in",
    "right",
    "left",
    "up",
    "down",
    "right_down",
    "right_up",
    "left_down",
    "left_up",
]
_ZOOM_ACTIONS = ["ZoomWide", "ZoomTele"]
_MOVE_1_ACTIONS = ["Right", "Left", "Up", "Down"]
_MOVE_2_ACTIONS = ["RightDown", "RightUp", "LeftDown", "LeftUp"]
_ACTION = _ZOOM_ACTIONS + _MOVE_1_ACTIONS + _MOVE_2_ACTIONS
 
_DEFAULT_TT = 0.2
 
_ATTR_PRESET = "preset"
_ATTR_COLOR_BW = "color_bw"
 
_CBW_COLOR = "color"
_CBW_AUTO = "auto"
_CBW_BW = "bw"
_CBW = [_CBW_COLOR, _CBW_AUTO, _CBW_BW]
 
_SRV_SCHEMA = vol.Schema({vol.Optional(ATTR_ENTITY_ID): cv.comp_entity_ids})
_SRV_GOTO_SCHEMA = _SRV_SCHEMA.extend(
    {vol.Required(_ATTR_PRESET): vol.All(vol.Coerce(int), vol.Range(min=1))}
)
_SRV_CBW_SCHEMA = _SRV_SCHEMA.extend({vol.Required(_ATTR_COLOR_BW): vol.In(_CBW)})
_SRV_PTZ_SCHEMA = _SRV_SCHEMA.extend(
    {
        vol.Required(_ATTR_PTZ_MOV): vol.In(_MOV),
        vol.Optional(_ATTR_PTZ_TT, default=_DEFAULT_TT): cv.small_float,
    }
)
 
CAMERA_SERVICES = {
    _SRV_EN_REC: (_SRV_SCHEMA, "async_enable_recording", ()),
    _SRV_DS_REC: (_SRV_SCHEMA, "async_disable_recording", ()),
    _SRV_EN_AUD: (_SRV_SCHEMA, "async_enable_audio", ()),
    _SRV_DS_AUD: (_SRV_SCHEMA, "async_disable_audio", ()),
    _SRV_EN_MOT_REC: (_SRV_SCHEMA, "async_enable_motion_recording", ()),
    _SRV_DS_MOT_REC: (_SRV_SCHEMA, "async_disable_motion_recording", ()),
    _SRV_GOTO: (_SRV_GOTO_SCHEMA, "async_goto_preset", (_ATTR_PRESET,)),
    _SRV_CBW: (_SRV_CBW_SCHEMA, "async_set_color_bw", (_ATTR_COLOR_BW,)),
    _SRV_TOUR_ON: (_SRV_SCHEMA, "async_start_tour", ()),
    _SRV_TOUR_OFF: (_SRV_SCHEMA, "async_stop_tour", ()),
    _SRV_PTZ_CTRL: (
        _SRV_PTZ_SCHEMA,
        "async_ptz_control",
        (_ATTR_PTZ_MOV, _ATTR_PTZ_TT),
    ),
}
 
_BOOL_TO_STATE = {True: STATE_ON, False: STATE_OFF}
 
 
async def async_setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    async_add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up an Amcrest IP Camera."""
    if discovery_info is None:
        return
 
    name = discovery_info[CONF_NAME]
    device = hass.data[DATA_AMCREST][DEVICES][name]
    entity = AmcrestCam(name, device, get_ffmpeg_manager(hass))
 
    async_add_entities([entity], True)
 
 
class CannotSnapshot(Exception):
    """Conditions are not valid for taking a snapshot."""
 
 
class AmcrestCommandFailed(Exception):
    """Amcrest camera command did not work."""
 
 
class AmcrestCam(Camera):
    """An implementation of an Amcrest IP camera."""
 
    _attr_should_poll = True  # Cameras default to False
    _attr_supported_features = CameraEntityFeature.ON_OFF | CameraEntityFeature.STREAM
 
    def __init__(self, name: str, device: AmcrestDevice, ffmpeg: FFmpegManager) -> None:
        """Initialize an Amcrest camera."""
        super().__init__()
        self._name_name = name
        self._api_api = device.api
        self._ffmpeg_ffmpeg = ffmpeg
        self._ffmpeg_arguments_ffmpeg_arguments = device.ffmpeg_arguments
        self._stream_source_stream_source = device.stream_source
        self._resolution_resolution = device.resolution
        self._channel_channel = device.channel
        self._token_token = self._auth_auth = device.authentication
        self._control_light_control_light = device.control_light
        self._is_recording: bool = False
        self._motion_detection_enabled: bool = False
        self._brand_brand: str | None = None
        self._model_model: str | None = None
        self._audio_enabled: bool | None = None
        self._motion_recording_enabled: bool | None = None
        self._color_bw: str | None = None
        self._rtsp_url_rtsp_url: str | None = None
        self._snapshot_task_snapshot_task: asyncio.tasks.Task | None = None
        self._unsub_dispatcher: list[Callable[[], None]] = []
 
    def _check_snapshot_ok(self) -> None:
        available = self.availableavailableavailableavailable
        if not available or not self.is_onis_onis_on:
            _LOGGER.warning(
                "Attempt to take snapshot when %s camera is %s",
                self.namenamename,
                "offline" if not available else "off",
            )
            raise CannotSnapshot
 
    async def _async_get_image(self) -> bytes | None:
        try:
            # Send the request to snap a picture and return raw jpg data
            # Snapshot command needs a much longer read timeout than other commands.
            return await self._api_api.async_snapshot(
                timeout=(COMM_TIMEOUT, SNAPSHOT_TIMEOUT)
            )
        except AmcrestError as error:
            log_update_error(_LOGGER, "get image from", self.namenamename, "camera", error)
            return None
        finally:
            self._snapshot_task_snapshot_task = None
 
    async def async_camera_image(
        self, width: int | None = None, height: int | None = None
    ) -> bytes | None:
        """Return a still image response from the camera."""
        _LOGGER.debug("Take snapshot from %s", self._name_name)
        try:
            # Amcrest cameras only support one snapshot command at a time.
            # Hence need to wait if a previous snapshot has not yet finished.
            # Also need to check that camera is online and turned on before each wait
            # and before initiating snapshot.
            while self._snapshot_task_snapshot_task:
                self._check_snapshot_ok_check_snapshot_ok()
                _LOGGER.debug("Waiting for previous snapshot from %s", self._name_name)
                await self._snapshot_task_snapshot_task
            self._check_snapshot_ok_check_snapshot_ok()
            # Run snapshot command in separate Task that can't be cancelled so
            # 1) it's not possible to send another snapshot command while camera is
            #    still working on a previous one, and
            # 2) someone will be around to catch any exceptions.
            self._snapshot_task_snapshot_task = self.hasshass.async_create_task(self._async_get_image_async_get_image())
            return await asyncio.shield(self._snapshot_task_snapshot_task)
        except CannotSnapshot:
            return None
 
    async def handle_async_mjpeg_stream(
        self, request: web.Request
    ) -> web.StreamResponse | None:
        """Return an MJPEG stream."""
        # The snapshot implementation is handled by the parent class
        if self._stream_source_stream_source == "snapshot":
            return await super().handle_async_mjpeg_stream(request)
 
        if not self.availableavailableavailableavailable:
            _LOGGER.warning(
                "Attempt to stream %s when %s camera is offline",
                self._stream_source_stream_source,
                self.namenamename,
            )
            return None
 
        if self._stream_source_stream_source == "mjpeg":
            # stream an MJPEG image stream directly from the camera
            websession = async_get_clientsession(self.hasshass)
            streaming_url = self._api_api.mjpeg_url(typeno=self._resolution_resolution)
            stream_coro = websession.get(
                streaming_url,
                auth=self._token_token,
                timeout=aiohttp.ClientTimeout(total=CAMERA_WEB_SESSION_TIMEOUT),
            )
 
            return await async_aiohttp_proxy_web(self.hasshass, request, stream_coro)
 
        # streaming via ffmpeg
        assert self._rtsp_url_rtsp_url is not None
        streaming_url = self._rtsp_url_rtsp_url
        stream = CameraMjpeg(self._ffmpeg_ffmpeg.binary)
        await stream.open_camera(streaming_url, extra_cmd=self._ffmpeg_arguments_ffmpeg_arguments)
 
        try:
            stream_reader = await stream.get_reader()
            return await async_aiohttp_proxy_stream(
                self.hasshass,
                request,
                stream_reader,
                self._ffmpeg_ffmpeg.ffmpeg_stream_content_type,
            )
        finally:
            await stream.close()
 
    # Entity property overrides
 
    @property
    def name(self) -> str:
        """Return the name of this camera."""
        return self._name_name
 
    @property
    def extra_state_attributes(self) -> dict[str, Any]:
        """Return the Amcrest-specific camera state attributes."""
        attr = {}
        if self._audio_enabled is not None:
            attr["audio"] = _BOOL_TO_STATE.get(self._audio_enabled)
        if self._motion_recording_enabled is not None:
            attr["motion_recording"] = _BOOL_TO_STATE.get(
                self._motion_recording_enabled
            )
        if self._color_bw is not None:
            attr[_ATTR_COLOR_BW] = self._color_bw
        return attr
 
    @property
    def available(self) -> bool:
        """Return True if entity is available."""
        return self._api_api.available
 
    # Camera property overrides
 
    @property
    def is_recording(self) -> bool:
        """Return true if the device is recording."""
        return self._is_recording
 
    @property
    def brand(self) -> str | None:
        """Return the camera brand."""
        return self._brand_brand
 
    @property
    def motion_detection_enabled(self) -> bool:
        """Return the camera motion detection status."""
        return self._motion_detection_enabled
 
    @property
    def model(self) -> str | None:
        """Return the camera model."""
        return self._model_model
 
    async def stream_source(self) -> str | None:
        """Return the source of the stream."""
        return self._rtsp_url_rtsp_url
 
    @property
    def is_on(self) -> bool:
        """Return true if on."""
        return self.is_streamingis_streaming
 
    # Other Entity method overrides
 
    @callback
    def async_on_demand_update(self) -> None:
        """Update state."""
        self.async_schedule_update_ha_stateasync_schedule_update_ha_state(True)
 
    async def async_added_to_hass(self) -> None:
        """Subscribe to signals and add camera to list."""
        self._unsub_dispatcher.extend(
            async_dispatcher_connect(
                self.hasshass,
                service_signal(service, self.entity_identity_id),
                getattr(self, callback_name),
            )
            for service, (_, callback_name, _) in CAMERA_SERVICES.items()
        )
        self._unsub_dispatcher.append(
            async_dispatcher_connect(
                self.hasshass,
                service_signal(SERVICE_UPDATE, self.namenamename),
                self.async_on_demand_updateasync_on_demand_update,
            )
        )
        self.hasshass.data[DATA_AMCREST][CAMERAS].append(self.entity_identity_id)
 
    async def async_will_remove_from_hass(self) -> None:
        """Remove camera from list and disconnect from signals."""
        self.hasshass.data[DATA_AMCREST][CAMERAS].remove(self.entity_identity_id)
        for unsub_dispatcher in self._unsub_dispatcher:
            unsub_dispatcher()
 
    async def async_update(self) -> None:
        """Update entity status."""
        if not self.availableavailableavailableavailable:
            return
        _LOGGER.debug("Updating %s camera", self.namenamename)
        try:
            if self._brand_brand is None:
                resp = await self._api_api.async_vendor_information
                _LOGGER.debug("Assigned brand=%s", resp)
                if resp:
                    self._brand_brand = resp
                else:
                    self._brand_brand = "unknown"
            if self._model_model is None:
                resp = await self._api_api.async_device_type
                _LOGGER.debug("Assigned model=%s", resp)
                if resp:
                    self._model_model = resp
                else:
                    self._model_model = "unknown"
            if self._attr_unique_id_attr_unique_id is None:
                serial_number = (await self._api_api.async_serial_number).strip()
                if serial_number:
                    self._attr_unique_id_attr_unique_id = (
                        f"{serial_number}-{self._resolution}-{self._channel}"
                    )
                    _LOGGER.debug("Assigned unique_id=%s", self._attr_unique_id_attr_unique_id)
            if self._rtsp_url_rtsp_url is None:
                self._rtsp_url_rtsp_url = await self._api_api.async_rtsp_url(typeno=self._resolution_resolution)
 
            (
                self._attr_is_streaming,
                self._is_recording,
                self._motion_detection_enabled,
                self._audio_enabled,
                self._motion_recording_enabled,
                self._color_bw,
            ) = await asyncio.gather(
                self._async_get_video_async_get_video(),
                self._async_get_recording_async_get_recording(),
                self._async_get_motion_detection_async_get_motion_detection(),
                self._async_get_audio_async_get_audio(),
                self._async_get_motion_recording_async_get_motion_recording(),
                self._async_get_color_mode_async_get_color_mode(),
            )
        except AmcrestError as error:
            log_update_error(_LOGGER, "get", self.namenamename, "camera attributes", error)
 
    # Other Camera method overrides
 
    async def async_turn_off(self) -> None:
        """Turn off camera."""
        await self._async_enable_video_async_enable_video(False)
 
    async def async_turn_on(self) -> None:
        """Turn on camera."""
        await self._async_enable_video_async_enable_video(True)
 
    async def async_enable_motion_detection(self) -> None:
        """Enable motion detection in the camera."""
        await self._async_enable_motion_detection_async_enable_motion_detection(True)
 
    async def async_disable_motion_detection(self) -> None:
        """Disable motion detection in camera."""
        await self._async_enable_motion_detection_async_enable_motion_detection(False)
 
    # Additional Amcrest Camera service methods
 
    async def async_enable_recording(self) -> None:
        """Call the job and enable recording."""
        await self._async_enable_recording_async_enable_recording(True)
 
    async def async_disable_recording(self) -> None:
        """Call the job and disable recording."""
        await self._async_enable_recording_async_enable_recording(False)
 
    async def async_enable_audio(self) -> None:
        """Call the job and enable audio."""
        await self._async_enable_audio_async_enable_audio(True)
 
    async def async_disable_audio(self) -> None:
        """Call the job and disable audio."""
        await self._async_enable_audio_async_enable_audio(False)
 
    async def async_enable_motion_recording(self) -> None:
        """Call the job and enable motion recording."""
        await self._async_enable_motion_recording_async_enable_motion_recording(True)
 
    async def async_disable_motion_recording(self) -> None:
        """Call the job and disable motion recording."""
        await self._async_enable_motion_recording_async_enable_motion_recording(False)
 
    async def async_goto_preset(self, preset: int) -> None:
        """Call the job and move camera to preset position."""
        await self._async_goto_preset_async_goto_preset(preset)
 
    async def async_set_color_bw(self, color_bw: str) -> None:
        """Call the job and set camera color mode."""
        await self._async_set_color_bw_async_set_color_bw(color_bw)
 
    async def async_start_tour(self) -> None:
        """Call the job and start camera tour."""
        await self._async_start_tour_async_start_tour(True)
 
    async def async_stop_tour(self) -> None:
        """Call the job and stop camera tour."""
        await self._async_start_tour_async_start_tour(False)
 
    async def async_ptz_control(self, movement: str, travel_time: float) -> None:
        """Move or zoom camera in specified direction."""
        code = _ACTION[_MOV.index(movement)]
 
        kwargs = {"code": code, "arg1": 0, "arg2": 0, "arg3": 0}
        if code in _MOVE_1_ACTIONS:
            kwargs["arg2"] = 1
        elif code in _MOVE_2_ACTIONS:
            kwargs["arg1"] = kwargs["arg2"] = 1
 
        try:
            await self._api_api.async_ptz_control_command(action="start", **kwargs)  # type: ignore[arg-type]
            await asyncio.sleep(travel_time)
            await self._api_api.async_ptz_control_command(action="stop", **kwargs)  # type: ignore[arg-type]
        except AmcrestError as error:
            log_update_error(
                _LOGGER, "move", self.namenamename, f"camera PTZ {movement}", error
            )
 
    # Methods to send commands to Amcrest camera and handle errors
 
    async def _async_change_setting(
        self, value: str | bool, description: str, attr: str | None = None
    ) -> None:
        func = description.replace(" ", "_")
        description = f"camera {description} to {value}"
        action = "set"
        max_tries = 3
        for tries in range(max_tries, 0, -1):
            try:
                await getattr(self, f"_async_set_{func}")(value)
                new_value = await getattr(self, f"_async_get_{func}")()
                if new_value != value:
                    raise AmcrestCommandFailed  # noqa: TRY301
            except (AmcrestError, AmcrestCommandFailed) as error:
                if tries == 1:
                    log_update_error(_LOGGER, action, self.namenamename, description, error)
                    return
                log_update_error(
                    _LOGGER, action, self.namenamename, description, error, logging.DEBUG
                )
            else:
                if attr:
                    setattr(self, attr, new_value)
                    self.schedule_update_ha_stateschedule_update_ha_state()
                return
 
    async def _async_get_video(self) -> bool:
        return await self._api_api.async_is_video_enabled(
            channel=0, stream=RESOLUTION_TO_STREAM[self._resolution_resolution]
        )
 
    async def _async_set_video(self, enable: bool) -> None:
        await self._api_api.async_set_video_enabled(
            enable, channel=0, stream=RESOLUTION_TO_STREAM[self._resolution_resolution]
        )
 
    async def _async_enable_video(self, enable: bool) -> None:
        """Enable or disable camera video stream."""
        # Given the way the camera's state is determined by
        # is_streaming and is_recording, we can't leave
        # recording on if video stream is being turned off.
        if self.is_recordingis_recordingis_recording and not enable:
            await self._async_enable_recording_async_enable_recording(False)
        await self._async_change_setting_async_change_setting(enable, "video", "_attr_is_streaming")
        if self._control_light_control_light:
            await self._async_change_light_async_change_light()
 
    async def _async_get_recording(self) -> bool:
        return (await self._api_api.async_record_mode) == "Manual"
 
    async def _async_set_recording(self, enable: bool) -> None:
        rec_mode = {"Automatic": 0, "Manual": 1}
        # The property has a str type, but setter has int type, which causes mypy confusion
        await self._api_api.async_set_record_mode(
            rec_mode["Manual" if enable else "Automatic"]
        )
 
    async def _async_enable_recording(self, enable: bool) -> None:
        """Turn recording on or off."""
        # Given the way the camera's state is determined by
        # is_streaming and is_recording, we can't leave
        # video stream off if recording is being turned on.
        if not self.is_streamingis_streaming and enable:
            await self._async_enable_video_async_enable_video(True)
        await self._async_change_setting_async_change_setting(enable, "recording", "_is_recording")
 
    async def _async_get_motion_detection(self) -> bool:
        return await self._api_api.async_is_motion_detector_on()
 
    async def _async_set_motion_detection(self, enable: bool) -> None:
        # The property has a str type, but setter has bool type, which causes mypy confusion
        await self._api_api.async_set_motion_detection(enable)
 
    async def _async_enable_motion_detection(self, enable: bool) -> None:
        """Enable or disable motion detection."""
        await self._async_change_setting_async_change_setting(
            enable, "motion detection", "_motion_detection_enabled"
        )
 
    async def _async_get_audio(self) -> bool:
        return await self._api_api.async_is_audio_enabled(
            channel=0, stream=RESOLUTION_TO_STREAM[self._resolution_resolution]
        )
 
    async def _async_set_audio(self, enable: bool) -> None:
        await self._api_api.async_set_audio_enabled(
            enable, channel=0, stream=RESOLUTION_TO_STREAM[self._resolution_resolution]
        )
 
    async def _async_enable_audio(self, enable: bool) -> None:
        """Enable or disable audio stream."""
        await self._async_change_setting_async_change_setting(enable, "audio", "_audio_enabled")
        if self._control_light_control_light:
            await self._async_change_light_async_change_light()
 
    async def _async_get_indicator_light(self) -> bool:
        return (
            "true"
            in (
                await self._api_api.async_command(
                    "configManager.cgi?action=getConfig&name=LightGlobal"
                )
            ).content.decode()
        )
 
    async def _async_set_indicator_light(self, enable: bool) -> None:
        await self._api_api.async_command(
            f"configManager.cgi?action=setConfig&LightGlobal[0].Enable={str(enable).lower()}"
        )
 
    async def _async_change_light(self) -> None:
        """Enable or disable indicator light."""
        await self._async_change_setting_async_change_setting(
            self._audio_enabled or self.is_streamingis_streaming, "indicator light"
        )
 
    async def _async_get_motion_recording(self) -> bool:
        return await self._api_api.async_is_record_on_motion_detection()
 
    async def _async_set_motion_recording(self, enable: bool) -> None:
        await self._api_api.async_set_motion_recording(enable)
 
    async def _async_enable_motion_recording(self, enable: bool) -> None:
        """Enable or disable motion recording."""
        await self._async_change_setting_async_change_setting(
            enable, "motion recording", "_motion_recording_enabled"
        )
 
    async def _async_goto_preset(self, preset: int) -> None:
        """Move camera position and zoom to preset."""
        try:
            await self._api_api.async_go_to_preset(preset_point_number=preset)
        except AmcrestError as error:
            log_update_error(
                _LOGGER, "move", self.namenamename, f"camera to preset {preset}", error
            )
 
    async def _async_get_color_mode(self) -> str:
        return _CBW[await self._api_api.async_day_night_color]
 
    async def _async_set_color_mode(self, cbw: str) -> None:
        await self._api_api.async_set_day_night_color(_CBW.index(cbw), channel=0)
 
    async def _async_set_color_bw(self, cbw: str) -> None:
        """Set camera color mode."""
        await self._async_change_setting_async_change_setting(cbw, "color mode", "_color_bw")
 
    async def _async_start_tour(self, start: bool) -> None:
        """Start camera tour."""
        try:
            await self._api_api.async_tour(start=start)
        except AmcrestError as error:
            log_update_error(
                _LOGGER, "start" if start else "stop", self.namenamename, "camera tour", error
            )