Domoticz

Code: Select all

    class Authorization(AlexaSmartHomeCall):

        def AcceptGrant(self, directive):
            print('AcceptGrant: ', directive)
            try:
                if 'grantee' in directive['payload']:
                    self.handler.setAccessToken(directive['payload']['grantee']['token'])
            except Exception: pass
            response = {
                'event': {
                    'header': {
                        'namespace': 'Alexa.Authorization',
                        'name': 'AcceptGrant.Response',
                        'payloadVersion': '3',
                        'messageId': str(uuid4())
                    },
                    'payload': {}
                }
            }
            return response

Code: Select all

#!/usr/bin/env python3
# (c) philchillbill, 2019. No rights reserved.

import sys, time, datetime, json, uuid, os
import requests
import AlexaSmartHome, DomoticzHandler

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"
LOCAL_DOMOTICZ_IP = "http://192.168.178.12:8080"

PREEMPTIVE_REFRESH_TTL_IN_SECONDS = 300
PATH = os.path.dirname(os.path.abspath(__file__))
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()
    device_id = argv[0]
    portal = DomoticzHandler.Domoticz(LOCAL_DOMOTICZ_IP)
    
    if device_id != '':
        device = portal.getDevice(device_id)
        endpoint = portal.getEndpointDevice(device)     
        data = AlexaSmartHome.api_reportState(endpoint, token)
        headers = { 'Content-Type': 'application/json' }
        headers['Authorization'] = 'Bearer ' + token
        response = requests.post(ALEXA_URI, headers=headers, json=data)
        print(response)
        

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