Python 3 & Flask Webhook Automation

Hello everyone,

For anyone looking to write a webhook automation in python3/flask please see this template below. After fumbling through all the different documentation and dissecting the bad examples in the docs, and reading some super helpful community posts, I finally came up with this. You should literally be able to copy paste it, start up Ngrok, and self publish.

The issue I was having was that the life cycle documentation is unclear if you are following the basic how to. It acts like you should be able to get to the self publish test as long as your app returns status code 200. I couldn’t get my public key until i started returning the challenge code posted by smartthings.

Unfortunately the “how to” is before the lifecycle documentation in the docs.

Here are the helpful links I used

How To:
https://smartthings.developer.samsung.com/develop/getting-started/automation.html

Life Cycle Basics (You need to have the ping response working to get past the first part of the how to)
https://smartthings.developer.samsung.com/develop/guides/smartapps/lifecycles.html

Detailed listing of lifecycle configurations
https://smartthings.developer.samsung.com/develop/api-ref/smartapps-v1.html#

Helpful community post response by Jim Anderson @Jim

It’s a super basic template that does the same example (sort of) that they do in the docs. It does handle all the life cycles, so you can just change them. You may also want to move the header verification before the “PING” IF statement after you get your public key. Thought I would share this since it took me a while, and might save someone else some time :stuck_out_tongue:

# Brendan Cain 2018
# brendancain99 at gmail.com

#Lifecycle Documentation (3 sources)

#from the developer guide page
#https://smartthings.developer.samsung.com/develop/guides/smartapps/lifecycles.html

#from the api page
#https://smartthings.developer.samsung.com/develop/api-ref/smartapps-v1.html#

#Jim anderson - because he is morehelpful than most the documentation :p
#https://community.smartthings.com/t/automation-webhook-asp-net-core/144749


from httpsig.verify import HeaderVerifier, Verifier
from flask import Flask, request, jsonify

app = Flask(__name__)


def get_public_key():
    with open('smartthings_rsa.pub', 'r') as f:
        return f.read()


@app.route('/', methods=['POST'])
def smarthings_requests():
    content = request.get_json()

    print(content)
    if (content['lifecycle'] == 'PING'):
        print("PING")
        challenge = content['pingData']["challenge"]
        data = {'pingData':{'challenge': challenge}}
        return jsonify(data)


    hv = HeaderVerifier(headers=request.headers, secret=get_public_key(), method='POST', path='/')

    if not (hv.verify()):
        # Invalid signature, return 403
        return '', 403


    if (content['lifecycle'] == 'CONFIGURATION' and content['configurationData']['phase'] == 'INITIALIZE'):
        print(content['configurationData']['phase'])
        data = {
                  "configurationData": {
                    "initialize": {
                      "name": "Webhook App Cain",
                      "description": "Webhook App Cain",
                      "id": "cain_webhook_app_page_1",
                      "permissions": [
                        # "l:devices"
                      ],
                      "firstPageId": "1"
                    }
                  }
                }
        return jsonify(data)

    elif (content['lifecycle'] == 'CONFIGURATION' and content['configurationData']['phase'] == 'PAGE'):
        print(content['configurationData']['phase'])
        pageId = content['configurationData']['pageId']

        data = {
                  "configurationData": {
                    "page": {
                      "pageId": "1",
                      "name": "On When Open/Off When Shut WebHook App",
                      "nextPageId": "null",
                      "previousPageId": "null",
                      "complete": "true",
                      "sections": [
                        {
                          "name": "When this opens/closes...",
                          "settings": [
                            {
                              "id": "contactSensor",
                              "name": "Which contact sensor?",
                              "description": "Tap to set",
                              "type": "DEVICE",
                              "required": "true",
                              "multiple": "false",
                              "capabilities": [
                                "contactSensor"
                              ],
                              "permissions": [
                                "r"
                              ]
                            }
                          ]
                        },
                        {
                          "name": "Turn on/off this light...",
                          "settings": [
                            {
                              "id": "lightSwitch",
                              "name": "Which switch?",
                              "description": "Tap to set",
                              "type": "DEVICE",
                              "required": "true",
                              "multiple": "false",
                              "capabilities": [
                                "switch"
                              ],
                              "permissions": [
                                "r",
                                "x"
                              ]
                            }
                          ]
                        }
                      ]
                    }
                  }
                }
        return jsonify(data)


    elif (content['lifecycle'] == 'UPDATE'):
        print(content['lifecycle'])
        data = {'updateData':{}}
        return jsonify(data)


    elif (content['lifecycle'] == 'INSTALL'):
        print(content['lifecycle'])
        data = {'installData':{}}
        return jsonify(data)


    elif (content['lifecycle'] == 'OAUTH_CALLBACK'):
        print(content['lifecycle'])
        data = {'oAuthCallbackData':{}}
        return jsonify(data)

    elif (content['lifecycle'] == 'EVENT'):
        print(content['lifecycle'])
        data = {'eventData':{}}
        return jsonify(data)


    elif (content['lifecycle'] == 'UNINSTALL'):
        print(content['lifecycle'])
        data = {'uninstallData':{}}
        return jsonify(data)


    else:
        return '',404



