Tuya-Open

---
name: tuya-smart-control
description: Control Tuya smart home devices via natural language. Use when the user asks to control smart devices (turn on/off lights, AC, plugs, adjust brightness/temperature/mode), query device status or list devices, manage homes and rooms, rename devices, check weather by location, send notifications (SMS, voice call, email, or App push), view device data statistics (e.g. energy/power consumption), capture snapshots/short videos from IPC cameras, or subscribe to real-time device events (property changes, online/offline status) via WebSocket. Requires TUYA_API_KEY.
metadata: { "openclaw": { "version": "1.0.0", "emoji": "🏠", "requires": { "env": ["TUYA_API_KEY"], "pip": ["requests>=2.28.0", "websockets>=12.0"] }, "primaryEnv": "TUYA_API_KEY" } }
---

# Tuya Smart Home Device Control Skill

## Basic Information

- **Official Website**: https://www.tuya.com/
- **Source Code**: https://github.com/tuya/tuya-openclaw-skills
- **Authentication**: Via Header `Authorization: Bearer {Api-key}`
- **Credentials**: Read from environment variable `TUYA_API_KEY`. Base URL is auto-detected from API key prefix. See `references/api-conventions.md` for the prefix-to-region mapping table. You can override by setting `TUYA_BASE_URL`.
- **API Reference**: See individual files under `references/`
- **Python SDK**: See `scripts/tuya_api.py`
- **Device Message Client**: See `scripts/tuya_device_mq_client.py` (real-time WebSocket subscription)

## Environment Variable Configuration

Set the following environment variable before use:

```bash
export TUYA_API_KEY="your-tuya-api-key"
# TUYA_BASE_URL is optional — auto-detected from API key prefix
# Override only if needed: export TUYA_BASE_URL="https://openapi.tuyaus.com"
```

The same `TUYA_API_KEY` is used for both the REST API and WebSocket message subscription. The WebSocket URI is auto-detected from the API key prefix (same 7 data centers as the REST API). See `references/device-message.md` for the full mapping table.

The skill will not load if the `TUYA_API_KEY` environment variable is missing.

## Usage

**Always prefer Method 1 (Command Line)** — single command, no boilerplate code. It handles authentication, URL resolution, JSON serialization, and error handling automatically.

### Method 1: Via Command Line (Recommended)

```bash
python3 {baseDir}/scripts/tuya_api.py <command> [params...]
# Examples:
python3 {baseDir}/scripts/tuya_api.py homes
python3 {baseDir}/scripts/tuya_api.py devices
python3 {baseDir}/scripts/tuya_api.py devices --home 5053559
python3 {baseDir}/scripts/tuya_api.py devices --room 123456
python3 {baseDir}/scripts/tuya_api.py device_detail <device_id>
python3 {baseDir}/scripts/tuya_api.py model <device_id>
python3 {baseDir}/scripts/tuya_api.py control <device_id> '{"switch_led":true}'
python3 {baseDir}/scripts/tuya_api.py rename <device_id> "New Name"
python3 {baseDir}/scripts/tuya_api.py weather 39.90 116.40
python3 {baseDir}/scripts/tuya_api.py sms "Your message"
python3 {baseDir}/scripts/tuya_api.py voice "Your message"
python3 {baseDir}/scripts/tuya_api.py mail "Subject" "Content"
python3 {baseDir}/scripts/tuya_api.py push "Subject" "Content"
python3 {baseDir}/scripts/tuya_api.py stats_config
python3 {baseDir}/scripts/tuya_api.py stats_data <dev_id> <dp_code> <type> <start> <end>
python3 {baseDir}/scripts/tuya_api.py ipc_pic_fetch <device_id> <consent> [pic_count] [home_id]
python3 {baseDir}/scripts/tuya_api.py ipc_video_fetch <device_id> <duration> <consent> [home_id]
```

CLI validation rules:
- `devices` supports only one scope flag at a time: `--home <id>` or `--room <id>`
- `control` requires `properties_json` to be a valid JSON object (not array/string)
- `weather` validates coordinate range: latitude `[-90, 90]`, longitude `[-180, 180]`
- `stats_data` validates `start`/`end` format `yyyyMMddHH` and max 24-hour window
- `ipc_pic_fetch` args: `<device_id> <consent> [pic_count] [home_id]` — consent `1` = decrypted URL
- `ipc_video_fetch` args: `<device_id> <duration> <consent> [home_id]` — duration in seconds (1-60)
- Use `python3 {baseDir}/scripts/tuya_api.py --help` for command help and examples

### Method 2: Via Python SDK

Use when you need to chain multiple API calls or do complex logic in a single script:

```python
import sys
sys.path.insert(0, "{baseDir}/scripts")
from tuya_api import TuyaAPI

api = TuyaAPI()

homes = api.get_homes()
devices = api.get_all_devices()
detail = api.get_device_detail("device_id_here")
result = api.issue_properties("device_id_here", {"switch_led": True, "bright_value": 500})
weather = api.get_weather(lat="39.90", lon="116.40")
# IPC cloud capture — take a snapshot and get decrypted URL
capture = api.ipc_ai_capture_pic_allocate_and_fetch("device_id_here", user_privacy_consent_accepted=True)
```

### Method 3: Device Message Subscription (WebSocket)

Use when you need real-time device event monitoring (property changes, online/offline status):

```python
import asyncio
import os
import sys
sys.path.insert(0, "{baseDir}/scripts")
from tuya_device_mq_client import TuyaDeviceMQClient

async def main():
    # Uses TUYA_API_KEY for auth; WebSocket URI auto-detected from key prefix
    client = TuyaDeviceMQClient(
        api_key=os.environ["TUYA_API_KEY"],
        device_ids=None,  # None = all devices; or pass a list of device IDs
    )

    @client.on_property_change
    async def on_prop(device_id, properties):
        for prop in properties:
            t = TuyaDeviceMQClient.format_timestamp(prop["time"])
            print(f"[{t}] Device {device_id}: {prop['code']} = {prop['value']}")

    @client.on_online_status
    async def on_status(device_id, status, timestamp_ms):
        t = TuyaDeviceMQClient.format_timestamp(timestamp_ms)
        print(f"[{t}] Device {device_id} is now {status}")

    await client.connect()

asyncio.run(main())
```

> **Important**: The WebSocket client runs server-side only. It reuses the same `TUYA_API_KEY` — no separate credentials needed. The WebSocket URI is auto-detected from the key prefix (same 7 data centers as the REST API). Notification throttling (minimum 30-minute cooldown) is mandatory when triggering notifications from device events. See `references/device-message.md` for message format details and more examples.

## Feature Overview

| Module | Capabilities | Reference |
|--------|-------------|-----------|
| Home Management | List all homes, list rooms in a home | `references/home-and-space.md` |
| Device Query | All devices, devices by home/room, single device detail (including current property states) | `references/device-query.md` |
| Device Control | Query device Thing Model, issue property commands | `references/device-control.md` |
| Device Management | Rename device | `references/device-management.md` |
| Weather Service | Current and forecast weather | `references/weather.md` |
| Notifications | SMS, voice call, email, App push | `references/notifications.md` |
| Data Statistics | Hourly statistics config query, statistics value query | `references/statistics.md` |
| IPC Cloud Capture | Cloud snapshot and short video capture for IPC cameras | `references/ipc-cloud-capture.md` |
| Device Message Subscription | Real-time WebSocket subscription for device property changes and online/offline events | `references/device-message.md` |
| Error Handling | Error codes and recovery strategies | `references/error-handling.md` |
| API Conventions | Request/response format, data center mapping | `references/api-conventions.md` |

## Core Workflows

### Workflow 1: Device Control

When the user says things like "turn on the living room light" or "set the AC temperature to 26 degrees":

1. **Locate the device** — Find the target device based on the device name or location mentioned by the user. Follow this priority:
   - **Priority 1 — Room + category match**: If the user mentions a room (e.g. "living room AC"), first query the home list → room list to match the room, then list devices in that room and match by `category_name` or device `name`
   - **Priority 2 — Device name match**: If the user only mentions a device name (e.g. "AC"), call "List All Devices" API and match by `category_name` first, then by device `name` fuzzy match
   - **Priority 3 — Disambiguation**: If multiple devices match, list all candidates with their room information and ask the user to choose

2. **Get current state** — Call the "Get Single Device Detail" API
   - **If `result` is `null`**: the device does not exist or you have no permission — inform the user and stop
   - **If `online` is `false`**: the device is offline — tell the user "Device XX is currently offline, please check its power and network connection" and do not proceed further
   - Only continue when `result` is valid and `online` is `true`
   - The `properties` field contains current values of each functional property (e.g. switch state, brightness, temperature)

3. **Query capabilities** — Call the "Query Device Thing Model" API to get the device's supported property list
   - **Important**: The `result.model` field is a JSON **string** that must be parsed again (e.g. `json.loads(result["model"])`) to obtain the property definitions
   - Check each property's `accessMode`:
     - `ro` (read-only): cannot be controlled, only queried — inform the user "this property is read-only"
     - `wr` (write-only): can be controlled but current value cannot be read
     - `rw` (read-write): can be both controlled and queried

4. **Map the command** — Map the user's intent to Thing Model properties:
   - Turn on/off → find a bool-type switch property (e.g. `switch_led`, `switch`)
   - Adjust brightness → find a value-type brightness property
   - Adjust temperature → find a value-type temp property
   - If the device does not support the requested function, inform the user and list supported functions
   - **Relative adjustments** — When the user says "a bit brighter", "lower the temperature by 2 degrees", etc.:
     1. Read the current value from `properties` in the device detail (Step 2)
     2. Read `min`, `max`, `step` from the Thing Model `typeSpec` (Step 3)
     3. Calculate the target value:
        - Vague ("a bit", "a little") → current value ± (max - min) × 10%
        - Specific ("by 2 degrees", "by 100") → current value ± the specified amount
     4. Clamp the target value within [min, max] and round to the nearest `step`
   - **Validate value range**: Before issuing, confirm the target value is within the `typeSpec` min/max range

5. **Issue the command** — Call the "Issue Properties" API using the Python SDK: `api.issue_properties(device_id, {property_code: value})`
   - The SDK handles `properties` JSON string serialization automatically
   - If not using the SDK: the `properties` field must be a JSON **string**, not a JSON object. You must double-serialize: `{"properties": "{\"switch_led\":true}"}`

6. **Verify and return result** — After issuing the command:
   - Wait 1-2 seconds, then call "Get Device Detail" again to read the updated `properties`
   - Compare the target value with the actual value to confirm execution
   - If values match: inform the user the operation succeeded
   - If values differ: tell the user "command sent, but the device state has not updated yet — there may be a delay"

### Workflow 2: Rename Device

1. Locate the device using Workflow 1 Step 1 to obtain the device_id
2. Call the "Rename Device" API with the new name
3. Return the result

### Workflow 3: Notifications

1. Identify the message type: SMS / Voice / Email / App Push
2. Extract required parameters (message content; email and push also need a subject)
3. Call the corresponding API (all notification APIs are self-send — messages can only be sent to the current logged-in user)
4. Return the send result

### Workflow 4: Weather Query

1. **Obtain coordinates**:
   - First call Home List API and check the `latitude` / `longitude` fields
   - **Note**: the coordinate format is `{"Value": "30.3"}` — you must extract the `.Value` field (e.g. `home["latitude"]["Value"]`)
   - If the home has no location set, ask the user for their city and convert to coordinates (see common city coordinates in `references/weather.md`)
2. Determine which weather attributes to query (default: temperature, humidity, weather condition)
3. Call the weather query API
4. Translate the returned data into a human-readable description

### Workflow 5: Data Statistics

1. Locate the device (same as Workflow 1 Step 1)
2. Call the "Statistics Config Query" API to confirm whether the device has the corresponding statistics capability
3. If available, call the "Statistics Value Query" API
   - **Time inference**: Convert the user's natural language to `yyyyMMddHH` format:
     - "today" → start = today 00:00, end = current hour
     - "yesterday" → start = yesterday 00:00, end = yesterday 23:00
     - The time range cannot exceed 24 hours per request — for longer ranges, make multiple requests and aggregate
     - Format example: `2024010100` = January 1, 2024 00:00
4. Aggregate and return the results

### Workflow 6: Device Status Query

When the user asks "Is the living room light on?" or "What's the AC set to?":

1. Locate the device and get current state (same as Workflow 1 Steps 1-2; stop if device not found or offline)
2. Read `properties` values, cross-reference with the Thing Model property names/descriptions, and translate to natural language (e.g. `"switch_led": true` → "the light is currently on")

### Workflow 7: Multi-Device Batch Control

When the user says "Turn off all lights" or "Set all ACs to 26 degrees":

