Cloud endpoint

tuya_cloud.py

import requests
import hashlib
import time
import uuid
import os
import copy
import json

# Fixed up version of my previous code to work with the Cloud endpoints.
# Hopefully this works.
# Had a quick look at their public cloud API implementation at:
# https://github.com/TuyaInc/TuyaDemo/
# to fix the issue.

## Use the region your device is registered in.

host = "https://a1.tuyaeu.com/api.json"

# Random and not really checked. Should keep persistent if you are using
# sessions.
device_id = os.urandom(32).encode('hex')
sid = ""
# Needs to be set, but they don't care what it is.
os = ("Linux", "0.1.2", "TEST")
# Your API keys.
appKey = "<BLANKED>"
appSecret = "<BLANKED>"

# This is the implementation of "sign".
def generate_request_sign(pairs):
    # This are the values that get "signed" in a request, worth checking if I
    # missed one in this list.
    values_to_hash = ["a", "v", "lat", "lon", "lang", "deviceId", "imei",
                        "imsi", "appVersion", "ttid", "isH5", "h5Token", "os",
                        "clientId", "postData", "time", "n4h5", "sid", "sp"]
    out = []
    sorted_pairs = sorted(pairs)
    for item in sorted_pairs:
        if item not in values_to_hash:
            continue
        if pairs[item] == "":
            continue
        out += [item + "=" + str(pairs[item])]
    sign_request = appSecret+"|".join(out)
    h = hashlib.md5()
    h.update(sign_request)
    return h.hexdigest()

# This will give you a dict with all the request parameters.
def url_generator(action, version, post_data=None, sid=None, time_param=None):
    client_id = appKey
    lang = "zh-Hans"
    if not time_param:
        time_param = int(time.time())
    request_id = uuid.uuid4()
    pairs = {
        "a": action,
        "deviceId": device_id,
        "os": os[0],
        "v": version,
        "clientId": client_id,
        "lang": lang,
        #"requestId": request_id,
        "time": time_param,
    }

    if sid:
        pairs['sid'] = sid
    if post_data:
        pairs['postData'] = post_data
    pairs['sign'] = generate_request_sign(pairs)
    return pairs

# Call the endpoint.
# * action is the name as defined in the tuya docs
# * version is the version they say to provide, normally "1.0"
# * data is the JSON data you want to pass to the action
# * requires_sid means it adds a session_id to the parameters. Used in mobile endpoints.
def preform_action(action, version, data=None, requires_sid=False):
    if requires_sid is True and not sid:
        return None
    params = url_generator(action, version, data, sid)
    print params
    endpoint = host
    headers = {
        # Maybe set a user agent or something here.
    }
    if data:
        r = requests.post(endpoint, params=params,
                          headers=headers)
    else:
        r = requests.post(endpoint, params=params, headers=headers)
    return r.status_code, r.json(), r.headers


if __name__ == "__main__":
    deviceID = "<DEVICE_ID_HERE>"
    print preform_action("tuya.p.weather.city.info.list", "1.0",
                         json.dumps({"countryCode":"CN"}))
    print preform_action("tuya.cloud.device.get", "1.0",
                         json.dumps({"devId": deviceID}))