if __name__ == '__main__':
    app.run('0.0.0.0',debug=True)
4 Likes

Awesome!

I’ve passed this feedback on to the documentation team as well, so they can consider ways to improve the pain points you went through.

1 Like

Just want to say thanks so much for this helpful
Post and information. This is exactly what I need after trying to get through all the info.

1 Like

I’m just glad it was useful to someone else!

1 Like

Hi again Brendan. I’ve got my app published and installed in smart things thanks to your template so much appreciated again. With regard to the configuration for my app I’m looking to see if I can get some help.

Basically I have been using a separate Smart app for some time called open dash which is a Smart App based API for Smartthings. I’ve been polling the “All devices end point” on this API which has been proven to be slow for me hence, the need for me to require a webhook/listening approach to the same functionality. See my post here: Pi Codesys Project

Essentially what I want my newly installed smart app to do is to listen for any event on the devices I have selected within the smartapp and in turn trigger a GET to the Smart things API for those devices. Once I have the return string from this GET I can pass it into my own PLC based application where I handle parsing and everything else.

I understand this seems like non standard usecase for a smartapp but I’m hoping I can get some advice from someone. Regardless I will continue my own learning and research into building this kind of smart app.

If anyone has any input it would be much appreciated.

Have you tried using the smart things api here?

I use it to poll my temp sensors every 5 minutes. I just use the status call on each device ID. You will have to run “locations” and then “devices” so you have those ID’s, then you can just call for status on each device. Then if something meets the requirements, send your GET request.

Hopefully that helps

Thanks for your reply. Yes this is essentially what I’m looking to do.

I’m thinking I will just need to subscribe to all devices I care about within the SmartApp and trigger this GET immediately when I receive an event from these devices. What I need is a stream of events that are the triggers for my GET so I think this will work.

My challenge is that I don’t know node and I’m also still learning Python. Most examples In the docs are in node so it will take some time to get this to work.

By any chance do you have any example code in your python smart app where your subscribing to events.

Thanks again.

So here is my script, but just realize that some of the functions never get called because I only had to use them once to get my IDs. So just change what is inside the “try:” statement to “get_locations()” for example, and then it will print out your location IDs. Then change it to “get_devices_ids()” to get the name and id of each device. get_devices() will give you a ton of other info, do get_device_ids() just pulls the name and ID out

#Brendan Cain 2018
#brendancain99@gmail.com


# https://auth-global.api.smartthings.com
#from flask import Flask, request, Response
from functools import wraps
import csv
from pathlib import Path
import datetime
import requests
import json
import time

