Here are some updates from this saga.
I thought that the problem was being caused by copying and pasting the code from the webpage into tuya.py inside the studio code server addon.
Restored from a backup, pasted the code from the page into editpad, converted it to uft8 and pasted into configuration.yaml.
Here is the code:
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" + \
EMPTY_BODY + "\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
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" + \
EMPTY_BODY + "\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
EMPTY_BODY = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
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 = "my client id"
client_secret = "my client secret"
access_token = get_access_token()
attributes = get_device_properties(access_token, device_id)
json_output = json.dumps(attributes)
print(json_output)
Here is the command line sensor inside configuration.yaml:
command_line:
- sensor:
name: Medidor Energia Chuveiro Dependencia
unique_id: tuya_power_clamp
command: "python3 /config/tuya.py device_id"
device_class: power
state_class: MEASUREMENT
unit_of_measurement: W
scan_interval: 10
value_template: "{{ value_json.cur_power1 | float(0) / 10 }}"
json_attributes:
- sync_request
- sync_response
- device_state
- add_ele1
- cur_power1
- cur_current1
- cur_voltage1
- total_energy1
- today_acc_energy1
- power_type1
- warn_power1
- today_energy_add1
- net_state
And it worked! For half a day…
I did this this morning and when I went to HA, the error appeared again 1h ago according the logs.
I don’t think that the error is inside the python script… something tells me that the data received from tuya cloud might arrive in a way that is corrupting the core.device_registry. I don’t know… 
Any suggestions would be greatly appreciated.
thanks in advance!
