ha-xthings-cloud

Async Python client for the Xthings Cloud API, designed for Home Assistant integration.

Installation

pip install ha-xthings-cloud

Usage

API Client

import aiohttp
from ha_xthings_cloud import XthingsCloudApiClient

async with aiohttp.ClientSession() as session:
    client = XthingsCloudApiClient(session)

    # Login (supports 2FA)
    result = await client.async_login("user@example.com", "password")

    # Get devices
    devices = await client.async_get_devices()

    # Control devices
    await client.async_switch_on("device-uuid")
    await client.async_brite_brightness("device-uuid", 80)
    await client.async_lock_lock("device-uuid")

WebSocket (Real-time Push)

from ha_xthings_cloud import XthingsCloudWebSocket

def on_status(device_uuid, status):
    print(f"Device {device_uuid}: {status}")

async def on_token_expired():
    # refresh token logic
    pass

ws = XthingsCloudWebSocket(session, token, on_status, on_token_expired)
await ws.async_start()

KVS Signaling (Camera WebRTC)

from ha_xthings_cloud import KvsSignalingClient

kvs = KvsSignalingClient(session, region, channel_arn, credentials)
answer_sdp = await kvs.async_get_answer_sdp(offer_sdp, on_ice_candidate=callback)

Features

• Async HTTP client with aiohttp.ClientSession injection
• Authentication: login, token refresh, 2FA (email/phone)
• Device control: Switch, Plug, Light (brightness/HS color/color temp), Lock, Camera
• WebSocket: real-time device status push with auto-reconnect
• KVS Signaling: AWS Kinesis Video Streams WebRTC SDP exchange with SigV4 auth
• FRP remote access configuration

API Reference

XthingsCloudApiClient

XthingsCloudWebSocket

KvsSignalingClient

License

Apache-2.0