Skip to content

API.md

Latest commit

 

History

History
482 lines (388 loc) · 11.4 KB

API.md

File metadata and controls

482 lines (388 loc) · 11.4 KB

Tuya Cloud API Documentation

This document describes how the Tuya Cloud API integration works.

API Overview

API Version: v2.1 (for historical logs), v1.0 (for device info)

Authentication: HMAC-SHA256 signature + Bearer token

Rate Limits: Varies by endpoint (typically 100-1000 requests/minute)

Base Endpoints:

  • Europe: https://openapi.tuyaeu.com
  • US: https://openapi.tuyaus.com
  • China: https://openapi.tuyacn.com
  • India: https://openapi.tuyain.com

Authentication Flow

1. Get Access Token

Endpoint: POST /v1.0/token?grant_type=1

Request:

GET /v1.0/token?grant_type=1 HTTP/1.1
Host: openapi.tuyaeu.com
client_id: {access_id}
sign: {calculated_signature}
t: {timestamp_ms}
sign_method: HMAC-SHA256

Signature Calculation:

import hashlib
import hmac
import time

def calculate_sign(access_id, access_secret):
    timestamp = str(int(time.time() * 1000))
    message = f"{access_id}{timestamp}"
    sign = hmac.new(
        access_secret.encode('utf-8'),
        message.encode('utf-8'),
        hashlib.sha256
    ).hexdigest().upper()
    return sign, timestamp

Response:

{
  "success": true,
  "result": {
    "access_token": "...",
    "expire_time": 7200,
    "refresh_token": "...",
    "uid": "..."
  }
}

Token Lifetime: 2 hours (7200 seconds)

Refresh: Use refresh_token before expiry to get new token

2. API Request with Token

Headers:

GET /v1.0/devices/{device_id} HTTP/1.1
Host: openapi.tuyaeu.com
client_id: {access_id}
access_token: {token_from_step_1}
sign: {calculated_signature}
t: {timestamp_ms}
sign_method: HMAC-SHA256

Signature for authenticated requests:

def calculate_authenticated_sign(access_id, access_secret, access_token,
                                  method, url, body=""):
    timestamp = str(int(time.time() * 1000))

    # String to sign format
    str_to_sign = f"{method}\n"  # GET, POST, etc.
    str_to_sign += hashlib.sha256(body.encode('utf-8')).hexdigest() + "\n"
    str_to_sign += "\n"  # Headers (empty for most requests)
    str_to_sign += url

    # Combine with client_id, access_token, timestamp
    message = f"{access_id}{access_token}{timestamp}{str_to_sign}"

    # Calculate HMAC-SHA256
    sign = hmac.new(
        access_secret.encode('utf-8'),
        message.encode('utf-8'),
        hashlib.sha256
    ).hexdigest().upper()

    return sign, timestamp

API Endpoints Used

Device Information

Get Device Details

Endpoint: GET /v1.0/devices/{device_id}

Response:

{
  "success": true,
  "result": {
    "id": "bf7b00f283462b0e20eyhi",
    "name": "smart_socket",
    "local_key": "...",
    "category": "cz",
    "product_id": "...",
    "product_name": "Smart Socket",
    "sub": false,
    "online": true,
    "active_time": 1706442123,
    "create_time": 1706442000,
    "update_time": 1706442123,
    "model": "PS-16-EU",
    "icon": "smart/icon/...",
    "ip": "...",
    "time_zone": "+01:00"
  }
}

Get Device Specifications

Endpoint: GET /v1.0/devices/{device_id}/specifications

Response:

{
  "success": true,
  "result": {
    "category": "cz",
    "functions": [
      {
        "code": "switch_1",
        "type": "Boolean",
        "values": "{}"
      },
      {
        "code": "countdown_1",
        "type": "Integer",
        "values": "{\"unit\":\"s\",\"min\":0,\"max\":86400,\"scale\":0,\"step\":1}"
      }
    ],
    "status": [
      {
        "code": "cur_power",
        "type": "Integer",
        "values": "{\"unit\":\"W\",\"min\":0,\"max\":50000,\"scale\":1,\"step\":1}"
      },
      {
        "code": "cur_current",
        "type": "Integer",
        "values": "{\"unit\":\"mA\",\"min\":0,\"max\":30000,\"scale\":3,\"step\":1}"
      },
      {
        "code": "cur_voltage",
        "type": "Integer",
        "values": "{\"unit\":\"V\",\"min\":0,\"max\":3000,\"scale\":1,\"step\":1}"
      },
      {
        "code": "add_ele",
        "type": "Integer",
        "values": "{\"unit\":\"kwh\",\"min\":0,\"max\":50000000,\"scale\":3,\"step\":1}"
      }
    ]
  }
}

Functions vs Status:

  • Functions: Commands you can send to device (switch_1, countdown_1)
  • Status: Properties device reports (cur_power, cur_current, add_ele)

Get Device Functions

Endpoint: GET /v1.0/devices/{device_id}/functions

Response: Same structure as specifications functions array.

Get Shadow Properties

Endpoint: GET /v2.0/cloud/thing/{device_id}/shadow/properties

Purpose: Get ALL data points (DPs) supported by device, including undocumented ones.

Response:

{
  "success": true,
  "result": {
    "properties": [
      {
        "code": "1",
        "type": "bool",
        "value": true
      },
      {
        "code": "4",
        "type": "value",
        "value": 195
      },
      {
        "code": "CH1_RealTemp",
        "type": "value",
        "value": 331
      }
    ]
  }
}

Why use shadow properties?

  • Devices often have undocumented DPs
  • Specifications API may not list all properties
  • Shadow gives complete picture of device capabilities

Code mapping: Numeric codes (1, 4, etc.) vs named codes (CH1_RealTemp)

  • System handles both formats
  • Applies smart defaults when specs missing

Historical Data

Get Report Logs

Endpoint: GET /v2.1/cloud/thing/{device_id}/report-logs

Parameters:

  • start_time: Start timestamp (milliseconds) - usually 0
  • end_time: End timestamp (milliseconds)
  • size: Max events to return (1-100, default: 100)
  • type: Query type (optional, usually omitted)
  • query_key: Specific property to query (optional, omit for all)

Example Request:

GET /v2.1/cloud/thing/bf7b00f283462b0e20eyhi/report-logs?start_time=0&end_time=1706442123000&size=100 HTTP/1.1

Response:

{
  "success": true,
  "result": {
    "has_more": false,
    "list": [
      {
        "code": "cur_power",
        "value": "195",
        "event_time": 1706442100000
      },
      {
        "code": "cur_current",
        "value": "850",
        "event_time": 1706442100100
      },
      {
        "code": "add_ele",
        "value": "1234",
        "event_time": 1706442100200
      },
      {
        "code": "cur_power",
        "value": "200",
        "event_time": 1706442110000
      }
    ],
    "total": 4
  }
}

CRITICAL: API returns most recent N events in time window, NOT chronological!

Example:

Request: start_time=0, end_time=NOW, size=100
Returns: 100 most recent events before NOW (in descending order)

Request: start_time=0, end_time=NOW-1hour, size=100
Returns: 100 most recent events before NOW-1hour (NOT first 100 after 0!)

Response Fields:

  • code: Property name (e.g., "cur_power", "switch_1")
  • value: Raw value as string (needs scaling)
  • event_time: Timestamp in milliseconds
  • has_more: Whether more events exist before oldest returned
  • total: Number of events in this response

Backward Walking Algorithm

Problem: Forward walking creates gaps (misses recent events).

Solution: Walk backward from NOW:

