272 lines
9.5 KiB
Python
272 lines
9.5 KiB
Python
"""Notifications for Android TV notification service."""
|
|
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 = f"http://{remoteip}:7676"
|
|
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
|