Das deutsche Python-Forum

Code: Alles auswählen

import sys
import re
import serial

KEYWORDS = {
    'A+': '1-0:1.8.0', # Meter reading import
	'A-': '1-0:2.8.0', # Meter reading export
	'L1': '1-0:21.7.255', # Power L1
	'L2': '1-0:41.7.255', # Power L2
	'L3': '1-0:61.7.255', # Power L3
	'In': '1-0:1.7.255', # Power total in
	'Out': '1-0:1.7.255', # Power total out
}

if len(sys.argv) != 2:
    print "wrong arguments."
    print "Usage: %s [KEYWORD]" % sys.argv[0]
    sys.exit()

keyword = sys.argv[1]

with serial.Serial(port='/dev/ttyUSB0', baudrate=9600, bytesize=7, parity='E', timeout=1) as ser:
    while True:
        reading = ser.read(300).decode("utf-8")
        ser.flushInput()
        if reading.startswith('/'):
            break
        # Try again if reading was unsuccessful

try:
    pattern = KEYWORDS[keyword]
except KeyError:
    print"Aborting: %s is not a valid argument" % keyword
    sys.exit()

match = re.search(r"%s.*?\((.*?)(?:\*(.*?))?\)" % pattern, reading)
value, unit = match.groups()
value = float(value)

# Adapt acutal value
if keyword == 'In':
    value = max(value, 0)
elif keyword == 'Out':
    value = max(-value, 0)

print value, unit

Code: Alles auswählen

import re
import serial
import logging
import multiprocessing
import json
import datetime

from config import CONFIG

KEYWORDS = {
    '1-0:1.8.0': {'name': 'A+', 'type': float}, # Meter reading import
    #'A-': '1-0:2.8.0', # Meter reading export
    '1-0:21.7.255': {'name': 'L1', 'type': float}, # Power L1
    '1-0:41.7.255': {'name': 'L2', 'type': float}, # Power L1
    '1-0:61.7.255': {'name': 'L3', 'type': float}, # Power L1
    '1-0:1.7.255': {'name': 'In', 'type': float}, # Power L1
    '0-0:96.1.255': {'name': 'SERIAL', 'type': str}, # Serial number
    #'Out': '1-0:1.7.255', # Power total out
}
extract_all = re.compile(r"(\d-\d:\d+\.\d+\.\d+).*?\((.*?)(?:\*(.*?))?\)").findall

TS_FORMAT = '%Y-%m-%d %H:%M:%S'

SQLITE_CREATE = """CREATE TABLE IF NOT EXISTS meter_data
                   ( meter TEXT
                    ,l1 NUMERIC
                    ,l2 NUMERIC
                    ,l3 NUMERIC
                    ,load NUMERIC
                    ,kwh NUMERIC
                    ,TS TEXT DEFAULT CURRENT_TIMESTAMP);"""

def read():
    with serial.Serial(port=CONFIG['dev'], baudrate=9600, bytesize=7, parity='E', timeout=2.5, exclusive=True) as ser:
        reading = ser.read_until(b'!').decode("utf-8")
        if not reading.startswith('/'):
            raise ValueError(reading)
        return reading

def read_meter(task_queues):
    logger = multiprocessing.get_logger()
    while True:
        try:
            reading = read()
            logger.debug(f'reading: {reading}, len: {len(reading)}')

            if CONFIG['utc']:
                ts = datetime.datetime.utcnow()
            else:
                ts = datetime.datetime.now()
            data = {'ts': ts.strftime(TS_FORMAT)}
            for keyword, value, unit in extract_all(reading):
                if keyword in KEYWORDS:
                    info = KEYWORDS[keyword]
                    data[info['name']] = info['type'](value)
                logger.debug((keyword,value,unit))
            #put the reading_dict into all publishing queues
            for queue in task_queues:
                queue.put(data)
        except:
            logger.exception('Error in worker_read_meter')

def worker_publish_mqtt(task_queue):
    import paho.mqtt.client as mqtt
    logger = multiprocessing.get_logger()
    client = mqtt.Client()
    if CONFIG['mqtt']['auth']['enabled']:
          client.username_pw_set(CONFIG['mqtt']['auth']['username'],
                                 CONFIG['mqtt']['auth']['password'])

    client.connect(host=CONFIG['mqtt']['host'], 
                   port=CONFIG['mqtt']['port'],
                   keepalive=CONFIG['mqtt']['keepalive'],
                   bind_address="")
 
    while True:
        try:
            reading = task_queue.get()
            client.publish(topic=CONFIG['mqtt']['topic'], 
                           payload=json.dumps(reading),
                           qos=CONFIG['mqtt']['qos'],
                           retain=CONFIG['mqtt']['retain'])
            logger.debug('worker_publish_mqtt' + json.dumps(reading))
        except:
            logger.exception('Error in worker_publish_mqtt')

