wrapper for rtlamr and rtl_tcp, sends data to graphite

rtlamr2graphite.py

#!/usr/bin/python2.7
import os
import time
import json
import socket
import struct
import pickle
import subprocess

CARBON_HOST = 'localhost'
CARBON_PICKLE_PORT = 2004
CARBON_PREFIX = 'meter.electric'
GOPATH = os.path.join(os.environ['HOME'],'gocode') #adjust for your environment, or just fetch os.environ['GOPATH'] if your env is better configured
RTLAMR = (os.path.join(GOPATH, 'bin','rtlamr'),'-centerfreq=915500000','-freqcorrection=47','-tunergainmode=true','-format=json','-msgtype=idm','-quiet')
RTLTCP = ('/usr/local/bin/rtl_tcp', '-a', '0.0.0.0')

def feed_carbon(path, readings):
   #prepare pickle payload
   tuples = list()
   for reading in readings:
       tuples.append((path, reading))
   payload = pickle.dumps(tuples)

   #prepare message
   header = struct.pack("!L", len(payload))
   message = header + payload

   #send message
   sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
   sock.connect((CARBON_HOST, CARBON_PICKLE_PORT))
   sock.send(message)
   sock.close()


def save_message(message):
    now = time.time()

    meter_serial = message['ERTSerialNumber']
    last_reading = message['LastConsumptionCount']
    last_reading_time = now - (now % 300) #round down for 5 minute resolution

    #use the differential consumption values to back-fill data
    readings = list()
    readings.append( (last_reading_time, last_reading) )
    for delta in message['DifferentialConsumptionIntervals'][:-6]: #3.5 hr
        last_reading = last_reading - delta
        last_reading_time = last_reading_time - 300
        readings.append( (last_reading_time, last_reading) )

    prefix = '%s.%s' % (CARBON_PREFIX, meter_serial)
    feed_carbon(prefix, readings)


if __name__ == "__main__":
    print "starting processes..."
    rtl_proc = subprocess.Popen(RTLTCP)
    time.sleep(5)

    os.environ['GOPATH'] = GOPATH #yay, golang?
    amr_proc = subprocess.Popen(RTLAMR,stdout=subprocess.PIPE)

    #FIXME rtl_tcp doesn't respond responsibly to SIGTERM or SIGINT.
    #      maybe there is a workaround to reap it.

    meters = dict()
    print "rtlamr2graphite running..."
    while True:
        line = amr_proc.stdout.readline()
        packet = json.loads(line)
        message = packet['Message']
        serial = message['ERTSerialNumber']
        last_count = meters.get(serial,-1) #default -1 ensures index 0 updates
        this_count = message['ConsumptionIntervalCount'] #this update index
        did_roll = True if last_count - this_count > 127 else False #rolled!

        print '%s %08d: (%03d<%03d) %s %08d%s' % (
          time.strftime('%Y-%m-%dT%H:%M:%S'),
          serial,
          last_count,
          this_count, '=' if last_count < this_count else '!',
          message['LastConsumptionCount'],
          ' rolled!' if did_roll else ''
        )

        #update only if never seen or update count has incremented or rolled
        if last_count < this_count or did_roll:
            save_message(message)
            meters[serial] = message['ConsumptionIntervalCount']