tinytuya
tinytuya copied to clipboard
jasonacox
New Test Appender
Hello Jason,
Hope all is well! Not really an Issue...
I am just getting back into my Home automation add for Universal Devices Inc for TreatLife Devices. I built this test program from Wizard today for device separation based on DPS numbering using Pandas. I didn't need to save json i needed to "dumps" them and read them in the same function so i stripped that and still need to clean it up to insert into my Node Server work. I needed to be able to separate LED's and Switches so i made this to do so and now am going to implement it on UDI. Then I can finally stop copying json files for identity. Hopefully it is useful to you and others trying to identify Device Types on their networks.
Cheers on the module fantastic to work with, THANK YOU for all of your valuable time!
Regards, Steve
`import requests
import time
import hmac
import hashlib
import json
import pprint
import logging
import tinytuya
import pandas as pd
import numpy as np
###### Custom Parameters ######
new_sign_algorithm = True,
token = None,
headers = None,
body = None,
apiRegion = "us",
apiKey = "",
apiSecret = "",
uri = "token?grant_type = 1",
def tuyaPlatform(apiRegion, apiKey, apiSecret, uri, token=None, new_sign_algorithm=True, body=None, headers=None):
###### Authentacation ######
url = "https://openapi.tuya%s.com/v1.0/%s" % (apiRegion, uri)
now = int(time.time()*1000)
headers = dict(list(headers.items(
)) + [('Signature-Headers', ":".join(headers.keys()))]) if headers else {}
if (token == None):
payload = apiKey + str(now)
headers['secret'] = apiSecret
else:
payload = apiKey + token + str(now)
# If running the post 6-30-2021 signing algorithm update the payload to include it's data
if new_sign_algorithm:
payload += ('GET\n' + # HTTPMethod
# Content-SHA256
hashlib.sha256(bytes((body or "").encode('utf-8'))).hexdigest() + '\n' +
''.join(['%s:%s\n' % (key, headers[key]) # Headers
for key in headers.get("Signature-Headers", "").split(":")
if key in headers]) + '\n' +
'/' + url.split('//', 1)[-1].split('/', 1)[-1])
# Sign Payload
signature = hmac.new(
apiSecret.encode('utf-8'),
msg=payload.encode('utf-8'),
digestmod=hashlib.sha256
).hexdigest().upper()
# Create Header Data
headers['client_id'] = apiKey
headers['sign'] = signature
headers['t'] = str(now)
headers['sign_method'] = 'HMAC-SHA256'
if (token != None):
headers['access_token'] = token
# Get Token
response = requests.get(url, headers=headers)
# print(f"LINE 65 Token: {response}")
try:
response_dict = json.loads(response.content.decode())
except:
try:
response_dict = json.dumps(response.content)
# print(response_dict)
except:
print("Failed to get valid JSON response")
return (response_dict)
# Custom Parameters Credentials
###### TO TEST ADD YOUR CREDNTIALS HERE ######
config = {}
config['apiKey'] = ''
config['apiSecret'] = ''
config['apiRegion'] = 'us'
config['apiDeviceID'] = ''
needconfigs = False
if config['apiKey'] == '':
print('PLEASE ADD YOUR CREDENTIALS')
# First Message
print('LINE 80 TinyTuya Device Appender ' +
' [%s]' % (tinytuya.version))
print('')
if (config['apiKey'] != '' and config['apiSecret'] != '' and
config['apiRegion'] != '' and config['apiDeviceID'] != ''):
needconfigs = False
# Second Message
print(" " + "LINE 87 Existing settings:" +
"\n API Key=%s \n Secret=%s\n DeviceID=%s\n Region=%s" %
(config['apiKey'], config['apiSecret'], config['apiDeviceID'],
config['apiRegion']))
# Write Config
json_object = json.dumps(config, indent=4)
print(json_object)
KEY = config['apiKey']
SECRET = config['apiSecret']
DEVICEID = config['apiDeviceID']
REGION = config['apiRegion']
LANG = 'en' # us, eu, cn, in
# en or zh
# Get Oauth Token from tuyaPlatform
uri = 'token?grant_type=1'
response_dict = tuyaPlatform(REGION, KEY, SECRET, uri)
print(f"LINE 105 RESPONSE DICT ACCESS TOKEN : {response_dict}")
if not response_dict['success']:
print('\n\n' 'Error from Tuya server: ' + response_dict['msg'])
pass
token = response_dict['result']['access_token']
# Get UID from sample Device ID
uri = 'devices/%s' % DEVICEID
response_dict = tuyaPlatform(REGION, KEY, SECRET, uri, token)
if not response_dict['success']:
print('\n\n' + 'Error from Tuya server: ' + response_dict['msg'])
pass
uid = response_dict['result']['uid']
# Use UID to get list of all Devices for User
uri = 'users/%s/devices' % uid
json_data = tuyaPlatform(REGION, KEY, SECRET, uri, token)
# Here internet IP address everything
output = json.dumps(json_data, indent=4)
print(f"\nLINE 126 Future Cloud Control json: {output}\n")
# Filter to only Name, ID and Key
tuyadevices = []
for i in json_data['result']:
item = {}
item['name'] = i['name'].strip()
item['id'] = i['id']
item['key'] = i['local_key']
tuyadevices.append(item)
# Display device list
output = json.dumps(tuyadevices, indent=4) # sort_keys=True')
print("\n\n" + "LINE 137 Device Listing\n")
print(output)
# Find out if we should poll all devices REMOVE THIS
answer = 'Y' # input(subbold + '\nPoll local devices? ' +
# normal + '(Y/n): ')
if (answer[0:1].lower() != 'n'):
# Scan network for devices and provide polling data
print("\nLINE 147 Scanning local network for Tuya devices...")
devices = tinytuya.deviceScan(False, 20)
def getIP(d, gwid):
for ip in d:
if 'gwId' in d[ip]:
if (gwid == d[ip]['gwId']):
return (ip, d[ip]['version'])
return (0, 0)
polling = []
print("LINE 196 Polling local devices...")
for i in tuyadevices:
item = {}
name = i['name']
(ip, ver) = getIP(devices, i['id'])
item['name'] = name
item['ip'] = ip
item['ver'] = ver
item['id'] = i['id']
item['key'] = i['key']
if (ip == 0):
print(" %s[%s] - %s%s - %sError: No IP found%s" %
(name, ip, ))
else:
try:
d = tinytuya.OutletDevice(i['id'], ip, i['key'])
if ver == "3.3":
d.set_version(3.3)
data = d.status()
if 'dps' in data:
item['dps'] = data
state = "Off"
try:
if '1' in data['dps'] or '20' in data['dps']:
if '1' in data['dps']:
if data['dps']['1'] == True:
state = "On"
if '20' in data['dps']:
if data['dps']['20'] == True:
state = "On"
# print(" %s[%s] - %s%s - %s - DPS: %r" %
# (name, ip, state, data['dps']))
else:
# print(" %s[%s] - %s%s - DPS: %r" %
# (name, ip, data['dps']))
pass
except:
# print(" %s[%s] - %s%s - %sNo Response" %
# (name, ip))
pass
else:
# print(" %s[%s] - %s%s - %sNo Response" %
# (name, ip,))
pass
except:
# print(" %s[%s] - %s%s - %sNo Response" %
# (name, ip))
pass
polling.append(item)
# for loop
# Save polling data snapsot
current = {'timestamp': time.time(), 'devices': polling}
output1 = json.dumps(current, indent=4) # indent=4
# print(f"LINE 257 Local Device json {output1}")
print('\nThis is POLLED Add DPS sorter here\n')
df = pd.read_json(output1)
df = pd.json_normalize(df['devices']) # Works with Tuple Inserted
df = df.fillna(-1)
# print(f"LINE 219 DF: {df}")
df['type'] = None
try:
df['type'] = np.where(df['dps.dps.20'] != -1, 'light', df['type'])
except:
pass
try:
df['type'] = np.where(df['dps.dps.1'] != -1, 'switch', df['type'])
except:
pass
try:
df['type'] = np.where(df['dps.101'] != -1, 'tuya', df['type'])
except:
pass
lights = df[df['type'] == 'light'].reset_index(drop=True)
switches = df[df['type'] == 'switch'].reset_index(drop=True)
tuya = df[df['type'] == 'tuya'].reset_index(drop=True)
print(f"LINE 285 LIGHTS: {lights}")
print(f"LINE 286 SWITCHES: {switches}")
device_list = [lights]
for device in device_list:
for idx, row in device.iterrows():
name = row['name']
id = row['id']
id_new = id
ip = row['ip']
key = row['key']
ver = row['ver']
# id_new = id
address = row['type'] + '_%s' % (idx+1)
print('{name}\n{id_new}\n{ip}\n{key}\n{ver}\n{address}\n'.format(
name=name, id_new=id_new, ip=ip, key=key, ver=ver, address=address,))
# node = tuya_light_node.LightNode(
# self.poly, self.address, address, name, id_new, ip, key)
# self.poly.addNode(node)
# self.wait_for_node_event()
device_list = [switches]
for device in device_list:
for idx, row in device.iterrows():
name = row['name']
id = row['id']
id_new = id
ip = row['ip']
key = row['key']
ver = row['ver']
# id_new = id
address = row['type'] + '_%s' % (idx+1)
print('{name}\n{id_new}\n{ip}\n{key}\n{ver}\n{address}\n'.format(
name=name, id_new=id_new, ip=ip, key=key, ver=ver, address=address,))
# node = tuya_switch_node.SwitchNode(
# self.poly, self.address, address, name, id_new, ip, key)
# self.poly.addNode(node)
# self.wait_for_node_event()
device_list = [tuya]
for device in device_list:
for idx, row in device.iterrows():
name = row['name']
id = row['id']
id_new = id
ip = row['ip']
key = row['key']
ver = row['ver']
# id_new = id
address = row['type'] + '_%s' % (idx+1)
print('{name}\n{id_new}\n{ip}\n{key}\n{ver}\n{address}\n'.format(
name=name, id_new=id_new, ip=ip, key=key, ver=ver, address=address,))
# node = TuyaNode(
# self.poly, self.address, address, name, value)
# self.poly.addNode(node)
# self.wait_for_node_event()
print("\nDone.\n")`
https://user-images.githubusercontent.com/52646125/200663384-8604af9d-692e-48dc-9045-1bdcf8d1ee27.mov
Hello Jason! You're welcome!@ jamespauly has taken over the endeavor for UDI.i am now hopeful contributing to his efforts, openly on his repository.The best to you and Thank you for all of Your hard work!Highest Regards!On Nov 16, 2022, at 10:14 PM, Jason Cox @.***> wrote: Thanks @sjpbailey !
—Reply to this email directly, view it on GitHub, or unsubscribe.You are receiving this because you were mentioned.Message ID: @.***>