"""Tuya OpenAPI client + Smart-Life device control.

Used by the network scanner's "Identify" button to flash Tuya bulbs/plugs/switches

on/off so the user can physically locate which IP belongs to which device.

Uses Tuya's *cloud* API rather than the local LAN protocol — the cloud route requires

only stdlib (urllib + hmac + hashlib) and avoids the local protocol's AES + per-version

handshake mess. Trade-off: each command takes ~1s and needs internet. Fine for the

"flash a bulb so I can find it" use case.

Setup is a one-time CLI flow (`tuya_setup.py`). Runtime control happens via this

module's `identify_by_mac()`.

Config files (all in repo root, gitignored):

- tuya_config.json — {endpoint, access_id, access_secret}

- tuya_devices.json — {mac → {device_id, name, category, on_code}}

- tuya_token.json — cached access token (auto-refreshed)

"""

from __future__ import annotations

import hashlib

import hmac

import json

import time

import urllib.error

import urllib.parse

import urllib.request

from pathlib import Path

from threading import Lock

ROOT = Path(__file__).resolve().parent

CONFIG_FILE = ROOT / "tuya_config.json"

DEVICES_FILE = ROOT / "tuya_devices.json"

TOKEN_FILE = ROOT / "tuya_token.json"

# Tuya regional OpenAPI endpoints. Pick whichever matches your Tuya/Smart-Life

# account region — it's not a free choice; each account is tied to one region.

ENDPOINTS = {

"us": "openapi.tuyaus.com",

"us-east": "openapi-ueaz.tuyaus.com",

"eu": "openapi.tuyaeu.com",

"eu-west": "openapi-weaz.tuyaeu.com",

"cn": "openapi.tuyacn.com",

"in": "openapi.tuyain.com",

}

# Per-category default on/off command code. Tuya devices expose typed function codes;

# bulbs use "switch_led", plugs use "switch_1" or "switch", etc. Setup time we query

# each device's spec for the actual code, but these defaults seed the lookup.

DEFAULT_ON_CODES = {

"dj": "switch_led", # bulbs

"dd": "switch_led", # light strips

"fwl": "switch_led", # ambient lights

"kg": "switch_1", # switches

"tgkg": "switch_1", # touch switches

"cz": "switch_1", # plugs

"pc": "switch_1", # power strips

"tdq": "switch_1", # breakers

}

_lock = Lock()

def _read_json(path: Path) -> dict:

if not path.exists():

return {}

try:

return json.loads(path.read_text())

except (OSError, json.JSONDecodeError):

return {}

def _write_json(path: Path, data: dict) -> None:

tmp = path.with_suffix(path.suffix + ".tmp")

tmp.write_text(json.dumps(data, indent=2))

tmp.replace(path)

def load_config() -> dict:

return _read_json(CONFIG_FILE)

def save_config(cfg: dict) -> None:

_write_json(CONFIG_FILE, cfg)

def load_devices() -> dict:

return _read_json(DEVICES_FILE)

def save_devices(devs: dict) -> None:

_write_json(DEVICES_FILE, devs)

def is_configured() -> bool:

cfg = load_config()

return bool(cfg.get("access_id") and cfg.get("access_secret") and cfg.get("endpoint"))

def normalize_mac(mac: str) -> str:

"""Match the scanner's MAC normalization — lowercase, colon-separated."""

s = "".join(c for c in mac.lower() if c in "0123456789abcdef")

if len(s) != 12:

return ""

return ":".join(s[i:i + 2] for i in range(0, 12, 2))

# ---------------------------------------------------------------------------

# Cloud client

# ---------------------------------------------------------------------------

class TuyaError(Exception):

pass

class TuyaCloud:

def __init__(self, access_id: str, access_secret: str, endpoint: str):

self.access_id = access_id

self.access_secret = access_secret

self.endpoint = endpoint # hostname only, no scheme

# -- Signing -----------------------------------------------------------

# Tuya's "v2" signature algorithm (in use since ~2019). The string-to-sign

