SmartThings Schema Connector Python SDK

The SmartThings Schema Connector Python SDK is a package that simplify resources of Schema Connector instances through built-in interfaces.

Installation

Install it using pip:

pip install st-schema-python

SchemaConnector structure

Using class inheritance, we'll gain access to a series of resources to control the request and response data of Interaction types.

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

Note: If any resource handler is not implemented but gets used by the SchemaConnector. interaction_handler built-in method, a NotImplementedError exception will be raised.

SchemaDevice definition.

SchemaDevice instance supporting the minimal requirements to create a virtual device at the SmartThings ecosystem.

1. Device definition using the SchemaDevice class and the set_mn instance method to specify the manufacturer's information:

from stschema import SchemaDevice


my_device = SchemaDevice(
    '{{external_device_id}}',
    '{{friendly_name}}',
    '{{device_handler_type}}'
)

my_device.set_mn(
    '{{manufacturer_name}}',
    '{{model_name}}'
)

2. States definition applying the set_state instance method:

my_device.set_state(
    'st.{{capability_id}}',
    '{{attribute}}',
    '{{value}}'
)

SchemaConnector as a web-service with the Http.server built-in module.

Using the Python's built-in module Http.server, here's an example of an application that will host our Webhook endpoint and our SchemaConnector instance to create and control a virtual switch at the SmartThings app.

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()

Notice that the SchemaConnector.grant_callback_access built-in resource hasn't been implemented. In this case, when the Schema Connector instance gets integrated for the first time at the SmartThings ecosystem, the NotImplementedError exception will be raised as following:

...
NotImplementedError: [grant_callback_access] - Interaction resource handler not implemented

Developer Note.

Before pushing any updates into the SmartThings Schema Connector Python SDK, please install pytest and execute the follwing command to run the full test suite.

python3 -m pytest -p no:cacheprovider

To learn more about SmartThings Schema Connector integrations, visit the SmartThings Community Forums.