Domoticz

Code: Select all

#!/usr/bin/env python3

import sys
import time
import datetime
import json
import uuid
import os
import requests
from uuid import uuid4

ALEXA_URI = "https://api.eu.amazonalexa.com/v3/events"
CODE = "RHwmsruJWOeBfqHdFEfN"
CLIENT_ID = "the.client.id.from.the.permissions.for.send.alexa.events" 
CLIENT_SECRET = "the.client.secret.from.the.permissions.for.send.alexa.events"
REDIRECT_URI =  "https://www.amazon.com/ap/oa/?redirect_url=https://layla.amazon.com/api/skill/link/xxxxxxxxxxxxxxxxxx"

PREEMPTIVE_REFRESH_TTL_IN_SECONDS = 300

PATH = os.path.dirname(os.path.abspath(__file__))
if ("scripts" in PATH):
    TOKEN_FILENAME = "/home/phil/domoticz/scripts/" + CODE + ".txt"
else:
    TOKEN_FILENAME = CODE + ".txt"
UTC_FORMAT = "%Y-%m-%dT%H:%M:%S.00Z"
LWA_TOKEN_URI = "https://api.amazon.com/auth/o2/token"
LWA_HEADERS = {
    "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"}


def get_utc_timestamp(seconds=None):
    return time.strftime(UTC_FORMAT, time.gmtime(seconds))


def get_utc_timestamp_from_string(string):
    return datetime.datetime.strptime(string, UTC_FORMAT)


def get_uuid():
    return str(uuid.uuid4())


def get_need_new_token():
    """Checks whether the access token is missing or needed to be refreshed"""
    need_new_token_response = {
        "need_new_token": False,
        "access_token": "",
        "refresh_token": ""
    }

    if os.path.isfile(TOKEN_FILENAME):
        with open(TOKEN_FILENAME, 'r') as infile:
            last_line = infile.readlines()[-1]
        token = last_line.split("***")
        parts = len(token)
        # sometimes we get extra or zero LFs in the last line so compensate
        if parts == 3:
            real_last_line = last_line.split("}")[1] + "}"
            token = real_last_line.split("***")
        token_received_datetime = get_utc_timestamp_from_string(token[0])
        token_json = json.loads(token[1])
        token_expires_in = token_json["expires_in"] - \
            PREEMPTIVE_REFRESH_TTL_IN_SECONDS
        token_expires_datetime = token_received_datetime + \
            datetime.timedelta(seconds=token_expires_in)
        current_datetime = datetime.datetime.utcnow()

        need_new_token_response["need_new_token"] = current_datetime > token_expires_datetime
        need_new_token_response["access_token"] = token_json["access_token"]
        need_new_token_response["refresh_token"] = token_json["refresh_token"]
    else:
        need_new_token_response["need_new_token"] = True

    return need_new_token_response


def get_access_token():
    """Performs access token or token refresh request as needed and returns valid access token"""

    need_new_token_response = get_need_new_token()
    access_token = ""

    if need_new_token_response["need_new_token"]:
        if os.path.isfile(TOKEN_FILENAME):
            with open(TOKEN_FILENAME, 'a') as outfile:
                outfile.write("\n")

            lwa_params = {
                "grant_type": "refresh_token",
                "refresh_token": need_new_token_response["refresh_token"],
                "client_id": CLIENT_ID,
                "client_secret": CLIENT_SECRET,
                "redirect_uri": REDIRECT_URI
            }
        else:
            lwa_params = {
                "grant_type": "authorization_code",
                "code": CODE,
                "client_id": CLIENT_ID,
                "client_secret": CLIENT_SECRET,
                "redirect_uri": REDIRECT_URI
            }

        response = requests.post(
            LWA_TOKEN_URI, headers=LWA_HEADERS, data=lwa_params, allow_redirects=True)

        if response.status_code != 200:
            return None

        token = get_utc_timestamp() + "***" + response.text
        with open(TOKEN_FILENAME, 'a') as outfile:
            outfile.write(token)

        access_token = json.loads(response.text)["access_token"]
    else:
        access_token = need_new_token_response["access_token"]

    return access_token


def main(argv):
    token = get_access_token()

    # The full endpointId such as Contact-356 (basically, type and idx separated by hyphen)
    endpointId = argv[0]
    # The type of the sensor to report, e.g. ContactSensor or MotionSensor
    devType = argv[1]
    # This is DETECTED if the sensor is active/open and NOT_DETECTED if it is inactive/closed
    devValue = argv[2]

    data = {'event': {'header': {'namespace': 'Alexa', 'name': 'ChangeReport', 'messageId': str(uuid4()),
                                 'payloadVersion': '3'}, 'payload': {'change': {'cause': {'type': 'PHYSICAL_INTERACTION'},
                                                                                'properties': [{'name': 'connectivity', 'namespace': 'Alexa.EndpointHealth', 'value': {'value': 'OK'},
                                                                                                'timeOfSample': datetime.datetime.now().replace(microsecond=0).isoformat()+"Z", 'uncertaintyInMilliseconds': 100},
                                                                                               {'name': 'detectionState', 'namespace': 'Alexa.'+devType, 'value': devValue,
                                                                                                   'timeOfSample': datetime.datetime.now().replace(microsecond=0).isoformat()+"Z", 'uncertaintyInMilliseconds': 100}]}},
                      'endpoint': {
        'scope': {'type': 'BearerToken', 'token': token}, 'endpointId': endpointId}}, 'context': {}}

    headers = {'Content-Type': 'application/json'}
    headers['Authorization'] = 'Bearer ' + token
    response = requests.post(ALEXA_URI, headers=headers, json=data)
    print(data, response)


if __name__ == "__main__":
    main(sys.argv[1:])

Code: Select all

@INTERFACES.register('Alexa.EndpointHealth')
class AlexaEndpointHealth(AlexaInterface):
    def name(self):
        return 'Alexa.EndpointHealth'

    def propertiesSupported(self):
        return [{'name': 'connectivity'}]

Code: Select all

            if prop_name == 'connectivity':
                prop_value = { 'value': 'OK' }

Code: Select all

    def __init__(self, endpointId, friendlyName='', description='', manufacturerName=''):
        super().__init__(endpointId, friendlyName, description, manufacturerName)
        self.addCapability(AlexaEndpointHealth(self, 'Alexa.EndpointHealth',[{'name': 'connectivity'}]))
        self._device_ = None