# is: client_id + (access_token || "") + t + nonce + stringToSign

# where stringToSign = METHOD + "\n" + sha256(body) + "\n" + (signed-headers) + "\n" + path?query

# We don't use signed headers here, so that piece is empty.

def _sign(self, ts: str, nonce: str, method: str, body: str,

full_path: str, token: str = "") -> str:

body_hash = hashlib.sha256(body.encode()).hexdigest()

string_to_sign = f"{method}\n{body_hash}\n\n{full_path}"

msg = self.access_id + token + ts + nonce + string_to_sign

return hmac.new(

self.access_secret.encode(), msg.encode(), hashlib.sha256,

).hexdigest().upper()

def _request(self, method: str, path: str, *,

query: dict | None = None, body: dict | None = None,

token: str = "") -> dict:

if query:

qs = urllib.parse.urlencode(query)

full_path = f"{path}?{qs}"

else:

full_path = path

body_str = json.dumps(body) if body is not None else ""

ts = str(int(time.time() * 1000))

nonce = "" # Tuya allows empty nonce for these calls

sig = self._sign(ts, nonce, method, body_str, full_path, token=token)

headers = {

"client_id": self.access_id,

"sign": sig,

"sign_method": "HMAC-SHA256",

"t": ts,

"Content-Type": "application/json",

}

if token:

headers["access_token"] = token

url = f"https://{self.endpoint}{full_path}"

req = urllib.request.Request(

url, method=method, headers=headers,

data=body_str.encode() if body_str else None,

)

try:

with urllib.request.urlopen(req, timeout=10) as r:

payload = json.loads(r.read().decode())

except urllib.error.HTTPError as e:

try:

payload = json.loads(e.read().decode())

except (ValueError, OSError):

raise TuyaError(f"HTTP {e.code} from Tuya")

except (urllib.error.URLError, OSError, TimeoutError) as e:

raise TuyaError(f"Network error talking to Tuya: {e}")

if not payload.get("success"):

raise TuyaError(

f"Tuya API error: {payload.get('msg', payload)} "

f"(code={payload.get('code')})",

)

return payload.get("result", {})

# -- Token management --------------------------------------------------

def get_token(self) -> str:

"""Return a valid access token, refreshing if expired."""

cached = _read_json(TOKEN_FILE)

if cached.get("token") and cached.get("expires_at", 0) > time.time() + 30:

return cached["token"]

# /v1.0/token?grant_type=1 returns a fresh app-mode token (no refresh flow).

result = self._request("GET", "/v1.0/token", query={"grant_type": 1})

token = result["access_token"]

expires_at = time.time() + int(result.get("expire_time", 7200))

_write_json(TOKEN_FILE, {"token": token, "expires_at": expires_at})

return token

# -- Device API --------------------------------------------------------

def list_user_devices(self) -> list[dict]:

"""List every device in every Smart-Life user linked to this project."""

token = self.get_token()

# First, list users linked to the project.

users = self._request(

"GET", "/v1.0/iot-01/associated-users/users",

query={"page_no": 1, "page_size": 100}, token=token,

)

all_devices: list[dict] = []

# Some Tuya accounts return the user list nested differently — handle both shapes.

user_list = users if isinstance(users, list) else users.get("list", [])

if not user_list:

# Fallback: query devices by app schema using uid 0 — older API style.

return self._list_devices_v2(token)

for u in user_list:

uid = u.get("uid") or u.get("user_id")

if not uid:

continue

page = 1

while True:

resp = self._request(

"GET", f"/v1.0/users/{uid}/devices",

query={"page_no": page, "page_size": 100}, token=token,

)

# Response can be a list (older) or {list, total, has_more}.

if isinstance(resp, list):

devs = resp

has_more = False

else:

devs = resp.get("list", [])

has_more = bool(resp.get("has_more"))

all_devices.extend(devs)

if not has_more:

break

page += 1

return all_devices

def _list_devices_v2(self, token: str) -> list[dict]:

