Html5 push notifications with image

Have anyone successfully sent notifications with images? It would be awesome to get a notification with a photo of my entrance when people ring my doorbell, with possible action of unlocking the door (or maybe activate the sprinkler :laughing: )

I see that the image property is in specification, but it is not in source code. I tried to add it in my virtual environment with no luck.

quick and dirty, you could add it as a data url that opens when you click it. I havenā€™t messed with it to much just yet but I know that works.

I setup html5 yesterday and have not got them working.

Canā€™t get timestamp working. All others work OK

EDIT:
I made changes below for items that are tested and work. Image requires create the custom component until support added into HA

  action:
    - service: notify.notify
      data:
        title: 'Title Displays in Message'
        message: 'actual message'
        target: # use device name from html_push_registration.conf
          - 'device1'
          - 'device2'
        data:
          tag: 'allows replacing message with new Message'
          url: 'https://site.com' #url to open when click message
          renotify: 0 (do not notify when replace message)/1(notify again)
          vibrate: #vibrate 300ms,pause100,vibrate 400
            - 300
            - 100
            - 400
          timestamp: #i don't know expected data
          image: "https://My.HAServer.com/api/camera_proxy/camera.MyCamera?api_password=MyHAPassword"
1 Like

After checking closer i dont see support for image in html5.py
I see nothing that look like it take image and add to notification but I admit I am no python guru.

https://github.com/home-assistant/home-assistant/blob/dev/homeassistant/components/notify/html5.py

closet thing I saw that worked was ā€œiconā€ but image size was tiny

maybe feature request unless someone has working example?

Yup, I mentioned that it was not in the code in the first post. Vibrate worked though - I tested it using a JSON array, in YAML I assume you have to send it as a list with dashes.

Simple fix

1.create folder /config/custom_components/notify
2.create file html5v2.py in /config/custom_components/notify
3.add code below to file. All I did was add ā€œimageā€ to line 95 of standard html5.py
4.change your configuration.yaml or notify.yaml,etc (if you split config files) Push component from html5 to html5v2
5 restart HomeAssistant (all automations will work as is and it will use existing html5_push_registration file)

FYI. for some reason I had to restart my phone to make it work

"""
HTML5 Push Messaging notification service.
For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/notify.html5/
"""
import asyncio
import os
import logging
import json
import time
import datetime
import uuid

import voluptuous as vol
from voluptuous.humanize import humanize_error

from homeassistant.const import (HTTP_BAD_REQUEST, HTTP_INTERNAL_SERVER_ERROR,
                                 HTTP_UNAUTHORIZED, URL_ROOT)
from homeassistant.util import ensure_unique_string
from homeassistant.components.notify import (
    ATTR_TARGET, ATTR_TITLE, ATTR_TITLE_DEFAULT, ATTR_DATA,
    BaseNotificationService, PLATFORM_SCHEMA)
from homeassistant.components.http import HomeAssistantView
from homeassistant.components.frontend import add_manifest_json_key
from homeassistant.helpers import config_validation as cv

REQUIREMENTS = ['pywebpush==1.1.0', 'PyJWT==1.5.3']

DEPENDENCIES = ['frontend']

_LOGGER = logging.getLogger(__name__)

REGISTRATIONS_FILE = 'html5_push_registrations.conf'

ATTR_GCM_SENDER_ID = 'gcm_sender_id'
ATTR_GCM_API_KEY = 'gcm_api_key'

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Optional(ATTR_GCM_SENDER_ID): cv.string,
    vol.Optional(ATTR_GCM_API_KEY): cv.string,
})

ATTR_SUBSCRIPTION = 'subscription'
ATTR_BROWSER = 'browser'

ATTR_ENDPOINT = 'endpoint'
ATTR_KEYS = 'keys'
ATTR_AUTH = 'auth'
ATTR_P256DH = 'p256dh'
ATTR_EXPIRATIONTIME = 'expirationTime'

