Energenie ENER314 Raspberry Pi board

switch_once.py

#!/usr/bin/python2

import sys
import time

import RPi.GPIO as GPIO

# set the pins numbering mode
GPIO.setmode(GPIO.BOARD)

# Select the GPIO pins used for the encoder K0-K3 data inputs
GPIO.setup(11, GPIO.OUT)
GPIO.setup(15, GPIO.OUT)
GPIO.setup(16, GPIO.OUT)
GPIO.setup(13, GPIO.OUT)

# Select the signal to select ASK/FSK
GPIO.setup(18, GPIO.OUT)

# Select the signal used to enable/disable the modulator
GPIO.setup(22, GPIO.OUT)

# Disable the modulator by setting CE pin lo
GPIO.output (22, False)

# Set the modulator to ASK for On Off Keying 
# by setting MODSEL pin lo
GPIO.output (18, False)

# Initialise K0-K3 inputs of the encoder to 0000

def decode_bit(x):
    assert x in '01'
    return x == '1'

def encode(code, duration=0.25):
    print("sending {}".format(code))
    d3, d2, d1, d0 = map(decode_bit, code)

    # Set K0-K3
    GPIO.output (11, d0)
    GPIO.output (15, d1)
    GPIO.output (16, d2)
    GPIO.output (13, d3)

    # let it settle, encoder requires this
    time.sleep(0.1)	

    # Enable the modulator
    GPIO.output (22, True)
    # keep enabled for a period

    time.sleep(duration)

    # Disable the modulator
    GPIO.output (22, False)

codes = {
    "all": {"on": "1011", "off": "0011"},
    "1":   {"on": "1111", "off": "0111"},
    "2":   {"on": "1110", "off": "0110"},
    "3":   {"on": "1101", "off": "0101"},
    "4":   {"on": "1100", "off": "0100"},
}

try:
    _, socket, state = sys.argv
    encode(codes[socket][state])
finally:
    GPIO.cleanup()

switch_server.py

#!/usr/bin/python2

"""
A lil Python server for controlling cheap Energenie switches with the ENER314
Raspberry Pi board.

Make POST requests to its webserver, supplying a `switch` and `state` in JSON:

    $ curl -X POST http://pi.lan:4800 -d '{"switch": "1", "state": "on"}'

The `switch` can be "1", "2", "3", "4", or "all".

The `state` is "on" or "off.
    
The request returns once the command has been sent. Concurrent requests are not
supported (since the bard only supports encoding one command at a time).

If the Pi is far from the switch, you might need to run the modulator for
longer than the recommended default (which is 0.25 seconds). Supply a
`duration`:

    $ curl -X POST http://pi.lan:4800 -d '{"switch": "1", "state": "on", "duration": 1.0}'
"""

import json
import time
import os
from BaseHTTPServer import HTTPServer
from SimpleHTTPServer import SimpleHTTPRequestHandler

import RPi.GPIO as GPIO

def setup_board():
    # set the pins numbering mode
    GPIO.setmode(GPIO.BOARD)

    # Select the GPIO pins used for the encoder K0-K3 data inputs
    GPIO.setup(11, GPIO.OUT)
    GPIO.setup(15, GPIO.OUT)
    GPIO.setup(16, GPIO.OUT)
    GPIO.setup(13, GPIO.OUT)

    # Select the signal to select ASK/FSK
    GPIO.setup(18, GPIO.OUT)

    # Select the signal used to enable/disable the modulator
    GPIO.setup(22, GPIO.OUT)

    # Disable the modulator by setting CE pin lo
    GPIO.output (22, False)

    # Set the modulator to ASK for On Off Keying 
    # by setting MODSEL pin lo
    GPIO.output (18, False)

    # Initialise K0-K3 inputs of the encoder to 0000
    encode('0000')


def decode_bit(x):
    assert x in '01'
    return x == '1'

def send(code, duration=0.25):
    print("sending {}".format(code))
    encode(code)
    modulate(duration)

def encode(code, duration=0.25):
    d3, d2, d1, d0 = map(decode_bit, code)

    # Set K0-K3
    GPIO.output(11, d0)
    GPIO.output(15, d1)
    GPIO.output(16, d2)
    GPIO.output(13, d3)

    # let it settle, encoder requires this
    time.sleep(0.1)	

def modulate(duration=0.25):
    """Enable the modulator for a period"""
    GPIO.output(22, True)
    time.sleep(duration)
    GPIO.output(22, False)

codes = {
    "all": {"on": "1011", "off": "0011"},
    "1":   {"on": "1111", "off": "0111"},
    "2":   {"on": "1110", "off": "0110"},
    "3":   {"on": "1101", "off": "0101"},
    "4":   {"on": "1100", "off": "0100"},
}


class Handler(SimpleHTTPRequestHandler):
    def send_head(self):
        if self.command != "POST":
            self.send_error(405, "Only POST allowed")
            return

        if self.path != "/":
            self.send_error(404, "Only / allowed")
            return

    def do_POST(self):
        self.send_head()

        length = int(self.headers.get('Content-Length', 0))
        if not length:
            self.send_error(400, "No request body")
            return

        try:
            command = json.loads(self.rfile.read(length))
        except ValueError:
            self.send_error(400, "Bad JSON")
            return

        switch = command.get("switch")
        if switch not in codes:
            self.send_error(400, "Bad `switch` parameter")
            return

        state = command.get("state")
        if state not in ('on', 'off'):
            self.send_error(400, "Bad `state` parameter")
            return

        duration = command.get("duration", 0.25)
        if not isinstance(duration, float) or duration <= 0.0:
            self.send_error(400, "Bad `duration` parameter")
            return

        code = codes[switch][state]
        send(code, duration)

        self.send_response(200)
        self.send_header('Content-Length', '3')
        self.send_header('Content-Type', 'text/json')
        self.end_headers()
        self.wfile.write("{}\n")


try:
    port = int(os.environ.get('PORT', 4800))
    host = os.environ.get('HOST', '')
    server = HTTPServer((host, port), Handler)
    print("Listening on {}:{}".format(host, port))

    setup_board()
    print("Board set up")

    server.serve_forever()
finally:
    GPIO.cleanup()