"""Newer 'iot-03' device-listing endpoint as fallback."""

out: list[dict] = []

last_id = ""

while True:

q: dict = {"page_size": 100}

if last_id:

q["last_row_key"] = last_id

resp = self._request(

"GET", "/v2.0/cloud/thing/device", query=q, token=token,

)

devs = resp.get("data_list") or resp.get("list") or []

out.extend(devs)

if not resp.get("has_more"):

break

last_id = resp.get("last_row_key", "")

if not last_id:

break

return out

def device_specs(self, device_id: str) -> dict:

"""Get a device's function/status schema. Tells us what command codes it accepts."""

token = self.get_token()

return self._request(

"GET", f"/v1.0/devices/{device_id}/specifications", token=token,

)

def device_status(self, device_id: str) -> list[dict]:

token = self.get_token()

result = self._request(

"GET", f"/v1.0/devices/{device_id}/status", token=token,

)

return result if isinstance(result, list) else []

def send_command(self, device_id: str, code: str, value) -> bool:

token = self.get_token()

result = self._request(

"POST", f"/v1.0/devices/{device_id}/commands",

body={"commands": [{"code": code, "value": value}]}, token=token,

)

return result is True or (isinstance(result, dict) and result.get("result") is True)

# ---------------------------------------------------------------------------

# Helpers used by the setup wizard and the identify endpoint

# ---------------------------------------------------------------------------

def infer_on_code(specs: dict, category: str = "") -> str:

"""Find the boolean function code that toggles a Tuya device on/off.

Tuya devices declare their controllable functions in the 'functions' block of

their spec. The on/off code is usually 'switch_led' (bulbs), 'switch_1' (plugs/

switches), or just 'switch'. Pick the first bool function whose name starts

with 'switch'.

"""

funcs = specs.get("functions") or []

# Prefer category-specific default when present in functions.

default = DEFAULT_ON_CODES.get(category)

if default and any(f.get("code") == default for f in funcs):

return default

for f in funcs:

code = f.get("code", "")

ftype = f.get("type", "").lower()

if code.startswith("switch") and ftype == "bool":

return code

# Last resort — sometimes "switch_led_1" or similar; just take the first switch_*.

for f in funcs:

if f.get("code", "").startswith("switch"):

return f["code"]

return ""

def get_client() -> TuyaCloud:

cfg = load_config()

if not cfg.get("access_id") or not cfg.get("access_secret") or not cfg.get("endpoint"):

raise TuyaError("Tuya cloud not configured — run python3 tuya_setup.py")

return TuyaCloud(cfg["access_id"], cfg["access_secret"], cfg["endpoint"])

def identify_by_mac(mac: str, *, on_off_count: int = 4) -> dict:

"""Flash the Tuya device whose LAN MAC matches `mac`.

Returns a small status dict so the API can report what happened. Raises TuyaError

on configuration / lookup / network failures.

"""

mac = normalize_mac(mac)

devs = load_devices()

entry = devs.get(mac)

if not entry:

raise TuyaError(

f"No Tuya cloud mapping for MAC {mac}. "

"Re-run python3 tuya_setup.py to refresh.",

)

device_id = entry["device_id"]

code = entry.get("on_code") or "switch_led"

client = get_client()

# Capture current state so we can restore. Tuya status entries look like

# [{"code": "switch_led", "value": true}, ...].

try:

status = client.device_status(device_id)

except TuyaError:

status = []

original = None

for s in status:

if s.get("code") == code:

original = bool(s.get("value"))

break

sequence: list[bool] = []

for i in range(on_off_count):

sequence.append(bool(i % 2 == 0) ^ (original is True))

if original is not None:

sequence.append(original)

failures = 0

for state in sequence:

try:

client.send_command(device_id, code, state)

except TuyaError:

failures += 1

time.sleep(0.6)

return {

"protocol": "tuya",

"device_id": device_id,

"device_name": entry.get("name", ""),

"code": code,

"commands_sent": len(sequence),

"failures": failures,

}