### API WEBSITE FOR DOCUMENTATION https://smartthings.developer.samsung.com/develop/api-ref/st-api.html#operation/getDevices
open_weather_map_api_key = "YOUR OPEN WEATHERMAP API KEY HERE"
zipcode = "YOUR ZIPCODE HERE"

home_location =  "YOUR HOME ID HERE"

thermostat = "DEVICE ID HERE"
motion = "DEVICE ID HERE"
basement_door ="DEVICE ID HERE"
front_door ="DEVICE ID HERE"
back_door = "DEVICE ID HERE"
attic_fan = "DEVICE ID HERE"

token = "YOUR TOKEN HERE"

def get_locations():
    headers = {'Authorization': 'Bearer ' +token}
    r = requests.get('https://api.smartthings.com/v1/locations',headers=headers)
    device = r.json()
    print(device)

def get_devices():
    headers = {'Authorization': 'Bearer ' +token}
    r = requests.get('https://api.smartthings.com/v1/devices',headers=headers)
    device = r.json()
    print(device)
	
def get_devices_ids():
    headers = {'Authorization': 'Bearer ' +token}
    r = requests.get('https://api.smartthings.com/v1/devices',headers=headers)
    devices = r.json()
    items = devices["items"]
    for each in items:
        print(each['name'],each['deviceId'])

def get_this_device():
    headers = {'Authorization': 'Bearer ' +token}
    r = requests.get('https://api.smartthings.com/v1/devices',headers=headers)
    devices = r.json()
    items = devices["items"]
    this_item = items[6]
    print(this_item)

def get_temp(device):
    headers = {'Authorization': 'Bearer ' +token}
    url = 'https://api.smartthings.com/v1/devices/' + device + '/status'
    r = requests.get(url, headers=headers)
    # print(r.content)
    data = r.json()
    temp = data['components']['main']['temperatureMeasurement']['temperature']['value']
    return(temp)

def get_all_device_data(device):
    headers = {'Authorization': 'Bearer ' +token}
    url = 'https://api.smartthings.com/v1/devices/' + device + '/status'
    r = requests.get(url, headers=headers)
    # print(r.content)
    data = r.json()
    print(data)

def get_weather():
    r = requests.get('http://api.openweathermap.org/data/2.5/weather?zip='+ zipcode +',us&APPID='+open_weather_map_api_key)
    weather = r.json()
    temp = float(weather["main"]["temp"])
    temp = (temp * 9/5) - 459.67
    # print(temp, " F")
    return temp


def get_thermostat():
    headers = {'Authorization': 'Bearer ' +token}
    url = 'https://api.smartthings.com/v1/devices/' + thermostat + '/status'
    r = requests.get(url, headers=headers)
    data = r.json()
    return(data)


def get_attic_fan():
    headers = {'Authorization': 'Bearer ' +token}
    url = 'https://api.smartthings.com/v1/devices/' + attic_fan + '/status'
    r = requests.get(url, headers=headers)
    data = r.json()
    return(data)

def csv_log():
    basement_door_temp = str(get_temp(basement_door))
    front_door_temp = str(get_temp(front_door))
    back_door_temp = str(get_temp(back_door))
    motion_temp = str(get_temp(motion))
    thermostat_temp = str(get_temp(thermostat))
    fan_data = get_attic_fan()
    thermo_data = get_thermostat()
    thermo_mode = thermo_data['components']['main']['thermostatOperatingState']['thermostatOperatingState']['value']
    fan_status = fan_data['components']['main']['switch']['switch']['value']
    fan_power = fan_data['components']['main']['powerMeter']['power']['value']

    my_file = Path("log_temps.csv")
    my_weather = get_weather()
    if my_file.is_file() == False:
        with open('log_temps.csv', 'w',newline='') as outcsv:
            writer = csv.writer(outcsv)
            writer.writerow(["Date","Time", "Attic", "Ceiling", "Back Door", "Front Door" ,"Outdoor","thermostat","thermostat_mode","fan_status","fan_power"])

        #if file exists, append to it
    with open('log_temps.csv', 'a',newline='') as outcsv:
        writer = csv.writer(outcsv)
        writer.writerow([ datetime.datetime.now().date() ,datetime.datetime.now().time(), basement_door_temp,
                          motion_temp,back_door_temp,front_door_temp,str(my_weather),thermostat_temp,thermo_mode,fan_status,fan_power])