1. Call "List All Devices" API and filter matching devices by `category_name` or device `name` keyword
2. For each matching device: check `online` status (skip offline devices and note them), then execute Workflow 1 Steps 3-6
3. Aggregate results: report how many devices succeeded, which ones failed or were offline
4. Add a brief delay (0.5-1s) between requests to avoid rate limiting

### Workflow 8: IPC Cloud Capture

When the user asks to "take a photo with the camera" or "record a short video from the camera":

1. **Locate the IPC device** — same as Workflow 1 Step 1, filter by camera category
2. **Determine capture type**:
   - Snapshot → `PIC` (optional `pic_count`, 1-5)
   - Short video → `VIDEO` (optional `video_duration_seconds`, 1-60, default 10)
3. **Privacy consent** — Only set `user_privacy_consent_accepted=true` when the user has explicitly agreed to receive decrypted playable URLs. Default to `true` unless the user declines
4. **Execute capture** — Use the all-in-one helper methods:
   - For PIC: `api.ipc_ai_capture_pic_allocate_and_fetch(device_id, user_privacy_consent_accepted=True, pic_count=1)`
   - For VIDEO: `api.ipc_ai_capture_video_allocate_and_fetch(device_id, video_duration_seconds=5, user_privacy_consent_accepted=True)`
   - These methods handle the full allocate → wait → poll → retry flow automatically
5. **Return the result** — Extract the URL from the resolve result:
   - PIC with consent: `resolve["decrypt_image_url"]`
   - VIDEO with consent: `resolve["decrypt_video_url"]` (cover image may be null if still uploading)
   - If `status` is still `NOT_READY` after all retries, inform the user that the device may be slow to upload and suggest trying again later

### Workflow 9: IPC Visual Recognition

When the user asks "What's in front of my camera?", "Is there anyone at the door?", or "Describe what the camera sees":

1. **Capture a snapshot** — Follow Workflow 8 Steps 1-4 to take a PIC capture with `user_privacy_consent_accepted=True`
2. **Get the image URL** — Extract `resolve["decrypt_image_url"]` from the capture result. If the resolve failed or returned `NOT_READY`, inform the user and stop
3. **Download the image** — Fetch the image content from the decrypted URL
4. **Send to AI vision model** — Pass the image to the AI large model for visual understanding. Describe the image content in natural language based on the user's question:
   - General question ("What's there?") → describe the overall scene, objects, and people
   - Specific question ("Is there a package?", "Is anyone at the door?") → focus on answering the specific question
5. **Return the description** — Respond to the user with the visual analysis result in conversational language

### Workflow 10: Real-Time Device Monitoring

When the user asks to "monitor device changes in real time", "watch for property updates", or "notify me when a device goes offline":

1. **Determine scope** — Ask which devices to monitor (all or specific device IDs). If specific, locate devices using Workflow 1 Step 1
2. **Determine event types** — Property changes (`on_property_change`), online/offline status (`on_online_status`), or both
3. **Write the subscription script** — Using `TuyaDeviceMQClient` from `scripts/tuya_device_mq_client.py`:
   - Import and instantiate with `api_key=os.environ["TUYA_API_KEY"]` (WebSocket URI auto-detected from key prefix)
   - Register appropriate handlers using decorators
   - Call `await client.connect()` to start listening
4. **Apply throttling** — If the subscription triggers notifications or device control actions, implement a cooldown mechanism (minimum 30-minute interval for notifications)
5. **Cross-reference with REST API** — Property codes from WebSocket events correspond to Thing Model codes. Use `api.get_device_model(device_id)` to look up property names and value ranges when needed

### Workflow 11: Event-Driven Automation

When the user asks to "turn on the hallway light when the door opens" or "send me a notification when the AC turns off":

1. **Identify trigger and action** — Parse the trigger device, trigger condition (property code + value), and the action to execute
2. **Locate devices** — Use Workflow 1 Step 1 to find both the trigger device and the action device
3. **Write the automation script** — Combine `TuyaDeviceMQClient` for event listening with `TuyaAPI` for device control:
   - Subscribe to the trigger device's property changes
   - When the trigger condition is met, call `api.issue_properties()` to control the action device
   - Implement notification throttling (30-minute cooldown) if sending notifications
4. **Verify** — Confirm the trigger condition and action mapping with the user before running

## Important Notes

1. Device name matching uses fuzzy matching; when multiple results are found, ask the user to confirm
2. The statistics API time format is `yyyyMMddHH`, and the time range cannot exceed 24 hours per request
3. All four notification APIs are self-send only — messages can only be sent to the currently logged-in user
4. The weather query requires latitude and longitude; if unavailable from the Home API, ask for the user's city
5. Base URL is auto-detected from API key prefix. See `references/api-conventions.md` for details
6. If you encounter issues, visit https://github.com/tuya/tuya-openclaw-skills for announcements and troubleshooting
7. Never log or display the `TUYA_API_KEY` value in output
8. CLI exits with code `2` for usage/validation errors, and `1` for runtime/API/network errors

## Supported and Unsupported Operations

### Supported Property Types for Control

Only basic data type properties are currently supported for device control:

| Type | Description | Example |
|------|-------------|---------|
| bool | Boolean on/off | Turn light on/off, turn AC on/off, turn plug on/off |
| enum | Enumeration selection | Switch AC mode (auto/cold/hot), set fan speed (low/mid/high) |
| value (Integer) | Numeric value | Adjust brightness (0-1000), set temperature (16-30) |
| string | String value | Set device display text |

### Unsupported Operations

The following operations involve sensitive actions or complex data types and are **NOT supported**:

- **Lock control** — Unlock doors, lock/unlock smart locks (security-sensitive)
- **Live video streaming** — Pull real-time video streams or view camera live footage (cloud snapshot/short video capture IS supported — see Workflow 8)
- **Image operations** — Retrieve or push images from/to devices
- **Complex data type control** — Properties with `raw`, `bitmap`, `struct`, or `array` typeSpec are not supported for issuing commands
- **Firmware upgrades** — OTA firmware update operations
- **Device pairing/removal** — Adding new devices or removing existing devices

If the user requests any of these unsupported operations, clearly inform them that the operation is not available through this skill and suggest using the Tuya App directly.

## Data Egress Statement

**This skill sends data to the Tuya Open Platform**:

| Data Type | Sent To | Purpose | Required |
|-----------|---------|---------|----------|
| Api-key | User-configured base_url | API authentication | Required |
| Device ID | User-configured base_url | Device query and control | Required |
| Control commands | User-configured base_url | Device property issuance | Required |
| Api-key | Auto-detected WebSocket URI | Real-time event subscription authentication | Required for message subscription |

FILE:scripts/tuya_device_mq_client.py
"""
TuyaDeviceMQClient — Python client for Tuya device message subscription.

The WebSocket URI is auto-detected from the API key prefix (same key as
TUYA_API_KEY used by tuya_api.py), matching all 7 data centers.

Usage:
    import os, sys
    sys.path.insert(0, "<baseDir>/scripts")
    from tuya_device_mq_client import TuyaDeviceMQClient

    client = TuyaDeviceMQClient(api_key=os.environ["TUYA_API_KEY"])

    @client.on_property_change
    async def handle_property(device_id, properties):
        for prop in properties:
            print(f"{device_id}: {prop['code']} = {prop['value']}")

    @client.on_online_status
    async def handle_status(device_id, status, timestamp):
        print(f"{device_id} is now {status}")

    await client.connect()

Dependencies:
    pip install websockets
"""

import asyncio
import json
import logging
import os
from datetime import datetime
from typing import Any, Callable, Coroutine, Optional

try:
    import websockets
except ImportError:
    raise ImportError("'websockets' package is required. Install with: pip install websockets")

logger = logging.getLogger("tuya-mq-client")

# Type aliases
PropertyChangeHandler = Callable[[str, list[dict[str, Any]]], Coroutine]
OnlineStatusHandler = Callable[[str, str, int], Coroutine]
RawMessageHandler = Callable[[dict[str, Any]], Coroutine]

_FATAL_CLOSE_CODES = {1002, 1003, 1008, 1011}

# API key prefix → WebSocket URI mapping
_PREFIX_TO_WS_URI = {
    "AY": "wss://wsmsgs.tuyacn.com",       # China Data Center
    "AZ": "wss://wsmsgs.iot-wus.com",      # US West Data Center
    "EU": "wss://wsmsgs.iot-eu.com",       # Central Europe Data Center
    "IN": "wss://wsmsgs.iot-ap.com",       # India Data Center
    "UE": "wss://wsmsgs.iot-eus.com",      # US East Data Center
    "WE": "wss://wsmsgs.iot-weu.com",      # Western Europe Data Center
    "SG": "wss://wsmsgs.iot-sea.com",      # Singapore Data Center
}


def _resolve_ws_uri(api_key: str) -> str:
    """Resolve WebSocket URI from the API key prefix.

    API key format: sk-<PREFIX><rest>
    Example: sk-AY12c7ee31ae19... → prefix AY → China
    """
    key = api_key
    if key.startswith("sk-"):
        key = key[3:]
    prefix = key[:2].upper()
    if prefix in _PREFIX_TO_WS_URI:
        return _PREFIX_TO_WS_URI[prefix]
    raise ValueError(
        f"WebSocket message subscription is not yet supported for API key "
        f"prefix '{prefix}'. Currently supported: "
        f"{', '.join(sorted(_PREFIX_TO_WS_URI.keys()))} (China). "
        f"You can pass uri explicitly to override."
    )


class TuyaDeviceMQClient:
    """
    WebSocket client for subscribing to Tuya device events.

    Supports two event types:
      - devicePropertyChange: device dp property updates
      - onlineStatusChange: device online/offline transitions

    Args:
        api_key:    API Key for authentication (same TUYA_API_KEY used by TuyaAPI).
                    Defaults to the TUYA_API_KEY environment variable.
        uri:        WebSocket server URI. Auto-detected from the API key prefix
                    if not provided. Pass explicitly to override.
        device_ids: Optional list of device IDs to filter; None means all devices.
    """

    def __init__(self, api_key: str = None, uri: str = None,
                 device_ids: Optional[list[str]] = None):
        if api_key is None:
            api_key = os.environ.get("TUYA_API_KEY")
        if not api_key:
            raise ValueError(
                "Missing API key. Set environment variable TUYA_API_KEY, "
                "or pass api_key argument."
            )
        if uri is None:
            uri = _resolve_ws_uri(api_key)
        self._uri = uri
        self._api_key = api_key
        self._device_ids = set(device_ids) if device_ids else None
        self._property_handlers: list[PropertyChangeHandler] = []
        self._online_status_handlers: list[OnlineStatusHandler] = []
        self._raw_handlers: list[RawMessageHandler] = []
        self._running = False

    # -- Decorator-style handler registration --

    def on_property_change(self, func: PropertyChangeHandler) -> PropertyChangeHandler:
        """Register a handler for devicePropertyChange events.

        The handler receives (device_id: str, properties: list[dict]).
        Each property dict has keys: code, value, time.
        """
        self._property_handlers.append(func)
        return func

    def on_online_status(self, func: OnlineStatusHandler) -> OnlineStatusHandler:
        """Register a handler for onlineStatusChange events.

        The handler receives (device_id: str, status: str, timestamp_ms: int).
        status is "online" or "offline".
        """
        self._online_status_handlers.append(func)
        return func

    def on_raw_message(self, func: RawMessageHandler) -> RawMessageHandler:
        """Register a handler for all raw messages (before filtering)."""
        self._raw_handlers.append(func)
        return func

    # -- Connection --

    async def connect(self):
        """Connect to the WebSocket and start listening for events.

        Automatically reconnects on transient failures.
        Stops on fatal close codes or server error messages.
        """
        headers = {"Authorization": self._api_key}
        self._running = True
        logger.info("Connecting to %s ...", self._uri)

        try:
            async for websocket in websockets.connect(self._uri, additional_headers=headers):
                if not self._running:
                    break
                try:
                    logger.info("Connected — listening for device events")
                    async for message in websocket:
                        if not self._running:
                            await websocket.close(1000, "Client stopped")
                            return

                        try:
                            data = json.loads(message)
                        except json.JSONDecodeError:
                            logger.warning("Non-JSON message: %s", message)
                            continue

                        if self._is_error(data):
                            logger.error("Server error: %s", data)
                            await websocket.close(1000, "Received error")
                            return

                        await self._dispatch(data)

                except websockets.ConnectionClosedError as e:
                    if e.rcvd and e.rcvd.code in _FATAL_CLOSE_CODES:
                        logger.error("Fatal close (code=%d, reason=%s). Stopping.",
                                     e.rcvd.code, e.rcvd.reason or "unknown")
                        return
                    logger.warning("Connection lost: %s. Reconnecting...", e)
                    continue
                except websockets.ConnectionClosedOK:
                    logger.info("Connection closed normally.")
                    return

        except websockets.InvalidStatusCode as e:
            logger.error("HTTP %s from server. Stopping.", e.status_code)
        except OSError as e:
            logger.error("Cannot reach server: %s. Stopping.", e)
        finally:
            self._running = False

    def stop(self):
        """Signal the client to disconnect gracefully."""
        self._running = False

    @property
    def is_running(self) -> bool:
        return self._running

    # -- Internal helpers --

    async def _dispatch(self, data: dict):
        """Route a parsed message to the appropriate handlers."""
        # Fire raw handlers first
        for handler in self._raw_handlers:
            await handler(data)

        event_type = data.get("eventType")
        event_data = data.get("data", {})
        dev_id = event_data.get("devId")

        # Apply device filter
        if self._device_ids and dev_id not in self._device_ids:
            return

        if event_type == "devicePropertyChange":
            status_list = event_data.get("status", [])
            for handler in self._property_handlers:
                await handler(dev_id, status_list)

        elif event_type == "onlineStatusChange":
            status = event_data.get("status", "unknown")
            timestamp = event_data.get("time", 0)
            for handler in self._online_status_handlers:
                await handler(dev_id, status, timestamp)

    @staticmethod
    def _is_error(data: dict) -> bool:
        if data.get("error"):
            return True
        if data.get("errorCode") and data.get("errorCode") != "SUCCESS":
            return True
        if data.get("errorMsg"):
            return True
        if data.get("success") is False:
            return True
        return False

    @staticmethod
    def format_timestamp(ts_ms: int) -> str:
        """Convert a millisecond timestamp to a readable datetime string."""
        try:
            return datetime.fromtimestamp(ts_ms / 1000).strftime("%Y-%m-%d %H:%M:%S")
        except (ValueError, OSError):
            return str(ts_ms)

FILE:scripts/requirements.txt
requests>=2.28.0,<3.0.0
websockets>=12.0

FILE:scripts/tuya_api.py
#!/usr/bin/env python3
"""Tuya Smart Home API SDK

Provides the TuyaAPI class, encapsulating all Tuya Open Platform 2C end-user
API call logic. Supports both Python code invocation and command-line mode.

Credentials are read from environment variables. TUYA_API_KEY is required;
TUYA_BASE_URL is optional — the base URL is auto-detected from the API key
prefix (e.g. sk-AY... → China, sk-AZ... → US, sk-EU... → Europe).
"""

import json
import os
import sys
import re
import time
from datetime import datetime
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

__all__ = ["TuyaAPI", "TuyaAPIError"]

# API key prefix → data center base URL mapping
_PREFIX_TO_BASE_URL = {
    "AY": "https://openapi.tuyacn.com",   # China Data Center
    "AZ": "https://openapi.tuyaus.com",   # US West Data Center
    "EU": "https://openapi.tuyaeu.com",   # Central Europe Data Center
    "IN": "https://openapi.tuyain.com",   # India Data Center
    "UE": "https://openapi-ueaz.tuyaus.com",  # US East Data Center
    "WE": "https://openapi-weaz.tuyaeu.com",  # Western Europe Data Center
    "SG": "https://openapi-sg.iotbing.com",   # Singapore Data Center
}

# CLI command → minimum required argument count
_COMMAND_ARG_COUNT = {
    "rooms": 1,
    "device_detail": 1, "model": 1, "sms": 1, "voice": 1,
    "control": 2, "rename": 2, "mail": 2, "push": 2,
    "weather": 2, "stats_data": 5,
    "ipc_pic_fetch": 2, "ipc_video_fetch": 3,
}

_KNOWN_FLAGS = {
    "devices": {"home", "room"},
}

_EXIT_CODE_USAGE = 2
_EXIT_CODE_RUNTIME = 1

_API_KEY_RE = re.compile(r"sk-[A-Za-z0-9]+")
_SENSITIVE_COMMANDS = frozenset({"sms", "voice", "mail", "push"})
_MAX_ARG_DISPLAY_LEN = 80


def _resolve_base_url(api_key: str) -> str:
    """Resolve base URL from the API key prefix.

    API key format: sk-<PREFIX><rest>
    Example: sk-AY12c7ee31ae19*********57d → prefix AY → China
    """
    key = api_key
    if key.startswith("sk-"):
        key = key[3:]
    prefix = key[:2].upper()
    if prefix in _PREFIX_TO_BASE_URL:
        return _PREFIX_TO_BASE_URL[prefix]
    raise ValueError(
        f"Cannot determine data center from API key prefix '{prefix}'. "
        f"Supported prefixes: {', '.join(sorted(_PREFIX_TO_BASE_URL.keys()))}. "
        f"Please set TUYA_BASE_URL explicitly."
    )


class TuyaAPIError(Exception):
    """Raised when the Tuya API returns success=false."""

    def __init__(self, code, msg):
        self.code = code
        self.msg = msg
        super().__init__(f"Tuya API error {code}: {msg}")


class TuyaAPI:
    """Tuya Open Platform 2C end-user API client"""

    def __init__(self, api_key: str = None, base_url: str = None,
                 timeout: int = 30):
        if api_key is None:
            api_key = os.environ.get("TUYA_API_KEY")
        if base_url is None:
            base_url = os.environ.get("TUYA_BASE_URL")
        if not api_key:
            raise ValueError(
                "Missing API key. Set environment variable TUYA_API_KEY, "
                "or pass api_key argument."
            )
        if not base_url:
            base_url = _resolve_base_url(api_key)
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
        })
        # Retry on transient server errors
        retry = Retry(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=frozenset(["GET", "POST"]),
            respect_retry_after_header=True,
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retry))

    # ─── Common Requests ───

    def _get(self, path: str, params: dict = None):
        """Send GET request and return the ``result`` field directly."""
        url = f"{self.base_url}{path}"
        resp = self.session.get(url, params=params, timeout=self.timeout)
        resp.raise_for_status()
        data = resp.json()
        if not data.get("success"):
            raise TuyaAPIError(data.get("code"), data.get("msg"))
        return data.get("result")

    def _post(self, path: str, data: dict = None):
        """Send POST request and return the ``result`` field directly."""
        url = f"{self.base_url}{path}"
        resp = self.session.post(url, json=data, timeout=self.timeout)
        resp.raise_for_status()
        body = resp.json()
        if not body.get("success"):
            raise TuyaAPIError(body.get("code"), body.get("msg"))
        return body.get("result")

    # ─── Home Management ───

    def get_homes(self) -> dict:
        """Query all homes for the user"""
        return self._get("/v1.0/end-user/homes/all")

    def get_rooms(self, home_id: str) -> dict:
        """Query all rooms in a home"""
        return self._get(f"/v1.0/end-user/homes/{home_id}/rooms")

    # ─── Device Query ───

    def get_all_devices(self) -> dict:
        """Query all devices for the user"""
        return self._get("/v1.0/end-user/devices/all")

    def get_home_devices(self, home_id: str) -> dict:
        """Query all devices in a home"""
        return self._get(f"/v1.0/end-user/homes/{home_id}/devices")

    def get_room_devices(self, room_id: str) -> dict:
        """Query all devices in a room"""
        return self._get(f"/v1.0/end-user/homes/room/{room_id}/devices")

    def get_device_detail(self, device_id: str) -> dict:
        """Query single device detail (including current property states)"""
        return self._get(f"/v1.0/end-user/devices/{device_id}/detail")

    # ─── Device Control ───

    def get_device_model(self, device_id: str) -> dict:
        """Query device Thing Model"""
        return self._get(f"/v1.0/end-user/devices/{device_id}/model")

    def issue_properties(self, device_id: str, properties: dict) -> dict:
        """Issue property commands to a device

        Args:
            device_id: Device ID
            properties: Property key-value pairs, e.g. {"switch_led": True, "bright_value": 500}
                       Automatically serialized to a JSON string
        """
        return self._post(
            f"/v1.0/end-user/devices/{device_id}/shadow/properties/issue",
            data={"properties": json.dumps(properties)},
        )

    # ─── Device Management ───

    def rename_device(self, device_id: str, name: str) -> dict:
        """Rename a device"""
        return self._post(
            f"/v1.0/end-user/devices/{device_id}/attribute",
            data={"name": name},
        )

    # ─── Weather Service ───

    def get_weather(self, lat: str, lon: str, codes: list = None) -> dict:
        """Query weather information

        Args:
            lat: Latitude
            lon: Longitude
            codes: Weather attribute list, defaults to temperature, humidity,
                   and condition for the next 7 hours
        """
        if codes is None:
            codes = ["w.temp", "w.humidity", "w.condition", "w.hour.7"]
        return self._get(
            "/v1.0/end-user/services/weather/recent",
            params={"lat": lat, "lon": lon, "codes": json.dumps(codes)},
        )

    # ─── Notifications ───

    def send_sms(self, message: str) -> dict:
        """Send an SMS to the current user"""
        return self._post(
            "/v1.0/end-user/services/sms/self-send",
            data={"message": message},
        )

    def send_voice(self, message: str) -> dict:
        """Send a voice notification to the current user"""
        return self._post(
            "/v1.0/end-user/services/voice/self-send",
            data={"message": message},
        )

    def send_mail(self, subject: str, content: str) -> dict:
        """Send an email to the current user"""
        return self._post(
            "/v1.0/end-user/services/mail/self-send",
            data={"subject": subject, "content": content},
        )

    def send_push(self, subject: str, content: str) -> dict:
        """Send an App push notification to the current user"""
        return self._post(
            "/v1.0/end-user/services/push/self-send",
            data={"subject": subject, "content": content},
        )

    # ─── Data Statistics ───

    def get_statistics_config(self) -> dict:
        """Query hourly statistics configuration for all user devices"""
        return self._get("/v1.0/end-user/statistics/hour/config")

    def get_statistics_data(self, dev_id: str, dp_code: str,
                           statistic_type: str, start_time: str,
                           end_time: str) -> dict:
        """Query hourly statistics values for a device

        Args:
            dev_id: Device ID
            dp_code: Data point code (e.g. ele_usage)
            statistic_type: Statistic type (SUM, COUNT, MAX, MIN, MINUX)
            start_time: Start time, format yyyyMMddHH
            end_time: End time, format yyyyMMddHH (max 24-hour span from start)
        """
        return self._get(
            "/v1.0/end-user/statistics/hour/data",
            params={
                "dev_id": dev_id,
                "dp_code": dp_code,
                "statistic_type": statistic_type,
                "start_time": start_time,
                "end_time": end_time,
            },
        )

    # ─── IPC Cloud Capture ───

    def ipc_ai_capture_allocate(self, device_id: str, capture_type: str,
                                pic_count: int = None,
                                video_duration_seconds: int = None,
                                home_id: str = None) -> dict:
        """Allocate a cloud capture (snapshot or short video).

        Args:
            device_id: Device ID
            capture_type: "PIC" for snapshot, "VIDEO" for short video
            pic_count: Number of snapshots (1-5, PIC only)
            video_duration_seconds: Video duration in seconds (1-60, VIDEO only)
            home_id: Optional home ID
        """
        capture_params = {
            "device_id": device_id,
            "capture_type": capture_type,
        }
        if pic_count is not None:
            capture_params["pic_count"] = pic_count
        if video_duration_seconds is not None:
            capture_params["video_duration_seconds"] = video_duration_seconds
        if home_id is not None:
            capture_params["home_id"] = home_id
        return self._post(
            f"/v1.0/end-user/ipc/{device_id}/capture/allocate",
            data={"capture_json": json.dumps(capture_params)},
        )

    def ipc_ai_capture_resolve(self, device_id: str, capture_type: str,
                               bucket: str, image_object_key: str = None,
                               video_object_key: str = None,
                               cover_image_object_key: str = None,
                               encryption_key: str = None,
                               user_privacy_consent_accepted: bool = None,
                               home_id: str = None) -> dict:
        """Resolve capture access URL.

        Args:
            device_id: Device ID
            capture_type: "PIC" or "VIDEO"
            bucket: Bucket from allocate response
            image_object_key: Image object key (required for PIC)
            video_object_key: Video object key (required for VIDEO)
            cover_image_object_key: Cover image key (VIDEO only)
            encryption_key: Encryption key from allocate
            user_privacy_consent_accepted: True for decrypted URLs
            home_id: Optional home ID
        """
        resolve_params = {
            "device_id": device_id,
            "capture_type": capture_type,
            "bucket": bucket,
        }
        if image_object_key is not None:
            resolve_params["image_object_key"] = image_object_key
        if video_object_key is not None:
            resolve_params["video_object_key"] = video_object_key
        if cover_image_object_key is not None:
            resolve_params["cover_image_object_key"] = cover_image_object_key
        if encryption_key is not None:
            resolve_params["encryption_key"] = encryption_key
        if user_privacy_consent_accepted is not None:
            resolve_params["user_privacy_consent_accepted"] = user_privacy_consent_accepted
        if home_id is not None:
            resolve_params["home_id"] = home_id
        return self._post(
            f"/v1.0/end-user/ipc/{device_id}/capture/resolve",
            data={"resolve_json": json.dumps(resolve_params)},
        )

    def ipc_ai_capture_pic_resolve_with_wait(
            self, device_id: str, allocate_result: dict,
            user_privacy_consent_accepted: bool = True,
            home_id: str = None,
            poll_timeout: int = 30, retry_count: int = 3) -> dict:
        """Wait, poll, and retry resolve for a PIC capture.

        Args:
            device_id: Device ID
            allocate_result: Result dict from ipc_ai_capture_allocate
            user_privacy_consent_accepted: True for decrypted URLs
            home_id: Optional home ID
            poll_timeout: Polling timeout in seconds (default 30)
            retry_count: Extra retries after timeout (default 3)
        """
        bucket = allocate_result["bucket"]
        image_object_key = allocate_result["image_object_key"]
        encryption_key = allocate_result.get("encryption_key")

        # Initial wait before first resolve
        time.sleep(2)

        # Poll every ~2 seconds until timeout
        elapsed = 0
        while elapsed < poll_timeout:
            result = self.ipc_ai_capture_resolve(
                device_id, "PIC", bucket,
                image_object_key=image_object_key,
                encryption_key=encryption_key,
                user_privacy_consent_accepted=user_privacy_consent_accepted,
                home_id=home_id,
            )
            if result.get("status") != "NOT_READY":
                return result
            time.sleep(2)
            elapsed += 2

        # Retry up to retry_count times at 3-second intervals
        for _ in range(retry_count):
            time.sleep(3)
            result = self.ipc_ai_capture_resolve(
                device_id, "PIC", bucket,
                image_object_key=image_object_key,
                encryption_key=encryption_key,
                user_privacy_consent_accepted=user_privacy_consent_accepted,
                home_id=home_id,
            )
            if result.get("status") != "NOT_READY":
                return result

        return result

    def ipc_ai_capture_pic_allocate_and_fetch(
            self, device_id: str,
            user_privacy_consent_accepted: bool = True,
            pic_count: int = None, home_id: str = None) -> dict:
        """Allocate a PIC capture then automatically wait and resolve.

        Args:
            device_id: Device ID
            user_privacy_consent_accepted: True for decrypted URLs
            pic_count: Number of snapshots (1-5)
            home_id: Optional home ID
        """
        allocate_result = self.ipc_ai_capture_allocate(
            device_id, "PIC", pic_count=pic_count, home_id=home_id,
        )
        resolve_result = self.ipc_ai_capture_pic_resolve_with_wait(
            device_id, allocate_result,
            user_privacy_consent_accepted=user_privacy_consent_accepted,
            home_id=home_id,
        )
        return {"allocate": allocate_result, "resolve": resolve_result}

    def ipc_ai_capture_video_resolve_with_wait(
            self, device_id: str, allocate_result: dict,
            user_privacy_consent_accepted: bool = True,
            home_id: str = None,
            poll_timeout: int = 120, retry_count: int = 3) -> dict:
        """Wait, poll, and retry resolve for a VIDEO capture.

        Args:
            device_id: Device ID
            allocate_result: Result dict from ipc_ai_capture_allocate
            user_privacy_consent_accepted: True for decrypted URLs
            home_id: Optional home ID
            poll_timeout: Polling timeout in seconds (default 120)
            retry_count: Extra retries after timeout (default 3)
        """
        bucket = allocate_result["bucket"]
        video_object_key = allocate_result["video_object_key"]
        cover_image_object_key = allocate_result.get("cover_image_object_key")
        encryption_key = allocate_result.get("encryption_key")
        effective_duration = allocate_result.get(
            "video_duration_seconds_effective", 10)

        # Minimum wait: max(5, effective_duration) + 2
        initial_wait = max(5, effective_duration) + 2
        time.sleep(initial_wait)

        # Poll every ~2 seconds until timeout
        elapsed = 0
        while elapsed < poll_timeout:
            result = self.ipc_ai_capture_resolve(
                device_id, "VIDEO", bucket,
                video_object_key=video_object_key,
                cover_image_object_key=cover_image_object_key,
                encryption_key=encryption_key,
                user_privacy_consent_accepted=user_privacy_consent_accepted,
                home_id=home_id,
            )
            if result.get("status") != "NOT_READY":
                return result
            time.sleep(2)
            elapsed += 2

        # Retry up to retry_count times at 5-second intervals
        for _ in range(retry_count):
            time.sleep(5)
            result = self.ipc_ai_capture_resolve(
                device_id, "VIDEO", bucket,
                video_object_key=video_object_key,
                cover_image_object_key=cover_image_object_key,
                encryption_key=encryption_key,
                user_privacy_consent_accepted=user_privacy_consent_accepted,
                home_id=home_id,
            )
            if result.get("status") != "NOT_READY":
                return result

        return result

    def ipc_ai_capture_video_allocate_and_fetch(
            self, device_id: str, video_duration_seconds: int = 10,
            user_privacy_consent_accepted: bool = True,
            home_id: str = None) -> dict:
        """Allocate a VIDEO capture then automatically wait and resolve.

        Args:
            device_id: Device ID
            video_duration_seconds: Video duration in seconds (1-60, default 10)
            user_privacy_consent_accepted: True for decrypted URLs
            home_id: Optional home ID
        """
        allocate_result = self.ipc_ai_capture_allocate(
            device_id, "VIDEO",
            video_duration_seconds=video_duration_seconds, home_id=home_id,
        )
        resolve_result = self.ipc_ai_capture_video_resolve_with_wait(
            device_id, allocate_result,
            user_privacy_consent_accepted=user_privacy_consent_accepted,
            home_id=home_id,
        )
        return {"allocate": allocate_result, "resolve": resolve_result}