def fetch_all_historical_logs(device_id, start_time, end_time):
    all_events = []
    current_end = end_time

    while current_end > start_time:
        response = api_request(
            f"/v2.1/cloud/thing/{device_id}/report-logs",
            params={
                "start_time": 0,
                "end_time": current_end,
                "size": 100
            }
        )

        events = response['result']['list']
        if not events:
            break

        all_events.extend(events)

        # Move end_time to oldest event minus 1ms
        oldest_timestamp = min(e['event_time'] for e in events)
        current_end = oldest_timestamp - 1

        if not response['result']['has_more']:
            break

    return all_events

Why backward?

  1. API gives most recent N events
  2. Walk backward ensures we capture all events
  3. Forward walking would miss recent events when fetching old data

Typical Usage:

# Full history (7 days back)
now = int(time.time() * 1000)
week_ago = now - (7 * 24 * 60 * 60 * 1000)
events = fetch_all_historical_logs(device_id, week_ago, now)

# Incremental update (since last fetch)
last_timestamp = get_last_timestamp_from_csv(device_name)
now = int(time.time() * 1000)
new_events = fetch_all_historical_logs(device_id, last_timestamp, now)

Data Retention

Tuya Cloud retention: 7 days for free tier

Workaround: Run incremental updates daily to build local history:

# Add to cron (daily at 2am)
0 2 * * * cd /path/to/csv-monitor && csv_fetch_tuya

Rate Limits and Best Practices

Rate Limits

Per endpoint:

  • /v1.0/token: 100 requests/minute
  • /v1.0/devices/*: 1000 requests/minute
  • /v2.1/cloud/thing/*/report-logs: 300 requests/minute

Strategy to avoid limits:

  1. Cache access tokens (2-hour lifetime)
  2. Fetch devices in batches
  3. Use incremental updates (not full history every time)
  4. Implement exponential backoff on 429 errors

Error Handling

Common error codes:

Code Message Solution
1001 Signature mismatch Check HMAC calculation
1004 Token expired Refresh access token
1010 Token invalid Re-authenticate
2006 Device not found Check device ID
2008 Device offline Skip or retry later
2017 Permission denied Check Tuya project permissions

Example error response:

{
  "success": false,
  "code": 1004,
  "msg": "token is expired"
}

Retry Logic

import time

def api_request_with_retry(endpoint, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = make_request(endpoint)
            if response['success']:
                return response

            # Handle specific errors
            if response['code'] == 1004:  # Token expired
                refresh_token()
                continue

            if response['code'] == 2008:  # Device offline
                return None  # Skip this device

        except Exception as e:
            if attempt < max_retries - 1:
                sleep_time = 2 ** attempt  # Exponential backoff
                time.sleep(sleep_time)
                continue
            raise

    return None

API Quotas

Free tier limits (Tuya IoT Platform):

  • 1,000,000 API calls/month
  • 7-day data retention
  • 100 devices max

Paid tier (varies by region):

  • Unlimited API calls
  • Extended data retention (30-90 days)
  • Unlimited devices

Quota monitoring:

  • Check Tuya IoT Platform console for current usage
  • Set up alerts near quota limits
  • Use incremental updates to minimize API calls

Permissions and Troubleshooting

See TUYA_PERMISSIONS.md for required API permissions, device linking, and common error solutions.

API Version Notes

Why v2.1 for historical logs?

v2.0 endpoint (/v2.0/cloud/thing/{id}/report-logs):

  • Limited to specific property codes
  • Doesn't support all device types
  • Returns data in different format

v2.1 endpoint (/v2.1/cloud/thing/{id}/report-logs):

  • Supports all device codes
  • Works with all device categories
  • Consistent response format
  • Uses event_time field (not eventTime)

Endpoint compatibility

fetch_tuya.py uses:

  • v1.0 for authentication and device info
  • v2.0 for shadow properties (most complete)
  • v2.1 for historical logs (best compatibility)

This combination provides maximum device coverage and data completeness.