core/homeassistant/components/nfandroidtv/notify.py

"""
Notifications for Android TV notification service.

For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/notify.nfandroidtv/
"""
import base64
import io
import logging

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
import voluptuous as vol

from homeassistant.const import CONF_TIMEOUT, CONF_HOST
import homeassistant.helpers.config_validation as cv

from homeassistant.components.notify import (
    ATTR_DATA, ATTR_TITLE, ATTR_TITLE_DEFAULT, PLATFORM_SCHEMA,
    BaseNotificationService)

_LOGGER = logging.getLogger(__name__)

CONF_DURATION = 'duration'
CONF_FONTSIZE = 'fontsize'
CONF_POSITION = 'position'
CONF_TRANSPARENCY = 'transparency'
CONF_COLOR = 'color'
CONF_INTERRUPT = 'interrupt'

DEFAULT_DURATION = 5
DEFAULT_FONTSIZE = 'medium'
DEFAULT_POSITION = 'bottom-right'
DEFAULT_TRANSPARENCY = 'default'
DEFAULT_COLOR = 'grey'
DEFAULT_INTERRUPT = False
DEFAULT_TIMEOUT = 5
DEFAULT_ICON = (
    'iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR4nGP6zwAAAgcBApo'
    'cMXEAAAAASUVORK5CYII=')

ATTR_DURATION = 'duration'
ATTR_FONTSIZE = 'fontsize'
ATTR_POSITION = 'position'
ATTR_TRANSPARENCY = 'transparency'
ATTR_COLOR = 'color'
ATTR_BKGCOLOR = 'bkgcolor'
ATTR_INTERRUPT = 'interrupt'
ATTR_IMAGE = 'filename2'
ATTR_FILE = 'file'
# Attributes contained in file
ATTR_FILE_URL = 'url'
ATTR_FILE_PATH = 'path'
ATTR_FILE_USERNAME = 'username'
ATTR_FILE_PASSWORD = 'password'
ATTR_FILE_AUTH = 'auth'
# Any other value or absence of 'auth' lead to basic authentication being used
ATTR_FILE_AUTH_DIGEST = 'digest'

FONTSIZES = {
    'small': 1,
    'medium': 0,
    'large': 2,
    'max': 3
}

POSITIONS = {
    'bottom-right': 0,
    'bottom-left': 1,
    'top-right': 2,
    'top-left': 3,
    'center': 4,
}

TRANSPARENCIES = {
    'default': 0,
    '0%': 1,
    '25%': 2,
    '50%': 3,
    '75%': 4,
    '100%': 5,
}

COLORS = {
    'grey': '#607d8b',
    'black': '#000000',
    'indigo': '#303F9F',
    'green': '#4CAF50',
    'red': '#F44336',
    'cyan': '#00BCD4',
    'teal': '#009688',
    'amber': '#FFC107',
    'pink': '#E91E63',
}

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Required(CONF_HOST): cv.string,
    vol.Optional(CONF_DURATION, default=DEFAULT_DURATION): vol.Coerce(int),
    vol.Optional(CONF_FONTSIZE, default=DEFAULT_FONTSIZE):
        vol.In(FONTSIZES.keys()),
    vol.Optional(CONF_POSITION, default=DEFAULT_POSITION):
        vol.In(POSITIONS.keys()),
    vol.Optional(CONF_TRANSPARENCY, default=DEFAULT_TRANSPARENCY):
        vol.In(TRANSPARENCIES.keys()),
    vol.Optional(CONF_COLOR, default=DEFAULT_COLOR):
        vol.In(COLORS.keys()),
    vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): vol.Coerce(int),
    vol.Optional(CONF_INTERRUPT, default=DEFAULT_INTERRUPT): cv.boolean,
})


def get_service(hass, config, discovery_info=None):
    """Get the Notifications for Android TV notification service."""
    remoteip = config.get(CONF_HOST)
    duration = config.get(CONF_DURATION)
    fontsize = config.get(CONF_FONTSIZE)
    position = config.get(CONF_POSITION)
    transparency = config.get(CONF_TRANSPARENCY)
    color = config.get(CONF_COLOR)
    interrupt = config.get(CONF_INTERRUPT)
    timeout = config.get(CONF_TIMEOUT)

    return NFAndroidTVNotificationService(
        remoteip, duration, fontsize, position,
        transparency, color, interrupt, timeout, hass.config.is_allowed_path)


