jw)ndZddlmZddlZddlZ ddlmZdZexsejjd Z dd l m Z  eZered Zd ezxZZd Zej*eZej0dd k(ZddZdZddZdZdZdZdZ ddZ!ddZ"ddZ#y#e$rdZYwxYw#e$rYpwxYw)a Python module to interface with Tuya WiFi smart devices Author: Jason A. Cox For more information see https://github.com/jasonacox/tinytuya Core Classes and Helper Functions Classes * AESCipher - Cryptography Helpers * XenonDevice(...) - Base Tuya Objects and Functions XenonDevice(dev_id, address=None, local_key="", dev_type="default", connection_timeout=5, version="3.1", persist=False, cid/node_id=None, parent=None, connection_retry_limit=5, connection_retry_delay=5, max_simultaneous_dps=0) * Device(XenonDevice) - Tuya Class for Devices Module Functions set_debug(toggle, color) # Activate verbose debugging output pack_message(msg, hmac_key=None) # Packs a TuyaMessage() into a network packet, encrypting or adding a CRC if protocol requires unpack_message(data, hmac_key=None, header=None, no_retcode=False) # Unpacks a TuyaMessage() from a network packet, decrypting or checking the CRC if protocol requires parse_header(data) # Unpacks just the header part of a message into a TuyaHeader() find_device(dev_id=None, address=None) # Scans network for Tuya devices with either ID = dev_id or IP = address device_info(dev_id) # Searches DEVICEFILE (usually devices.json) for devices with ID = dev_id and returns just that device assign_dp_mappings(tuyadevices, mappings) # Adds mappings to all the devices in the tuyadevices list decrypt_udp(msg) # Decrypts a UDP network broadcast packet merge_dps_results(dest, src) # Merge multiple receive() responses into a single dict # `src` will be combined with and merged into `dest` Device Functions json = status() # returns json payload json = cached_status(historic=False, nowait=False) # When a persistent connection is open, this will return a cached version of the device status # if historic=True, all seen DPs are returned even if their values might be out of date # if historic=False, only DPs which are current are returned # if nowait=False (the default), a status() call will be made if no cached status is available. # if nowait=True, `None` will be returned immediately if no cached status is available. cache_clear() # Clears the cache, causing cached_status() to either call status() or return None subdev_query(nowait) # query sub-device status (only for gateway devices) set_version(version) # 3.1 [default], 3.2, 3.3 or 3.4 set_socketPersistent(False/True) # False [default] or True set_socketNODELAY(False/True) # False or True [default] set_socketRetryLimit(integer) # retry count limit [default 5] set_socketRetryDelay(integer) # retry delay [default 5] set_socketTimeout(timeout) # set connection timeout in seconds [default 5] set_dpsUsed(dps_to_request) # add data points (DPS) to request add_dps_to_request(index) # add data point (DPS) index set to None set_retry(retry=True) # retry if response payload is truncated set_status(on, switch=1, nowait) # Set status of switch to 'on' or 'off' (bool) set_value(index, value, nowait) # Set int value of any index. set_multiple_values(index_value_dict, nowait) # Set multiple values with a single request heartbeat(nowait) # Send heartbeat to device updatedps(index=[1], nowait) # Send updatedps command to device turn_on(switch=1, nowait) # Turn on device / switch # turn_off(switch=1, nowait) # Turn off set_timer(num_secs, nowait) # Set timer for num_secs set_sendWait(num_secs) # Time to wait after sending commands before pulling response detect_available_dps() # Return list of DPS available from device generate_payload(command, data,...)# Generate TuyaMessage payload for command with data send(payload) # Send payload to device (do not wait for response) receive() # Receive payload from device Credits * TuyaAPI https://github.com/codetheweb/tuyapi by codetheweb and blackrozes For protocol reverse engineering * PyTuya https://github.com/clach04/python-tuya by clach04 The origin of this python module (now abandoned) * LocalTuya https://github.com/rospogrigio/localtuya-homeassistant by rospogrigio Updated pytuya to support devices with Device IDs of 22 characters * Tuya Protocol 3.4 and 3.5 Support by uzlonewolf Enhancement to TuyaMessage logic for multi-payload messages )print_functionN)initTFwin) AESCipher)rrz%d.%d.%d jasonacoxc|rdndtrdjfd|D}|Sdjfd|D}|S)N c3>K|]}dt|fzywz%02X%sN)ord.0yspaces ?/DATA/.local/lib/python3.12/site-packages/tinytuya/core/core.py zbin2hex..zs?QSVUO3Qsc3,K|] }d|fz ywrrs rrzbin2hex..|s:1QJ.s)IS_PY2join)xprettyresultrs @rbin2hexrtsI  ?Q?? M::: McZtr|jdStj|S)Nhex)rdecodebytesfromhex)rs rhex2binr%s" xx}}Qrc|xrt}|rJ|r&tjdtjn%tjdtjtj tjtj dttj dtjtjtjsItj dtjtjtj dy tj dtjtjy tj tjy ) zEnable tinytuya verbose loggingz$%(levelname)s:%(message)s)formatlevelz%(levelname)s:%(message)szTinyTuya [%s] zPython %s on %szUsing %s %s for cryptozMWarning: Crypto library does not support AES-GCM, v3.5 devices will not work!z(Using %s %s for crypto, GCM is supportedN) HAVE_COLORlogging basicConfigDEBUGlogsetLeveldebug __version__sysversionplatformrCRYPTOLIB_HAS_GCM CRYPTOLIB CRYPTOLIB_VERNOTSET)togglecolors r set_debugr:s  jE    C7==    'B'-- X W]]# #[1 #S[[#,,?** II. 0C0CYE\E\ ] IIe f II@)BUBUW`WnWn o W^^$rct|tk7r td|r|sy|D]5} |d}|d}||vr |||d<tj d|d|d<7y#tj d|YSxYw)z Adds mappings to all the devices in the tuyadevices list Parameters: tuyadevices = list of devices mappings = dict containing the mappings Response: Nothing, modifies tuyadevices in place z'mappings' must be a dictNid product_idz9Cannot add DP mapping, no device id and/or product id: %rmappingzDevice %s has no mapping!)typedict ValueErrorr-r/) tuyadevicesmappingsdevdevid productids rassign_dp_mappingsrGs H~799 k IEL)I  %i0C N II2E ;!C N  IIRTW Y s AA8c`|dt|dzz tdt|dzz zzS)N)lenchrss rpadrNs1 SVb[ CSVb[(8$99 99rc>|dt|t|dz d S)Nr)rrJrLs runpadrPs& $Ac!fqjlO$$ %%rc$|d|vry|||d<y)NipTFr) newdevicedevicess r appenddevicerUs"'!(GIdO rc ~|xrt}|durdx}x}x}x}x}x}x}x}} nd}d}d}d}d}d}d }d }d } ||||||||| f S) NFr z z z z z z z z z )r)) r9boldsubboldnormaldimalertalertdimcyanredyellows r termcolorr`s  jE ~QSSSwSS#SSS4S#(#"&(+ " s5$s6 AArc8ddlm}|j|||y)z9Scans your network for Tuya devices with output to stdoutr scanner)scantimer9 forcescanN)r rcscan)maxretryr9rercs rrfrfs LL(%9LErc<ddlm}|j||||||S)akScans your network for Tuya devices and returns dictionary of devices discovered devices = tinytuya.deviceScan(verbose) Parameters: verbose = True or False, print formatted output to stdout [Default: False] maxretry = The number of loops to wait to pick up UDP from all devices color = True or False, print output in color [Default: True] poll = True or False, poll dps status for devices if possible forcescan = True or False, force network scan for device IP addresses Response: devices = Dictionary of all devices found To unpack data, you can do something like this: devices = tinytuya.deviceScan() for ip in devices: id = devices[ip]['gwId'] key = devices[ip]['productKey'] vers = devices[ip]['version'] dps = devices[ip]['dps'] r rb)verboserdr9pollrebyID)r rcrT)rirgr9rjrerkrcs r deviceScanrls%0 ??7XUQUajqu? vvr)F)TT)T)NTF)FNTTFF)$__doc__ __future__rr*r1coloramar HAVE_COLORAMA ImportErrorr3 startswithr) crypto_helperr raw_inputinput NameError version_tupler2r0 __author__ getLogger__name__r- version_inforrr%r:rGrNrPrUr`rfrlrrrr|sIX& M @#,,"9"9%"@@ $ E F "]22+  g!  !  !  %,"N:&B(FwkM  s#BB,B)(B),B43B4