ATTR_TAG = 'tag'
ATTR_ACTION = 'action'
ATTR_ACTIONS = 'actions'
ATTR_TYPE = 'type'
ATTR_URL = 'url'

ATTR_JWT = 'jwt'

# The number of days after the moment a notification is sent that a JWT
# is valid.
JWT_VALID_DAYS = 7

KEYS_SCHEMA = vol.All(dict,
                      vol.Schema({
                          vol.Required(ATTR_AUTH): cv.string,
                          vol.Required(ATTR_P256DH): cv.string
                          }))

SUBSCRIPTION_SCHEMA = vol.All(dict,
                              vol.Schema({
                                  # pylint: disable=no-value-for-parameter
                                  vol.Required(ATTR_ENDPOINT): vol.Url(),
                                  vol.Required(ATTR_KEYS): KEYS_SCHEMA,
                                  vol.Optional(ATTR_EXPIRATIONTIME):
                                      vol.Any(None, cv.positive_int)
                                  }))

REGISTER_SCHEMA = vol.Schema({
    vol.Required(ATTR_SUBSCRIPTION): SUBSCRIPTION_SCHEMA,
    vol.Required(ATTR_BROWSER): vol.In(['chrome', 'firefox'])
})

CALLBACK_EVENT_PAYLOAD_SCHEMA = vol.Schema({
    vol.Required(ATTR_TAG): cv.string,
    vol.Required(ATTR_TYPE): vol.In(['received', 'clicked', 'closed']),
    vol.Required(ATTR_TARGET): cv.string,
    vol.Optional(ATTR_ACTION): cv.string,
    vol.Optional(ATTR_DATA): dict,
})

NOTIFY_CALLBACK_EVENT = 'html5_notification'

# Badge and timestamp are Chrome specific (not in official spec)
HTML5_SHOWNOTIFICATION_PARAMETERS = (
    'actions', 'badge', 'body', 'dir', 'icon', 'image', 'lang', 'renotify',
    'requireInteraction', 'tag', 'timestamp', 'vibrate')


def get_service(hass, config, discovery_info=None):
    """Get the HTML5 push notification service."""
    json_path = hass.config.path(REGISTRATIONS_FILE)

    registrations = _load_config(json_path)

    if registrations is None:
        return None

    hass.http.register_view(
        HTML5PushRegistrationView(registrations, json_path))
    hass.http.register_view(HTML5PushCallbackView(registrations))

    gcm_api_key = config.get(ATTR_GCM_API_KEY)
    gcm_sender_id = config.get(ATTR_GCM_SENDER_ID)

    if gcm_sender_id is not None:
        add_manifest_json_key(
            ATTR_GCM_SENDER_ID, config.get(ATTR_GCM_SENDER_ID))

    return HTML5NotificationService(gcm_api_key, registrations, json_path)


def _load_config(filename):
    """Load configuration."""
    if not os.path.isfile(filename):
        return {}

    try:
        with open(filename, 'r') as fdesc:
            inp = fdesc.read()

        # In case empty file
        if not inp:
            return {}

        return json.loads(inp)
    except (IOError, ValueError) as error:
        _LOGGER.error("Reading config file %s failed: %s", filename, error)
        return None


class JSONBytesDecoder(json.JSONEncoder):
    """JSONEncoder to decode bytes objects to unicode."""

    # pylint: disable=method-hidden
    def default(self, obj):
        """Decode object if it's a bytes object, else defer to baseclass."""
        if isinstance(obj, bytes):
            return obj.decode()
        return json.JSONEncoder.default(self, obj)


def _save_config(filename, config):
    """Save configuration."""
    try:
        with open(filename, 'w') as fdesc:
            fdesc.write(json.dumps(
                config, cls=JSONBytesDecoder, indent=4, sort_keys=True))
    except (IOError, TypeError) as error:
        _LOGGER.error("Saving config file failed: %s", error)
        return False
    return True


