Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
SmartThings integration with Amazon Dash buttons
# This is the xinetd configuration file for the dashproxy file
service dashproxy
{
type = UNLISTED
socket_type = stream
protocol = tcp
wait = no
port = 9999
user = nobody
server = /usr/local/bin/dashproxy.py
per_source = 1 # Very important! This is what combines with the sleep in the python script to avoid duplicate events
}
#!/usr/bin/python
import sys
import syslog
import requests
import os
import string
import json
import time
import os
#from scapy.all import *
def oauth_setup():
# See http://docs.smartthings.com/en/latest/smartapp-web-services-developers-guide/tutorial-part2.html#appendix-just-the-urls-please
# for some idea of how to get these values
client_code="<your client code>" # From the first OAUTH request
oauth_client_id="00000000-38a8-4de7-bb19-2e3585aba2a2" # From the SmartApp page
oauth_client_secret="00000000-2aee-409f-af80-870763db0ded" # From the SmartApp page
# This is what the response of the second request gets you:
# {"access_token":"00000000-4a5a-4191-baa5-a03b9807329c","token_type":"bearer","expires_in":1576799999,"scope":"app"}
access_token = "00000000-4a5a-4191-baa5-a03b9807329c"
return { "access_token": access_token,
"token_type": "bearer",
"expires_in": 1576799999,
"client_code": client_code,
# The endpoint comes from the third API request
"endpoint_uri": "https://graph.api.smartthings.com/api/smartapps/installations/0000000-a8a2-4742-8c9e-4501eaf512a7",
"headers": {"Authorization": "Bearer %s" % access_token },
"scope": "app"}
def light_name_to_id(light_name):
info=oauth_setup()
url = "%s/%s" % (info['endpoint_uri'], "switches")
# syslog.syslog("Endpoint: %s" % url)
r = requests.get(url, headers=info['headers'])
if r.status_code != 200:
syslog.syslog("HTTP Status: %s" % r.status_code)
return None
# syslog.syslog( r.text)
data = json.loads(r.text)
return data.get(light_name)
def toggle_st(light_name):
info=oauth_setup()
switch_id=light_name_to_id(light_name)
url = "%s/switches/%s" % (info['endpoint_uri'], switch_id)
syslog.syslog("Toggling %s with URL: %s" % ( light_name, url ))
data = '{"command":"toggle"}'
r = requests.put(url, data=data, headers=info['headers'])
def ip_to_light(ip):
octet_lookup = {
'117': "VSDash-Bounty", # Bounty
'160': "VSDash-Huggies", # Huggies
'228': "VSDash-Ziploc", # Ziploc
'238': "VSDash-Glad" # Glad
}
last_octet = string.split(remote_ip, '.')[3]
return octet_lookup.get(last_octet)
remote_ip = os.environ.get('REMOTE_HOST')
if remote_ip == None:
syslog.syslog("No IP detected")
sys.exit(1)
light_name = ip_to_light(remote_ip)
if light_name == None:
syslog.syslog("Unknown IP connected - %s" % remote_ip)
sys.exit(2)
syslog.syslog("%s - %s" % (remote_ip, light_name))
toggle_st(light_name)
# This causes xinetd to close the connection
os.close(0)
os.close(1)
os.close(2)
# We keep sleeping so that the retry attempts are suppressed by xinetd
time.sleep(10)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment