Tuya api script (for home assistant)

command_line.yaml

- sensor:
    name: Tuya Power Clamp
    unique_id: tuya_power_clamp
    command: "python3 /config/tuya.py device_id"
    device_class: power
    state_class: total_increasing
    unit_of_measurement: Wh
    scan_interval: 60
    value_template: "{{ value_json.EnergyConsumed | float(0) * 10 }}"

configuration.yaml

# put this on the end of file

command_line: !include command_line.yaml

tuya.py

import sys
import hashlib
import hmac
import json
import urllib
import urllib.parse
import logging

from urllib.request import urlopen, Request
from datetime import datetime

def make_request(url, params=None, headers=None):
    if params:
        url = url + "?" + urllib.parse.urlencode(params)
    request = Request(url, headers=headers or {})

    try:
        with urlopen(request, timeout=10) as response:
            return response, response.read().decode("utf-8")

    except Exception as error:
        return error, ""

def get_timestamp(now = datetime.now()):
    return str(int(datetime.timestamp(now)*1000))

def get_sign(payload, key):
    byte_key = bytes(key, 'UTF-8')
    message = payload.encode()
    sign = hmac.new(byte_key, message, hashlib.sha256).hexdigest()
    return sign.upper()

def get_access_token():
    now = datetime.now()

    timestamp = get_timestamp(now)
    string_to_sign = client_id + timestamp + "GET\n" + \
        "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855\n" + \
        "\n" + \
        LOGIN_URL
    signed_string = get_sign(string_to_sign, client_secret)

    headers = {
            "client_id": client_id,
            "sign": signed_string,
            "t": timestamp,
            "mode": "cors",
            "sign_method": "HMAC-SHA256",
            "Content-Type": "application/json"
            }

    response, body = make_request(BASE_URL + LOGIN_URL, headers = headers)

    json_result = json.loads(body)["result"]
    access_token = json_result["access_token"]
    return access_token

'''
Get the current attributes of device as hash. Example:
{
    "attribute1": value,
    "attribute2": value,
    "attribute3": value
}
'''
def get_device_properties(access_token, device_id):
    url = ATTRIBUTES_URL.format(device_id=device_id)
    timestamp = get_timestamp()
    string_to_sign = client_id + access_token + timestamp + "GET\n" + \
        "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855\n" + \
        "\n" + \
        url
    signed_string = get_sign(string_to_sign, client_secret)
    headers = {
            "client_id": client_id,
            "sign": signed_string,
            "access_token": access_token,
            "t": timestamp,
            "mode": "cors",
            "sign_method": "HMAC-SHA256",
            "Content-Type": "application/json"
            }

    response, body = make_request(BASE_URL + url, headers = headers)

    json_result = json.loads(body)
    properties = json_result["result"]["properties"]
    output = {j['code']: j['value'] for j in properties}

    return output

BASE_URL       = "https://openapi.tuyaus.com"
LOGIN_URL      = "/v1.0/token?grant_type=1"
ATTRIBUTES_URL = "/v2.0/cloud/thing/{device_id}/shadow/properties"

if len(sys.argv) != 2:
    raise SystemExit("usage: python3 tuya.py device_id")
_, device_id = sys.argv

client_id     = "tuya_client_id"
client_secret = "tuya_client_secret"
access_token  = get_access_token()

attributes  = get_device_properties(access_token, device_id)
json_output = json.dumps(attributes)
print(json_output)

Hi

Thank you. I have this working without issue.

Would it be possible to expand on it to allow reporting of all other attributes? Voltage, Current etc.

Regards

Hi,

what is e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 ?

https://developer.tuya.com/en/docs/iot/singnature?id=Ka43a5mtx1gsc#title-6-Signature%20algorithm%20for%20token%20management%20operations

Hello, sounds nice but is there a way to make it work on HAOS ?
-bash: python3: command not found
....

I found an alternative solution for python in:
#https://developer.tuya.com/en/docs/iot/device-control-best-practice-python?id=Kav4zc0nphsn5
#pip3 install tuya-connector-python

see the examples:

import logging
from tuya_connector import TuyaOpenAPI, TUYA_LOGGER

ACCESS_ID = "xtu7m*****48ufod"
ACCESS_KEY = "479bcba6d*******d9c4e080f7"
API_ENDPOINT = "https://openapi.tuyacn.com"

Enable debug log

TUYA_LOGGER.setLevel(logging.DEBUG)

Init OpenAPI and connect

openapi = TuyaOpenAPI(API_ENDPOINT, ACCESS_ID, ACCESS_KEY)
openapi.connect()

Set up device_id

DEVICE_ID ="vdevo********74966"

Call APIs from Tuya

Get the device information

response = openapi.get("/v1.0/iot-03/devices/{}".format(DEVICE_ID))

Get the instruction set of the device

response = openapi.get("/v1.0/iot-03/devices/{}/functions".format(DEVICE_ID))

Send commands

commands = {'commands': [{'code': 'switch_led', 'value': False}]}
openapi.post('/v1.0/iot-03/devices/{}/commands'.format(DEVICE_ID), commands)

Get the status of a single device

response = openapi.get("/v1.0/iot-03/devices/{}/status".format(DEVICE_ID))

Yes, it works with tuya-connector-python lib, but I wrote the script to work without any dependency, just placing the script on HA and running it

Good morning! I would appreciate it if you could point out the error. I configured the data exactly as you said, but I still can't use it. I don't know if I had a problem with the language translation or if I just made a mistake. Thanks!

response: the datacenter is suspended

<http.client.HTTPResponse object at 0x000002A20A525930>
body :
{"code":28841107,"msg":"No permission. The data center is suspended.Please go to the cloud development platform to enable the data center. ","success":false,"t":1736847424064,"tid":"1cd852a7d25b11ef9229e22b915a863a"}

same issue than here:
rospogrigio/localtuya#1719

Hi,

what is e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855 ?

@mateuszbrzezinski
This looks like the SHA-256 encoding of an of an empty string. Needed in the concat.

Here is a fork https://gist.github.com/shamasis/a71d0471f968550cfbf71551f304ae42 for seeing all sensors + the script uses URL for India data center - you can change the URL to the data center of your choosing.