class HTML5PushRegistrationView(HomeAssistantView):
    """Accepts push registrations from a browser."""

    url = '/api/notify.html5'
    name = 'api:notify.html5'

    def __init__(self, registrations, json_path):
        """Init HTML5PushRegistrationView."""
        self.registrations = registrations
        self.json_path = json_path

    @asyncio.coroutine
    def post(self, request):
        """Accept the POST request for push registrations from a browser."""
        try:
            data = yield from request.json()
        except ValueError:
            return self.json_message('Invalid JSON', HTTP_BAD_REQUEST)

        try:
            data = REGISTER_SCHEMA(data)
        except vol.Invalid as ex:
            return self.json_message(
                humanize_error(data, ex), HTTP_BAD_REQUEST)

        name = ensure_unique_string('unnamed device', self.registrations)

        self.registrations[name] = data

        if not _save_config(self.json_path, self.registrations):
            return self.json_message(
                'Error saving registration.', HTTP_INTERNAL_SERVER_ERROR)

        return self.json_message('Push notification subscriber registered.')

    @asyncio.coroutine
    def delete(self, request):
        """Delete a registration."""
        try:
            data = yield from request.json()
        except ValueError:
            return self.json_message('Invalid JSON', HTTP_BAD_REQUEST)

        subscription = data.get(ATTR_SUBSCRIPTION)

        found = None

        for key, registration in self.registrations.items():
            if registration.get(ATTR_SUBSCRIPTION) == subscription:
                found = key
                break

        if not found:
            # If not found, unregistering was already done. Return 200
            return self.json_message('Registration not found.')

        reg = self.registrations.pop(found)

        if not _save_config(self.json_path, self.registrations):
            self.registrations[found] = reg
            return self.json_message(
                'Error saving registration.', HTTP_INTERNAL_SERVER_ERROR)

        return self.json_message('Push notification subscriber unregistered.')


class HTML5PushCallbackView(HomeAssistantView):
    """Accepts push registrations from a browser."""

    requires_auth = False
    url = '/api/notify.html5/callback'
    name = 'api:notify.html5/callback'

    def __init__(self, registrations):
        """Init HTML5PushCallbackView."""
        self.registrations = registrations

    def decode_jwt(self, token):
        """Find the registration that signed this JWT and return it."""
        import jwt

        # 1.  Check claims w/o verifying to see if a target is in there.
        # 2.  If target in claims, attempt to verify against the given name.
        # 2a. If decode is successful, return the payload.
        # 2b. If decode is unsuccessful, return a 401.

        target_check = jwt.decode(token, options={'verify_signature': False})
        if target_check[ATTR_TARGET] in self.registrations:
            possible_target = self.registrations[target_check[ATTR_TARGET]]
            key = possible_target[ATTR_SUBSCRIPTION][ATTR_KEYS][ATTR_AUTH]
            try:
                return jwt.decode(token, key)
            except jwt.exceptions.DecodeError:
                pass

        return self.json_message('No target found in JWT',
                                 status_code=HTTP_UNAUTHORIZED)

    # The following is based on code from Auth0
    # https://auth0.com/docs/quickstart/backend/python
    def check_authorization_header(self, request):
        """Check the authorization header."""
        import jwt
        auth = request.headers.get('Authorization', None)
        if not auth:
            return self.json_message('Authorization header is expected',
                                     status_code=HTTP_UNAUTHORIZED)

        parts = auth.split()

        if parts[0].lower() != 'bearer':
            return self.json_message('Authorization header must '
                                     'start with Bearer',
                                     status_code=HTTP_UNAUTHORIZED)
        elif len(parts) != 2:
            return self.json_message('Authorization header must '
                                     'be Bearer token',
                                     status_code=HTTP_UNAUTHORIZED)

        token = parts[1]
        try:
            payload = self.decode_jwt(token)
        except jwt.exceptions.InvalidTokenError:
            return self.json_message('token is invalid',
                                     status_code=HTTP_UNAUTHORIZED)
        return payload

    @asyncio.coroutine
    def post(self, request):
        """Accept the POST request for push registrations event callback."""
        auth_check = self.check_authorization_header(request)
        if not isinstance(auth_check, dict):
            return auth_check

        try:
            data = yield from request.json()
        except ValueError:
            return self.json_message('Invalid JSON', HTTP_BAD_REQUEST)

        event_payload = {
            ATTR_TAG: data.get(ATTR_TAG),
            ATTR_TYPE: data[ATTR_TYPE],
            ATTR_TARGET: auth_check[ATTR_TARGET],
        }

        if data.get(ATTR_ACTION) is not None:
            event_payload[ATTR_ACTION] = data.get(ATTR_ACTION)

        if data.get(ATTR_DATA) is not None:
            event_payload[ATTR_DATA] = data.get(ATTR_DATA)

        try:
            event_payload = CALLBACK_EVENT_PAYLOAD_SCHEMA(event_payload)
        except vol.Invalid as ex:
            _LOGGER.warning("Callback event payload is not valid: %s",
                            humanize_error(event_payload, ex))

        event_name = '{}.{}'.format(NOTIFY_CALLBACK_EVENT,
                                    event_payload[ATTR_TYPE])
        request.app['hass'].bus.fire(event_name, event_payload)
        return self.json({'status': 'ok',
                          'event': event_payload[ATTR_TYPE]})


