#!/usr/bin/python -u '''PowerMeter Data Processor for Brultech ECM-1240. Collect data from Brultech ECM-1240 power monitors. Print the data, save the data to database, or upload the data to WattsOn. Thanks to: Amit Snyderman bpwwer & tenholde from the cocoontech.com forums Kelvin Kakugawa brian jackson [http://fivejacksons.com/brian/] Marc MERLIN - http://marc.merlins.org/perso/solar/ Ben Example usage: ecmread.py --serial --serialport=/dev/ttyUSB0 -p Example output: 2010/06/07 20:22:48: ECM ID: 355555 2010/06/07 21:48:37: Volts: 120.90V 2010/06/07 21:48:37: Ch1 Amps: 0.25A 2010/06/07 21:48:37: Ch2 Amps: 3.24A 2010/06/07 21:48:37: Ch1 Watts: -124.586KWh ( 1536W) < PG&E 2010/06/07 21:48:37: Ch1 Positive Watts: 210.859KWh ( 1536W) 2010/06/07 21:48:37: Ch1 Negative Watts: 335.445KWh ( 0W) 2010/06/07 21:48:37: Ch2 Watts: -503.171KWh ( 0W) < PV 2010/06/07 21:48:37: Ch2 Positive Watts: 0.028KWh ( 0W) 2010/06/07 21:48:37: Ch2 Negative Watts: 503.199KWh ( 0W) 2010/06/07 21:48:37: Aux1 Watts: 114.854KWh ( 311W) < Computer Closet 2010/06/07 21:48:37: Aux2 Watts: 80.328KWh ( 523W) < MythTV/AV System 2010/06/07 21:48:37: Aux3 Watts: 13.014KWh ( 35W) < Computer Office/BR4 2010/06/07 21:48:37: Aux4 Watts: 4.850KWh ( 0W) < AC 2010/06/07 21:48:37: Aux5 Watts: 25.523KWh ( 137W) < Kitchen Fridge Database Configuration: This script writes to one or two tables: ecm and ecmwh. The ecm table contains instantaneous watt data, the ecmwh table contains watt-hour counts. Create the database and watt table by doing something like this: mysql -u root -p mysql> create database ecm; mysql> grant usage on *.* to ecmuser@ecmclient identified by 'ecmpass'; mysql> grant all privileges on ecm.* to ecmuser@ecmclient; mysql> create table ecm -> (id int primary key auto_increment, -> ecm_serial int, -> volts int, -> ch1_amps int, -> ch2_amps int, -> ch1_w int, -> ch2_w int, -> aux1_w int, -> aux2_w int, -> aux3_w int, -> aux4_w int, -> aux5_w int, -> time_created int); mysql>exit Create the watt-hour table by doing something like this: mysql> create table ecmwh -> (id int primary key auto_increment, -> ecm_serial int, -> ch1_wh int, -> ch2_wh int, -> aux1_wh int, -> aux2_wh int, -> aux3_wh int, -> aux4_wh int, -> aux5_wh int, -> time_created int); If you do not want the database to grow, then do not create the 'id' primary key. In that case the database is used for data transfer, not data capture. Changelog: - 2.1.1 20dec2011 mwall * incorporate packet reading changes from ben from marc's ecmread.py 0.1.5 for both serial and socket configurations - check for end header bytes. * validate each packet using checksum. ignore packet if checksum fails. * added debug output for diagnosing packet collisions * cleaned up serial and socket packet reading - 2.1.0 10dec2011 mwall * report volts and amps as well as watts * added option to save watt-hours to database in a separate table * rename columns from *_ws to *_w (we are recording watts, not watt-seconds) * to rename your columns, do this in mysql: alter table ecm change ch1_ws ch1_w - 2.0.0 08dec2011 mwall * support any number of ECM on a single bus when printing or pushing to db. this required a change to the database schema, specifically the addition of a field to record the ECM serial number. when uploading to wattson, multiple ECM devices are not distinguished. * display the ECM serial number when printing. * catch divide-by-zero exceptions when printing. * bump version to 2.0.0. the version distributed by marc merlins was listed as 1.4.1, but internally as 0.4.1. the changes to support multiple ECM qualify at least as a minor revision, but since they break previous schema we'll go with a major revision. - 0.1.5. 2011/08/25: Ben * improved binary packet parsing to better deal with end of packets, and remove some corrupted data. * TODO: actually check the CRC in the binary packet. - 0.1.4. 2010/06/06: modified screen output code to * Show Kwh counters for each input as well as instant W values * For channel 1 & 2, show positive and negative values. ''' __author__ = 'Brian Jackson; Kelvin Kakugawa; Marc MERLIN; mwall' __version__ = '2.1.1' import base64 import bisect import new import optparse import socket import sys import time import traceback import urllib import urllib2 import warnings warnings.filterwarnings('ignore', category=DeprecationWarning) # MySQLdb in 2.6.* # External (Optional) Dependencies try: import serial except Exception, e: serial = None try: import MySQLdb except Exception, e: MySQLdb = None try: import cjson as json # XXX: maintain compatibility w/ json module setattr(json, 'dumps', json.encode) setattr(json, 'loads', json.decode) except Exception, e: try: import simplejson as json except Exception, e: import json MINUTE = 60 HOUR = 60 * MINUTE DAY = 24 * HOUR # PACKET SETTINGS START_HEADER0 = hex(254) START_HEADER1 = hex(255) START_HEADER2 = hex(3) END_HEADER0 = hex(255) END_HEADER1 = hex(254) DATA_BYTES_LENGTH = 59 # does not include the start/end headers SEC_COUNTER_MAX = 16777216 # SERIAL SETTINGS SERIALPORT = "/dev/ttyUSB0" # the com/serial port the ecm is connected to (COM4, /dev/ttyS01, etc) BAUDRATE = 19200 # the baud rate we talk to the ecm # ETHERNET SETTINGS HOST = '' PORT = 8083 # default port that the EtherBee is pushing data to # DATABASE DEFAULTS DB_HOST = 'localhost' DB_USER = '' DB_PASSWD = '' DB_DATABASE = '' DB_RECORD_WH = 1 # WATTZON DEFAULTS WATTZON_API_URL = 'http://www.wattzon.com/api/2009-01-27/3' DEBUG = 0 # set this to 1 to debug packet collisions # Helper Functions def dbgmsg(msg): '''emit a debug message if debugging output is enabled''' if DEBUG: logmsg(msg) def logmsg(msg): '''put a timestamp on the message and spit it out''' ts = time.strftime("%Y/%m/%d %H:%M:%S", time.localtime()) print "%s %s" % (ts, msg) def getserial(packet): '''extract the ECM serial number from a compiled packet''' return "%d%d" % (packet['unit_id'], packet['ser_no']) def calculate_checksum(packet): '''calculate the packet checksum''' checksum = int(START_HEADER0,16) checksum += int(START_HEADER1,16) checksum += int(START_HEADER2,16) checksum += sum([ord(c) for c in packet]) checksum += int(END_HEADER0,16) checksum += int(END_HEADER1,16) checksum = checksum & 0xff checksum = hex(checksum) return checksum def calculate(now, prev): '''calc average watts/s between packets''' if now['secs'] < prev['secs']: now['secs'] += SEC_COUNTER_MAX # handle seconds counter overflow secs_delta = float(now['secs'] - prev['secs']) # CH1/2 Watts now['ch1_watts'] = (now['ch1_aws'] - prev['ch1_aws']) / secs_delta now['ch2_watts'] = (now['ch2_aws'] - prev['ch2_aws']) / secs_delta now['ch1_positive_watts'] = (now['ch1_pws'] - prev['ch1_pws']) / secs_delta now['ch2_positive_watts'] = (now['ch2_pws'] - prev['ch2_pws']) / secs_delta now['ch1_negative_watts'] = now['ch1_watts'] - now['ch1_positive_watts'] now['ch2_negative_watts'] = now['ch2_watts'] - now['ch2_positive_watts'] # All Watts increase no matter which way the current is going # Polar Watts only increase if the current is positive # Every Polar Watt does register as an All Watt too. # math comes to: Watts = 2x Polar Watts - All Watts now['ch1_pwh'] = now['ch1_pws'] / 3600000.0 now['ch2_pwh'] = now['ch2_pws'] / 3600000.0 now['ch1_nwh'] = (now['ch1_aws'] - now['ch1_pws']) / 3600000.0 now['ch2_nwh'] = (now['ch2_aws'] - now['ch2_pws']) / 3600000.0 now['ch1_wh'] = now['ch1_pwh'] - now['ch1_nwh'] now['ch2_wh'] = now['ch2_pwh'] - now['ch2_nwh'] now['aux1_wh'] = now['aux1_ws'] / 3600000.0 now['aux2_wh'] = now['aux2_ws'] / 3600000.0 now['aux3_wh'] = now['aux3_ws'] / 3600000.0 now['aux4_wh'] = now['aux4_ws'] / 3600000.0 now['aux5_wh'] = now['aux5_ws'] / 3600000.0 # Polar Watts' instant value's only purpose seems to help find out if # main watts are positive or negative. Polar Watts only goes up if the # sign is positive. If they are null, tha means the sign is negative. if (now['ch1_positive_watts'] == 0): now['ch1_watts'] *= -1 if (now['ch2_positive_watts'] == 0): now['ch2_watts'] *= -1 # AUX1-5 Watts now['aux1_watts'] = (now['aux1_ws'] - prev['aux1_ws']) / secs_delta now['aux2_watts'] = (now['aux2_ws'] - prev['aux2_ws']) / secs_delta now['aux3_watts'] = (now['aux3_ws'] - prev['aux3_ws']) / secs_delta now['aux4_watts'] = (now['aux4_ws'] - prev['aux4_ws']) / secs_delta now['aux5_watts'] = (now['aux5_ws'] - prev['aux5_ws']) / secs_delta now['time'] = time.strftime("%Y/%m/%d %H:%M:%S", time.localtime()) # Packet Server Classes class BasePacketServer(object): def __init__(self, packet_processor): self.packet_processor = packet_processor def _convert(self, bytes): return reduce(lambda x,y:x+y[0] * (256**y[1]), zip(bytes,xrange(len(bytes))),0) def _compile(self, packet): now = {} # Voltage Data (2 bytes) now['volts'] = 0.1 * self._convert(packet[1::-1]) # CH1-2 Absolute Watt-Second Counter (5 bytes each) now['ch1_aws'] = self._convert(packet[2:7]) now['ch2_aws'] = self._convert(packet[7:12]) # CH1-2 Polarized Watt-Second Counter (5 bytes each) now['ch1_pws'] = self._convert(packet[12:17]) now['ch2_pws'] = self._convert(packet[17:22]) # Reserved (4 bytes) # Device Serial Number (2 bytes) # now['ser_no'] = self._convert(packet[27:25:-1]) sn1 = self._convert(packet[26:27]) sn2 = self._convert(packet[27:28]) now['ser_no'] = sn1 + 256*sn2 # Reset and Polarity Information (1 byte) now['flag'] = self._convert(packet[28:29]) # Device Information (1 byte) now['unit_id'] = self._convert(packet[29:30]) # CH1-2 Current (2 bytes each) now['ch1_amps'] = 0.01 * self._convert(packet[30:32]) now['ch2_amps'] = 0.01 * self._convert(packet[32:34]) # Seconds (3 bytes) now['secs'] = self._convert(packet[34:37]) # AUX1-5 Watt-Second Counter (4 bytes each) now['aux1_ws'] = self._convert(packet[37:41]) now['aux2_ws'] = self._convert(packet[41:45]) now['aux3_ws'] = self._convert(packet[45:49]) now['aux4_ws'] = self._convert(packet[49:53]) now['aux5_ws'] = self._convert(packet[53:57]) return now def process(self, packet): packet = self._compile(packet) self.packet_processor.preprocess(packet) self.packet_processor.process(packet) def read(self): pass def run(self): try: self.packet_processor.setup() while True: try: self.read() except Exception, e: if type(e) == KeyboardInterrupt: # only break for KeyboardInterrupt raise e traceback.print_exc() if not self.packet_processor.handle(e): print 'Exception [in %s]: %s' % (self, e) except Exception, e: traceback.print_exc() sys.exit(1) finally: self.packet_processor.cleanup() class SerialPacketServer(BasePacketServer): def __init__(self, packet_processor, port=SERIALPORT, baudrate=BAUDRATE): super(SerialPacketServer, self).__init__(packet_processor) self._port = port self._baudrate = baudrate if not serial: print 'Error: pySerial module could not be imported.' sys.exit(1) self.conn = None def read(self): try: self.conn = serial.Serial(self._port, self._baudrate) self.conn.open() while True: data = self.conn.read(1) if not data: break byte = hex(ord(data)) if byte != START_HEADER0: dbgmsg("expected START_HEADER0 %s, got %s" % (START_HEADER0, byte)) continue data = self.conn.read(1) byte = hex(ord(data)) if byte != START_HEADER1: dbgmsg("expected START_HEADER1 %s, got %s" % (START_HEADER1, byte)) continue data = self.conn.read(1) byte = hex(ord(data)) if byte != START_HEADER2: dbgmsg("expected START_HEADER2 %s, got %s" % (START_HEADER2, byte)) continue packet = '' while len(packet) < DATA_BYTES_LENGTH: data = self.conn.read(DATA_BYTES_LENGTH-len(packet)) if not data: break; packet += data if len(packet) < DATA_BYTES_LENGTH: logmsg("incomplete packet: expected %d bytes, got %d" % (DATA_BYTES_LENGTH, len(packet))) continue; data = self.conn.read(1) byte = hex(ord(data)) if byte != END_HEADER0: dbgmsg("expected END_HEADER0 %s, got %s" % (END_HEADER0, byte)) continue data = self.conn.read(1) byte = hex(ord(data)) if byte != END_HEADER1: dbgmsg("expected END_HEADER1 %s, got %s" % (END_HEADER1, byte)) continue checksum = calculate_checksum(packet) data = self.conn.read(1) byte = hex(ord(data)) if byte != checksum: logmsg("bad checksum: expected %s, got %s" % (checksum, byte)) continue packet = [ord(c) for c in packet] self.process(packet) finally: if self.conn: self.conn.close() self.conn = None class SocketPacketServer(BasePacketServer): def __init__(self, packet_processor, host=HOST, port=PORT): super(SocketPacketServer, self).__init__(packet_processor) socket.setdefaulttimeout(60) # override None self._host = host self._port = port self.sock = None self.conn = None def read(self): try: self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) except: # REUSEPORT may not be supported on all systems pass self.sock.bind((self._host, self._port)) self.sock.listen(1) self.conn, addr = self.sock.accept() while True: data = self.conn.recv(1) if not data: break byte = hex(ord(data)) if byte != START_HEADER0: continue data = self.conn.recv(1) byte = hex(ord(data)) if byte != START_HEADER1: continue data = self.conn.recv(1) byte = hex(ord(data)) if byte != START_HEADER2: continue packet = '' while len(packet) < DATA_BYTES_LENGTH: data = self.conn.recv(DATA_BYTES_LENGTH-len(packet)) if not data: # No data left break packet += data if len(packet) < DATA_BYTES_LENGTH: logmsg("incomplete packet: expected %d bytes, got %d" % (DATA_BYTES_LENGTH, len(packet))) continue; data = self.conn.recv(1) byte = hex(ord(data)) if byte != END_HEADER0: continue data = self.conn.recv(1) byte = hex(ord(data)) if byte != END_HEADER1: continue checksum = calculate_checksum(packet) data = self.conn.recv(1) byte = hex(ord(data)) if byte != checksum: logmsg("bad checksum: expected %s, got %s" % (checksum, byte)) continue packet = [ord(c) for c in packet] self.process(packet) finally: if self.conn: self.conn.shutdown(socket.SHUT_RD) self.conn.close() self.conn = None if self.sock: self.sock.shutdown(socket.SHUT_RD) self.sock.close() self.sock = None # Packet Processor Classes class MovingBuffer(object): '''Maintain fixed-size buffer of data over time''' def __init__(self, max_timeframe=DAY): self.time_points = [] self.max_timeframe = max_timeframe def insert(self, timestamp, time_dict): bisect.insort(self.time_points, (timestamp, time_dict)) now = int(time.time()) cull_index = bisect.bisect(self.time_points, (now-self.max_timeframe, {})) del(self.time_points[:cull_index]) def data_over(self, time_delta): now = int(time.time()) delta_index = bisect.bisect(self.time_points, (now-time_delta, {})) return self.time_points[delta_index:] def delta_over(self, time_delta): now = int(time.time()) delta_index = bisect.bisect(self.time_points, (now-time_delta, {})) offset = self.time_points[delta_index][1] current = self.time_points[-1][1] calculate(current, offset) return current class CompoundBuffer(object): '''Variable number of moving buffers, each associated with an ID''' def __init__(self, max_timeframe=DAY): self.max_timeframe = max_timeframe self.buffers = {} def insert(self, timestamp, time_dict): ecm_serial = getserial(time_dict) return self.getbuffer(ecm_serial).insert(timestamp, time_dict) def data_over(self, ecm_serial, time_delta): return self.getbuffer(ecm_serial).data_over(time_delta) def delta_over(self, ecm_serial, time_delta): return self.getbuffer(ecm_serial).delta_over(time_delta) def getbuffer(self, ecm_serial): if not ecm_serial in self.buffers: self.buffers[ecm_serial] = MovingBuffer(self.max_timeframe) return self.buffers[ecm_serial] class BaseProcessor(object): def __init__(self, buffer_timeframe, *args, **kwargs): # self.buffer = MovingBuffer(buffer_timeframe) self.buffer = CompoundBuffer(buffer_timeframe) def setup(self): pass def preprocess(self, packet): self.buffer.insert(int(time.time()), packet) def process(self, packet): pass def handle(self, exception): return False def cleanup(self): pass # keep the previous packet in a dictionary keyed by the ECM serial number. # that way if there are multiple ECM then the calculation is applied to each # ECM independently of any other. class PrintMixin(object): def __init__(self, *args, **kwargs): super(PrintMixin, self).__init__(*args, **kwargs) self.print_out = kwargs.get('print_out', False) self.prev_packet = {} def process(self, packet): sn = getserial(packet) if sn in self.prev_packet: try: calculate(packet, self.prev_packet[sn]) except ZeroDivisionError, zde: print "not enough data in buffer for %s" % sn return # start with newline in case previous run was stopped in the middle of a line # this ensures that the new output is not attached to some old incompletely # written line print print packet['time']+": ECM: %s" % sn print packet['time']+": Volts: %9.2fV" % packet['volts'] print packet['time']+": Ch1 Amps: %9.2fA" % packet['ch1_amps'] print packet['time']+": Ch2 Amps: %9.2fA" % packet['ch2_amps'] print packet['time']+": Ch1 Watts: % 13.6fKWh (% 5dW)" % (packet['ch1_wh'] , packet['ch1_watts']) print packet['time']+": Ch1 Positive Watts: % 13.6fKWh (% 5dW)" % (packet['ch1_pwh'], packet['ch1_positive_watts']) print packet['time']+": Ch1 Negative Watts: % 13.6fKWh (% 5dW)" % (packet['ch1_nwh'], packet['ch1_negative_watts']) print packet['time']+": Ch2 Watts: % 13.6fKWh (% 5dW)" % (packet['ch2_wh'] , packet['ch2_watts']) print packet['time']+": Ch2 Positive Watts: % 13.6fKWh (% 5dW)" % (packet['ch2_pwh'], packet['ch2_positive_watts']) print packet['time']+": Ch2 Negative Watts: % 13.6fKWh (% 5dW)" % (packet['ch2_nwh'], packet['ch2_negative_watts']) print packet['time']+": Aux1 Watts: % 13.6fKWh (% 5dW)" % (packet['aux1_wh'], packet['aux1_watts']) print packet['time']+": Aux2 Watts: % 13.6fKWh (% 5dW)" % (packet['aux2_wh'], packet['aux2_watts']) print packet['time']+": Aux3 Watts: % 13.6fKWh (% 5dW)" % (packet['aux3_wh'], packet['aux3_watts']) print packet['time']+": Aux4 Watts: % 13.6fKWh (% 5dW)" % (packet['aux4_wh'], packet['aux4_watts']) print packet['time']+": Aux5 Watts: % 13.6fKWh (% 5dW)" % (packet['aux5_wh'], packet['aux5_watts']) self.prev_packet[sn] = packet super(PrintMixin, self).process(packet) class DatabaseMixin(object): def __init__(self, *args, **kwargs): super(DatabaseMixin, self).__init__(*args, **kwargs) self.db_host = kwargs.get('db_host') or DB_HOST self.db_user = kwargs.get('db_user') or DB_USER self.db_passwd = kwargs.get('db_passwd') or DB_PASSWD self.db_database = kwargs.get('db_database') or DB_DATABASE self.quiet = kwargs.get('quiet') if not MySQLdb: print 'Error: MySQLdb module could not be imported.' sys.exit(1) def setup(self): super(DatabaseMixin, self).setup() try: self.conn = MySQLdb.connect( host=self.db_host, user=self.db_user, passwd=self.db_passwd, db=self.db_database) except Exception, e: if type(e) == MySQLdb.Error: print 'MySQL Error: [#%d] %s' % (exception.args[0], exception.args[1]) else: traceback.print_exc() self.conn = None sys.exit(1) self.insert_period = MINUTE self.last_insert = {} self.insert_period_wh = HOUR self.last_insert_wh = {} def process(self, packet): super(DatabaseMixin, self).process(packet) sn = getserial(packet) now = int(time.time()) if sn in self.last_insert and now < (self.last_insert[sn]+self.insert_period): return try: delta = self.buffer.delta_over(sn, self.insert_period) except ZeroDivisionError, zde: return # not enough data in buffer cursor = self.conn.cursor() cursor.execute( '''INSERT INTO '''+self.db_database+'''.ecm ( ecm_serial, volts, ch1_amps, ch2_amps, ch1_w, ch2_w, aux1_w, aux2_w, aux3_w, aux4_w, aux5_w, time_created ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)''', ( int(sn), float(delta['volts']), float(delta['ch1_amps']), float(delta['ch2_amps']), int(delta['ch1_watts']), int(delta['ch2_watts']), int(delta['aux1_watts']), int(delta['aux2_watts']), int(delta['aux3_watts']), int(delta['aux4_watts']), int(delta['aux5_watts']), now)) cursor.close() self.last_insert[sn] = now if not self.quiet: print 'DB: insert @%s: sn: %s, v: %s, ch1a: %s, ch2a: %s, ch1: %s, ch2: %s, aux1: %s, aux2: %s, aux3: %s, aux4: %s, aux5: %s' % ( now, sn, delta['volts'], delta['ch1_amps'], delta['ch2_amps'], delta['ch1_watts'], delta['ch2_watts'], delta['aux1_watts'], delta['aux2_watts'], delta['aux3_watts'], delta['aux4_watts'], delta['aux5_watts'], ) if DB_RECORD_WH and (not sn in self.last_insert_wh or now < (self.last_insert_wh[sn]+self.insert_period_wh)): cursor = self.conn.cursor() cursor.execute( '''INSERT INTO '''+self.db_database+'''.ecmwh ( ecm_serial, ch1_wh, ch2_wh, aux1_wh, aux2_wh, aux3_wh, aux4_wh, aux5_wh, time_created ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)''', ( int(sn), int(delta['ch1_wh']), int(delta['ch2_wh']), int(delta['aux1_wh']), int(delta['aux2_wh']), int(delta['aux3_wh']), int(delta['aux4_wh']), int(delta['aux5_wh']), now)) cursor.close() self.last_insert_wh[sn] = now if not self.quiet: print 'DB: insert_wh @%s: sn: %s, ch1: %s, ch2: %s, aux1: %s, aux2: %s, aux3: %s, aux4: %s, aux5: %s' % ( now, sn, delta['ch1_wh'], delta['ch2_wh'], delta['aux1_wh'], delta['aux2_wh'], delta['aux3_wh'], delta['aux4_wh'], delta['aux5_wh'], ) def handle(self, exception): if type(exception) == MySQLdb.Error: print 'MySQL Error: [#%d] %s' % (exception.args[0], exception.args[1]) return True return super(DatabaseMixin, self).handle(exception) def cleanup(self): super(DatabaseMixin, self).cleanup() if not self.conn: return self.conn.commit() self.conn.close() class RequestPostWithContentType(urllib2.Request): def __init__(self, content_type, *args, **kwargs): urllib2.Request.__init__(self, *args, **kwargs) self._content_type = content_type urllib2.Request.add_header(self, 'Content-Type', content_type) def has_header(self, header_name): return header_name == 'Content-Type' or urllib2.Request.has_header(self, header_name) def get_header(self, header_name, default=None): return header_name == 'Content-Type' and self._content_type or urllib2.Request.get_header(self, header_name, default) class WattzOnMixin(object): def __init__(self, *args, **kwargs): super(WattzOnMixin, self).__init__(*args, **kwargs) self.api_key = kwargs.get('api_key') self.api_username = kwargs.get('api_username') self.api_passwd = kwargs.get('api_passwd') self.meter_ch1 = kwargs.get('meter_ch1') self.meter_ch2 = kwargs.get('meter_ch2') self.meter_aux1 = kwargs.get('meter_aux1') self.meter_aux2 = kwargs.get('meter_aux2') self.meter_aux3 = kwargs.get('meter_aux3') self.meter_aux4 = kwargs.get('meter_aux4') self.meter_aux5 = kwargs.get('meter_aux5') self.quiet = kwargs.get('quiet') def _create_url(self, meter_name): return '%s/user/%s/powermeter/%s/upload.json?key=%s' % ( WATTZON_API_URL, self.api_username, urllib.quote(meter_name), self.api_key ) def setup(self): super(WattzOnMixin, self).setup() if not (self.api_key and self.api_username and self.api_passwd and self.meter_ch1): print 'WattzOn Error: Insufficient credentials' if not self.api_key: print ' No API Key' if not self.api_username: print ' No API Username' if not self.api_passwd: print ' No API Passord' if not self.meter_ch1: print ' No Powermeter Name for CH1' sys.exit(1) p = urllib2.HTTPPasswordMgrWithDefaultRealm() p.add_password( 'WattzOn', WATTZON_API_URL, self.api_username, self.api_passwd) auth = urllib2.HTTPBasicAuthHandler(p) opener = urllib2.build_opener(auth) urllib2.install_opener(opener) self.ch1_url = self._create_url(self.meter_ch1) self.ch2_url = self.meter_ch2 and self._create_url(self.meter_ch2) or '' self.aux1_url = self.meter_aux1 and self._create_url(self.meter_aux1) or '' self.aux2_url = self.meter_aux2 and self._create_url(self.meter_aux2) or '' self.aux3_url = self.meter_aux3 and self._create_url(self.meter_aux3) or '' self.aux4_url = self.meter_aux4 and self._create_url(self.meter_aux4) or '' self.aux5_url = self.meter_aux5 and self._create_url(self.meter_aux5) or '' self.call_period = MINUTE self.last_call = 0 def _make_call(self, url, timestamp, magnitude): data = { 'updates': [ { 'timestamp': timestamp, 'power': { 'magnitude': int(magnitude), # truncated by WattzOn API, anyway 'unit': 'W', } }, ] } req = RequestPostWithContentType('application/json', url, json.dumps(data)) f = urllib2.urlopen(req) return f.read() def process(self, packet): super(WattzOnMixin, self).process(packet) now = int(time.time()) if not self.last_call: self.last_call = now return if now < (self.last_call+self.call_period): return self.last_call = now timestamp = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()) sn = getserial(packet) try: delta = self.buffer.delta_over(sn, self.call_period) except ZeroDivisionError, zde: return # not enough data in buffer result = self._make_call(self.ch1_url, timestamp, delta['ch1_watts']) if not self.quiet: print 'WattzOn: %s' % (timestamp,) print ' [%s] magnitude: %s w/ result: %s' % ( self.meter_ch1, delta['ch1_watts'], result) for meter_type in ['ch2', 'aux1', 'aux2', 'aux3', 'aux4', 'aux5']: if getattr(self, meter_type+'_url', None) and delta[meter_type+'_watts']: result = self._make_call( getattr(self, meter_type+'_url'), timestamp, delta[meter_type+'_watts']) if not self.quiet: print ' [%s] magnitude: %s w/ result: %s' % ( getattr(self, 'meter_'+meter_type), delta[meter_type+'_watts'], result) def handle(self, exception): if type(exception) == urllib2.HTTPError: print 'HTTPError: ', exception print ' URL: ', self.update_url print ' username: ', self.api_username print ' passwd: ', self.api_passwd print ' API key: ', self.api_key print ' powermeter: ', self.powermeter return True return super(WattzOnMixin, self).handle(exception) if __name__ == '__main__': parser = optparse.OptionParser() parser.add_option('--serial', action='store_true', dest='serial_read', default=False, help='read from Serial Port') parser.add_option('--serialport', dest='serial_port', help='serial port') parser.add_option('-b', '--baudrate', dest='baudrate', help='baud rate') parser.add_option('--ip', action='store_true', dest='ip_read', default=False, help='read from EtherBee') parser.add_option('--host', dest='ip_host', help='ip host') parser.add_option('--port', dest='ip_port', help='ip port') parser.add_option('-p', '--print', action='store_true', dest='print_out', default=False, help='print data to screen') parser.add_option('-d', '--database', action='store_true', dest='db_write', default=False, help='write data to db') parser.add_option('--db-host', dest='db_host', help='db host') parser.add_option('--db-user', dest='db_user', help='db user') parser.add_option('--db-passwd', dest='db_passwd', help='db passwd') parser.add_option('--db-database', dest='db_database', help='db database') parser.add_option('--wattzon', action='store_true', dest='wattzon_out', default=False, help='upload data via WattzOn API') parser.add_option('--wattzon-user', dest='wattzon_user', help='WattzOn username') parser.add_option('--wattzon-pass', dest='wattzon_pass', help='WattzOn password') parser.add_option('--wattzon-key', dest='wattzon_key', help='WattzOn API key') parser.add_option('--wattzon-ch1', dest='wattzon_ch1', help='WattzOn powermeter name for CH1') parser.add_option('--wattzon-ch2', dest='wattzon_ch2', help='WattzOn powermeter name for CH2') parser.add_option('--wattzon-aux1', dest='wattzon_aux1', help='WattzOn powermeter name for AUX1') parser.add_option('--wattzon-aux2', dest='wattzon_aux2', help='WattzOn powermeter name for AUX2') parser.add_option('--wattzon-aux3', dest='wattzon_aux3', help='WattzOn powermeter name for AUX3') parser.add_option('--wattzon-aux4', dest='wattzon_aux4', help='WattzOn powermeter name for AUX4') parser.add_option('--wattzon-aux5', dest='wattzon_aux5', help='WattzOn powermeter name for AUX5') parser.add_option('-q', '--quiet', action='store_true', dest='quiet', default=False, help='quiet output') (options, args) = parser.parse_args() # Packet Processor Setup if not (options.print_out or options.db_write or options.wattzon_out): print 'Please choose a processing option (or \'-h\' for more help):' print ' -p (print to screen)' print ' -d (write to databse)' print ' --wattzon (update WattzOn)' sys.exit(1) buffer_timeframe = 5*MINUTE bases = [BaseProcessor] kwargs = { 'quiet': options.quiet, } if options.print_out: bases.insert(0, PrintMixin) if options.db_write: bases.insert(0, DatabaseMixin) kwargs.update({ 'db_host': options.db_host, 'db_user': options.db_user, 'db_passwd': options.db_passwd, 'db_database': options.db_database, }) if options.wattzon_out: bases.insert(0, WattzOnMixin) kwargs.update({ 'api_key': options.wattzon_key, 'api_username': options.wattzon_user, 'api_passwd': options.wattzon_pass, 'meter_ch1': options.wattzon_ch1, 'meter_ch2': options.wattzon_ch2, 'meter_aux1': options.wattzon_aux1, 'meter_aux2': options.wattzon_aux2, 'meter_aux3': options.wattzon_aux3, 'meter_aux4': options.wattzon_aux4, 'meter_aux5': options.wattzon_aux5, }) Processor = new.classobj('Processor', tuple(bases), {}) processor = Processor( buffer_timeframe, **kwargs) # Packet Server Setup if options.serial_read: options.serial_port = options.serial_port and options.serial_port or SERIALPORT options.baudrate = options.baudrate and options.baudrate or BAUDRATE server = SerialPacketServer(processor, options.serial_port, options.baudrate) elif options.ip_read: options.ip_host = options.ip_host and options.ip_host or HOST options.ip_port = options.ip_port and options.ip_port or PORT server = SocketPacketServer(processor, options.ip_host, options.ip_port) else: print 'Please choose a data feed (or \'-h\' for more help):' print ' --serial (read from serial)' print ' --ip (read from EtherBee)' sys.exit(1) server.run() sys.exit(0)