try:

   csv_log()

except:
  print("File Open, Please close")
1 Like

Brendan. Thanks a lot for sharing your script. From looking through it you are not using the Smart App approach and just hitting the Smart Things API passing in your personal token.

Is this correct?. Are you calling this script in a schedule every 5 minutes or are you calling it with triggers from another script?.

Our use cases are somewhat similar in that you are delivering your device data to a csv file whereas I am delivering my data to a PLC application.

Since I am writing all my own automations in a separate PLC program I need the response from switches and motion sensors to be fast so I think I will need to use the standard smart app approach where subscribed events provide me with an stream. Those functions in your script will be very helpful for me so thanks again for sharing.

I need to keep digging into the smart app configurations and subscriptions, I just wish there were some examples in Python. Alas its probably more my beginners experience with Python.

Looks like the docs here have a great starting example for subscriptions and events. Just really need to see how this can be written in a python app!

SourceURL:https://smartthings.developer.samsung.com/docs/guides/smartapps/subscriptions.html

const express = require('express');
const bodyParser = require('body-parser');
const request = require('request');
const baseUrl = 'https://api.smartthing.com';

app.use(bodyParser.json());

// handle all incoming requests to our app
app.post('/', function(req, resp, err) {
  let evt = req.body;
  let lifecycle = evt.lifecycle;
  let res = null;

  switch(lifecycle) {
    case 'CONFIGURE':
      res = handleConfig(evt.configurationData);
      resp.json({statusCode: 200, configurationData: res});
      break;
    case 'INSTALL':
      handleInstall(evt.installData.installedApp, evt.installData.authToken);
      resp.json({statusCode: 200, installData: {}});
      break;
    case 'UPDATE':
      handleUpdate(evt.updateData.installedApp, evt.authToken);
      resp.json({statusCode: 200, updateData: {}});
      break;

    // handle other lifecycles...

    default:
      console.log(`lifecycle ${lifecycle} not supported`);
  }
});

function handleConfig(configData) {
  if (!configData.config) {
    throw new Error('No config section set in request.');
  }
  let config = {};
  const phase = configData.phase;
  const pageId = configData.pageId;
  const settings = configData.config;
  switch (phase) {
    case 'INITIALIZE':
      config.initialize = createConfigInitializeSetting();
      break;
    case 'PAGE':
      config.page = createConfigPage(pageId, settings);
      break;
    default:
      throw new Error(`Unsupported config phase: ${phase}`);
      break;
  }
  return config;
}

function createConfigInitializeSetting() {
  return {
    name: 'Your app name',
    description: 'Some app description',
    id: 'app',
    firstPageId: '1'
  }
}

/**
 * Creates a simple one page configuration screen where the user can
 * select a contact sensor device, and we will request read access to this
 * device.
 */
function createConfigPage(pageId, currentConfig) {
  if (pageId !== '1') {
    throw new Error(`Unsupported page name: ${pageId}`);
  }

  return {
    pageId: '1',
    name: 'Some page name',
    nextPageId: null,
    previousPageId: null,
    complete: true, // last page
    sections: [
      {
        name: 'When this opens...',
        settings: [
          {
            id: 'contactSensor',
            name: 'Which contact sensor?',
            description: 'Tap to set',
            type: 'DEVICE',
            required: true,
            multiple: false,
            capabilities: ['contactSensor'],
            permissions: ['r'] // need read permission to create subscriptions!
          }
        ]
      }
    ]
  };
}