class HTML5NotificationService(BaseNotificationService):
    """Implement the notification service for HTML5."""

    def __init__(self, gcm_key, registrations, json_path):
        """Initialize the service."""
        self._gcm_key = gcm_key
        self.registrations = registrations
        self.registrations_json_path = json_path

    @property
    def targets(self):
        """Return a dictionary of registered targets."""
        targets = {}
        for registration in self.registrations:
            targets[registration] = registration
        return targets

    def send_message(self, message="", **kwargs):
        """Send a message to a user."""
        import jwt
        from pywebpush import WebPusher

        timestamp = int(time.time())
        tag = str(uuid.uuid4())

        payload = {
            'badge': '/static/images/notification-badge.png',
            'body': message,
            ATTR_DATA: {},
            'icon': '/static/icons/favicon-192x192.png',
            ATTR_TAG: tag,
            'timestamp': (timestamp*1000),  # Javascript ms since epoch
            ATTR_TITLE: kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
        }

        data = kwargs.get(ATTR_DATA)

        if data:
            # Pick out fields that should go into the notification directly vs
            # into the notification data dictionary.

            data_tmp = {}

            for key, val in data.items():
                if key in HTML5_SHOWNOTIFICATION_PARAMETERS:
                    payload[key] = val
                else:
                    data_tmp[key] = val

            payload[ATTR_DATA] = data_tmp

        if (payload[ATTR_DATA].get(ATTR_URL) is None and
                payload.get(ATTR_ACTIONS) is None):
            payload[ATTR_DATA][ATTR_URL] = URL_ROOT

        targets = kwargs.get(ATTR_TARGET)

        if not targets:
            targets = self.registrations.keys()

        for target in list(targets):
            info = self.registrations.get(target)
            if info is None:
                _LOGGER.error("%s is not a valid HTML5 push notification"
                              " target", target)
                continue

            jwt_exp = (datetime.datetime.fromtimestamp(timestamp) +
                       datetime.timedelta(days=JWT_VALID_DAYS))
            jwt_secret = info[ATTR_SUBSCRIPTION][ATTR_KEYS][ATTR_AUTH]
            jwt_claims = {'exp': jwt_exp, 'nbf': timestamp,
                          'iat': timestamp, ATTR_TARGET: target,
                          ATTR_TAG: payload[ATTR_TAG]}
            jwt_token = jwt.encode(jwt_claims, jwt_secret).decode('utf-8')
            payload[ATTR_DATA][ATTR_JWT] = jwt_token

            response = WebPusher(info[ATTR_SUBSCRIPTION]).send(
                json.dumps(payload), gcm_key=self._gcm_key, ttl='86400')

            # pylint: disable=no-member
            if response.status_code == 410:
                _LOGGER.info("Notification channel has expired")
                reg = self.registrations.pop(target)
                if not _save_config(self.registrations_json_path,
                                    self.registrations):
                    self.registrations[target] = reg
                    _LOGGER.error("Error saving registration.")
                else:
                    _LOGGER.info("Configuration saved")