class NFAndroidTVNotificationService(BaseNotificationService):
    """Notification service for Notifications for Android TV."""

    def __init__(self, remoteip, duration, fontsize, position, transparency,
                 color, interrupt, timeout, is_allowed_path):
        """Initialize the service."""
        self._target = 'http://{}:7676'.format(remoteip)
        self._default_duration = duration
        self._default_fontsize = fontsize
        self._default_position = position
        self._default_transparency = transparency
        self._default_color = color
        self._default_interrupt = interrupt
        self._timeout = timeout
        self._icon_file = io.BytesIO(base64.b64decode(DEFAULT_ICON))
        self.is_allowed_path = is_allowed_path

    def send_message(self, message="", **kwargs):
        """Send a message to a Android TV device."""
        _LOGGER.debug("Sending notification to: %s", self._target)

        payload = dict(filename=('icon.png', self._icon_file,
                                 'application/octet-stream',
                                 {'Expires': '0'}), type='0',
                       title=kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT),
                       msg=message, duration='%i' % self._default_duration,
                       fontsize='%i' % FONTSIZES.get(self._default_fontsize),
                       position='%i' % POSITIONS.get(self._default_position),
                       bkgcolor='%s' % COLORS.get(self._default_color),
                       transparency='%i' % TRANSPARENCIES.get(
                           self._default_transparency),
                       offset='0', app=ATTR_TITLE_DEFAULT, force='true',
                       interrupt='%i' % self._default_interrupt,)

        data = kwargs.get(ATTR_DATA)
        if data:
            if ATTR_DURATION in data:
                duration = data.get(ATTR_DURATION)
                try:
                    payload[ATTR_DURATION] = '%i' % int(duration)
                except ValueError:
                    _LOGGER.warning("Invalid duration-value: %s",
                                    str(duration))
            if ATTR_FONTSIZE in data:
                fontsize = data.get(ATTR_FONTSIZE)
                if fontsize in FONTSIZES:
                    payload[ATTR_FONTSIZE] = '%i' % FONTSIZES.get(fontsize)
                else:
                    _LOGGER.warning("Invalid fontsize-value: %s",
                                    str(fontsize))
            if ATTR_POSITION in data:
                position = data.get(ATTR_POSITION)
                if position in POSITIONS:
                    payload[ATTR_POSITION] = '%i' % POSITIONS.get(position)
                else:
                    _LOGGER.warning("Invalid position-value: %s",
                                    str(position))
            if ATTR_TRANSPARENCY in data:
                transparency = data.get(ATTR_TRANSPARENCY)
                if transparency in TRANSPARENCIES:
                    payload[ATTR_TRANSPARENCY] = '%i' % TRANSPARENCIES.get(
                        transparency)
                else:
                    _LOGGER.warning("Invalid transparency-value: %s",
                                    str(transparency))
            if ATTR_COLOR in data:
                color = data.get(ATTR_COLOR)
                if color in COLORS:
                    payload[ATTR_BKGCOLOR] = '%s' % COLORS.get(color)
                else:
                    _LOGGER.warning("Invalid color-value: %s", str(color))
            if ATTR_INTERRUPT in data:
                interrupt = data.get(ATTR_INTERRUPT)
                try:
                    payload[ATTR_INTERRUPT] = '%i' % cv.boolean(interrupt)
                except vol.Invalid:
                    _LOGGER.warning("Invalid interrupt-value: %s",
                                    str(interrupt))
            filedata = data.get(ATTR_FILE) if data else None
            if filedata is not None:
                # Load from file or URL
                file_as_bytes = self.load_file(
                    url=filedata.get(ATTR_FILE_URL),
                    local_path=filedata.get(ATTR_FILE_PATH),
                    username=filedata.get(ATTR_FILE_USERNAME),
                    password=filedata.get(ATTR_FILE_PASSWORD),
                    auth=filedata.get(ATTR_FILE_AUTH))
                if file_as_bytes:
                    payload[ATTR_IMAGE] = (
                        'image', file_as_bytes,
                        'application/octet-stream', {'Expires': '0'})

        try:
            _LOGGER.debug("Payload: %s", str(payload))
            response = requests.post(
                self._target, files=payload, timeout=self._timeout)
            if response.status_code != 200:
                _LOGGER.error("Error sending message: %s", str(response))
        except requests.exceptions.ConnectionError as err:
            _LOGGER.error("Error communicating with %s: %s",
                          self._target, str(err))

    def load_file(self, url=None, local_path=None, username=None,
                  password=None, auth=None):
        """Load image/document/etc from a local path or URL."""
        try:
            if url is not None:
                # Check whether authentication parameters are provided
                if username is not None and password is not None:
                    # Use digest or basic authentication
                    if ATTR_FILE_AUTH_DIGEST == auth:
                        auth_ = HTTPDigestAuth(username, password)
                    else:
                        auth_ = HTTPBasicAuth(username, password)
                    # Load file from URL with authentication
                    req = requests.get(
                        url, auth=auth_, timeout=DEFAULT_TIMEOUT)
                else:
                    # Load file from URL without authentication
                    req = requests.get(url, timeout=DEFAULT_TIMEOUT)
                return req.content

            if local_path is not None:
                # Check whether path is whitelisted in configuration.yaml
                if self.is_allowed_path(local_path):
                    return open(local_path, "rb")
                _LOGGER.warning("'%s' is not secure to load data from!",
                                local_path)
            else:
                _LOGGER.warning("Neither URL nor local path found in params!")

        except OSError as error:
            _LOGGER.error("Can't load from url or local path: %s", error)

        return None