heimanhome-mqtt 1.0.1

HeimanHome MQTT SDK - Python SDK for Heiman IoT Platform

HeimanHome MQTT SDK

用于与海曼智能家居 MQTT 网关进行通信的 Python SDK，提供设备属性读写、功能调用、事件上报、子设备管理、场景自动化、固件升级等完整功能。

功能特性

• MQTT 连接管理：支持 SSL/TLS 连接、自动重连
• 设备属性操作：读取、写入、上报设备属性
• 设备功能调用：远程调用设备功能/方法
• 事件上报：设备事件实时上报到云端
• 网关与子设备管理：子设备注册、上下线、属性透传
• 场景与自动化：场景的增删改查与执行
• 固件管理：固件版本上报与升级
• 安全防护：布防/撤防、告警处理

安装

pip install heimanhome-mqtt

Python 版本要求：>=3.10

快速开始

import asyncio
from heimanhome_mqtt import HeimanMqttSdk

async def main():
    # 初始化 SDK
    sdk = HeimanMqttSdk(
        secure_id="your_secure_id",
        secure_key="your_secure_key",
        access_token="your_access_token",  # 可选
        user_id="your_user_id",            # 可选
        mqtt_broker="spmqtt.heiman.cn",
        mqtt_port=1884,
    )

    # 连接到 MQTT Broker
    await sdk.connect()
    print(f"已连接，client_id={sdk.client_id}")

    # 上报设备属性
    await sdk.device.report_properties(
        product_id="your_product_id",
        device_id="your_device_id",
        properties={"temperature": 25.5, "humidity": 60},
    )

    # 断开连接
    await sdk.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

初始化参数

参数	类型	必填	默认值	说明
secure_id	str	是	—	MQTT 认证 Secure ID
secure_key	str	是	—	MQTT 密码生成 Secure Key
access_token	str	否	None	OAuth2 Access Token
user_id	str	否	None	用户 ID，用于 Topic 订阅
mqtt_broker	str	否	spmqtt.heiman.cn	MQTT Broker 地址
mqtt_port	int	否	1884	MQTT Broker 端口
ssl_enabled	bool	否	True	是否启用 SSL/TLS
keepalive	int	否	60	心跳保活间隔（秒）
auto_reconnect	bool	否	True	是否自动重连

核心功能

设备操作 (sdk.device)

上报设备属性

await sdk.device.report_properties(
    product_id="your_product_id",
    device_id="your_device_id",
    properties={"temperature": 25.5, "humidity": 60, "battery": 85},
)

读取设备属性

result = await sdk.device.read_properties(
    product_id="your_product_id",
    device_id="your_device_id",
    property_identifiers=["temperature", "humidity"],
    timeout=10.0,
)
print(result)

写入/控制设备属性

success = await sdk.device.write_properties(
    product_id="your_product_id",
    device_id="your_device_id",
    properties={"targetTemperature": 26.0, "mode": "cool"},
)

调用设备功能

result = await sdk.device.invoke_function(
    product_id="your_product_id",
    device_id="your_device_id",
    function_id="playVoice",
    inputs=[{"name": "text", "value": "Hello World"}],
)

上报设备事件

await sdk.device.report_event(
    product_id="your_product_id",
    device_id="your_device_id",
    event_id="LowBatteryAlarm",
    data={"batteryLevel": 15, "alarmType": "low_battery"},
)

时间同步

await sdk.device.time_sync(
    product_id="your_product_id",
    device_id="your_device_id",
)

网关操作 (sdk.gateway)

子设备注册

await sdk.gateway.child_register(
    product_id="gateway_product_id",
    device_id="gateway_device_id",
    child_device_id="child_device_id",
    child_product_id="child_product_id",
    child_device_name="Child Sensor",
    child_name="sensor_001",
)

子设备上下线

# 子设备上线
await sdk.gateway.child_online(
    product_id="gateway_product_id",
    device_id="gateway_device_id",
    child_device_id="child_device_id",
    child_product_id="child_product_id",
    child_device_name="sensor_001",
    keep_online_timeout=-1,
    keep_online=1,
)

# 子设备下线
await sdk.gateway.child_offline(
    product_id="gateway_product_id",
    device_id="gateway_device_id",
    child_device_id="child_device_id",
)

子设备属性上报

await sdk.gateway.child_report_properties(
    gateway_product_id="gateway_product_id",
    gateway_device_id="gateway_device_id",
    child_device_id="child_device_id",
    properties={"temperature": 22.5, "humidity": 55, "battery": 90},
)

查询网关拓扑

await sdk.gateway.get_topo(
    product_id="gateway_product_id",
    device_id="gateway_device_id",
)

场景操作 (sdk.scene)

# 查询场景列表
await sdk.scene.list_scenes(product_id="your_product_id", device_id="your_device_id")

# 添加场景
await sdk.scene.add_scene(
    product_id="your_product_id",
    device_id="your_device_id",
    scene_data={"name": "离家模式", "actions": [...]},
)

# 执行场景
await sdk.scene.execute_scene(
    product_id="your_product_id",
    device_id="your_device_id",
    scene_id="scene_001",
)

固件操作 (sdk.firmware)

await sdk.firmware.report_version(
    product_id="your_product_id",
    device_id="your_device_id",
    version="1.2.3",
    properties={"MCU": "1.0.1", "WIFI": "2.1.0"},
)

回调注册

注册回调函数以接收设备消息和事件：

def on_device_update(device_id: str, data: dict):
    print(f"设备 {device_id} 属性更新: {data}")

def on_event(device_id: str, event_type: str, data: dict):
    print(f"设备 {device_id} 事件: {event_type} - {data}")

sdk.register_device_callback(on_device_update)
sdk.register_event_callback(on_event)

异常处理

SDK 定义了以下异常类，建议在实际使用中进行捕获处理：

异常类	说明
HeimanAuthError	认证失败（如缺少 secure_id/secure_key）
HeimanConnectionError	网络连接错误
HeimanMqttError	MQTT 操作错误
HeimanTimeoutError	请求超时
HeimanPayloadError	无效 Payload
HeimanTopicError	无效 Topic

from heimanhome_mqtt import (
    HeimanMqttSdk,
    HeimanAuthError,
    HeimanConnectionError,
    HeimanTimeoutError,
)

try:
    sdk = HeimanMqttSdk(secure_id="...", secure_key="...")
    await sdk.connect()
except HeimanAuthError as e:
    print(f"认证失败: {e}")
except HeimanConnectionError as e:
    print(f"连接失败: {e}")
except HeimanTimeoutError as e:
    print(f"请求超时: {e}")

完整示例

详见 examples/ 目录：

• examples/basic_usage.py — 设备属性读写、功能调用、事件上报
• examples/gateway_usage.py — 网关子设备注册、上下线、拓扑查询

许可证

MIT License