ClusterM

TuyaAPI Client on Python

Dec 7th, 2021 (edited)
719
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 4.98 KB | None | 0 0
  1. import os
  2. import time
  3. import hashlib
  4. import hmac
  5. import json
  6. import requests
  7. import logging
  8.  
  9. logger = logging.getLogger()
  10.  
  11. class TuyaAPI:
  12.     def __init__(self, api_region, access_id, api_secret, token_cache=None):
  13.         self.api_region = api_region
  14.         self.access_id = access_id
  15.         self.api_secret = api_secret
  16.         self.token = None
  17.         self.token_cache = token_cache
  18.         if self.token_cache and os.path.isfile(self.token_cache):
  19.             with open(self.token_cache, 'r') as f:
  20.                 self.token = json.loads(f.read())
  21.                 logger.info(f"Tuya - loaded cached token from {self.token_cache}")
  22.  
  23.         api_region = api_region.lower()
  24.         if api_region == "ch":
  25.             self.urlhost = "openapi.tuyacn.com"      # China Data Center
  26.         elif api_region == "us":
  27.             self.urlhost = "openapi.tuyaus.com"      # Western America Data Center
  28.         elif api_region == "us-e":
  29.             self.urlhost = "openapi-ueaz.tuyaus.com" # Eastern America Data Center
  30.         elif api_region == "eu":
  31.             self.urlhost = "openapi.tuyaeu.com"      # Central Europe Data Center
  32.         elif api_region == "eu-w":
  33.             self.urlhost = "openapi-weaz.tuyaeu.com" # Western Europe Data Center
  34.         elif api_region == "in":
  35.             self.urlhost = "openapi.tuyain.com"      # India Datacenter
  36.         else:
  37.             raise NotImplementedError(f"Unknown region '{api_region}'")
  38.  
  39.     def request(self, method, uri, body=None, headers=None, no_token=False):
  40.         # Build URL
  41.         while uri.startswith("/"): uri = uri[1:]
  42.         url = "https://%s/%s" % (self.urlhost, uri)
  43.  
  44.         # Debugging stuff
  45.         if logging.DEBUG >= logger.level:
  46.             d = body
  47.             if body and type(body) == dict: d = json.dumps(body, sort_keys=False, indent=4)
  48.             logger.debug(f"Tuya - {method} request to {url}" + (f":\n{d}" if body else ""))
  49.  
  50.         # Check body
  51.         if body and type(body) == dict:
  52.             body = json.dumps(body)
  53.  
  54.         # Build Header
  55.         now = int(time.time() * 1000)
  56.         headers = dict(list(headers.items()) + [('Signature-Headers', ":".join(headers.keys()))]) if headers else {}
  57.         if no_token:
  58.             payload = self.access_id + str(now)
  59.             headers['secret'] = self.api_secret
  60.         else:
  61.             self.refresh_token()
  62.             payload = self.access_id + self.token["access_token"] + str(now)
  63.  
  64.         payload += (method + '\n' +                                                  # HTTPMethod
  65.             hashlib.sha256(bytes((body or "").encode('utf-8'))).hexdigest() + '\n' + # Content-SHA256
  66.             ''.join(['%s:%s\n'%(key, headers[key])                                   # Headers
  67.                      for key in headers.get("Signature-Headers", "").split(":")
  68.                      if key in headers]) + '\n' +
  69.             '/' + url.split('//', 1)[-1].split('/', 1)[-1])
  70.  
  71.         # Sign Payload
  72.         signature = hmac.new(
  73.             self.api_secret.encode('utf-8'),
  74.             msg=payload.encode('utf-8'),
  75.             digestmod=hashlib.sha256
  76.         ).hexdigest().upper()
  77.  
  78.         # Create Header Data
  79.         headers['client_id'] = self.access_id
  80.         headers['sign'] = signature
  81.         headers['t'] = str(now)
  82.         headers['sign_method'] = 'HMAC-SHA256'
  83.  
  84.         if not no_token:
  85.             headers['access_token'] = self.token["access_token"]
  86.         if method == "GET":
  87.             response = requests.get(url, headers=headers)
  88.         elif method == "POST":
  89.             response = requests.post(url, headers=headers, data=body)
  90.         elif method == "PUT":
  91.             response = requests.put(url, headers=headers, data=body)
  92.         elif method == "DELETE":
  93.             response = requests.delete(url, headers=headers)
  94.         else:
  95.             raise NotImplementedError(f"Unknown method '{method}'")
  96.  
  97.         response_dict = response.json()
  98.         # More debugging stuff
  99.         if logging.DEBUG >= logger.level:
  100.             logger.debug(f"Tuya - response:\n{json.dumps(response_dict, sort_keys=False, indent=4)}")
  101.         success = response_dict["success"]
  102.         if not success: raise TuyaError(response_dict["msg"])
  103.         return response_dict["result"]
  104.  
  105.     def save_token(self):
  106.         if self.token_cache:
  107.             with open(self.token_cache, 'w') as f:
  108.                 f.write(json.dumps(self.token))
  109.  
  110.     def refresh_token(self):
  111.         if self.token and time.time() < self.token["token_time"] + self.token["expire_time"]:
  112.             return
  113.         if not self.token:
  114.             logger.info(f"Tuya - token is empty, requesting new token...")
  115.         else:
  116.             logger.info(f"Tuya - token expired, requesting new one...")
  117.         uri = "v1.0/token?grant_type=1"
  118.         self.token = self.request("GET", uri, no_token=True)
  119.         self.token["token_time"] = time.time()
  120.         self.save_token()
  121.         logger.info(f"Tuya - new token received")
  122.  
  123. def TuyaError(Exception):
  124.     pass
  125.  
Add Comment
Please, Sign In to add comment