#!/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 a server. Includes support for uploading to the following services: * WattzOn * PlotWatt * MyEnerSave * eragy * PeoplePower 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 How to specify options: Options can be specified via command line or in a configuration file. Use -h or --help to see the complete list of options. The configuration file is INI format, with parameter names as shown in the help output. For example: [general] serial_read = true serial_port = COM1 ip_read = false ip_host = ip_port = [database] db_out = true db_host = localhost db_user = ecm db_passwd = ecm db_database = ecm [plotwatt] plotwatt_out = true pw_map = 311111_ch1,123 pw_house_id = 1234 pw_api_key = 12345 [myenersave] myenersave_out = false mes_token = 123 Database Configuration: When saving data to database, this script writes power (watts) to a table called ecm and and optionally writes energy (watt-hours) to a table called ecmwh. The parameter DB_RECORD_WH (enabled by default) determines whether energy data will be recorded to database. In the default configuration, power is recorded about once per minute and energy is recorded about once per hour. Create the database 'ecm' 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; Create the power table 'ecm' by doing something like this: 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); Create the energy table 'ecmwh' 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, and make the ecm_serial the primary key without auto_increment. In that case the database is used for data transfer, not data capture. WattzOn Configuration: Use the WATTZON_* variables in this script to set the wattzon parameters instead of specifying them on the command line, or specify the parameters in a configuration file. PlotWatt Configuration: First register for a plotwatt account at www.plotwatt.com. You should receive a house ID and an api key. Then configure 'meters' at the plotwatt web site that correspond to channels on each ECM. Using curl to create 2 meters looks something like this: curl -d "number_of_new_meters=2" http://API_KEY:@plotwatt.com/api/v2/new_meters Using curl to list meters looks something like this: curl http://API_KEY:@plotwatt.com/api/v2/list_meters Use the meter identifiers provided by plotwatt when uploading data, with each meter associated with a channel/aux of an ECM. For example, to upload data from ch1 and ch2 from ecm serial 399999 to meters 1000 and 1001, respectively, run ecmread like this: ecmread.py --serial --serialport=/dev/ttyUSB0 --plotwatt --pw-house-id XXXX --pw-api-key XXXXXXXXXXXXXXXX --pw-map 399999_ch1,1000,399999_ch2,1001 Use the PLOTWATT_* variables in this script to set the plotwatt parameters instead of specifying them on the command line, or specify the parameters in a configuration file. MyEnerSave Configuration: Use the MES_* variables in this script to set the myEnerSave parameters instead of specifying them on the command line, or specify the parameters in a configuration file. PeoplePower Configuration: Borrow someone's iPhone or droid, run the People Power app, register for an account, enter TED as the device type. An activation key will be sent to your email account. Use the activation key to obtain a device authorization token. Create an XML file with the activation key, a 'hub ID' (this is the identifier for your system - use any number), and a device type (1004 is the type for a TED-5000 and seems to work). The ecmread.py script acts as the hub, and each channel of each ECM is a device. For example, create the file req.xml with the following contents: xxx:yyyyyyyyyyyyyyy 1 1004 Send the file using HTTP POST to the hub activation URL: curl -X POST -d @req.xml http://esp.peoplepowerco.com/espapi/rest/hubActivation You should get a response with resultCode 0 and a deviceAuthToken. The response also contains pieces of the URL that you should use for subsequent configuration, for example: https://esp.peoplepowerco.com:8443/deviceio/ml Now add each channel of each ECM as a new ppco device. This can be done manually, or let ecmread do it based on the channel-to-device map. Use the device type 1005, which is a generic TED MTU. Create an arbitrary deviceId for each ECM channel that will be tracked. It looks like the deviceId must contain hex characters, so use c1 and c2 for channels 1 and 2 and use aN for the aux. The seq field is a monotonically increasing nonce. Send the file to the URL you received on activation. curl -H "Content-Type: text/xml" -H "PPCAuthorization: esp token=TOKEN" -d @add.xml https://esp.peoplepowerco.com:8443/deviceio/ml To get a list of devices that ppco recognizes: curl https://esp.peoplepowerco.com/espapi/rest/deviceTypes Use the PPCO_* variables in this script to set the People Power parameters instead of specifying them on the command line, or specify the parameters in a configuration file. Eragy Configuration: Use the ERAGY_* variables in this script to set the eragy parameters instead of specifying them on the command line, or specify the parameters in a configuration file. Changelog: - 2.3.0 26dec2011 mwall * use containment not polymorphism to control processing of multiple outputs * added support for peoplepower - 2.2.0 - mwall * consolidate packet reading so socket and serial share the same code * reject any packet that is not an ecm-1240 packet * enable multi-ecm support for wattzon * added support for plotwatt (supercedes plotwatt_v0.1.py) * added support for myEnerSave (supercedes myEnerSave.py) * added default settings within script to reduce need for command-line options * use UTC throughout, but display local time * added option to read parameters from configuration file * refactor options to make them consistent * support use of multiple packet processors (e.g. upload to multiple services) * fixed wattzon usage of urllib to enable multiple, concurrent cloud services - 2.1.2 22dec2011 mwall * indicate ecm serial number in checksum mismatch log messages * use simpler form of extraction for ser_no * eliminate superfluous use of hex/str/int conversions * added to packet compilation the DC voltage on aux5 * display reset counter state when printing packets * respect the reset flag - if a reset has ocurred, skip calculation until another packet is added to the buffer - 2.1.1 20dec2011 mwall * incorporate ben's packet reading changes 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 columns in a database table, 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; ben; mwall' __version__ = '2.3.0' # set debug to 1 to enable much more logging. note that this will affect the # application behavior, especially when sampling the serial line, as it changes # the timing of read operations. DEBUG = 0 MINUTE = 60 HOUR = 60 * MINUTE DAY = 24 * HOUR BUFFER_TIMEFRAME = 5*MINUTE # serial settings # the com/serial port the ecm is connected to (COM4, /dev/ttyS01, etc) SERIAL_PORT = "/dev/ttyUSB0" SERIAL_BAUD = 19200 # the baud rate we talk to the ecm # ethernet settings IP_HOST = '' IP_PORT = 8083 # default port that the EtherBee is pushing data to SOCKET_TIMEOUT = 60 # database defaults DB_HOST = 'localhost' DB_USER = '' DB_PASSWD = '' DB_DATABASE = '' DB_RECORD_WH = 1 DB_INSERT_PERIOD_W = MINUTE # how often to record power to database DB_INSERT_PERIOD_WH = HOUR # how often to record energy to database # WattzOn defaults WATTZON_API_URL = 'http://www.wattzon.com/api/2009-01-27/3' WATTZON_UPLOAD_PERIOD = MINUTE WATTZON_TIMEOUT = 15 # seconds # comma-delimited list of channel,meter pairs, e.g. 311111_ch1,living room WATTZON_MAP = '' WATTZON_API_KEY = '' WATTZON_USER = '' WATTZON_PASS = '' # PlotWatt defaults # https://plotwatt.com/docs/api # Recommended upload period is one minute to a few minutes. Recommended # sampling as often as possible, no faster than once per second. plotwatt # provides a separate API key for brultech users, but this script uses the # standard API key. PLOTWATT_BASE_URL = 'http://plotwatt.com' PLOTWATT_UPLOAD_URL = '/api/v2/push_readings' PLOTWATT_UPLOAD_PERIOD = MINUTE PLOTWATT_TIMEOUT = 15 # seconds # comma-delimited list of channel,meter pairs, e.g. 311111_ch1,1001 PLOTWATT_MAP = '' PLOTWATT_HOUSE_ID = '' PLOTWATT_API_KEY = '' # myEnerSave defaults MES_URL = 'http://data.myenersave.com/fetcher/data' MES_UPLOAD_PERIOD = MINUTE MES_TIMEOUT = 15 # seconds MES_TOKEN = '' # PeoplePower defaults # http://developer.peoplepowerco.com/docs # Recommended upload period is 15 minutes. # Requires header auth: # PPCAuthorization: esp token=[token] # Timestamps are in the format: # YYYY-MM-DDThh:mm:ss[Z|(+|-)hh:mm] PPCO_URL = 'https://esp.peoplepowerco.com:8443/deviceio/ml' PPCO_UPLOAD_PERIOD = 15 * MINUTE PPCO_TIMEOUT = 15 # seconds PPCO_TOKEN = '' PPCO_HUBID = 1 PPCO_MAP = '' PPCO_FIRST_NONCE = 3 PPCO_DEVICE_TYPE = 1005 # eragy defaults ERAGY_UPLOAD_PERIOD = 15 * MINUTE ERAGY_TIMEOUT = 15 # seconds 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 try: import ConfigParser except Exception, e: ConfigParser = None # PACKET SETTINGS START_HEADER0 = 254 START_HEADER1 = 255 START_HEADER2 = 3 END_HEADER0 = 255 END_HEADER1 = 254 DATA_BYTES_LENGTH = 59 # does not include the start/end headers SEC_COUNTER_MAX = 16777216 ECM1240_CHANNELS = ['ch1', 'ch2', 'aux1', 'aux2', 'aux3', 'aux4', 'aux5'] class CounterResetError(Exception): def __init__(self, msg): self.msg = msg def __str__(self): return repr(self.msg) # 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 getgmtime(): return int(time.time()) def cleanvalue(s): '''ensure that values read from configuration file are sane''' s = s.replace('\n', '') # we never want newlines s = s.replace('\r', '') if s.lower() == 'false': s = False elif s.lower() == 'true': s = True return s def pairs2dict(s): '''conver comma-delimited name,value pairs to a dictionary''' items = s.split(',') m = {} for k, v in zip(items[::2], items[1::2]): m[k] = v return m def getresetcounter(byte): '''extract the reset counter from a byte''' return byte & 0b00000111 # same as 0x07 def getserial(packet): '''extract the ECM serial number from a compiled packet''' return "%d%d" % (packet['unit_id'], packet['ser_no']) def getserialraw(packet): '''extract the ECM serial number from a raw packet''' sn1 = ord(packet[26:27]) sn2 = ord(packet[27:28]) * 256 id1 = ord(packet[29:30]) return "%d%d" % (id1, sn1+sn2) def obfuscate_serial(sn): '''obfuscate a brultech serial number - keep the last 3 digits only''' n = len(sn) return 'XXX%s' % sn[n-3:n] def calculate_checksum(packet): '''calculate the packet checksum''' checksum = START_HEADER0 checksum += START_HEADER1 checksum += START_HEADER2 checksum += sum([ord(c) for c in packet]) checksum += END_HEADER0 checksum += END_HEADER1 return checksum & 0xff def calculate(now, prev): '''calc average watts/s between packets''' # if reset counter has changed since last packet, skip the calculation c0 = getresetcounter(prev['flag']) c1 = getresetcounter(now['flag']) if c1 != c0: raise CounterResetError("old: %d new: %d" % (c0, c1)) if now['secs'] < prev['secs']: now['secs'] += SEC_COUNTER_MAX # handle seconds counter overflow secs_delta = float(now['secs'] - prev['secs']) ret = now # CH1/2 Watts ret['ch1_watts'] = (ret['ch1_aws'] - prev['ch1_aws']) / secs_delta ret['ch2_watts'] = (ret['ch2_aws'] - prev['ch2_aws']) / secs_delta ret['ch1_positive_watts'] = (ret['ch1_pws'] - prev['ch1_pws']) / secs_delta ret['ch2_positive_watts'] = (ret['ch2_pws'] - prev['ch2_pws']) / secs_delta ret['ch1_negative_watts'] = ret['ch1_watts'] - ret['ch1_positive_watts'] ret['ch2_negative_watts'] = ret['ch2_watts'] - ret['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 ret['ch1_pwh'] = ret['ch1_pws'] / 3600000.0 ret['ch2_pwh'] = ret['ch2_pws'] / 3600000.0 ret['ch1_nwh'] = (ret['ch1_aws'] - ret['ch1_pws']) / 3600000.0 ret['ch2_nwh'] = (ret['ch2_aws'] - ret['ch2_pws']) / 3600000.0 ret['ch1_wh'] = ret['ch1_pwh'] - ret['ch1_nwh'] ret['ch2_wh'] = ret['ch2_pwh'] - ret['ch2_nwh'] ret['aux1_wh'] = ret['aux1_ws'] / 3600000.0 ret['aux2_wh'] = ret['aux2_ws'] / 3600000.0 ret['aux3_wh'] = ret['aux3_ws'] / 3600000.0 ret['aux4_wh'] = ret['aux4_ws'] / 3600000.0 ret['aux5_wh'] = ret['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 (ret['ch1_positive_watts'] == 0): ret['ch1_watts'] *= -1 if (ret['ch2_positive_watts'] == 0): ret['ch2_watts'] *= -1 # AUX1-5 Watts ret['aux1_watts'] = (ret['aux1_ws'] - prev['aux1_ws']) / secs_delta ret['aux2_watts'] = (ret['aux2_ws'] - prev['aux2_ws']) / secs_delta ret['aux3_watts'] = (ret['aux3_ws'] - prev['aux3_ws']) / secs_delta ret['aux4_watts'] = (ret['aux4_ws'] - prev['aux4_ws']) / secs_delta ret['aux5_watts'] = (ret['aux5_ws'] - prev['aux5_ws']) / secs_delta return ret # Packet Server Classes class BasePacketServer(object): def __init__(self, packet_processor): self.packet_buffer = CompoundBuffer(BUFFER_TIMEFRAME) 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[26:28]) # 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]) # DC voltage on AUX5 (2 bytes) now['aux5_volts'] = self._convert(packet[57:59]) return now def process(self, packet): packet = self._compile(packet) self.packet_buffer.insert(getgmtime(), packet) for p in self.packet_processor: try: p.process(packet, self.packet_buffer) except Exception, e: if not p.handle(e): print 'Exception [in %s]: %s' % (self, e) # Read the indicated number of bytes. This should be overridden by derived # classes to do the actual reading of bytes. def readbytes(self, count): return '' # Loop until no more data are available. Put the data into a packet, check # the data, then compile the packet into a structure. def read(self): while True: data = self.readbytes(1) if not data: break byte = ord(data) if byte != START_HEADER0: dbgmsg("expected START_HEADER0 %s, got %s" % (hex(START_HEADER0), hex(byte))) continue data = self.readbytes(1) byte = ord(data) if byte != START_HEADER1: dbgmsg("expected START_HEADER1 %s, got %s" % (hex(START_HEADER1), hex(byte))) continue data = self.readbytes(1) byte = ord(data) if byte != START_HEADER2: dbgmsg("expected START_HEADER2 %s, got %s" % (hex(START_HEADER2), hex(byte))) continue packet = '' while len(packet) < DATA_BYTES_LENGTH: data = self.readbytes(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.readbytes(1) byte = ord(data) if byte != END_HEADER0: dbgmsg("expected END_HEADER0 %s, got %s" % (hex(END_HEADER0), hex(byte))) continue data = self.readbytes(1) byte = ord(data) if byte != END_HEADER1: dbgmsg("expected END_HEADER1 %s, got %s" % (hex(END_HEADER1), hex(byte))) continue # we only handle ecm-1240 packets uid = ord(packet[29:30]) if uid != 3: logmsg("unrecognized unit id: expected %s, got %s" % (hex(3), hex(uid))) continue # if the checksum is incorrect, ignore the packet checksum = calculate_checksum(packet) data = self.readbytes(1) byte = ord(data) if byte != checksum: logmsg("bad checksum for %s: expected %s, got %s" % (getserialraw(packet), hex(checksum), hex(byte))) continue packet = [ord(c) for c in packet] self.process(packet) # Loop forever collecting data then processing packets. Break only for # keyboard interrupts. def run(self): try: for p in self.packet_processor: p.setup() while True: try: self.read() except Exception, e: if type(e) == KeyboardInterrupt: raise e traceback.print_exc() except Exception, e: traceback.print_exc() sys.exit(1) finally: for p in self.packet_processor: p.cleanup() class SerialPacketServer(BasePacketServer): def __init__(self, packet_processor, port=SERIAL_PORT, rate=SERIAL_BAUD): super(SerialPacketServer, self).__init__(packet_processor) if not serial: print 'Error: serial module could not be imported.' sys.exit(1) self._port = port self._baudrate = rate self.conn = None def readbytes(self, count): return self.conn.read(count) def read(self): try: self.conn = serial.Serial(self._port, self._baudrate) self.conn.open() super(SerialPacketServer, self).read(); finally: if self.conn: self.conn.close() self.conn = None class SocketPacketServer(BasePacketServer): def __init__(self, packet_processor, host=IP_HOST, port=IP_PORT): super(SocketPacketServer, self).__init__(packet_processor) socket.setdefaulttimeout(SOCKET_TIMEOUT) # override None self._host = host self._port = port self.sock = None self.conn = None def readbytes(self, count): return self.conn.recv(count) 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() super(SocketPacketServer, self).read(); 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 # Buffer 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 = getgmtime() cull_index = bisect.bisect(self.time_points, (now-self.max_timeframe, {})) del(self.time_points[:cull_index]) def data_over(self, time_delta): now = getgmtime() delta_index = bisect.bisect(self.time_points, (now-time_delta, {})) return self.time_points[delta_index:] def delta_over(self, time_delta): now = getgmtime() delta_index = bisect.bisect(self.time_points, (now-time_delta, {})) offset = self.time_points[delta_index][1] current = self.time_points[-1][1] return calculate(current, offset) def size(self): return len(self.time_points) 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 size(self, ecm_serial): return self.getbuffer(ecm_serial).size() 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] # Packet Processor Classes class BaseProcessor(object): def __init__(self, *args, **kwargs): self.quiet = kwargs.get('quiet') pass def setup(self): pass def process(self, packet, packet_buffer): pass def handle(self, exception): return False def cleanup(self): pass class PrintProcessor(BaseProcessor): def __init__(self, *args, **kwargs): super(PrintProcessor, self).__init__(*args, **kwargs) self.prev_packet = {} def process(self, packet, packet_buffer): sn = getserial(packet) if sn in self.prev_packet: try: p = calculate(packet, self.prev_packet[sn]) except ZeroDivisionError, zde: print "not enough data in buffer for %s" % sn return except CounterResetError, cre: print "counter reset for %s: %s" % (sn, cre.msg) return ts = time.strftime("%Y/%m/%d %H:%M:%S", time.localtime()) # 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 ts+": ECM: %s" % sn print ts+": Counter: %d" % getresetcounter(p['flag']) print ts+": Volts: %9.2fV" % p['volts'] print ts+": Ch1 Amps: %9.2fA" % p['ch1_amps'] print ts+": Ch2 Amps: %9.2fA" % p['ch2_amps'] print ts+": Ch1 Watts: % 13.6fKWh (% 5dW)" % (p['ch1_wh'] , p['ch1_watts']) print ts+": Ch1 Positive Watts: % 13.6fKWh (% 5dW)" % (p['ch1_pwh'], p['ch1_positive_watts']) print ts+": Ch1 Negative Watts: % 13.6fKWh (% 5dW)" % (p['ch1_nwh'], p['ch1_negative_watts']) print ts+": Ch2 Watts: % 13.6fKWh (% 5dW)" % (p['ch2_wh'] , p['ch2_watts']) print ts+": Ch2 Positive Watts: % 13.6fKWh (% 5dW)" % (p['ch2_pwh'], p['ch2_positive_watts']) print ts+": Ch2 Negative Watts: % 13.6fKWh (% 5dW)" % (p['ch2_nwh'], p['ch2_negative_watts']) print ts+": Aux1 Watts: % 13.6fKWh (% 5dW)" % (p['aux1_wh'], p['aux1_watts']) print ts+": Aux2 Watts: % 13.6fKWh (% 5dW)" % (p['aux2_wh'], p['aux2_watts']) print ts+": Aux3 Watts: % 13.6fKWh (% 5dW)" % (p['aux3_wh'], p['aux3_watts']) print ts+": Aux4 Watts: % 13.6fKWh (% 5dW)" % (p['aux4_wh'], p['aux4_watts']) print ts+": Aux5 Watts: % 13.6fKWh (% 5dW)" % (p['aux5_wh'], p['aux5_watts']) self.prev_packet[sn] = packet class DatabaseProcessor(BaseProcessor): def __init__(self, *args, **kwargs): super(DatabaseProcessor, 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 if not MySQLdb: print 'Error: MySQLdb module could not be imported.' sys.exit(1) def setup(self): 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 = DB_INSERT_PERIOD_W self.last_insert = {} self.insert_period_wh = DB_INSERT_PERIOD_WH self.last_insert_wh = {} def process(self, packet, packet_buffer): sn = getserial(packet) now = getgmtime() if sn in self.last_insert and now < (self.last_insert[sn] + self.insert_period): return try: delta = packet_buffer.delta_over(sn, self.insert_period) except ZeroDivisionError, zde: return # not enough data in buffer except CounterResetError, cre: return # counter reset so skip calculation 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, e): if type(e) == MySQLdb.Error: print 'MySQL Error: [#%d] %s' % (e.args[0], e.args[1]) return True return super(DatabaseProcessor, self).handle(e) def cleanup(self): if not self.conn: return self.conn.commit() self.conn.close() class WattzOnProcessor(BaseProcessor): def __init__(self, *args, **kwargs): super(WattzOnProcessor, self).__init__(*args, **kwargs) self.api_key = kwargs.get('wo_api_key') or WATTZON_API_KEY self.username = kwargs.get('wo_user') or WATTZON_USER self.password = kwargs.get('wo_pass') or WATTZON_PASS self.map_str = kwargs.get('wo_map') or WATTZON_MAP def _create_url(self, meter_name): return '%s/user/%s/powermeter/%s/upload.json?key=%s' % ( WATTZON_API_URL, self.username, urllib.quote(meter_name), self.api_key ) def _make_call(self, meter, timestamp, magnitude): data = { 'updates': [ { 'timestamp': timestamp, 'power': { 'magnitude': int(magnitude), # truncated by WattzOn API 'unit': 'W', } }, ] } url = self._create_url(meter) req = urllib2.Request(url) req.add_header("Content-Type", "application/json") req.add_header("User-Agent", "ecmread/%s" % __version__) return self.urlopener.open(req, json.dumps(data), WATTZON_TIMEOUT) def setup(self): if not (self.api_key and self.username and self.password and self.map_str): print 'WattzOn Error: Insufficient parameters' if not self.api_key: print ' No API key' if not self.username: print ' No username' if not self.password: print ' No passord' if not self.map_str: print ' No mapping between ECM channels and WattzOn meters' sys.exit(1) self.map = pairs2dict(self.map_str) if not self.map: print 'WattzOn Error: cannot determine channel-meter map' sys.exit(1) p = urllib2.HTTPPasswordMgrWithDefaultRealm() p.add_password('WattzOn', WATTZON_API_URL, self.username, self.password) auth = urllib2.HTTPBasicAuthHandler(p) self.urlopener = urllib2.build_opener(auth) self.upload_period = WATTZON_UPLOAD_PERIOD self.last_upload = {} def process(self, packet, packet_buffer): sn = getserial(packet) now = getgmtime() if sn in self.last_upload and now < (self.last_upload[sn] + self.upload_period): return self.last_upload[sn] = now data = packet_buffer.data_over(sn, self.upload_period) for a,b in zip(data[0:], data[1:]): p = calculate(b[1],a[1]) for c in ECM1240_CHANNELS: key = sn + '_' + c if key in self.map: meter = self.map[key] timestamp = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(b[0])) result = self._make_call(meter, timestamp, p[c+'_watts']) if not self.quiet: logmsg('WattzOn: %s' % timestamp) logmsg(' [%s] magnitude: %s result: %s' % ( meter, p[c+'_watts'], result.read())) dbgmsg('WattzOn: %s' % result.info()) def handle(self, e): if type(e) == urllib2.HTTPError: print 'HTTPError: ', e print ' username: ', self.username print ' password: ', self.password print ' API key: ', self.api_key return True return super(WattzOnProcessor, self).handle(e) class PlotWattProcessor(BaseProcessor): def __init__(self, *args, **kwargs): super(PlotWattProcessor, self).__init__(*args, **kwargs) self.api_key = kwargs.get('pw_api_key') or PLOTWATT_API_KEY self.house_id = kwargs.get('pw_house_id') or PLOTWATT_HOUSE_ID self.map_str = kwargs.get('pw_map') or PLOTWATT_MAP b64s = base64.encodestring('%s:%s' % (self.api_key, ''))[:-1] self.authheader = "Basic %s" % b64s self.url = PLOTWATT_BASE_URL + PLOTWATT_UPLOAD_URL def _post(self, s): if len(s) == 0: return payload = ','.join(s) try: req = urllib2.Request(self.url) req.add_header("Content-Type", "text/xml") req.add_header("Authorization", self.authheader) req.add_header("User-Agent", "ecmread/%s" % __version__) result = urllib2.urlopen(req, payload, PLOTWATT_TIMEOUT) if not self.quiet: logmsg('PlotWatt: post complete') dbgmsg('PlotWatt: url: %s\n info: %s' % (result.geturl(), result.info())) except urllib2.HTTPError, he: print 'PlotWatt Error: ', he print ' URL: ', self.url print ' API key: ', self.api_key print ' house ID: ', self.house_id print ' payload: ', payload def setup(self): if not (self.api_key and self.house_id and self.map_str): print 'PlotWatt Error: Insufficient parameters' if not self.api_key: print ' No API key' if not self.house_id: print ' No house ID' if not self.map_str: print ' No mapping between ECM channels and PlotWatt meters' sys.exit(1) self.map = pairs2dict(self.map_str) if not self.map: print 'PlotWatt Error: cannot determine channel-meter map' sys.exit(1) self.upload_period = PLOTWATT_UPLOAD_PERIOD self.last_upload = {} def process(self, packet, packet_buffer): sn = getserial(packet) now = getgmtime() if sn in self.last_upload and now < (self.last_upload[sn] + self.upload_period): return self.last_upload[sn] = now s = [] data = packet_buffer.data_over(sn, self.upload_period) for a,b in zip(data[0:], data[1:]): p = calculate(b[1],a[1]) for c in ECM1240_CHANNELS: key = sn + '_' + c if key in self.map: meter = self.map[key] # format for each meter is: meter-id,kW,gmt-timestamp s.append("%s,%1.4f,%d" % (meter, p[c+'_watts']/1000, b[0])) self._post(s) class MyEnerSaveProcessor(BaseProcessor): def __init__(self, *args, **kwargs): super(MyEnerSaveProcessor, self).__init__(*args, **kwargs) self.url = kwargs.get('mes_url') or MES_URL self.token = kwargs.get('mes_token') or MES_TOKEN def _post(self, s): if len(s) == 0: return s.insert(0, '') payload = ''.join(s) try: req = urllib2.Request(self.url) req.add_header("Content-Type", "application/xml") req.add_header("Token", self.token) req.add_header("User-Agent", "ecmread/%s" % __version__) result = urllib2.urlopen(req, payload, MES_TIMEOUT) if not self.quiet: logmsg('MyEnerSave: post complete') dbgmsg('MyEnerSave: url: %s\n info: %s' % (result.geturl(), result.info())) except urllib2.HTTPError, he: print 'MyEnerSave Error: ', he print ' URL: ', self.url print ' token: ', self.token print ' payload: ', payload def setup(self): if not (self.url and self.token): print 'MyEnerSave Error: Insufficient parameters' if not self.url: print ' No URL' if not self.token: print ' No token' sys.exit(1) self.upload_period = MES_UPLOAD_PERIOD self.last_upload = {} def process(self, packet, packet_buffer): sn = getserial(packet) now = getgmtime() if sn in self.last_upload and now < (self.last_upload[sn] + self.upload_period): return self.last_upload[sn] = now s = [] data = packet_buffer.data_over(sn, self.upload_period) for a,b in zip(data[0:], data[1:]): p = calculate(b[1],a[1]) for c in ECM1240_CHANNELS: s.append("\r\n" % (b[0], p[c+'_wh'])) if len(s): s.insert(0, '') s.append('') self._post(s) class PeoplePowerProcessor(BaseProcessor): def __init__(self, *args, **kwargs): super(PeoplePowerProcessor, self).__init__(*args, **kwargs) self.url = kwargs.get('pp_url') or PPCO_URL self.token = kwargs.get('pp_token') or PPCO_TOKEN self.hub_id = kwargs.get('pp_hub_id') or PPCO_HUBID self.map_str = kwargs.get('pp_map') or PPCO_MAP self.nonce = PPCO_FIRST_NONCE self.dev_type = PPCO_DEVICE_TYPE def _post(self, s): if len(s) == 0: return s.insert(0, '') s.insert(1, '' % (self.hub_id, self.nonce)) s.append('') payload = ''.join(s) try: req = urllib2.Request(self.url) req.add_header("Content-Type", "text/xml") req.add_header("PPCAuthorization", "esp token=%s" % self.token) req.add_header("User-Agent", "ecmread/%s" % __version__) result = urllib2.urlopen(req, payload, PPCO_TIMEOUT) self.nonce += 1 if not self.quiet: logmsg('PeoplePower: post complete') dbgmsg('PeoplePower: url: %s\n info: %s' % (result.geturl(), result.info())) except urllib2.HTTPError, he: print 'PeoplePower Error: ', he print ' URL: ', self.url print ' token: ', self.token print ' hub ID: ', self.hub_id print ' payload: ', payload def _add_devices(self): s = [] nd = 0 for key in self.map.keys(): s.append('' % ( self.map[key], self.dev_type)) nd += 1 self._post(s) def setup(self): if not (self.url and self.token and self.hub_id and self.map_str): print 'PeoplePower Error: Insufficient parameters' if not self.url: print ' No URL' if not self.token: print ' No token' if not self.hub_id: print ' No hub ID' if not self.map_str: print ' No mapping between ECM channels and PeoplePower devices' sys.exit(1) self.map = pairs2dict(self.map_str) if not self.map: print 'PeoplePower Error: cannot determine channel-meter map' sys.exit(1) self.upload_period = PPCO_UPLOAD_PERIOD self.last_upload = {} self._add_devices() def process(self, packet, packet_buffer): sn = getserial(packet) now = getgmtime() if sn in self.last_upload and now < (self.last_upload[sn] + self.upload_period): return self.last_upload[sn] = now s = [] nm = 0 data = packet_buffer.data_over(sn, self.upload_period) for a,b in zip(data[0:], data[1:]): p = calculate(b[1],a[1]) for c in ECM1240_CHANNELS: key = sn + '_' + c if key in self.map: ts = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(b[0])) s.append('' % (self.map[key], self.dev_type, ts)) s.append(' %1.4f' % p[c+'_watts']) s.append(' %1.4f' % p[c+'_wh']) s.append('') nm += 1 self._post(s) class EragyProcessor(BaseProcessor): def __init__(self, *args, **kwargs): super(EragyProcessor, self).__init__(*args, **kwargs) def setup(self): self.upload_period = ERAGY_UPLOAD_PERIOD self.last_upload = {} def process(self, packet, packet_buffer): sn = getserial(packet) now = getgmtime() if sn in self.last_upload and now < (self.last_upload[sn] + self.upload_period): return self.last_upload[sn] = now if __name__ == '__main__': parser = optparse.OptionParser(version=__version__) 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='serial_baud', help='serial baud rate') parser.add_option('--ip', action='store_true', dest='ip_read', default=False, help='read from TCP/IP source such as 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_out', default=False, help='write data to database') parser.add_option('--db-host', dest='db_host', help='database host') parser.add_option('--db-user', dest='db_user', help='database user') parser.add_option('--db-passwd', dest='db_passwd', help='database passwd') parser.add_option('--db-database', dest='db_database', help='database name') parser.add_option('--wattzon', action='store_true', dest='wattzon_out', default=False, help='upload data using WattzOn API') parser.add_option('--wo-user', dest='wo_user', help='WattzOn username') parser.add_option('--wo-pass', dest='wo_pass', help='WattzOn password') parser.add_option('--wo-key', dest='wo_api_key', help='WattzOn API key') parser.add_option('--wo-map', dest='wo_map', help='WattzOn channel-to-meter mapping') parser.add_option('--plotwatt', action='store_true', dest='plotwatt_out', default=False, help='upload data using PlotWatt API') parser.add_option('--pw-house-id', dest='pw_house_id', help='PlotWatt house ID') parser.add_option('--pw-api-key', dest='pw_api_key', help='PlotWatt API key') parser.add_option('--pw-map', dest='pw_map', help='PlotWatt channel-to-meter mapping') parser.add_option('--myenersave', action='store_true', dest='myenersave_out', default=False, help='upload data using MyEnerSave API') parser.add_option('--mes-token', dest='mes_token', help='MyEnerSave token') parser.add_option('--mes-url', dest='mes_url', help='MyEnerSave URL') parser.add_option('--peoplepower', action='store_true', dest='peoplepower_out', default=False, help='upload data using PeoplePower API') parser.add_option('--pp-token', dest='pp_token', help='PeoplePower auth token') parser.add_option('--pp-hub-id', dest='pp_hub_id', help='PeoplePower hub ID') parser.add_option('--pp-url', dest='pp_url', help='PeoplePower URL') parser.add_option('--pp-map', dest='pp_map', help='PeoplePower channel-to-device mapping') parser.add_option('-q', '--quiet', action='store_true', dest='quiet', default=False, help='quiet output') parser.add_option('-v', '--verbose', action='store_false', dest='quiet', default=False, help='verbose output') parser.add_option('-c', '--config-file', dest='configfile', help='read configuration from FILE', metavar='FILE') (options, args) = parser.parse_args() # if there is a configration file, read the parameters from file and set # values on the options object. if options.configfile: if not ConfigParser: print 'ConfigParser not loaded, cannot parse config file' sys.exit(1) config = ConfigParser.ConfigParser() config.read(options.configfile) for section in config.sections(): # section names do not matter for name,value in config.items(section): if not getattr(options, name): setattr(options, name, cleanvalue(value)) # Packet Processor Setup if not (options.print_out or options.db_out or options.wattzon_out or options.plotwatt_out or options.myenersave_out or options.peoplepower_out): print 'Please specify one or more processing options (or \'-h\' for help):' print ' -p print to screen' print ' -d write to databse' print ' --wattzon update WattzOn' print ' --plotwatt update PlotWatt' print ' --myenersave update MyEnerSave' print ' --peoplepower update PeoplePower' sys.exit(1) procs = [] if options.print_out: procs.append(PrintProcessor(args, **{'quiet': options.quiet})) if options.db_out: procs.append(DatabaseProcessor(args, **{ 'quiet': options.quiet, 'db_host': options.db_host, 'db_user': options.db_user, 'db_passwd': options.db_passwd, 'db_database': options.db_database, })) if options.wattzon_out: procs.append(WattzOnProcessor(args, **{ 'quiet': options.quiet, 'wo_api_key': options.wo_api_key, 'wo_user': options.wo_user, 'wo_pass': options.wo_pass, 'wo_map': options.wo_map, })) if options.plotwatt_out: procs.append(PlotWattProcessor(args, **{ 'quiet': options.quiet, 'pw_api_key': options.pw_api_key, 'pw_house_id': options.pw_house_id, 'pw_map': options.pw_map, })) if options.myenersave_out: procs.append(MyEnerSaveProcessor(args, **{ 'quiet': options.quiet, 'mes_token': options.mes_token, 'mes_url': options.mes_url, })) if options.peoplepower_out: procs.append(PeoplePowerProcessor(args, **{ 'quiet': options.quiet, 'pp_url': options.pp_url, 'pp_token': options.pp_token, 'pp_hub_id': options.pp_hub_id, 'pp_map': options.pp_map, })) # Packet Server Setup if options.serial_read: options.serial_port = options.serial_port and options.serial_port or SERIAL_PORT options.serial_baud = options.serial_baud and options.serial_baud or SERIAL_BAUD server = SerialPacketServer(procs, options.serial_port, options.serial_baud) elif options.ip_read: options.ip_host = options.ip_host and options.ip_host or IP_HOST options.ip_port = options.ip_port and options.ip_port or IP_PORT server = SocketPacketServer(procs, options.ip_host, options.ip_port) else: print 'Please specify a data source (or \'-h\' for help):' print ' --serial read from serial' print ' --ip read from TCP/IP e.g. EtherBee' sys.exit(1) server.run() sys.exit(0)