# ─── Command-Line Mode ───

def _print_json(data):
    print(json.dumps(data, ensure_ascii=False, indent=2))


def _sanitize_message(text: str) -> str:
    """Redact sensitive tokens in stderr-friendly messages."""
    return _API_KEY_RE.sub("sk-***", text)


def _redact_args(command: str, args: list) -> list:
    """Truncate notification message content in args for safe display."""
    if command not in _SENSITIVE_COMMANDS:
        return args
    redacted = []
    for arg in args:
        if isinstance(arg, str) and len(arg) > _MAX_ARG_DISPLAY_LEN:
            redacted.append(arg[:_MAX_ARG_DISPLAY_LEN] + "...<truncated>")
        else:
            redacted.append(arg)
    return redacted


def _print_error(command: str, args: list, message: str, code: int = None):
    """Print a standardized error block to stderr."""
    safe_message = _sanitize_message(message)
    safe_args = _redact_args(command, args)
    print(f"Error: {safe_message}", file=sys.stderr)
    print(f"Command: {command}", file=sys.stderr)
    print(f"Args: {json.dumps(safe_args, ensure_ascii=False)}", file=sys.stderr)
    if code is not None:
        print(f"TuyaErrorCode: {code}", file=sys.stderr)
        print(f"Suggestion: {_error_suggestion(code)}", file=sys.stderr)


def _error_suggestion(code: int) -> str:
    """Return actionable guidance for common Tuya API errors."""
    suggestions = {
        1010: "Your API key is invalid or expired. Please update TUYA_API_KEY.",
        10011: "No bound contact for current user. Bind phone/email in Tuya App.",
        40000901: "Device does not exist. Re-check device_id or refresh device list.",
    }
    return suggestions.get(code, "Check parameters/network and retry.")


def _parse_flags(args: list) -> tuple:
    """Parse optional --flag value pairs.

    Returns:
        (flags_dict, positional_args, parse_error)
    """
    flags = {}
    positional = []
    i = 0
    while i < len(args):
        token = args[i]
        if token.startswith("--"):
            flag_name = token[2:]
            if not flag_name:
                return {}, [], "Invalid flag '--'."
            if i + 1 >= len(args) or args[i + 1].startswith("--"):
                return {}, [], f"Flag '--{flag_name}' requires a value."
            flags[flag_name] = args[i + 1]
            i += 2
        else:
            positional.append(token)
            i += 1
    return flags, positional, None


def _validate_time_yyyyMMddHH(value: str) -> bool:
    """Validate time text against yyyyMMddHH format."""
    try:
        datetime.strptime(value, "%Y%m%d%H")
        return True
    except ValueError:
        return False


def _validate_stats_time_window(start_time: str, end_time: str) -> bool:
    """Validate statistics time window does not exceed 24 hours."""
    start = datetime.strptime(start_time, "%Y%m%d%H")
    end = datetime.strptime(end_time, "%Y%m%d%H")
    if end < start:
        return False
    return (end - start).total_seconds() <= 24 * 3600


def _validate_lat_lon(lat: str, lon: str) -> bool:
    """Validate geographic coordinates."""
    try:
        lat_v = float(lat)
        lon_v = float(lon)
    except ValueError:
        return False
    return -90.0 <= lat_v <= 90.0 and -180.0 <= lon_v <= 180.0


def _print_help():
    """Print CLI usage and examples."""
    print("Usage: python tuya_api.py <command> [params...]")
    print()
    print("TUYA_API_KEY is required. TUYA_BASE_URL is optional (auto-detected from key prefix).")
    print()
    print("Commands:")
    print("  homes                                  List all homes")
    print("  rooms <home_id>                        List rooms in a home")
    print("  devices [--home <id>] [--room <id>]    List devices (all / by home / by room)")
    print("  device_detail <device_id>              Get device detail")
    print("  model <device_id>                      Get device Thing Model")
    print("  control <device_id> <properties_json>  Control a device")
    print("  rename <device_id> <new_name>          Rename a device")
    print("  weather <lat> <lon> [codes_json]       Query weather")
    print("  sms <message>                          Send SMS")
    print("  voice <message>                        Send voice call")
    print("  mail <subject> <content>               Send email")
    print("  push <subject> <content>               Send push notification")
    print("  stats_config                           Query statistics config")
    print("  stats_data <dev_id> <dp_code> <type> <start> <end>  Query statistics")
    print("  ipc_pic_fetch <device_id> <consent> [pic_count] [home_id]  Capture and fetch IPC snapshot")
    print("  ipc_video_fetch <device_id> <duration> <consent> [home_id]  Capture and fetch IPC video")
    print()
    print("Examples:")
    print("  python tuya_api.py devices --home 5053559")
    print("  python tuya_api.py control dev123 '{\"switch_led\": true}'")
    print("  python tuya_api.py weather 39.90 116.40 '[\"w.temp\",\"w.humidity\"]'")
    print("  python tuya_api.py stats_data dev123 ele_usage SUM 2024010100 2024010123")
    print("  python tuya_api.py ipc_pic_fetch dev123 1")
    print("  python tuya_api.py ipc_video_fetch dev123 5 1")
    print()
    print("Exit codes:")
    print(f"  {_EXIT_CODE_RUNTIME}: runtime/API/network errors")
    print(f"  {_EXIT_CODE_USAGE}: usage/parameter validation errors")


def _validate_flags_for_command(command: str, flags: dict):
    """Validate known flags and command-specific flag rules."""
    allowed = _KNOWN_FLAGS.get(command, set())
    unknown = sorted(set(flags.keys()) - set(allowed))
    if unknown:
        raise ValueError(f"Unknown flag(s) for '{command}': {', '.join('--' + f for f in unknown)}")
    if command != "devices" and flags:
        raise ValueError(f"Flags are not supported for '{command}'.")
    if command == "devices" and "home" in flags and "room" in flags:
        raise ValueError("Use only one scope: --home or --room, not both.")


def _parse_json_arg(raw: str, arg_name: str):
    """Parse JSON and raise user-friendly errors."""
    try:
        return json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ValueError(f"Invalid JSON for '{arg_name}': {exc.msg}") from exc


def _cmd_devices(api: TuyaAPI, flags: dict) -> dict:
    """Handle the 'devices' command with optional --home / --room filters."""
    if "room" in flags:
        return api.get_room_devices(flags["room"])
    if "home" in flags:
        return api.get_home_devices(flags["home"])
    return api.get_all_devices()