/**
 * Once the user has selected the device and agreed to the requested
 * permissions, our app will create a subscription for the "open" value
 * of the "contact" attribute for the contact sensor.
 */
function handleInstall(installedApp, authToken) {
    let deviceConfig = installedApp.config.contactSensor[0].deviceConfig;
    createSubscription(deviceConfig);
}

function createSubscription(deviceConfig, authToken) {
  const path = `/installedapps/${installedApp.installedAppId}/subscriptions`;

  let subRequest = {
    sourceType: 'DEVICE',
    device: {
      componentId: deviceConfig.componentId,
      deviceId: deviceConfig.deviceId,
      capability: 'contactSensor',
      attribute: 'contact',
      stateChangeOnly: true,
      subscriptionName: "contact_subscription",
      value: "open"
    }
  };
  request.post({
    url: `${ baseUrl }${ path }`,
    json: true,
    body: subRequest,
    headers: {
      'Authorization': 'Bearer ' + authToken
    }
  },
  function (error, response, body) {
    if (!error && response.statusCode == 200) {
      console.log('subscription created')
    } else {
      console.log('failed to created subscriptions');
      console.log(error);
    }
  });  
}

/**
 * If the user has updated their configuration, for example they may
 * have selected a different contact sensor, we need to delete the
 * old subscriptions and create a new subscription.
 */
function handleUpdate(installedApp, authToken) {
  const path = `/installedapps/${installedApp.installedAppId}/subscriptions`;

  // delete all subscriptions, since some may have changed.
  request.delete({
    url: `${ baseUrl }${ path }`,
    json: true,
    headers: {
      'Authorization': 'Bearer ' + authToken
    }},
    function (error, response, body) {
      if (!error && response.statusCode == 200) {
        console.log('subscriptions deleted');
        // now create new subscription for (possibly) changed device configuration
        createSubscription(installedApp.config.contactSensor[0].deviceConfig, authToken);  
      } else {
        console.log('failed to delete subscriptions');
        console.log(error);
      }
    });  
}

Paul,

I use a cron job to run the script. That way I don’t have to rely on my script staying alive in case some crazy error is thrown, and I know that it will still run after reboot. It runs on a Raspberry Pi.

And yes I pass the personal token in the script. Typically you would set it as an OS variable, but for this I was pretty confident no one else would be on my Pi (shame on me).

Try this script. All you need to do is put your token in. Make sure you have the requests library for python3 installed (pip install requests). Then run it!

#Brendan Cain 2019
#brendancain99@gmail.com

# https://auth-global.api.smartthings.com

import requests

### API WEBSITE FOR DOCUMENTATION https://smartthings.developer.samsung.com/develop/api-ref/st-api.html#operation/getDevices

token = "YOUR TOKEN GOES HERE"


def get_locations():
    headers = {'Authorization': 'Bearer ' +token}
    r = requests.get('https://api.smartthings.com/v1/locations',headers=headers)
    device = r.json()
    return device


def get_devices():
    headers = {'Authorization': 'Bearer ' +token}
    r = requests.get('https://api.smartthings.com/v1/devices',headers=headers)
    device = r.json()
    print(device)


def get_devices_ids():
    devices_names = []
    headers = {'Authorization': 'Bearer ' +token}
    r = requests.get('https://api.smartthings.com/v1/devices',headers=headers)
    devices = r.json()
    items = devices["items"]
    for each in items:
        #print(each['name'],each['deviceId'])
        devices_names.append({'name':each['name'],'id': each['deviceId']})
    return devices_names


def get_status(device):
    headers = {'Authorization': 'Bearer ' +token}
    url = 'https://api.smartthings.com/v1/devices/' + device + '/status'
    r = requests.get(url, headers=headers)
    data = r.json()
    return(data)



print("""

************************************
This is all of your smarthings data!
************************************

""")

location_json = get_locations()

print("These are all of your locations:",location_json)
print()

location_id = location_json['items'][0]["locationId"]

