RESTful wifi plugs

Wifi plugs can control mains devices over a wireless network. This has a lot of safety advantages over breaking into mains circuitry to wire up relays. But to interact programmatically without relying upon proprietary apps or IFTTT we need to bind the plug and then expose it using web services.

flash

The TP-Link HS100 is pretty easy to set up via the app. It becomes much more powerful once you can control it from Python. Fortunately the hard work of Python binding the plug has been done for us by GadgetReactor’s pyHS100 package.

What we’ve done on top of this is:

  1. Create new functions to bypass slightly untidy errors. A bit bodgy, but it works.
  2. Expose core operations via REST. This means you can switch the device on and off over the internet.

We’ve also added an element of security, although it’s not yet fully-fledged. We could easily add Basic Auth and SSL, but that’s for another tutorial. Don’t rely on the security here for really sensitive applications.

The code below binds the device (find your plug’s IP by installing the pyHS100 package via pip then running “pyHS100 discover” from the console), wraps a tricky error in try-except loops, then exposes four endpoints via Flask-RESTful. The endpoints are:

  • on – to turn the plug on
  • off – to turn the plug off
  • switch – to flick the switch
  • state – to find out whether the switch is on or off
from flask import Flask, request
from pyHS100 import SmartPlug
import time, json

app = Flask(__name__)

USER = <your username>
PASS = <your password>

#----------------------------------------------------------------------------------------
#The following functions are necessary to correct weird errors (these even occur from the console)
def turn_on(plug):
     success = False
     failures = 0
     while not success:
          try:
               plug.turn_on()
               success = True
          except:
               failures = failures + 1
     print 'Turned on after', failures, 'failures'
     return success

def turn_off(plug):
     success = False
     failures = 0
     while not success:
          try:
               plug.turn_off()
               success = True
          except:
               failures = failures + 1
     print 'Turned off after', failures, 'failures'
     return success

def get_state(plug):
     success = False
     failures = 0
     while not success:
          try:
               state = plug.state
               success = True
          except Exception as e:
               print 'Failed to get state', str(e)
               failures = failures + 1
     print 'Caught state after', failures, 'failures'
     return str(state)
#----------------------------------------------------------------------------------------

#Basic validation
def validate(username,password):
     if username == USER:
          if password == PASS:
               print 'Hello', username
               return True
          else:
               return False
     else:
          return False

#REST endpoint to turn the switch on
@app.route('/on', methods=['POST'])
def on():
     json_data = request.get_json(force=True)
     username = str(json_data['username'])
     password = str(json_data['password'])
     plugIP = json_data['plug']
     if not validate:
          return json.dumps('Go away')

     #Attach plug
     plug = SmartPlug(plugIP)
     state = get_state(plug)

     #Turn plug off
     if state == 'ON':
          return json.dumps("Plug is already on")
     else:
          turn_on(plug)
          return json.dumps("Plug was turned on")

#REST endpoint to turn the switch off
@app.route('/off', methods=['POST'])
def off():
     json_data = request.get_json(force=True)
     username = str(json_data['username'])
     password = str(json_data['password'])
     plugIP = json_data['plug']
    
     if not validate:
          return json.dumps('Go away')

     #Attach plug
     plug = SmartPlug(plugIP)
     state = get_state(plug)

     #Turn plug off
     if state == 'OFF':
          return json.dumps("Plug is already off")
     else:
          turn_off(plug)
          return json.dumps("Plug was turned off")

#REST endpoint to flick the switch
@app.route('/switch', methods=['POST'])
def switch():
     json_data = request.get_json(force=True)
     username = str(json_data['username'])
     password = str(json_data['password'])
     plugIP = json_data['plug']

     if not validate:
          return json.dumps('Go away')

     #Attach plug
     plug = SmartPlug(plugIP)
     state = get_state(plug)
     print 'Current:', state
    
     #Turn plug off
     if state == 'OFF':
          turn_on(plug)
          return json.dumps("Plug switched on")
     else:
          turn_off(plug)
          return json.dumps("Plug switched off")

#REST endpoint to report the current state (ON/OFF)
@app.route('/state', methods=['POST'])
def state():
     json_data = request.get_json(force=True)
     username = str(json_data['username'])
     password = str(json_data['password'])
     plugIP = json_data['plug']

     if not validate:
          return json.dumps('Go away')

     #Attach plug
     plug = SmartPlug(plugIP)
     state = get_state(plug)
     return json.dumps(state)

#Start the service
if __name__ == '__main__':
     app.run(host='0.0.0.0',debug=False, port=8080, threaded=False)

Running this from the console puts up the service on localhost port 8080. You can then hit the different endpoints using POST. Try Postman to check your interface works. For example, you can POST this raw JSON…

{"username":"<your username>", "password":"<your password>", "plug":"192.168.1.9"}

…to http://localhost:8080/switch to flick the switch.

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s