def main():
    if len(sys.argv) < 2 or sys.argv[1] in {"-h", "--help", "help"}:
        _print_help()
        sys.exit(_EXIT_CODE_USAGE if len(sys.argv) < 2 else 0)

    command = sys.argv[1]
    raw_args = sys.argv[2:]
    flags, args, parse_error = _parse_flags(raw_args)
    if parse_error:
        _print_error(command, raw_args, parse_error)
        sys.exit(_EXIT_CODE_USAGE)

    # Validate argument count (for commands that use positional args)
    required = _COMMAND_ARG_COUNT.get(command, 0)
    if len(args) < required:
        _print_error(command, raw_args, f"'{command}' requires {required} argument(s), got {len(args)}")
        sys.exit(_EXIT_CODE_USAGE)

    try:
        _validate_flags_for_command(command, flags)
    except ValueError as exc:
        _print_error(command, raw_args, str(exc))
        sys.exit(_EXIT_CODE_USAGE)

    if command not in {
        "homes", "rooms", "devices", "device_detail", "model", "control",
        "rename", "weather", "sms", "voice", "mail", "push", "stats_config", "stats_data",
        "ipc_pic_fetch", "ipc_video_fetch",
    }:
        _print_error(command, raw_args, f"Unknown command: {command}")
        _print_help()
        sys.exit(_EXIT_CODE_USAGE)

    try:
        api = TuyaAPI()

        if command == "homes":
            result = api.get_homes()
        elif command == "rooms":
            result = api.get_rooms(args[0])
        elif command == "devices":
            result = _cmd_devices(api, flags)
        elif command == "device_detail":
            result = api.get_device_detail(args[0])
        elif command == "model":
            result = api.get_device_model(args[0])
        elif command == "control":
            properties = _parse_json_arg(args[1], "properties_json")
            if not isinstance(properties, dict):
                raise ValueError("control properties_json must be a JSON object.")
            result = api.issue_properties(args[0], properties)
        elif command == "rename":
            result = api.rename_device(args[0], args[1])
        elif command == "weather":
            if not _validate_lat_lon(args[0], args[1]):
                raise ValueError("weather requires valid lat/lon ranges: lat [-90,90], lon [-180,180].")
            codes = _parse_json_arg(args[2], "codes_json") if len(args) > 2 else None
            if codes is not None and not isinstance(codes, list):
                raise ValueError("weather codes_json must be a JSON array.")
            result = api.get_weather(args[0], args[1], codes)
        elif command == "sms":
            result = api.send_sms(args[0])
        elif command == "voice":
            result = api.send_voice(args[0])
        elif command == "mail":
            result = api.send_mail(args[0], args[1])
        elif command == "push":
            result = api.send_push(args[0], args[1])
        elif command == "stats_config":
            result = api.get_statistics_config()
        elif command == "stats_data":
            start_time = args[3]
            end_time = args[4]
            if not (_validate_time_yyyyMMddHH(start_time) and _validate_time_yyyyMMddHH(end_time)):
                raise ValueError("stats_data start/end must use yyyyMMddHH format.")
            if not _validate_stats_time_window(start_time, end_time):
                raise ValueError("stats_data time window must be within 24 hours and end >= start.")
            result = api.get_statistics_data(args[0], args[1], args[2], start_time, end_time)
        elif command == "ipc_pic_fetch":
            consent = args[1] == "1"
            pic_count = int(args[2]) if len(args) > 2 else None
            home_id = args[3] if len(args) > 3 else None
            result = api.ipc_ai_capture_pic_allocate_and_fetch(
                args[0], user_privacy_consent_accepted=consent,
                pic_count=pic_count, home_id=home_id,
            )
        elif command == "ipc_video_fetch":
            duration = int(args[1])
            consent = args[2] == "1"
            home_id = args[3] if len(args) > 3 else None
            result = api.ipc_ai_capture_video_allocate_and_fetch(
                args[0], video_duration_seconds=duration,
                user_privacy_consent_accepted=consent, home_id=home_id,
            )

        _print_json(result)
    except ValueError as e:
        _print_error(command, raw_args, str(e))
        sys.exit(_EXIT_CODE_USAGE)
    except TuyaAPIError as e:
        _print_error(command, raw_args, str(e), e.code)
        sys.exit(_EXIT_CODE_RUNTIME)
    except requests.exceptions.Timeout:
        _print_error(command, raw_args, "Request timed out. Please try again later.")
        sys.exit(_EXIT_CODE_RUNTIME)
    except requests.exceptions.ConnectionError:
        _print_error(command, raw_args, "Unable to connect to Tuya API. Please check your network.")
        sys.exit(_EXIT_CODE_RUNTIME)
    except requests.exceptions.HTTPError as e:
        _print_error(command, raw_args, f"HTTP error from Tuya API: {e}")
        sys.exit(_EXIT_CODE_RUNTIME)
    except requests.exceptions.RequestException as e:
        _print_error(command, raw_args, f"Request failed: {e}")
        sys.exit(_EXIT_CODE_RUNTIME)
    except json.JSONDecodeError:
        _print_error(command, raw_args, "Received non-JSON response from Tuya API.")
        sys.exit(_EXIT_CODE_RUNTIME)


if __name__ == "__main__":
    main()

FILE:references/notifications.md
# Notifications

All notification APIs are **self-send** mode — they can only send messages to the currently logged-in user.

## 1. Send SMS

Send an SMS to the current user's bound phone number.

**Request**

```
POST /v1.0/end-user/services/sms/self-send
```

**Request Body**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| message | String | Yes | SMS body text (without signature) |

**Request Example**

```json
{
  "message": "Smart home security alert: Your living room door/window sensor detected an abnormal opening. Please verify immediately."
}
```

**Business Rules**
- The phone number must match the one bound to the current user's account
- SMS signature is fixed as "Smart Life" and does not need to be passed as a parameter

**Error Codes**

| Code | Description |
|------|-------------|
| 20001 | Invalid phone number |
| 20002 | Same phone number exceeded 15 messages within 24 hours |
| 20003 | Same phone number sent identical content more than 2 times within 50 seconds |
| 20005 | Can only send SMS to the phone number bound to this account |

---

## 2. Send Voice Call

Send a voice notification to the current user's bound phone number.

**Request**

```
POST /v1.0/end-user/services/voice/self-send
```

**Request Body**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| message | String | Yes | Voice broadcast content |

**Request Example**

```json
{
  "message": "Smart home security alert: Your living room door/window sensor detected an abnormal opening. Please verify immediately."
}
```

**Error Codes**

| Code | Description |
|------|-------------|
| 20001 | Invalid phone number |
| 40002 | Same phone number exceeded 15 voice calls within 24 hours |
| 40003 | Same phone number sent identical content more than 2 times within 50 seconds |
| 40005 | Can only send voice calls to the phone number bound to this account |

---

## 3. Send Email

Send an email to the current user's bound email address.

**Request**

```
POST /v1.0/end-user/services/mail/self-send
```

**Request Body**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| subject | String | Yes | Email subject |
| content | String | Yes | Email body |

**Request Example**

```json
{
  "subject": "Device Offline Notification",
  "content": "Your living room smart air conditioner has gone offline. Please check the device network connection."
}
```

**Error Codes**

| Code | Description |
|------|-------------|
| 30001 | Invalid email address |
| 30002 | Same email exceeded 30 emails within 24 hours |
| 30003 | Same email sent identical content more than 2 times within 50 seconds |
| 30004 | Can only send emails to the email address bound to this account |

---

## 4. Send App Push Notification

Send an App notification bar push to the current user.

**Request**

```
POST /v1.0/end-user/services/push/self-send
```

**Request Body**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| subject | String | Yes | Push notification title |
| content | String | Yes | Push notification content |

**Request Example**

```json
{
  "subject": "Security Alert Notification",
  "content": "Your porch smart camera detected a moving object. Please check the live feed."
}
```

**Error Codes**

| Code | Description |
|------|-------------|
| 50001 | Can only send push notifications to this account |

---

## Common Error Codes

The following error codes apply to all notification APIs:

| Code | Description |
|------|-------------|
| 500 | System error |
| 10001 | Invalid parameter |
| 10010 | End user does not exist |
| 10011 | End user has no bound contact method |

FILE:references/device-query.md
# Device Query

## 1. List All User Devices

Get all devices under the current user's authorization.

**Request**

```
GET /v1.0/end-user/devices/all
```

**Request Parameters**: None

**Response**

```json
{
  "success": true,
  "result": {
    "devices": [
      {
        "category": "dj",
        "category_name": "Light Source",
        "device_id": "0620068884f3eb414579",
        "name": "Living Room Smart Ceiling Light",
        "online": true,
        "product_id": "bcsbx******ss4xxx"
      }
    ],
    "total": 3
  }
}
```

**Device Fields**

| Field | Type | Description |
|-------|------|-------------|
| device_id | String | Device ID |
| name | String | Device name |
| category | String | Product category code |
| category_name | String | Category display name |
| product_id | String | Product ID |
| online | Boolean | Whether the device is online |
| total | Integer | Total device count (sibling of devices) |

---

## 2. List All Devices in a Home

Get all devices under a specified home.

**Request**

```
GET /v1.0/end-user/homes/{home_id}/devices
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| home_id | String | Yes | Home ID |

**Response**

```json
{
  "success": true,
  "result": {
    "devices": [
      {
        "category": "dj",
        "category_name": "Light Source",
        "device_id": "0620068884f3eb414579",
        "name": "Living Room Smart Ceiling Light",
        "online": true,
        "room_id": "123123"
      }
    ]
  }
}
```

**Device Fields**

| Field | Type | Description |
|-------|------|-------------|
| device_id | String | Device ID |
| name | String | Device name |
| category | String | Product category code |
| category_name | String | Category display name |
| online | Boolean | Whether the device is online |
| room_id | String | Room ID (returned when the device is assigned to a room, optional) |

---

## 3. List All Devices in a Room

Get all devices under a specified room.

**Request**

```
GET /v1.0/end-user/homes/room/{room_id}/devices
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| room_id | String | Yes | Room ID |

**Response**

```json
{
  "success": true,
  "result": {
    "devices": [
      {
        "category": "dj",
        "category_name": "Light Source",
        "device_id": "0620068884f3eb414579",
        "name": "Living Room Smart Ceiling Light",
        "online": true
      }
    ]
  }
}
```

**Device Fields**

| Field | Type | Description |
|-------|------|-------------|
| device_id | String | Device ID |
| name | String | Device name |
| category | String | Product category code |
| category_name | String | Category display name |
| online | Boolean | Whether the device is online |

---

## 4. Get Single Device Detail

Query detailed information of a single device, including current property states.

**Request**

```
GET /v1.0/end-user/devices/{device_id}/detail
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| device_id | String | Yes | Device ID |

**Response (device exists)**

```json
{
  "success": true,
  "t": 1773473029779,
  "result": {
    "device_id": "0620068884f3eb414579",
    "name": "Living Room Smart Ceiling Light",
    "category": "dj",
    "category_name": "Light Source",
    "product_name": "WiFi Smart Light",
    "online": true,
    "firmware_version": "1.0.0",
    "firmware_update_available": false,
    "properties": {
      "bright_value": 100,
      "control_data": "1000000000000000a010a",
      "countdown": 0,
      "do_not_disturb": false,
      "switch_led": true,
      "work_mode": "colour"
    }
  }
}
```

**Response (device not found)**

```json
{
  "success": true,
  "t": 0,
  "result": null
}
```

**Detail Fields**

| Field | Type | Description | Default |
|-------|------|-------------|---------|
| device_id | String | Device ID | "" |
| name | String | Device name | "" |
| category | String | Product category code | "" |
| category_name | String | Category display name | "" |
| product_name | String | Product name | "" |
| online | Boolean | Whether the device is online | false |
| firmware_version | String | Firmware version | "" |
| firmware_update_available | Boolean | Whether a firmware update is available | false |
| properties | Map\<String, Object\> | Current values of the device's functional properties. Keys are property codes (dp codes), values are current values | {} |

> The `properties` field allows you to directly read the device's current state without making a separate Thing Model query. Property codes (keys) correspond to the `code` field of properties in the Thing Model. Value types depend on the `typeSpec` definition of each property (bool returns true/false, value returns a number, enum returns an enum string, etc.).

> All device list APIs return the full set of devices in a single response (no pagination). The `total` field indicates the total device count.

FILE:references/ipc-cloud-capture.md
# IPC AI — Cloud Capture

Cloud snapshot / short video capture for IPC (camera) devices.

> **Capture flow**: `allocate` requests the device to capture and upload → wait a few seconds → poll `resolve` until the media URL is ready. **Do not** call resolve immediately after allocate.

## 1. Allocate Cloud Capture

Request the device to take a snapshot or record a short video and upload it to cloud storage. Returns storage coordinates only — no media URL.

**Request**

```
POST /v1.0/end-user/ipc/{device_id}/capture/allocate
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| device_id | String | Yes | Device ID |

**Request Body**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| capture_json | String | Yes | JSON string containing capture parameters |

**`capture_json` Fields**

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| device_id | String | Yes | Device ID (must match path parameter) |
| capture_type | String | Yes | `PIC` for snapshot, `VIDEO` for short video |
| pic_count | Integer | No | Number of snapshots. Server clamps to **1–5** |
| video_duration_seconds | Integer | No | Video duration in seconds. Default 10, server clamps to **1–60** |
| home_id | String | No | Home ID. If provided, must match the device owner |

**Request Example**

```json
{
  "capture_json": "{\"device_id\":\"6c95a7a3...\",\"capture_type\":\"PIC\",\"pic_count\":1}"
}
```

**Response**

```json
{
  "success": true,
  "result": {
    "success": true,
    "status": "ACCEPTED",
    "device_id": "6c95a7a3...",
    "capture_type": "PIC",
    "bucket": "ty-cn-storage30",
    "image_object_key": "/path/to/1234567890_0.jpg",
    "encryption_key": "abc123...",
    "upload_async_hint": "bucket/objectKey only mean upload slots were allocated; the device may still be uploading.",
    "pic_count_requested": 1,
    "pic_count_effective": 1,
    "pic_slots": [
      {
        "success": true,
        "image_object_key": "/path/to/1234567890_0.jpg",
        "encryption_key": "abc123..."
      }
    ]
  }
}
```

**Response Fields**

| Field | Type | Description |
|-------|------|-------------|
| success | Boolean | Whether allocation succeeded |
| status | String | `ACCEPTED` or `REJECTED` |
| bucket | String | Cloud storage bucket name |
| image_object_key | String | Image object key (PIC) |
| video_object_key | String | Video object key (VIDEO) |
| cover_image_object_key | String | Video cover image key (VIDEO) |
| encryption_key | String | File encryption key |
| upload_async_hint | String | Reminder that file may still be uploading |
| pic_slots | List | Per-snapshot details (PIC only) |
| video_duration_seconds_effective | Integer | Actual clamped duration (VIDEO only) |

---

## 2. Resolve Capture Access URL

Poll this API after allocate to obtain accessible media URLs. Returns `NOT_READY` while the device is still uploading.

**Request**

```
POST /v1.0/end-user/ipc/{device_id}/capture/resolve
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| device_id | String | Yes | Device ID |

**Request Body**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| resolve_json | String | Yes | JSON string containing resolve parameters |

**`resolve_json` Fields**

| Field | Type | Required | Description |
|-------|------|----------|-------------|
| device_id | String | Yes | Device ID |
| capture_type | String | Yes | `PIC` or `VIDEO` (must match allocate) |
| bucket | String | Yes | Bucket from allocate response |
| image_object_key | String | PIC: Yes | Image object key from allocate |
| video_object_key | String | VIDEO: Yes | Video object key from allocate |
| cover_image_object_key | String | No | Cover image key (VIDEO only) |
| encryption_key | String | No | Encryption key from allocate |
| user_privacy_consent_accepted | Boolean | No | `true`: return decrypted playable URLs; `false`: return raw presigned URLs (encrypted) |
| home_id | String | No | Home ID |

**Request Example**

```json
{
  "resolve_json": "{\"device_id\":\"6c95a7a3...\",\"capture_type\":\"PIC\",\"bucket\":\"ty-cn-storage30\",\"image_object_key\":\"/path/to/1234567890_0.jpg\",\"encryption_key\":\"abc123...\",\"user_privacy_consent_accepted\":true}"
}
```

**Response (ready)**

```json
{
  "success": true,
  "result": {
    "status": "ACCEPTED",
    "decrypt_image_url": "https://...",
    "message_for_user": "ok"
  }
}
```

**Response (not ready — keep polling)**

```json
{
  "success": true,
  "result": {
    "status": "NOT_READY",
    "error_code": "OBJECT_NOT_READY",
    "message_for_user": "object not uploaded yet or empty (check contentLength)"
  }
}
```

**Response Fields**

| Field | Type | Description |
|-------|------|-------------|
| status | String | `ACCEPTED` (URL ready), `NOT_READY` (keep polling) |
| decrypt_image_url | String | Decrypted playable image URL (when consent = true, PIC) |
| decrypt_video_url | String | Decrypted playable video URL (when consent = true, VIDEO) |
| decrypt_cover_image_url | String | Decrypted cover image URL (VIDEO). May be null if cover is not ready yet — check message_for_user |
| raw_presigned_image_url | String | Raw presigned URL (when consent = false, PIC) |
| raw_presigned_video_url | String | Raw presigned URL (when consent = false, VIDEO) |
| raw_presigned_cover_image_url | String | Raw presigned cover URL (VIDEO). May be null if cover is not ready yet |
| error_code | String | `OBJECT_NOT_READY` when file not yet uploaded |
| message_for_user | String | `ok` when all ready; `ok (cover image not ready yet)` when video ready but cover still uploading |

---

## 3. Fusion-level errors (`success: false` on the outer envelope)

Both capture endpoints are implemented by Hawkeye **`IIpcAiCaptureOpenFusionService`**.

Empty/invalid JSON or missing required fields cause Atop to **throw HawkeyeException**; business exceptions from the Biz layer also propagate as exceptions. All exceptions are caught and mapped to **`FusionResult.buildFail`** by **`HawkeyeFusionService`**. The outer **`code` / `msg`** are determined by the exception type and the Fusion / gateway layer (commonly numeric API error numbers such as **`1109`** for bad input).

---

## 4. Recommended Usage

1. **Capture**: call `allocate` → wait a few seconds → poll `resolve` until `status` is not `NOT_READY`.
2. **Video resolve**: when `status` is `ACCEPTED`, video URL is guaranteed. Cover image URL may be null if still uploading — check `message_for_user` for `"ok (cover image not ready yet)"`. You can call resolve again later to get the cover URL.
3. **Privacy**: only set `user_privacy_consent_accepted=true` when the user has explicitly agreed.
4. **Outer failures**: if `success` is `false`, read **`code`** and **`msg`** per §3. Parameter errors and Biz exceptions all produce outer `success: false` with a numeric error code.

---

## 5. Pic Capture: Wait and Retry

The device must finish capturing and uploading before resolve can return an image URL.

- **Initial wait**: sleep **2 seconds** before the first resolve call.
- **Polling**: call resolve every **~2 seconds** until an image URL appears or the timeout (default **30 seconds**) elapses.
- **Retries after timeout**: if polling times out without a URL, retry resolve up to **3 more times** at **3-second intervals**.

**Helper methods** (`scripts/tuya_api.py`):

| Method | Description |
|--------|-------------|
| `ipc_ai_capture_pic_resolve_with_wait(...)` | Wait, poll, and retry resolve given allocate results |
| `ipc_ai_capture_pic_allocate_and_fetch(...)` | Allocate PIC then automatically wait and resolve |

**CLI example**:

```bash
python3 scripts/tuya_api.py ipc_pic_fetch <device_id> 1
# consent=1 (decrypted URL); optional 3rd arg: pic_count; optional 4th: home_id
```

---

## 6. Video Capture: Wait and Retry

The device must finish recording and uploading before resolve can return a playable URL.

- **Minimum wait**: before the first resolve call, wait at least **`max(5, video_duration_seconds_effective) + 2` seconds**.
- **Polling**: call resolve every **~2 seconds** until a video URL appears or polling reaches the timeout (default **120 seconds**).
- **Retries after timeout**: if polling times out without a URL, retry resolve up to **3 more times** at **5-second intervals**.

**Helper methods** (`scripts/tuya_api.py`):

| Method | Description |
|--------|-------------|
| `ipc_ai_capture_video_resolve_with_wait(...)` | Wait, poll, and retry resolve given allocate results |
| `ipc_ai_capture_video_allocate_and_fetch(...)` | Allocate VIDEO then automatically wait and resolve |

**CLI example**:

```bash
python3 scripts/tuya_api.py ipc_video_fetch <device_id> 5 1
# 5-second video, consent=1 (decrypted URL); optional 4th arg: home_id
```

FILE:references/statistics.md
# Data Statistics

## 1. Query Hourly Statistics Configuration

Query the hourly statistics capabilities configured for all devices under the current user. Use this to verify whether a device supports a specific type of data aggregation.

**Request**

```
GET /v1.0/end-user/statistics/hour/config
```

**Request Parameters**: None (user identity is automatically obtained from the login context)

**Response**

```json
{
  "success": true,
  "result": [
    {
      "dev_id": "0620068884f3eb414579",
      "dp_id": 17,
      "dp_code": "ele_usage",
      "statistic_type": "SUM",
      "interval": "hour"
    }
  ]
}
```

**Response Fields**

| Field | Type | Description |
|-------|------|-------------|
| dev_id | String | Device ID |
| dp_id | Integer | Data point ID |
| dp_code | String | Data point code |
| statistic_type | String | Statistic type: SUM (sum), COUNT (count), MAX (maximum), MIN (minimum) |
| interval | String | Statistics interval, fixed as "hour" |

---

## 2. Query Hourly Statistics Values

Get hourly statistics data for a specified device within a specified time range.

**Request**

```
GET /v1.0/end-user/statistics/hour/data
```

**Query Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| dev_id | String | Yes | Device ID |
| dp_code | String | Yes | Data point code (e.g. `ele_usage`) |
| statistic_type | String | Yes | Statistic type (e.g. `SUM`) |
| start_time | String | Yes | Start time, format: `yyyyMMddHH` (e.g. `2024010110`) |
| end_time | String | Yes | End time, format: `yyyyMMddHH` (e.g. `2024010123`) |

> The time range from start_time to end_time cannot exceed 24 hours, and `end_time` must be later than or equal to `start_time`.

**Request Example**

```
GET /v1.0/end-user/statistics/hour/data?dev_id=0620068884f3eb414579&dp_code=ele_usage&statistic_type=SUM&start_time=2024010110&end_time=2024010123
```

**Response**

```json
{
  "success": true,
  "result": [
    {"2024010110": "123.45"},
    {"2024010111": "234.56"},
    {"2024010112": "345.67"}
  ]
}
```

**Response Description**

The return value is an array where each element is a key-value pair:
- **key**: Time in `yyyyMMddHH` format
- **value**: Statistics value, String type

### Usage Workflow

1. First call the "Statistics Configuration Query" to confirm which statistics items are available for the device (dp_code + statistic_type)
2. Then use the dp_code and statistic_type from the configuration to call this API for the actual data
3. If you need statistics spanning more than 24 hours, make multiple requests and aggregate the results yourself
4. For CLI usage, `start_time` and `end_time` are pre-validated locally before sending the request:
   - Must follow `yyyyMMddHH`
   - Must satisfy `end_time >= start_time`
   - Must not exceed 24 hours

FILE:references/device-control.md
# Device Control

## 1. Query Device Thing Model

Query the Thing Model definition of a specified device to understand which functional properties it supports.

**Request**

```
GET /v1.0/end-user/devices/{device_id}/model
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| device_id | String | Yes | Device ID |

**Response**

```json
{
  "success": true,
  "t": 1673423649773,
  "result": {
    "model": "{\"modelId\":\"000004****\",\"services\":[{\"code\":\"\",\"description\":\"\",\"name\":\"\",\"properties\":[{\"code\":\"switch_led\",\"description\":\"Light switch\",\"accessMode\":\"rw\",\"name\":\"Switch\",\"typeSpec\":{\"type\":\"bool\"},\"abilityId\":20},{\"code\":\"bright_value\",\"description\":\"Brightness adjustment\",\"accessMode\":\"rw\",\"name\":\"Brightness\",\"typeSpec\":{\"max\":1000,\"scale\":0,\"type\":\"value\",\"unit\":\"\",\"min\":10,\"step\":1},\"abilityId\":22}]}]}"
  }
}
```

> `result.model` is a JSON string that needs to be parsed again.

**Parsed model structure**

```json
{
  "modelId": "000004****",
  "services": [
    {
      "code": "",
      "name": "",
      "description": "",
      "properties": [
        {
          "abilityId": 20,
          "code": "switch_led",
          "name": "Switch",
          "description": "Light switch",
          "accessMode": "rw",
          "typeSpec": {
            "type": "bool"
          }
        },
        {
          "abilityId": 22,
          "code": "bright_value",
          "name": "Brightness",
          "description": "Brightness adjustment",
          "accessMode": "rw",
          "typeSpec": {
            "type": "value",
            "min": 10,
            "max": 1000,
            "step": 1,
            "unit": "",
            "scale": 0
          }
        }
      ]
    }
  ]
}
```

### Thing Model Structure

**service**

| Field | Type | Description |
|-------|------|-------------|
| code | String | Service identifier; empty string represents the default service |
| name | String | Service name |
| description | String | Service description |
| properties | List | Property definition list |

**property**

| Field | Type | Description |
|-------|------|-------------|
| abilityId | Integer | Ability identifier ID |
| code | String | Property code (used as the key when issuing properties) |
| name | String | Property display name |
| description | String | Property description |
| accessMode | String | Access mode: `ro` read-only, `wr` write-only, `rw` read-write |
| typeSpec | Object | Data type specification |

### typeSpec Type Definitions

**value (numeric)**

```json
{ "type": "value", "min": 0, "max": 100, "step": 1, "unit": "%", "scale": 0 }
```

- `scale`: Decimal multiplier. Actual value = input value / 10^scale. Example: if `scale` is `1` and raw value is `255`, actual value = 255 / 10^1 = 25.5

**bool (boolean)**

```json
{ "type": "bool" }
```

- Values: `true` or `false`

**enum (enumeration)**

```json
{ "type": "enum", "range": ["auto", "cold", "hot", "wind"] }
```

- Value must be within the `range` list

**string**

```json
{ "type": "string", "maxlen": 100 }
```

**Other types**: `float`, `double`, `date`, `raw`, `bitmap`, `struct`, `array` — refer to the Thing Model documentation for details.

**Error Codes**

| Code | Message | Description |
|------|---------|-------------|
| 40000901 | The device does not exist | Device not found |
| 40000903 | The modelId of the device does not exist | Device model not found |

---

## 2. Issue Properties

Send control commands to a specified device.

**Request**

```
POST /v1.0/end-user/devices/{device_id}/shadow/properties/issue
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| device_id | String | Yes | Device ID |

**Request Body**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| properties | String | Yes | Property data in **JSON string format**. Content is a key-value map from property code (dp code) to property value |

> The value of `properties` is a string, not a JSON object. You must serialize the key-value pairs into a string before passing it.

**Request Example**

```json
{
  "properties": "{\"switch_led\": true, \"bright_value\": 500}"
}
```

**Response (success)**

```json
{
  "success": true,
  "t": 1710234567890,
  "result": {}
}
```

### Common Control Scenarios and Property Codes

| User Intent | Common Property Code | Type | Example Value |
|------------|---------------------|------|---------------|
| Turn light on/off | switch_led | bool | true / false |
| Adjust brightness | bright_value | value | 10-1000 |
| Adjust color temperature | temp_value | value | 0-1000 |
| Turn AC on/off | switch | bool | true / false |
| Set temperature | temp_set | value | 16-30 |
| Switch mode | mode | enum | "auto" / "cold" / "hot" |
| Turn plug on/off | switch_1 | bool | true / false |

> These are common examples only. The actual property codes and value ranges are determined by the "Query Device Thing Model" API response.

FILE:references/device-management.md
# Device Management

## Rename Device

Modify a device's custom name by device ID.

**Request**

```
POST /v1.0/end-user/devices/{device_id}/attribute
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| device_id | String | Yes | Device ID |

**Request Body**

| Parameter | Type | Required | Description | Constraints |
|-----------|------|----------|-------------|-------------|
| name | String | Yes | New device name | Max 50 characters, cannot be empty |

**Request Example**

```json
{
  "name": "Master Bedroom Smart Bedside Lamp"
}
```

**Response (success)**

```json
{
  "success": true,
  "t": 1706256000000,
  "result": {
    "device_id": "0620068884f3eb414579",
    "name": "Master Bedroom Smart Bedside Lamp"
  }
}
```

**Response (device not found)**

```json
{
  "success": false,
  "t": 1706256000000,
  "code": "DEVICE_NOT_EXIST_V2",
  "msg": "Device does not exist"
}
```

**Response Fields**

| Field | Type | Description |
|-------|------|-------------|
| result.device_id | String | Device ID |
| result.name | String | Updated device name |

FILE:references/home-and-space.md
# Home and Space Management

## 1. List All Homes

Get all homes the current user has created or joined.

**Request**

```
GET /v1.0/end-user/homes/all
```

**Request Parameters**: None

**Response**

```json
{
  "success": true,
  "result": {
    "homes": [
      {
        "home_id": "123456",
        "name": "Smart Home Apartment",
        "role": "admin",
        "create_time": 1593661208,
        "latitude": {"Value": "30.3"},
        "longitude": {"Value": "120.07"}
      }
    ]
  }
}
```

**Response Fields**

| Field | Type | Description |
|-------|------|-------------|
| home_id | String | Home ID |
| name | String | Home name |
| role | String | User's role in this home. Values: `owner` (home creator), `admin` (administrator), `member` (regular member) |
| create_time | Long | Creation time (Unix timestamp in seconds) |
| latitude | Object | Home latitude (optional). Format: `{"Value": "30.3"}`. Only returned when the home has location set. Can be used as the `lat` parameter for the Weather Query API |
| longitude | Object | Home longitude (optional). Format: `{"Value": "120.07"}`. Only returned when the home has location set. Can be used as the `lon` parameter for the Weather Query API |

---

## 2. List All Rooms in a Home

Get all rooms under a specified home.

**Request**

```
GET /v1.0/end-user/homes/{home_id}/rooms
```

**Path Parameters**

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| home_id | String | Yes | Home ID |

**Response**

```json
{
  "success": true,
  "result": {
    "rooms": [
      {
        "room_id": "123123",
        "name": "Living Room"
      }
    ]
  }
}
```

**Response Fields**

| Field | Type | Description |
|-------|------|-------------|
| room_id | String | Room ID |
| name | String | Room name |

FILE:references/device-message.md
# Device Message Subscription

Real-time WebSocket subscription for Tuya device events. Enables monitoring property changes (brightness, switch state, temperature, etc.) and online/offline transitions as they happen.

> **Server-side only.** The WebSocket client must run on a backend server. Never connect from a browser or mobile app. To push real-time data to a frontend, subscribe server-side and relay via SSE, your own WebSocket, or polling.

**Dependency:** `pip install websockets` (Python 3.7+)

---

## Authentication & Connection

The WebSocket client reuses the same `TUYA_API_KEY` used by the REST API (`tuya_api.py`). No separate credentials are needed.

The WebSocket URI is **auto-detected** from the API key prefix, matching the same convention as the REST base URL:

| API Key Prefix | Region | WebSocket URI |
|---------------|--------|---------------|
| `AY` | China | `wss://wsmsgs.tuyacn.com` |
| `AZ` | US West | `wss://wsmsgs.iot-wus.com` |
| `EU` | Central Europe | `wss://wsmsgs.iot-eu.com` |
| `IN` | India | `wss://wsmsgs.iot-ap.com` |
| `UE` | US East | `wss://wsmsgs.iot-eus.com` |
| `WE` | Western Europe | `wss://wsmsgs.iot-weu.com` |
| `SG` | Singapore | `wss://wsmsgs.iot-sea.com` |

You can pass `uri` explicitly to override if needed.

---

## Client API Summary

| Capability | Usage |
|-----------|-------|
| Register property change handler | `@client.on_property_change` |
| Register online/offline handler | `@client.on_online_status` |
| Register raw message handler | `@client.on_raw_message` |
| Connect (blocking, auto-reconnect) | `await client.connect()` |
| Graceful stop | `client.stop()` |
| Format timestamp | `TuyaDeviceMQClient.format_timestamp(ts_ms)` |

Fatal close codes that stop reconnection: 1002, 1003, 1008, 1011.

Server error detection: any frame with `error`, `errorMsg`, `errorCode` (non-SUCCESS), or `success: false`.

---

## Message Format

All WebSocket messages are JSON objects with two top-level fields:

```json
{
  "data": { ... },
  "eventType": "devicePropertyChange" | "onlineStatusChange"
}
```

### devicePropertyChange

Pushed when one or more device property values change.

```json
{
  "data": {
    "devId": "36040531cc50e35ee60d",
    "status": [
      { "code": "led_switch", "value": true, "time": 1773668532000 },
      { "code": "bright_value", "value": 255, "time": 1773668532000 }
    ]
  },
  "eventType": "devicePropertyChange"
}
```

| Field | Type | Description |
|-------|------|-------------|
| data.devId | String | Device ID (matches `device_id` in REST API) |
| data.status | Array | One or more property changes in this event |
| data.status[].code | String | Property dp code (matches Thing Model `code`) |
| data.status[].value | Any | New value — type depends on the property's typeSpec |
| data.status[].time | Long | Change timestamp (milliseconds) |

#### Value formats by typeSpec

| typeSpec.type | Format | Example |
|---------------|--------|---------|
| bool | `true` / `false` | `"code": "switch_led", "value": true` |
| value | Integer | `"code": "bright_value", "value": 255` |
| enum | String within range | `"code": "work_mode", "value": "scene_3"` |
| string | String | `"code": "flash_scene_3", "value": "ffff5001ff0000"` |

#### Common property codes

| Code | Meaning | Value Type | Example Values |
|------|---------|------------|----------------|
| led_switch / switch_led | Light switch | bool | true, false |
| switch | Generic switch | bool | true, false |
| bright_value | Brightness | value | 10-1000 |
| temp_value | Color temperature | value | 0-1000 |
| work_mode | Working mode | enum | "white", "colour", "scene_1" |
| temp_set | Target temperature (AC) | value | 16-30 |
| mode | AC mode | enum | "auto", "cold", "hot" |
| doorcontact_state | Door/window sensor | bool | true (open), false (closed) |
| pir | Motion detection | enum | "pir" (motion detected) |

Actual codes depend on each device's Thing Model.

### onlineStatusChange

Pushed when a device connects to or disconnects from the network.

```json
{
  "data": {
    "devId": "6c8b3a57470efd4d9cun7h",
    "status": "online",
    "time": 1773668467323
  },
  "eventType": "onlineStatusChange"
}
```

| Field | Type | Description |
|-------|------|-------------|
| data.devId | String | Device ID |
| data.status | String | `"online"` or `"offline"` |
| data.time | Long | Event timestamp (milliseconds) |

### Mapping to REST API

| Message Field | REST API Equivalent | Query Method |
|--------------|---------------------|--------------|
| data.devId | device_id | `GET /v1.0/end-user/devices/{device_id}/detail` |
| status[].code | Thing Model property code | `GET /v1.0/end-user/devices/{device_id}/model` |
| status[].value | Current property value | Also in device detail `properties` field |

---

## Usage

### Quick Start

```python
import asyncio
import os
import sys
sys.path.insert(0, "{baseDir}/scripts")
from tuya_device_mq_client import TuyaDeviceMQClient

async def main():
    client = TuyaDeviceMQClient(api_key=os.environ["TUYA_API_KEY"])

    @client.on_property_change
    async def on_prop(device_id, properties):
        for prop in properties:
            t = TuyaDeviceMQClient.format_timestamp(prop["time"])
            print(f"[{t}] Device {device_id}: {prop['code']} = {prop['value']}")

    @client.on_online_status
    async def on_status(device_id, status, timestamp_ms):
        t = TuyaDeviceMQClient.format_timestamp(timestamp_ms)
        print(f"[{t}] Device {device_id} is now {status}")

    await client.connect()

asyncio.run(main())
```

### Filtering Specific Devices

```python
client = TuyaDeviceMQClient(
    api_key=os.environ["TUYA_API_KEY"],
    device_ids=["device_id_1", "device_id_2"],
)
```

### Property Change with Human-Readable Values

```python
def format_property_value(code: str, value) -> str:
    """Format property values into human-readable descriptions."""
    bool_codes = {"led_switch", "switch_led", "switch", "doorcontact_state"}
    if code in bool_codes:
        if code == "doorcontact_state":
            return "open" if value else "closed"
        return "on" if value else "off"

    mode_labels = {
        "white": "white mode", "colour": "color mode",
        "auto": "auto mode", "cold": "cooling mode", "hot": "heating mode",
    }
    if isinstance(value, str) and value in mode_labels:
        return mode_labels[value]

    return str(value)

@client.on_property_change
async def handle(device_id, properties):
    for prop in properties:
        t = TuyaDeviceMQClient.format_timestamp(prop["time"])
        readable = format_property_value(prop["code"], prop["value"])
        print(f"[{t}] Device {device_id} | {prop['code']}: {prop['value']} ({readable})")
```

### Online Status Monitor with Counters

```python
from collections import defaultdict

stats: dict[str, dict[str, int]] = defaultdict(lambda: {"online": 0, "offline": 0})

@client.on_online_status
async def handle(device_id, status, timestamp_ms):
    t = TuyaDeviceMQClient.format_timestamp(timestamp_ms)
    stats[device_id][status] += 1
    print(f"[{t}] Device {device_id}: {status}")
    s = stats[device_id]
    print(f"         Total: online {s['online']} times, offline {s['offline']} times")
```