print("This is your location ID, assuming you only have one:",location_id)
print()

devices_names = get_devices_ids()
print("These are your devices and their IDs:")
print()
for each in devices_names:
    print(each)

print()


#Just change the zero to 1, to get you second device and so on...
first_device = devices_names[0]['id']
first_device_status = get_status(first_device)
print("This is the status of your first device:")
print()
print(first_device_status)


#UNCOMMENT THE BELOW LINE TO SEE ALL DATA FOR EVERY DEVICE
#devices = get_devices()
1 Like

Brendan. Huge thanks for taking time out to streamline and simplify your script for all my device info in python. You’ve given me more than enough python examples and options to integrate into my smart app. This is extremely helpful. I hope I can return the favour in future development.

The only thing i’m missing now are the events triggers that i’m hoping to achieve by implementing event subscriptions. As soon as I learn more on the events/subscriptions side to smart apps using python, ill report back here to hopefully help some more users.

Again, thanks again for your help.

1 Like

My pleasure! Happy coding :slight_smile:

So ive been working away on this and think im nearly there with subscriptions in my smart app but no matter what I try im getting a 403 error when trying to create the “All Contact Sensor” subscription.

See my code below…

    elif (content['lifecycle'] == 'UPDATE'):
     print(content['lifecycle'])
    print('The auth token is:' + str(content['updateData']['authToken']))
    print('The app id is:' + str(content['updateData']['installedApp']['installedAppId']))
    #Build Post URL
    BaseURL = 'https://api.smartthings.com/v1/installedapps/'
    AppID = content['updateData']['installedApp']['installedAppId']
    EndURL = '/subscriptions'
    
    #Assign Token to variable
    authtoken = content['updateData']['authToken']
    
    #Subscribe to all contact sensor events
    headers = {'Authorization': 'Bearer '+authtoken}
    datasub = {
                    "sourceType": "CAPABILITY",
                    "capability": {
                    "locationId": LocationID,
                    "capability": "contactSensor",
                    "attribute": "contact",
                    "value": "*",
                    "stateChangeOnly": "true",
                    "subscriptionName": "all_contacts_sub"
                                   }
                    }
    r = requests.post(BaseURL+str(AppID)+EndURL,headers=headers,json=datasub)
    #error checking for creation of subscription
    print(r.status_code)
    print('The below is the content for the fault code')
    print(r.content)
    if r.status_code == 200:
        print('Subscription successfully created')
    else:
        print('Error Creating subscription because you prob dont have the token')

I can confirm that the location ID is correct. The auth token is correctly retrieved during the update lifecycle along with the application iD.

Ive also looked and I need to have r:devices set during the Initialization lifecycle. I have this done as per the code below…

When I do this the app gives me an error when authorizing.
image

Maybe I need to chase this ticket before troubleshooting more?. Banging my head at this stage!

If anyone can see something wrong in this code id appreciate the feedback.

Thanks.

Comment out the r:devices in your permissions of the data. I believe I had a similar problem, and the app will end up asking you what permissions you wanted anyway. That is how my script runs right now. Not sure if it is intended behavior, but it worked for me :stuck_out_tongue:

Brendan.

Thanks for the reply.

The docs explicitly tell me I need to add this if I’m creating this kind of subscription.

When I do remove it I’m getting the forbidden error (403) from the subscription POST request.

Either the docs are wrong or I still have some kind of issue with my token or appID but I’m convinced they are correct.

I’ll have to keep digging I guess. Thanks again.

Hi Brendan.

Just so I’m not missing something. Have you created a capability subscription with r:devices commented out in INITIALIZATION in your app?

Just want to sure I’m not missing something from my side.

Thanks.

I have this same issue. A 403 when trying to subscribe.

From what I can see the 403 was occurring because I did not have r:devices in the INITIALIZE lifecycle code. If I do add it my app crashes so there’s no way to create this kind of subscription.

Could you add this to your application and let me know what happens?