core/homeassistant/components/netatmo/camera.py

"""Support for the Netatmo cameras."""
import logging

import pyatmo
import requests
import voluptuous as vol

from homeassistant.components.camera import (
    DOMAIN as CAMERA_DOMAIN,
    SUPPORT_STREAM,
    Camera,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON
import homeassistant.helpers.config_validation as cv
from homeassistant.util import Throttle

from .const import (
    ATTR_PSEUDO,
    AUTH,
    DATA_PERSONS,
    DOMAIN,
    MANUFACTURER,
    MIN_TIME_BETWEEN_EVENT_UPDATES,
    MIN_TIME_BETWEEN_UPDATES,
    MODELS,
)

_LOGGER = logging.getLogger(__name__)

CONF_HOME = "home"
CONF_CAMERAS = "cameras"
CONF_QUALITY = "quality"

DEFAULT_QUALITY = "high"

VALID_QUALITIES = ["high", "medium", "low", "poor"]

_BOOL_TO_STATE = {True: STATE_ON, False: STATE_OFF}

SCHEMA_SERVICE_SETLIGHTAUTO = vol.Schema(
    {vol.Optional(ATTR_ENTITY_ID): cv.entity_domain(CAMERA_DOMAIN)}
)


async def async_setup_entry(hass, entry, async_add_entities):
    """Set up the Netatmo camera platform."""

    def get_entities():
        """Retrieve Netatmo entities."""
        entities = []
        try:
            camera_data = CameraData(hass, hass.data[DOMAIN][entry.entry_id][AUTH])
            for camera in camera_data.get_all_cameras():
                _LOGGER.debug("Setting up camera %s %s", camera["id"], camera["name"])
                entities.append(
                    NetatmoCamera(
                        camera_data, camera["id"], camera["type"], True, DEFAULT_QUALITY
                    )
                )
            camera_data.update_persons()
        except pyatmo.NoDevice:
            _LOGGER.debug("No cameras found")
        return entities

    async_add_entities(await hass.async_add_executor_job(get_entities), True)


async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
    """Set up the Netatmo camera platform."""
    return


class NetatmoCamera(Camera):
    """Representation of a Netatmo camera."""

    def __init__(self, data, camera_id, camera_type, verify_ssl, quality):
        """Set up for access to the Netatmo camera images."""
        super().__init__()
        self._data = data
        self._camera_id = camera_id
        self._camera_name = self._data.camera_data.get_camera(cid=camera_id).get("name")
        self._name = f"{MANUFACTURER} {self._camera_name}"
        self._camera_type = camera_type
        self._unique_id = f"{self._camera_id}-{self._camera_type}"
        self._verify_ssl = verify_ssl
        self._quality = quality
        self._vpnurl = None
        self._localurl = None
        self._status = None
        self._sd_status = None
        self._alim_status = None
        self._is_local = None

    def camera_image(self):
        """Return a still image response from the camera."""
        try:
            if self._localurl:
                response = requests.get(
                    f"{self._localurl}/live/snapshot_720.jpg", timeout=10
                )
            elif self._vpnurl:
                response = requests.get(
                    f"{self._vpnurl}/live/snapshot_720.jpg",
                    timeout=10,
                    verify=self._verify_ssl,
                )
            else:
                _LOGGER.error("Welcome/Presence VPN URL is None")
                self._data.update()
                (self._vpnurl, self._localurl) = self._data.camera_data.camera_urls(
                    cid=self._camera_id
                )
                return None
        except requests.exceptions.RequestException as error:
            _LOGGER.info("Welcome/Presence URL changed: %s", error)
            self._data.update()
            (self._vpnurl, self._localurl) = self._data.camera_data.camera_urls(
                cid=self._camera_id
            )
            return None
        return response.content

    @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 name(self):
        """Return the name of this Netatmo camera device."""
        return self._name

    @property
    def device_info(self):
        """Return the device info for the sensor."""
        return {
            "identifiers": {(DOMAIN, self._camera_id)},
            "name": self._camera_name,
            "manufacturer": MANUFACTURER,
            "model": MODELS[self._camera_type],
        }

    @property
    def device_state_attributes(self):
        """Return the Netatmo-specific camera state attributes."""
        attr = {}
        attr["id"] = self._camera_id
        attr["status"] = self._status
        attr["sd_status"] = self._sd_status
        attr["alim_status"] = self._alim_status
        attr["is_local"] = self._is_local
        attr["vpn_url"] = self._vpnurl

        return attr

    @property
    def available(self):
        """Return True if entity is available."""
        return bool(self._alim_status == "on")

    @property
    def supported_features(self):
        """Return supported features."""
        return SUPPORT_STREAM

    @property
    def is_recording(self):
        """Return true if the device is recording."""
        return bool(self._status == "on")

    @property
    def brand(self):
        """Return the camera brand."""
        return MANUFACTURER

    @property
    def motion_detection_enabled(self):
        """Return the camera motion detection status."""
        return bool(self._status == "on")

    @property
    def is_on(self):
        """Return true if on."""
        return self.is_streaming

    async def stream_source(self):
        """Return the stream source."""
        url = "{0}/live/files/{1}/index.m3u8"
        if self._localurl:
            return url.format(self._localurl, self._quality)
        return url.format(self._vpnurl, self._quality)

    @property
    def model(self):
        """Return the camera model."""
        if self._camera_type == "NOC":
            return "Presence"
        if self._camera_type == "NACamera":
            return "Welcome"
        return None

    @property
    def unique_id(self):
        """Return the unique ID for this sensor."""
        return self._unique_id

    def update(self):
        """Update entity status."""
        self._data.update()

        camera = self._data.camera_data.get_camera(cid=self._camera_id)

        self._vpnurl, self._localurl = self._data.camera_data.camera_urls(
            cid=self._camera_id
        )
        self._status = camera.get("status")
        self._sd_status = camera.get("sd_status")
        self._alim_status = camera.get("alim_status")
        self._is_local = camera.get("is_local")
        self.is_streaming = self._alim_status == "on"


class CameraData:
    """Get the latest data from Netatmo."""

    def __init__(self, hass, auth):
        """Initialize the data object."""
        self._hass = hass
        self.auth = auth
        self.camera_data = None

    def get_all_cameras(self):
        """Return all camera available on the API as a list."""
        self.update()
        cameras = []
        for camera in self.camera_data.cameras.values():
            cameras.extend(camera.values())
        return cameras

    def get_modules(self, camera_id):
        """Return all modules for a given camera."""
        return self.camera_data.get_camera(camera_id).get("modules", [])

    def get_camera_type(self, camera_id):
        """Return camera type for a camera, cid has preference over camera."""
        return self.camera_data.cameraType(cid=camera_id)

    def update_persons(self):
        """Gather person data for webhooks."""
        for person_id, person_data in self.camera_data.persons.items():
            self._hass.data[DOMAIN][DATA_PERSONS][person_id] = person_data.get(
                ATTR_PSEUDO
            )

    @Throttle(MIN_TIME_BETWEEN_UPDATES)
    def update(self):
        """Call the Netatmo API to update the data."""
        self.camera_data = pyatmo.CameraData(self.auth, size=100)
        self.update_persons()

    @Throttle(MIN_TIME_BETWEEN_EVENT_UPDATES)
    def update_event(self, camera_type):
        """Call the Netatmo API to update the events."""
        self.camera_data.updateEvent(devicetype=camera_type)