1 Like

FYI

https://github.com/home-assistant/home-assistant/issues/9832

support will probably be added by next release (after 0.55.0)

1 Like

Great! As explained in the first post, I already tried to do this fix locally, but I did not delete the .pyc file. That might by why it didnā€™t work.

Do you know when it will be released? Iā€™m on v0.56.2 now

It was released in v0.56 :slight_smile: What have you tried?

- service: notify.google
  data:
    message: 'This is a test message'
    data:
      attachment:
        type: 'image'
        payload: http://user:password@IP-adress/cgi-bin/snapshot.cgi?1
    target:
      - '+31612345678'

This is my automation.yaml
When I manually try to send a message it works (but without snapshot).

Check documentation. I think it should be:

data:
  message: 'This is a test message'
  data:
    image: 'http://....'

No succes unfortunately.
I checked the documentation (HTML5 Push Notifications - Home Assistant) but Iā€™m still not getting any further.

Try calling the service with this data. Also note, on my phone I have to expand the notification by dragging down on it to show the image.

{
  "message": "Message",
  "data": {
"body": "Body",
"image": "https://i.pinimg.com/736x/9a/50/78/9a5078871bd0972ba625b9ec65b36aa4--steel-exterior-doors-steel-doors.jpg",
    "actions": [
      {
        "action": "open",
        "icon": "/static/icons/favicon-192x192.png",
        "title": "Open Home Assistant"
      },
      {
        "action": "open_door",
        "title": "Open door"
      }
    ]
  }
}

The image you are using will not work in my experience.
try using actual file ending .jpg, .png, etc

Service should be ā€œnotify.notifyā€ and ā€œpayload not neededā€

- service: notify.notify
  data:
    message: 'This is a test message'
    target:
       - '+31612345678' #this is name from your html_push_registration.conf
    data:
      image: 'http://user:password@IP-adress/cgi-bin/snapshot.cgi?1' #this file type wont work

i documented what worked and was tested by me in my first post

thnx tmjpugh, I think the problem is indeed the file type. But that is the url of my IP camera. Is there a way to work around this file type?

some camera may provide direct link to file location and this can be used.
OR
create an automation that saves the image file to static location.
You may combine this with this notify automation and call SaveImage.script service before notify.notify service for example

I think save automation is enough to give image in notification. Then use ā€œurlā€ so that you may click notification and be taken to camera live video, camera feed in HA or camera feed on NVR

SUCCESS
must use HA api to send the image from camera in HA
below is working for me
When Gate open I receive notification with image from my camera

- alias: 'Notify Gate Closed'
  trigger:
    - platform: state
      entity_id: binary_sensor.entry_gate
      from: 'on'
      to: 'off'
  action:
    - service: notify.notify
      data:
        title: 'GATE'
        message: 'Closed'
        data:
          tag: alert
          url: 'https://my_HAserver.com'
          image: "https://my_HAserver.com/api/camera_proxy/camera.mycamera?api_password=MyHAPassword"
          vibrate:
            - 300
            - 100
            - 400
          renotify: 0

per documenation the api outputs an image by my understanding

3 Likes

I canā€™t get vibrate to work, is that not supported per default? In my log I get:
Invalid service data for notify.notification: extra keys not allowed @ data[ā€˜vibrateā€™]

How do I get my notifications to work with vibrate? :slight_smile:

Never mind, it was my action that was incorrect. It works now!