""" 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 logging import io import base64 import requests from requests.auth import HTTPBasicAuth from requests.auth import HTTPDigestAuth import voluptuous as vol from homeassistant.components.notify import ( ATTR_TITLE, ATTR_TITLE_DEFAULT, ATTR_DATA, BaseNotificationService, PLATFORM_SCHEMA) from homeassistant.const import CONF_TIMEOUT import homeassistant.helpers.config_validation as cv _LOGGER = logging.getLogger(__name__) CONF_IP = 'host' 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_IP): 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_IP) 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 elif 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