original_docstring: Send a message to a user.
docstring: Send a message to a user.
docstring_tokens: Send a message to a user .
code: 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
  
              # Only pass the gcm key if we're actually using GCM
              # If we don't, notifications break on FireFox
              gcm_key = self._gcm_key \
                  if 'googleapis.com' in info[ATTR_SUBSCRIPTION][ATTR_ENDPOINT] \
                  else None
              response = WebPusher(info[ATTR_SUBSCRIPTION]).send(
                  json.dumps(payload), gcm_key=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_json(self.registrations_json_path,
                                   self.registrations):
                      self.registrations[target] = reg
                      _LOGGER.error("Error saving registration")
                  else:
                      _LOGGER.info("Configuration saved")
code_tokens: def send_message ( self , message = "" , ** kwargs ) : 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 ) , ATTR_TITLE : kwargs . get ( ATTR_TITLE , ATTR_TITLE_DEFAULT ) } data = kwargs . get ( ATTR_DATA ) if data : 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 gcm_key = self . _gcm_key if 'googleapis.com' in info [ ATTR_SUBSCRIPTION ] [ ATTR_ENDPOINT ] else None response = WebPusher ( info [ ATTR_SUBSCRIPTION ] ) . send ( json . dumps ( payload ) , gcm_key = gcm_key , ttl = '86400' ) if response . status_code == 410 : _LOGGER . info ( "Notification channel has expired" ) reg = self . registrations . pop ( target ) if not save_json ( self . registrations_json_path , self . registrations ) : self . registrations [ target ] = reg _LOGGER . error ( "Error saving registration" ) else : _LOGGER . info ( "Configuration saved" )
short_docstring: Send a message to a user.
short_docstring_tokens: Send a message to a user .
comment: """Send a message to a user."""$SEP$# Javascript ms since epoch$SEP$# Pick out fields that should go into the notification directly vs$SEP$# into the notification data dictionary.$SEP$# Only pass the gcm key if we're actually using GCM$SEP$# If we don't, notifications break on FireFox$SEP$# pylint: disable=no-member