### Event-Driven Automation with Throttling

Use case: "When the door sensor opens, turn on the hallway light."

```python
import time
import sys
sys.path.insert(0, "{baseDir}/scripts")
from tuya_device_mq_client import TuyaDeviceMQClient
from tuya_api import TuyaAPI

# Configuration
TRIGGER_DEVICE_ID = "your_trigger_device_id"  # e.g. door sensor
TRIGGER_CODE = "doorcontact_state"
TRIGGER_VALUE = True  # door opened

ACTION_DEVICE_ID = "your_action_device_id"  # e.g. hallway light
ACTION_PROPERTIES = {"switch_led": True}

NOTIFICATION_COOLDOWN_SECONDS = 30 * 60  # 30 minutes

last_notification_time: dict[str, float] = {}

def should_notify(event_key: str) -> bool:
    """Check whether a notification should be sent (throttle control)."""
    now = time.time()
    last_time = last_notification_time.get(event_key, 0)
    if now - last_time >= NOTIFICATION_COOLDOWN_SECONDS:
        last_notification_time[event_key] = now
        return True
    return False

api = TuyaAPI()

client = TuyaDeviceMQClient(
    api_key=os.environ["TUYA_API_KEY"],
    device_ids=[TRIGGER_DEVICE_ID],
)

@client.on_property_change
async def handle(device_id, properties):
    for prop in properties:
        if prop["code"] == TRIGGER_CODE and prop["value"] == TRIGGER_VALUE:
            # Execute device control via REST API
            api.issue_properties(ACTION_DEVICE_ID, ACTION_PROPERTIES)
            print(f"Automation triggered: turned on {ACTION_DEVICE_ID}")

            # Throttled notification
            event_key = f"{device_id}:{TRIGGER_CODE}"
            if should_notify(event_key):
                api.send_push("Automation Triggered", "Door opened — hallway light turned on.")
```

---

## Constraints

- **Server-side only** — credentials must never reach the frontend.
- **Notification throttling is mandatory** — property events fire frequently; minimum 30-minute cooldown for notifications.
- **Same API key** — uses `TUYA_API_KEY`, same as the REST API. No separate WebSocket credentials needed. URI auto-detected from key prefix for all 7 data centers.
- **Complementary to device control** — this handles real-time events; use `tuya_api.py` for device queries and commands.

FILE:references/error-handling.md
# Error Handling

## Common Error Scenarios

| Scenario | How to Handle |
|----------|--------------|
| Device not found (`result` is `null`) | Tell the user: "Cannot find a device named XX, please verify the device name" |
| Device offline (`online` is `false`) | Tell the user: "Device XX is currently offline, please check its power and network connection." Do not attempt to issue commands |
| Device does not support the function | Tell the user: "Device XX does not support XX function", and list the device's supported functions |
| Home/room not found | Tell the user: "Cannot find a home/room named XX" |
| Multiple devices match | List all matching devices with room information and ask the user to choose |
| Notification send failure | Return the specific error code explanation (rate limit, format error, etc.) |
| Property value out of range | Tell the user: "The value XX is outside the supported range (min-max). Please provide a value within range" |
| Read-only property (`accessMode: ro`) | Tell the user: "This property is read-only and cannot be controlled" |
| Invalid CLI JSON input (`control` / `weather` codes) | Return clear parameter error and ask user to provide valid JSON |
| Invalid statistics time window (`stats_data`) | Tell the user start/end must use `yyyyMMddHH`, end >= start, and window <= 24 hours |

## API Error Codes

| Code | Message | Description | Recovery |
|------|---------|-------------|----------|
| 1010 | token invalid | API key has expired | Tell the user the API key needs to be updated |
| 1108 | uri path invalid | API path is incorrect | Check whether the API path is correct |
| 10001 | Invalid parameter | Request parameter error | Verify the parameters format and values |
| 10010 | End user does not exist | User account issue | Check the API key is valid |
| 10011 | End user has no bound contact | Missing phone/email | Tell the user to bind a contact method in the Tuya App |
| 40000901 | The device does not exist | Device not found | Verify the device_id |
| 40000903 | The modelId of the device does not exist | Device model not found | Device may not support Thing Model queries |
| 500 | System error | Server-side error | Retry after a brief delay |
| 429 | Too Many Requests | API rate limited | Retry with exponential backoff and honor `Retry-After` when present |

> API error codes may appear as numeric integers (e.g. `1010`) or string identifiers (e.g. `"DEVICE_NOT_EXIST_V2"`) depending on the endpoint. Handle both forms when matching error codes.

## Notification-Specific Error Codes

For channel-specific error codes (SMS, Voice, Email, App Push), see `references/notifications.md`.

## General Recovery Strategy

When encountering an unresolvable error, guide the user to visit the GitHub repository for the latest announcements and troubleshooting: https://github.com/tuya/tuya-openclaw-skills

## CLI Error Output Contract

The command-line SDK outputs a standardized error block to `stderr`:
- `Error`: human-readable error message
- `Command`: command name
- `Args`: argument summary (JSON)
- `TuyaErrorCode` and `Suggestion`: included when Tuya API returns a known error code

Exit codes:
- `2`: usage or parameter validation error
- `1`: runtime/API/network error

FILE:references/weather.md
# Weather Query

Get current and forecast weather data for a specified latitude/longitude location.

## Request

```
GET /v1.0/end-user/services/weather/recent
```

## Query Parameters

| Parameter | Type | Required | Description |
|-----------|------|----------|-------------|
| codes | String | Yes | JSON array string specifying which weather attributes to query. E.g. `["w.temp","w.humidity","w.condition","w.hour.7"]` |
| lat | String | Yes | Latitude |
| lon | String | Yes | Longitude |

## Request Example

```
GET /v1.0/end-user/services/weather/recent?codes=["w.temp","w.humidity","w.condition","w.hour.7"]&lat=39.9042&lon=116.4074
```

> When constructing URLs manually, the `codes` value must be URL-encoded (e.g. `%5B%22w.temp%22%5D`). The Python SDK handles encoding automatically.

## Response Example

```json
{
  "success": true,
  "result": {
    "data": {
      "w.condition.0": "mixed precipitation",
      "w.condition.1": "mixed precipitation",
      "w.temp.0": 7,
      "w.temp.1": 6,
      "w.temp.2": 6,
      "w.humidity.0": 44,
      "w.humidity.1": 46,
      "w.humidity.2": 48
    },
    "expiration": 15
  }
}
```

## Response Field Description

Response data keys follow the format `{attribute}.{time_index}`:

- `w.temp.0` — Current temperature; `w.temp.1` — Temperature in 1 hour; and so on
- `w.humidity.0` — Current humidity
- `w.condition.0` — Current weather condition (English description)
- `expiration` — Cache expiration time (minutes)

The time granularity is determined by `w.hour.N` in the `codes` parameter. For example, `w.hour.7` returns data for the next 7 hourly time points.

## Available Attribute Codes

Supported attribute codes for this API:

| Code | Description |
|------|-------------|
| w.temp | Temperature |
| w.humidity | Humidity |
| w.condition | Weather condition description (English) |
| w.conditionNum | Weather condition numeric code |
| w.pressure | Atmospheric pressure |
| w.realFeel | Feels-like temperature |
| w.uvi | UV index |
| w.windDir | Wind direction |
| w.windLevel | Wind level |
| w.windSpeed | Wind speed |
| w.hour.N | Time granularity (N = hours, e.g. `w.hour.7` = next 7 hours) |

> Full supported list: condition, conditionNum, dateTime, humidity, mark, pressure, realFeel, sunRiseTimeStamp, sunSetTimeStamp, temp, uvi, windDir, windLevel, windSpeed

## Weather Condition Mapping

| English Description | Chinese | Category |
|--------------------|---------|----------|
| no precipitation | Sunny (晴) | Sunny |
| clear | Clear (晴朗) | Clear |
| partly cloudy | Partly Cloudy (少云) | Cloud |
| mostly cloudy | Mostly Cloudy (多云) | Cloud |
| mixed precipitation | Mostly Sunny (大部晴朗) | Cloud |
| overcast | Overcast (阴) | Cloud |
| foggy | Foggy (雾) | Fog |
| haze fog | Haze (霾) | Cloud |
| drizzle / light rain | Light Rain (小雨) | Rainy |
| precipitation | Rain (雨) | Rainy |
| rain | Moderate Rain (中雨) | Rainy |
| heavy rain | Heavy Rain (大雨) | Rainy |
| heavy precipitation | Rainstorm (暴雨) | Rainy |
| rainy big | Heavy Rainstorm (大暴雨) | Rainy |
| very heavy snow | Extreme Rainstorm (特大暴雨) | Rainy |
| shower | Shower (阵雨) | Rainy |
| shower with thunder | Thundershower (雷阵雨) | Rainy |
| shower with hail | Thundershower with Hail (雷阵雨伴有冰雹) | Snow |
| thunderstorms | Thunderstorm (雷暴) | Rainy |
| light snow | Light Snow (小雪) | Snow |
| snow | Snow / Moderate Snow (雪/中雪) | Snow |
| heavy snow | Heavy Snow (大雪) | Snow |
| blizzard | Blizzard (暴雪) | Snow |
| flurries | Snow Showers (阵雪) | Rainy |
| possible flurries | Light Snow Showers (小阵雪) | Snow |
| light sleet | Sleet (雨夹雪) | Snow |
| ice rain | Freezing Rain (冻雨) | Rainy |
| frozen fog | Frozen Fog (冻雾) | Snow |
| ice particle | Ice Pellets (冰粒) | Snow |
| heavy sleet | Needle Ice (冰针) | Snow |
| sleet | Hail (冰雹) | Snow |
| blowing sand | Blowing Sand (扬沙) | Sunny |
| float dust | Floating Dust (浮尘) | Sunny |
| dangerously windy | Sandstorm (沙尘暴) | Sunny |
| strong sand storm | Severe Sandstorm (强沙尘暴) | Rainy |
| dust devil | Dust Devil (尘卷风) | Sunny |

## Common City Coordinates

This API requires latitude and longitude. Here are reference coordinates for common cities:

| City | Latitude (lat) | Longitude (lon) |
|------|----------------|-----------------|
| Beijing | 39.9042 | 116.4074 |
| Shanghai | 31.2304 | 121.4737 |
| Guangzhou | 23.1291 | 113.2644 |
| Shenzhen | 22.5431 | 114.0579 |
| Hangzhou | 30.2741 | 120.1551 |
| Chengdu | 30.5728 | 104.0668 |
| Wuhan | 30.5928 | 114.3055 |
| Nanjing | 32.0603 | 118.7969 |
| Chongqing | 29.5630 | 106.5516 |
| Xi'an | 34.3416 | 108.9398 |

> If the user does not provide coordinates, ask for their city and use the corresponding latitude/longitude to call the API.

FILE:references/api-conventions.md
# API Conventions

## API Key Prefix → Data Center Mapping

The base URL is automatically resolved from the first two characters after `sk-` in the API key:

| Prefix | Region | Base URL |
|--------|--------|----------|
| AY | China Data Center | https://openapi.tuyacn.com |
| AZ | US West Data Center | https://openapi.tuyaus.com |
| EU | Central Europe Data Center | https://openapi.tuyaeu.com |
| IN | India Data Center | https://openapi.tuyain.com |
| UE | US East Data Center | https://openapi-ueaz.tuyaus.com |
| WE | Western Europe Data Center | https://openapi-weaz.tuyaeu.com |
| SG | Singapore Data Center | https://openapi-sg.iotbing.com |

How to obtain an API key:
- China Mainland users: Get from https://tuyasmart.com/
- International users: Get from https://tuya.ai/
- Different regions use different API service domains, which must match your account registration region

## Request Format

All APIs use the configured Base URL concatenated with the path. Examples:

```
GET {base_url}/v1.0/end-user/homes/all
POST {base_url}/v1.0/end-user/devices/{device_id}/shadow/properties/issue
```

Authentication is via the `Authorization: Bearer {Api-key}` header (handled automatically by the Python SDK).

## Response Format

APIs return a unified structure:

**Success response**:
```json
{
  "success": true,
  "t": 1710234567890,
  "result": { ... }
}
```

**Error response**:
```json
{
  "success": false,
  "code": 1108,
  "msg": "uri path invalid"
}
```

- When `success` is `true`, the result is in the `result` field
- When `success` is `false`, error details are in the `code` and `msg` fields
- The Python SDK automatically checks `success` and raises `TuyaAPIError` on failure
- HTTP 429 and transient 5xx responses are retried automatically with backoff