SmartThings 架构连接器 Python SDK
项目描述
SmartThings 架构连接器 Python SDK
SmartThings Schema Connector Python SDK是一个 通过内置接口简化Schema Connector实例资源的包。
安装
使用安装它pip：
pip install st-schema-python
SchemaConnector 结构
使用类继承，我们将获得一系列资源来控制交互类型的请求和响应数据。
from stschema import SchemaConnector
class MyConnector(SchemaConnector):
def __init__(self, *opts):
SchemaConnector.__init__(self, enable_logger=True)
def discovery_handler(self, request_id, access_token):
# The discovery_handler built-in method
# gives access to discoveryRequest data.
#
# SchemaDevice instances must be passed
# as a list argument to discovery_response
# built-in method.
declared_devices = [...]
return self.discovery_response(declared_devices, request_id)
def state_refresh_handler(self, devices, request_id, access_token):
# The state_refresh_handler gives access to
# stateRefreshRequest data.
# A filtered list of SchemaDevice instances
# must be passed as response to the
# state_refresh_response built-in method.
filtered_devices = [...]
return self.state_refresh_response(filtered_devices, request_id)
def command_handler(self, devices, request_id, access_token):
# The command_handler gives access to the
# commandRequest data.
# A list of an updated SchemaDevice instance
# must be passed as response to the
# command_response built-in method.
updated_device = [...]
return self.command_response(updated_device, request_id)
def grant_callback_access(self, callback_authentication, callback_urls):
# Built-in method triggered with the
# grantCallbackAccess interaction type.
pass
def integration_deleted(self, callback_authentication):
# Built-in method triggered with the
# integrationDeleted interactionType.
pass
def interaction_result_handler(self, interaction_result, origin):
# The interaction_result_handler provides
# a description of the error triggered
# between interaction type responses.
pass
注意：如果任何资源处理程序未实现但被SchemaConnector. interaction_handler内置方法使用，NotImplementedError则会引发异常。
SchemaDevice 定义。
SchemaDevice 实例支持在SmartThings 生态系统中创建虚拟设备的最低要求。
- 设备定义使用 SchemaDevice 类和
set_mn实例方法来指定制造商的信息：
from stschema import SchemaDevice
my_device = SchemaDevice(
'{{external_device_id}}',
'{{friendly_name}}',
'{{device_handler_type}}'
)
my_device.set_mn(
'{{manufacturer_name}}',
'{{model_name}}'
)
- 应用
set_state实例方法的状态定义：
my_device.set_state(
'st.{{capability_id}}',
'{{attribute}}',
'{{value}}'
)
SchemaConnector 作为具有Http.server内置模块的 Web 服务。
使用 Python 的内置模块Http.server，这是一个应用程序示例，它将托管我们的 Webhook 端点和我们的SchemaConnector实例，以在SmartThings应用程序中创建和控制虚拟交换机。
from http.server import BaseHTTPRequestHandler, HTTPServer
from stschema import SchemaConnector, SchemaDevice
import json
# MyConnector definition
class MyConnector(SchemaConnector):
def __init__(self, *opts):
SchemaConnector.__init__(self, enable_logger=True)
def discovery_handler(self, request_id, access_token):
# Device definition using the SchemaDevice class
my_switch = SchemaDevice( # Device info
'xyz_example_id_xyz',
'Office light',
'c2c-switch')
my_switch.set_mn( # Manufacturer info
'Switch Mn Example',
'Model X1')
my_switch.set_context(
'Office',
[],
['light'])
declared_devices = [my_switch]
return self.discovery_response(declared_devices, request_id)
def state_refresh_handler(self, devices, request_id, access_token):
# State Refresh Request information
device_id = devices[0]['externalDeviceId']
# SchemaDevice Instance
# and state definition.
my_device = SchemaDevice(device_id)
my_device.set_state(
'st.switch',
'switch',
'on'
)
# Collection of devices, in this
# case, just my_device instance.
filtered_devices = [my_device]
return self.state_refresh_response(filtered_devices, request_id)
def command_handler(self, devices, request_id, access_token):
# Command Request information
device_id = devices[0]['externalDeviceId']
command = devices[0]['commands'][0]
# SchemaDevice instance applying
# the updated state as commanded.
my_device = SchemaDevice(device_id)
my_device.set_state(
'st.switch',
'switch',
command['command'] # 'on' or 'off'
)
# Updated device passed as a list argument.
updated_device = [my_device]
return self.command_response(updated_device, request_id)
def interaction_result_handler(self, interaction_result: dict, origin: str):
print(interaction_result, origin)
pass
# MyConnector instance
my_connector = MyConnector()
class WebhookServer(BaseHTTPRequestHandler):
"""
This class will serve as endpoint to handle
the POST Http Requests sent to the
registered Target Url. Notice that this
webhook instance won't differentiate endpoints.
"""
def do_POST(self):
# POST Http Request handler.
content_length = int(self.headers['Content-Length'])
req_body = self.rfile.read(content_length).decode('utf-8')
# getting JSON body from request.
json_data = json.loads(req_body)
return self._set_response(json_data)
def _set_response(self, data):
# interaction_handler is a
# SchemaConnector built-in method
# that takes the JSON body as
# argument.
connector_handler = my_connector.interaction_handler(data)
# JSON Interaction types responses
res_data = json.dumps(connector_handler).encode('utf-8')
self._set_headers()
self.wfile.write(res_data)
def _set_headers(self):
# Declare application/json
# headers to parse JSON string
# response.
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
if __name__ == '__main__':
server_address = ('', 8000)
http = HTTPServer(server_address, WebhookServer)
http.serve_forever()
请注意，SchemaConnector.grant_callback_access内置资源尚未实现。在这种情况下，当 Schema Connector 实例在 SmartThings 生态系统中首次集成时，NotImplementedError将引发以下异常：
...
NotImplementedError: [grant_callback_access] - Interaction resource handler not implemented
开发人员说明。
在将任何更新推送到SmartThings Schema Connector Python SDK之前，请安装pytest并执行以下命令以运行完整的测试套件。
python3 -m pytest -p no:cacheprovider
要了解有关SmartThings 架构连接器集成的更多信息，请访问SmartThings 社区论坛。