def worker_sqlite(task_queue):
    import sqlite3
    logger = multiprocessing.get_logger()

    previous_filename = None
    db_connection = None
    while True:
        try:
            reading = task_queue.get()
            logger.debug(reading)
            #build sqlite filename with timestamp
            ts = datetime.datetime.strptime(reading['ts'], TS_FORMAT)
                    reading['ts_datetime'] = ts
            db_filename = ts.strftime(CONFIG['sqlite']['fname'])
            logger.debug(db_filename)
            if db_filename != previous_filename:
                previous_filename = db_filename
                if db_connection is not None:
                    db_connection.close()
                db_connection = sqlite3.connect(db_filename)
                logger.debug(f'connected to {db_filename}')
                logger.debug('setting up new db...')
                cursor = db_connection.cursor()
                cursor.execute(SQLITE_CREATE)
                    
            #build insert stmnt
            sql = """INSERT INTO meter_data
                     (meter, l1, l2, l3, kwh, ts) 
                     VALUES (?,?,?,?,?,?);"""
            params = [reading['SERIAL'], reading['L1'],
               reading['L2'], reading['L3'], reading['A+'], 
               reading['ts']]

            cursor = db_connection.cursor()
            cursor.execute(sql, params)
            cursor.close()
            db_connection.commit()
        except:
            logger.exception('Error in worker_sqlite')

def worker_logfile(task_queue):
    raise NotImplementedError

def run():
    multiprocessing.log_to_stderr(CONFIG['loglevel'])
    multiprocessing.get_logger().setLevel(CONFIG['loglevel'])

    #target functions for publishing services
    targets ={'mqtt': worker_publish_mqtt,
              'logfile': worker_logfile,
              'sqlite': worker_sqlite} 

    #prepare workers (create queues, link target functions)
    task_queues = []
    for name, function in targets.items():
        if CONFIG[name]['enabled']:
            queue = multiprocessing.Queue()
            task_queues.append(queue)
            p = multiprocessing.Process(target=function,
                                        args=(queue,))
            p.daemon = True #main process kills children before it will be terminated
            p.start()

    read_meter(task_queues)

if __name__ == '__main__':
    run()


""" 
/ESY5Q3DA1004 V3.02
1-0:0.0.0*255(0273011003684)
1-0:1.8.0*255(00026107.7034231*kWh)
1-0:21.7.255*255(000200.13*W)
1-0:41.7.255*255(000122.31*W)
1-0:61.7.255*255(000014.01*W)
1-0:1.7.255*255(000336.45*W)
1-0:96.5.5*255(82)
0-0:96.1.255*255(1ESY1011003684) """

Code: Alles auswählen

                    create_new = False
                    if not os.path.exists(filename):
                        create_new = True

                # =>

                    create_new = not os.path.exists(filename)

Code: Alles auswählen

#!/usr/bin/env python3
import datetime
import json
import multiprocessing
import os
import re
import sqlite3
import time
from collections import defaultdict
from contextlib import closing

import paho.mqtt.client as mqtt
import serial
from config import CONFIG

KEYWORDS = {
    "A+": {"keyword": "1-0:1.8.0", "dtype": float},  # Meter reading import
    # "A-": "1-0:2.8.0",  # Meter reading export
    "L1": {"keyword": "1-0:21.7.255", "dtype": float},  # Power L1
    "L2": {"keyword": "1-0:41.7.255", "dtype": float},  # Power L2
    "L3": {"keyword": "1-0:61.7.255", "dtype": float},  # Power L3
    "In": {"keyword": "1-0:1.7.255", "dtype": float},  # Power total in
    "SERIAL": {"keyword": "0-0:96.1.255", "dtype": str},  # Serial number
    # "Out": "1-0:1.7.255",  # Power total out
}

TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"

DB_SCHEMA = """CREATE TABLE IF NOT EXISTS meter_data
                   (meter TEXT,
                    l1 NUMERIC,
                    l2 NUMERIC,
                    l3 NUMERIC,
                    load NUMERIC,
                    kwh NUMERIC,
                    TS TIMESTAMP DEFAULT CURRENT_TIMESTAMP);"""


def read():
    with serial.Serial(
        port=CONFIG["dev"],
        bytesize=serial.SEVENBITS,
        parity=serial.PARITY_EVEN,
        timeout=2.5,
        exclusive=True,
    ) as connnection:
        reading = connnection.read_until(b"!").decode("utf-8")
        connnection.reset_input_buffer()

        if reading.startswith("/"):
            return reading
        time.sleep(0.5)  # wait to reach the right cycle
        raise IOError(f"reading failed {reading!r}")


def extract(keyword, reading):
    match = re.search(
        re.escape(KEYWORDS[keyword]["keyword"]) + r".*?\((.*?)(?:\*(.*?))?\)",
        reading,
    )
    value, unit = match.groups()
    return KEYWORDS[keyword]["dtype"](value), unit


def worker_read_meter(task_queues):
    logger = multiprocessing.get_logger()
    while True:
        try:
            reading_source = read()
            logger.debug(
                "reading: %r, len: %s", reading_source, len(reading_source)
            )
            if CONFIG["utc"]:
                timestamp = datetime.datetime.utcnow()
            else:
                timestamp = datetime.datetime.now()
            #
            # TODO Don't put a string into the data here but the datetime
            # object.
            #
            reading = {"ts": timestamp.strftime(TIMESTAMP_FORMAT)}
            #
            # TODO Instead of calling extract for each key, extract all
            # key(word)/value+unit pairs and then picke the ones we are
            # interested in.
            #
            for key in KEYWORDS:
                value, unit = extract(key, reading_source)
                reading[key] = value
                logger.debug("%r", (key, value, unit))

            for queue in task_queues:
                queue.put(reading)
        except:
            logger.exception("Error in worker_read_meter")


