Touch to Identify

An example of how to use the REST API with a simple script that identifies devices by touch.


A popular feature in DT Studio for both our customers and us internally is the ability to locate a sensor by simply touching it. Like any other function in DT Studio, the implementation is based on our REST API, and we will here show how simple it is to recreate using the stream endpoint.


  • Basic Auth For simplicity, we here use Basic Auth for authentication. We recommend replacing this with an OAuth2 flow for integrations more complex than local experimentation.

  • Service Account Credentials You must create and know the credentials of a Service Account. Any role will suffice.

  • Streaming Best Practices While this example is based on our other example for Streaming Events, it does not implement the same retry policy or other best practices as it is not the focus here. You are, however, free to combine the two examples for a more robust touch-event-listening loop.

Example Code

The following points summarize the provided example code.

  • Sends a GET request to the REST API to initialize an event stream.

  • Keep the TCP connection open while receiving events.

  • When receiving a Touch Event, fetch and print the source device information.

  • Break the stream.

Environment Setup

If you wish to run the code locally, make sure you have a working runtime environment.

Python 3.9
Python API
Node.js 14
Python 3.9

The following packages are required by the example code and must be installed.

pip install requests==2.25.1
Python API

The latest version of our Python API can be installed through pip.

pip install --upgrade disruptive
Node.js 14

The following modules are required by the example code and must be installed.

npm install [email protected]
npm install [email protected]
npm install [email protected]


The following code snippet implements the touch-to-identify listening loop.

Python 3.9
Python API
Node.js 14
Python 3.9
import os
import json
import requests # pip install requests==2.25.1
# Service Account credentials.
# Construct API URL.
PROJECT_ID = os.environ.get('DT_PROJECT_ID')
DEVICES_STREAM_URL = '{}projects/{}/devices:stream'.format(
if __name__ == '__main__':
# Set up a stream connection.
print('Waiting for touch event...')
stream = requests.get(
'event_types': ['touch'],
# Iterate through the events as they come in (one event per line).
for line in stream.iter_lines():
# Decode the response payload and isolate event dictionary.
payload = json.loads(line.decode('ascii'))
# Halt at errors
if 'result' not in payload.keys():
# Fetch touched device blob for more detailed information.
event = payload['result']['event']
device = requests.get(
url=API_BASE + event['targetName'],
# Print some device information.
print('\nTouch event received:')
print(json.dumps(device, indent=4))
# Stop stream as we've found a device.
Python API
import os
import disruptive as dt
# Environment variables for authentication credentials and target.
KEY_ID = os.environ.get('DT_SERVICE_ACCOUNT_KEY_ID')
PROJECT_ID = os.environ.get('DT_PROJECT_ID')
if __name__ == '__main__':
# Authenticate using Service Account credentials.
dt.default_auth = dt.Auth.service_account(KEY_ID, SECRET, EMAIL)
# Enable logging to see what's happening under the hood.
dt.log_level = 'debug'
# Set up a stream for all touch events in a project.
for e in dt.Stream.event_stream(PROJECT_ID, event_types=['touch']):
# When a touch event is received, fetch the source device.
touched_device = dt.Device.get_device(e.device_id)
# Print information about the touched device.
# Stop the stream as we've found a device.
Node.js 14
const EventSource = require("eventsource") // npm install [email protected]
const jwt = require('jsonwebtoken') // npm install [email protected]
const axios = require('axios').default // npm install [email protected]
// Service Account credentials
const serviceAccountKeyID = process.env.DT_SERVICE_ACCOUNT_KEY_ID
const serviceAccountSecret = process.env.DT_SERVICE_ACCOUNT_SECRET
const serviceAccountEmail = process.env.DT_SERVICE_ACCOUNT_EMAIL
// Construct API URL
const projectID = process.env.DT_PROJECT_ID
const apiBase = ''
const devicesStreamUrl = apiBase + `projects/${projectID}/devices:stream`
async function main() {
let retryCount = 0
let stream
async function setupStream() {
// Since EventSource does not support setting HTTP headers, we need to use
// OAuth for authentication. The received access token will be set as a
// `token` query parameter when setting up the stream in the next step.
const accessToken = await getAccessToken(
// Add query parameters to the URL
let url = devicesStreamUrl
url += `?eventTypes=touch` // Filters out all events except touch
url += `&token=${accessToken}` // The access token for authentication
// Set up a new stream with callback functions for messages
console.log('Waiting for touch event...')
stream = new EventSource(url)
stream.onmessage = handleStreamMessage
async function handleStreamMessage(message) {
// Parse the payload as JSON
const data = JSON.parse(
if (!data || !data.result) {
// Parse the event object
const event = data.result.event
// Close stream as we've found a device
// Send GET request to fetch device information.
const response = await axios({
method: 'GET',
url: apiBase + event.targetName,
auth: {
username: process.env.DT_SERVICE_ACCOUNT_KEY_ID,
password: process.env.DT_SERVICE_ACCOUNT_SECRET,
// Print response contents.
console.log('\nTouch event received:')
console.log(JSON.stringify(, null, 2))
main().catch((err) => {console.log(err)})
// Fetches an access token. See the following guide for documentation:
async function getAccessToken(keyID, email, secret) {
// Construct the JWT header
let jwtHeaders = {
'alg': 'HS256',
'kid': keyID,
// Construct the JWT payload
let jwtPayload = {
'iat': Math.floor( / 1000), // current unixtime
'exp': Math.floor( / 1000) + 3600, // expiration unixtime
'aud': '',
'iss': email,
// Sign and encode JWT with the secret
const jwtEncoded = jwt.sign(
header: jwtHeaders,
algorithm: 'HS256',
// Prepare POST request data
const requestObject = {
'assertion': jwtEncoded,
'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
// Converts the requestObject to a Form URL-Encoded string
const requestData = Object.keys(requestObject).map(function(key) {
return encodeURIComponent(key)+'='+encodeURIComponent(requestObject[key])
// Exchange JWT for access token
const accessTokenResponse = await axios({
method: 'POST',
url: '',
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
data: requestData,
}).catch(function (error) {
if (error.response) {
throw error
// Return the access token in the request

Expected Output

Once a touch event has been received, the device information is printed and stream terminated.

Waiting for touch event...
Touch event received:
"name": "projects/c0md3mm0c7bet3vico8g/devices/emuc0uc989qdqebrvv29so0",
"type": "touch",
"labels": {
"name": "my-favorite-sensor",
"virtual-sensor": ""
"reported": {
"networkStatus": {
"signalStrength": 99,
"rssi": 0,
"updateTime": "2021-03-03T17:21:53.080314Z",
"cloudConnectors": [
"id": "emulated-ccon",
"signalStrength": 99,
"rssi": 0
"transmissionMode": "LOW_POWER_STANDARD_MODE"
"batteryStatus": null,
"touch": {
"updateTime": "2021-03-03T17:22:51.059514Z"
Terminating stream.