""" This code is released under the GNU Affero General Public License. OpenEnergyMonitor project: http://openenergymonitor.org """ import serial import time import datetime import logging import socket import select import emonhub_coder as ehc """class EmonHubInterfacer Monitors a data source. This almost empty class is meant to be inherited by subclasses specific to their data source. """ class EmonHubInterfacer(object): def __init__(self, name): # Initialize logger self._log = logging.getLogger("EmonHub") # Initialise settings self.name = name self.init_settings = {} self._defaults = {'pause': 'off', 'interval': 0, 'datacode': '0', 'timestamped': 'False'} self._settings = {} self._packet_counter = 0 # This line will stop the default values printing to logfile at start-up # unless they have been overwritten by emonhub.conf entries # comment out if diagnosing a startup value issue self._settings.update(self._defaults) # Initialize interval timer's "started at" timestamp self._interval_timestamp = 0 def close(self): """Close socket.""" pass def read(self): """Read data from socket and process if complete line received. Return data as a list: [NodeID, val1, val2] """ pass def _process_frame(self, frame, timestamp=0.0): """Process a frame of data f (string): 'NodeID val1 val2 ...' This function splits the string into numbers and check its validity. 'NodeID val1 val2 ...' is the generic data format. If the source uses a different format, override this method. Return data as a list: [NodeID, val1, val2] """ # Discard the frame if 'pause' set to 'all' or 'in' if 'pause' in self._settings and \ str.lower(self._settings['pause']) in ['all', 'in']: return # Add timestamp if not done already if not timestamp: timestamp = round(time.time(), 2) # Assign a "Packet" reference number self._packet_counter +=1 ref = self._packet_counter # Log data self._log.debug(str(ref) + " NEW FRAME : " + str(timestamp) + " " + frame) # Get an array out of the space separated string frame = frame.strip().split(' ') # create a RSSI variable self.rssi = False # Validate frame validated = self._validate_frame(ref, frame) if not validated: #self._log.debug('Discard RX Frame "Failed validation"') return else: frame = self._decode_frame(ref, validated) if frame: self._log.debug(str(ref) + " Timestamp : " + str(timestamp)) self._log.debug(str(ref) + " Node : " + str(frame[0])) self._log.debug(str(ref) + " Values : " + str(frame[1:])) frame = [timestamp] + frame # Append RSSI only if value is not 'False' if self.rssi: self._log.debug(str(ref) + " RSSI : " + str(self.rssi)) frame += [self.rssi] frame += [ref] else: return # pause output if 'pause' set to 'all' or 'out' if 'pause' in self._settings \ and str(self._settings['pause']).lower() in ['all', 'out']: return return frame def _validate_frame(self, ref, received): """Validate a frame of data This function performs logical tests to filter unsuitable data. Each test discards frame with a log entry if False Returns True if data frame passes tests. """ # Discard if frame not of the form [node, val1, ...] # with number of elements at least 2 if len(received) < 2: self._log.warning(str(ref) + " Discarded RX frame 'string too short' : " + str(received)) return False # Discard if anything non-numerical found try: [float(val) for val in received] except Exception: self._log.warning(str(ref) + " Discarded RX frame 'non-numerical content' : " + str(received)) return False # Discard if first value is not a valid node id n = float(received[0]) if n % 1 != 0 or n < 0 or n > 31: self._log.warning(str(ref) + " Discarded RX frame 'node id outside scope' : " + str(received)) return False # If it passes all the checks return return received def _decode_frame(self, ref, data): """Decodes a frame of data Performs decoding of data types Returns decoded string of data. """ node = data[0] data = data[1:] decoded = [] # check if node is listed and has individual datacodes for each value if node in ehc.nodelist and 'datacodes' in ehc.nodelist[node]: # fetch the string of datacodes datacodes = ehc.nodelist[node]['datacodes'] # fetch a string of data sizes based on the string of datacodes datasizes = [] for code in datacodes: datasizes.append(ehc.check_datacode(code)) # Discard the frame & return 'False' if it doesn't match the summed datasizes if len(data) != sum(datasizes): self._log.warning(str(ref) + " RX data length: " + str(len(data)) + " is not valid for datacodes " + str(datacodes)) return False else: # Determine the expected number of values to be decoded count = len(datacodes) # Set decoder to "Per value" decoding using datacode 'False' as flag datacode = False else: # if node is listed, but has only a single default datacode for all values if node in ehc.nodelist and 'datacode' in ehc.nodelist[node]: datacode = ehc.nodelist[node]['datacode'] else: # when node not listed or has no datacode(s) use the interfacers default if specified datacode = self._settings['datacode'] # Ensure only int 0 is passed not str 0 if datacode == '0': datacode = 0 # when no (default)datacode(s) specified, pass string values back as numerical values if not datacode: for val in data: if float(val) % 1 != 0: val = float(val) else: val = int(float(val)) decoded.append(val) # Discard frame if total size is not an exact multiple of the specified datacode size. elif len(data) % ehc.check_datacode(datacode) != 0: self._log.warning(str(ref) + " RX data length: " + str(len(data)) + " is not valid for datacode " + str(datacode)) return False else: # Determine the number of values in the frame of the specified code & size count = len(data) / ehc.check_datacode(datacode) # Decode the string of data one value at a time into "decoded" if not decoded: bytepos = int(0) for i in range(0, count, 1): # Use single datacode unless datacode = False then use datacodes dc = datacode if not datacode: dc = datacodes[i] # Determine the number of bytes to use for each value by it's datacode size = int(ehc.check_datacode(dc)) try: value = ehc.decode(dc, [int(v) for v in data[bytepos:bytepos+size]]) except: self._log.warning(str(ref) + " Unable to decode as values incorrect for datacode(s)") return False bytepos += size decoded.append(value) # Insert node ID before data decoded.insert(0, int(node)) return decoded def set(self, **kwargs): """Set configuration parameters. **kwargs (dict): settings to be sent. Example: {'setting_1': 'value_1', 'setting_2': 'value_2'} pause (string): pause status 'pause' = all pause Interfacer fully, nothing read, processed or posted. 'pause' = in pauses the input only, no input read performed 'pause' = out pauses output only, input is read, processed but not posted to buffer 'pause' = off pause is off and Interfacer is fully operational (default) """ for key, setting in self._defaults.iteritems(): if key in kwargs.keys(): setting = kwargs[key] else: setting = self._defaults[key] if key in self._settings and self._settings[key] == setting: continue elif key == 'pause' and str(setting).lower() in ['all', 'in', 'out', 'off']: pass elif key == 'interval' and str(setting).isdigit(): pass elif key == 'datacode' and str(setting) in ['0', 'b', 'B', 'h', 'H', 'L', 'l', 'f']: pass elif key == 'timestamped' and str(setting).lower() in ['true', 'false']: pass else: self._log.warning("'%s' is not a valid setting for %s: %s" % (str(setting), self.name, key)) continue self._settings[key] = setting self._log.debug("Setting " + self.name + " " + key + ": " + str(setting)) def run(self): """Placeholder for background tasks. Allows subclasses to specify actions that need to be done on a regular basis. This should be called in main loop by instantiater. """ pass def _open_serial_port(self, com_port, com_baud): """Open serial port com_port (string): path to COM port """ #if not int(com_baud) in [75, 110, 300, 1200, 2400, 4800, 9600, 19200, 38400, 57600, 115200]: # self._log.debug("Invalid 'com_baud': " + str(com_baud) + " | Default of 9600 used") # com_baud = 9600 try: s = serial.Serial(com_port, com_baud, timeout=0) self._log.debug("Opening serial port: " + str(com_port) + " @ "+ str(com_baud) + " bits/s") except serial.SerialException as e: self._log.error(e) raise EmonHubInterfacerInitError('Could not open COM port %s' % com_port) else: return s def _open_socket(self, port_nb): """Open a socket port_nb (string): port number on which to open the socket """ self._log.debug('Opening socket on port %s', port_nb) try: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.bind(('', int(port_nb))) s.listen(1) except socket.error as e: self._log.error(e) raise EmonHubInterfacerInitError('Could not open port %s' % port_nb) else: return s """class EmonhubSerialInterfacer Monitors the serial port for data """ class EmonHubSerialInterfacer(EmonHubInterfacer): def __init__(self, name, com_port='', com_baud=9600): """Initialize interfacer com_port (string): path to COM port """ # Initialization super(EmonHubSerialInterfacer, self).__init__(name) # Open serial port self._ser = self._open_serial_port(com_port, com_baud) # Initialize RX buffer self._rx_buf = '' def close(self): """Close serial port""" # Close serial port if self._ser is not None: self._log.debug("Closing serial port") self._ser.close() def read(self): """Read data from serial port and process if complete line received. Return data as a list: [NodeID, val1, val2] """ # Read serial RX self._rx_buf = self._rx_buf + self._ser.readline() # If line incomplete, exit if '\r\n' not in self._rx_buf: return # Remove CR,LF f = self._rx_buf[:-2] # Reset buffer self._rx_buf = '' # Discard empty frames if not f: self._log.warning("Discarded empty frame") return # unix timestamp t = round(time.time(), 2) # Process data frame return self._process_frame(f, t) """class EmonHubJeeInterfacer Monitors the serial port for data from "Jee" type device """ class EmonHubJeeInterfacer(EmonHubSerialInterfacer): def __init__(self, name, com_port='/dev/ttyAMA0', com_baud=0): """Initialize Interfacer com_port (string): path to COM port """ # Initialization if com_baud != 0: super(EmonHubJeeInterfacer, self).__init__(name, com_port, com_baud) else: for com_baud in (38400, 9600): super(EmonHubJeeInterfacer, self).__init__(name, com_port, com_baud) self._ser.write("?") time.sleep(2) self._rx_buf = self._rx_buf + self._ser.readline() if '\r\n' in self._rx_buf or '\x00' in self._rx_buf: self._ser.flushInput() self._rx_buf="" break elif self._ser is not None: self._ser.close() continue # Display device firmware version and current settings self.info = ["",""] if self._ser is not None: self._ser.write("v") time.sleep(2) self._rx_buf = self._rx_buf + self._ser.readline() if '\r\n' in self._rx_buf: self._rx_buf="" info = self._rx_buf + self._ser.readline()[:-2] if info != "": # Split the returned "info" string into firmware version & current settings self.info[0] = info.strip().split(' ')[0] self.info[1] = info.replace(str(self.info[0]), "") self._log.info( self.name + " device firmware version: " + self.info[0]) self._log.info( self.name + " device current settings: " + str(self.info[1])) else: # since "v" command only v11> recommend firmware update ? #self._log.info( self.name + " device firmware is pre-version RFM12demo.11") self._log.info( self.name + " device firmware version & configuration: not available") else: self._log.warning("Device communication error - check settings") self._rx_buf="" self._ser.flushInput() # Initialize settings self._defaults.update({'pause': 'off', 'interval': 0, 'datacode': 'h'}) # This line will stop the default values printing to logfile at start-up # unless they have been overwritten by emonhub.conf entries # comment out if diagnosing a startup value issue self._settings.update(self._defaults) # Jee specific settings to be picked up as changes not defaults to initialise "Jee" device self._jee_settings = ({'baseid': '15', 'frequency': '433', 'group': '210', 'quiet': 'True'}) self._jee_prefix = ({'baseid': 'i', 'frequency': '', 'group': 'g', 'quiet': 'q'}) # Pre-load Jee settings only if info string available for checks if all(i in self.info[1] for i in (" i", " g", " @ ", " MHz")): self._settings.update(self._jee_settings) def read(self): """Read data from serial port and process if complete line received. Return data as a list: [NodeID, val1, val2] """ # Read serial RX self._rx_buf = self._rx_buf + self._ser.readline() # If line incomplete, exit if '\r\n' not in self._rx_buf: return # Remove CR,LF f = self._rx_buf[:-2] # Reset buffer self._rx_buf = '' # Discard empty frames if not f: self._log.warning("Discarded empty frame") return # Discard information messages if (f[0] == '>'): self._log.debug(self.name + " acknowledged command: " + str(f)) return if (len(f)>2 and f[0:3] == ' ->'): self._log.debug(self.name + " confirmed sent packet size: " + str(f)) return if f[0] == '\x01': #self._log.debug("Ignoring frame consisting of SOH character" + str(f)) return if " i" and " g" and " @ " and " MHz" in f: self.info[1] = f self._log.debug( self.name + " device settings updated: " + str(self.info[1])) return # unix timestamp t = round(time.time(), 2) # Process data frame return self._process_frame(f, t) def _validate_frame(self, ref, received): """Validate a frame of data This function performs logical tests to filter unsuitable data. Each test discards frame with a log entry if False Returns True if data frame passes tests. """ if received[0] == '?'and str(received[-1])[0]=='(' and str(received[-1])[-1]==')': self._log.info(str(ref) + " Discard RX frame 'unreliable content' : RSSI " + str(received[-1])) return False # Strip 'OK' from frame if needed if received[0]=='OK': received = received[1:] # extract RSSI if packet is from RFM69 type Jee Device if str(received[-1])[0]=='(' and str(received[-1])[-1]==')': self.rssi = int(received[-1][1:-1]) received = received[:-1] return received else: # set RSSI false for standard frames so RSSI is not re-appended later self.rssi = False # include checks from parent if not super(EmonHubJeeInterfacer, self)._validate_frame(ref, received): return False return received def set(self, **kwargs): """Send configuration parameters to the "Jee" type device through COM port **kwargs (dict): settings to be modified. Available settings are 'baseid', 'frequency', 'group'. Example: {'baseid': '15', 'frequency': '4', 'group': '210'} """ for key, setting in self._jee_settings.iteritems(): # Decide which setting value to use if key in kwargs.keys(): setting = kwargs[key] else: setting = self._jee_settings[key] # convert bools to ints if str.capitalize(str(setting)) in ['True', 'False']: setting = int(setting == "True") # confirmation string always contains baseid, group anf freq if " i" and " g" and " @ " and " MHz" in self.info[1]: # If setting confirmed as already set, continue without changing if (self._jee_prefix[key] + str(setting)) in self.info[1].split(): continue elif key in self._settings and self._settings[key] == setting: continue if key == 'baseid' and int(setting) >=1 and int(setting) <=26: command = setting + 'i' elif key == 'frequency' and setting in ['433','868','915']: command = setting[:1] + 'b' elif key == 'group'and int(setting) >=0 and int(setting) <=212: command = setting + 'g' elif key == 'quiet' and int(setting) >=0 and int(setting) <2: command = str(setting) + 'q' else: self._log.warning("'%s' is not a valid setting for %s: %s" % (str(setting), self.name, key)) continue self._settings[key] = setting self._log.info("Setting " + self.name + " %s: %s" % (key, setting) + " (" + command + ")") self._ser.write(command) # Wait a sec between two settings time.sleep(1) # include kwargs from parent super(EmonHubJeeInterfacer, self).set(**kwargs) def run(self): """Actions that need to be done on a regular basis. This should be called in main loop by instantiater. """ now = time.time() # Broadcast time to synchronize emonGLCD interval = int(self._settings['interval']) if interval: # A value of 0 means don't do anything if now - self._interval_timestamp > interval: self._send_time() self._interval_timestamp = now def _send_time(self): """Send time over radio link to synchronize emonGLCD The radio module can be used to broadcast time, which is useful to synchronize emonGLCD in particular. Beware, this is know to garble the serial link on RFM2Piv1 sendtimeinterval defines the interval in seconds between two time broadcasts. 0 means never. """ now = datetime.datetime.now() self._log.debug(self.name + " broadcasting time: %02d:%02d" % (now.hour, now.minute)) self._ser.write("00,%02d,%02d,00,s" % (now.hour, now.minute)) """class EmonHubSocketInterfacer Monitors a socket for data, typically from ethernet link """ class EmonHubSocketInterfacer(EmonHubInterfacer): def __init__(self, name, port_nb=50011): """Initialize Interfacer port_nb (string): port number on which to open the socket """ # Initialization super(EmonHubSocketInterfacer, self).__init__(name) # Open socket self._socket = self._open_socket(port_nb) # Initialize RX buffer for socket self._sock_rx_buf = '' def close(self): """Close socket.""" # Close socket if self._socket is not None: self._log.debug('Closing socket') self._socket.close() def read(self): """Read data from socket and process if complete line received. Return data as a list: [NodeID, val1, val2] """ # Check if data received ready_to_read, ready_to_write, in_error = \ select.select([self._socket], [], [], 0) # If data received, add it to socket RX buffer if self._socket in ready_to_read: # Accept connection conn, addr = self._socket.accept() # Read data self._sock_rx_buf = self._sock_rx_buf + conn.recv(1024) # Close connection conn.close() # If there is at least one complete frame in the buffer if '\r\n' in self._sock_rx_buf: # Process and return first frame in buffer: f, self._sock_rx_buf = self._sock_rx_buf.split('\r\n', 1) if str(self._settings['timestamped']).lower() == "true": f = f.split(" ") t = float(f[0]) f = ' '.join(map(str, f[1:])) return self._process_frame(f, t) else: return self._process_frame(f) """class EmonHubInterfacerInitError Raise this when init fails. """ class EmonHubInterfacerInitError(Exception): pass