def worker_publish_mqtt(task_queue):
    logger = multiprocessing.get_logger()
    client = mqtt.Client()
    while True:
        try:
            reading = task_queue.get()
            if CONFIG["mqtt"]["auth"]["enabled"]:
                client.username_pw_set(
                    CONFIG["mqtt"]["auth"]["username"],
                    CONFIG["mqtt"]["auth"]["password"],
                )
            client.connect(
                host=CONFIG["mqtt"]["host"],
                port=CONFIG["mqtt"]["port"],
                keepalive=CONFIG["mqtt"]["keepalive"],
                bind_address="",
            )
            payload = json.dumps(reading)
            client.publish(
                topic=CONFIG["mqtt"]["topic"],
                payload=payload,
                qos=CONFIG["mqtt"]["qos"],
                retain=CONFIG["mqtt"]["retain"],
            )
            logger.debug("worker_publish_mqtt %s", payload)
        except:
            logger.exception("Error in worker_publish_mqtt")
            time.sleep(0.1)


def worker_sqlite(task_queue):
    #
    # TODO This function does too much — split up.
    #
    logger = multiprocessing.get_logger()
    while True:
        try:
            #
            # TODO: Take care the queue doesnt get too large (in case of insert
            # issues here)
            #
            if task_queue.qsize() >= CONFIG["sqlite"]["min_rows_insert"]:
                # get readings and build sqlite filenames (maybe different
                # filenames because of timestamp)
                filename_to_readings = defaultdict(list)
                while not task_queue.empty():
                    reading = task_queue.get()
                    logger.debug(reading)
                    timestamp = datetime.datetime.strptime(
                        reading["ts"], TIMESTAMP_FORMAT
                    )
                    reading["ts_datetime"] = timestamp
                    filename = timestamp.strftime(CONFIG["sqlite"]["fname"])
                    logger.debug(filename)
                    filename_to_readings[filename].append(reading)

                for filename, readings in filename_to_readings.items():
                    is_new_database = not os.path.exists(filename)
                    with sqlite3.connect(filename) as connection:
                        with closing(connection.cursor()) as cursor:
                            logger.debug("connected to %s", filename)
                            try:
                                if is_new_database:
                                    logger.debug("setting up new db...")
                                    cursor.execute(DB_SCHEMA)

                                cursor.executemany(
                                    "INSERT INTO meter_data"
                                    " (meter, l1, l2, l3, kwh, ts)"
                                    " VALUES (?,?,?,?,?,?)",
                                    [
                                        (
                                            reading["SERIAL"],
                                            reading["L1"],
                                            reading["L2"],
                                            reading["L3"],
                                            reading["A+"],
                                            reading["ts_datetime"],
                                        )
                                        for reading in readings
                                    ],
                                )
                                connection.commit()
                                logger.debug(
                                    "insert into %s was successful. "
                                    "Inserted %s readings.",
                                    filename,
                                    len(readings),
                                )
                            except:
                                logger.exception(
                                    "insert into %s failed", filename
                                )
                                for reading in readings:
                                    logger.debug(
                                        "add %s to queue again", reading
                                    )
                                    task_queue.put(reading)
                    logger.debug("closed connection to %s", filename)
        except:
            logger.exception("Error in worker_sqlite")
        time.sleep(1)


def worker_logfile(task_queue):
    raise NotImplementedError


def run():
    multiprocessing.log_to_stderr(CONFIG["loglevel"])
    multiprocessing.get_logger().setLevel(CONFIG["loglevel"])

    # target functions for publishing services
    targets = {
        "mqtt": worker_publish_mqtt,
        "logfile": worker_logfile,
        "sqlite": worker_sqlite,
    }
    # prepare workers (create queues, link target functions)
    workers = []
    for key in targets:
        if CONFIG[key]["enabled"]:
            workers.append((multiprocessing.Queue(), targets[key]))
    # now add worker_read_meter and give it a ref to all queues as argument
    workers.append(([queue for queue, _ in workers], worker_read_meter))

    for queue, target in workers:
        process = multiprocessing.Process(
            target=target, args=[queue], daemon=True
        )
        process.start()

    worker_read_meter([queue for queue, _ in workers])


if __name__ == "__main__":
    run()


# /ESY5Q3DA1004 V3.02

# 1-0:0.0.0*255(0273011003684)
# 1-0:1.8.0*255(00026107.7034231*kWh)
# 1-0:21.7.255*255(000200.13*W)
# 1-0:41.7.255*255(000122.31*W)
# 1-0:61.7.255*255(000014.01*W)
# 1-0:1.7.255*255(000336.45*W)
# 1-0:96.5.5*255(82)
# 0-0:96.1.255*255(1ESY1011003684)