<commit_before>import logging
import functools
import xml.etree.ElementTree
import io
import time
import struct
import pickle
import socket

REGISTRY = {}

logger = logging.getLogger(__name__)


def route(path):
    def wrapper(func):
        @functools.wraps(func)
        def wrapped(*args, **kwargs):
            return func(*args, **kwargs)
        assert path not in REGISTRY, 'Cannot register {} path'.format(path)
        REGISTRY[path] = func
        return wrapped
    return wrapper
'''
<InstantaneousDemand>
    <DeviceMacId>0x00158d0000000004</DeviceMacId>
    <MeterMacId>0x00178d0000000004</MeterMacId>
    <TimeStamp>0x185adc1d</TimeStamp>
    <Demand>0x001738</Demand>
    <Multiplier>0x00000001</Multiplier>
    <Divisor>0x000003e8</Divisor>
    <DigitsRight>0x03</DigitsRight>
    <DigitsLeft>0x00</DigitsLeft>
    <SuppressLeadingZero>Y</SuppressLeadingZero>
</InstantaneousDemand>
'''

t_s = time.time()
queue = []
conn = socket.create_connection(('nyx.lan', 2004))


def post_metric(path, value, timestamp=None):
    '''
    [(path, (timestamp, value)), ...
    '''
    global t_s, conn
    queue.append((path, (timestamp or time.time(), value,)))

    if queue and (time.time() - t_s > 10 or len(queue) > 10):
        payload = pickle.dumps(queue, protocol=2)
        queue[:] = []

        header = struct.pack("!L", len(payload))
        message = header + payload
        try:
            conn.send(message)
        except Exception:
            conn = socket.create_connection(('nyx.lan', 2004))
        t_s = time.time()


@route('/ingest')
def consume(request):
    body = io.BytesIO(request.body)
    body.seek(0)
    try:
        e = xml.etree.ElementTree.parse(body).getroot()
    except Exception:
        logger.exception('Fault in decoding {!s}'.format(request.body))
    else:
        logger.info('Got body: {}'.format(request.body))
        for item in e:
            if item.tag == 'InstantaneousDemand':
                item = {
                    x.tag: x.text for x in item
                }
                # time_stamp = int(item['TimeStamp'], 16)
                instant_demand = \
                    int(item['Demand'], 16) * \
                    int(item['Multiplier'], 16) / int(item['Divisor'], 16)
                post_metric('power.instant_demand', instant_demand)
    return request.Response(text='')<commit_msg>Log the data in Redis so I can go to bed and examine it tomorrow<commit_after>import logging
import functools
import xml.etree.ElementTree
import io
import time
import struct
import pickle
import redis
import socket

REGISTRY = {}

db = redis.Redis()

logger = logging.getLogger(__name__)


def route(path):
    def wrapper(func):
        @functools.wraps(func)
        def wrapped(*args, **kwargs):
            return func(*args, **kwargs)
        assert path not in REGISTRY, 'Cannot register {} path'.format(path)
        REGISTRY[path] = func
        return wrapped
    return wrapper
'''
<InstantaneousDemand>
    <DeviceMacId>0x00158d0000000004</DeviceMacId>
    <MeterMacId>0x00178d0000000004</MeterMacId>
    <TimeStamp>0x185adc1d</TimeStamp>
    <Demand>0x001738</Demand>
    <Multiplier>0x00000001</Multiplier>
    <Divisor>0x000003e8</Divisor>
    <DigitsRight>0x03</DigitsRight>
    <DigitsLeft>0x00</DigitsLeft>
    <SuppressLeadingZero>Y</SuppressLeadingZero>
</InstantaneousDemand>
'''

t_s = time.time()
queue = []
conn = socket.create_connection(('nyx.lan', 2004))


def post_metric(path, value, timestamp=None):
    '''
    [(path, (timestamp, value)), ...
    '''
    global t_s, conn
    queue.append((path, (timestamp or time.time(), value,)))

    if queue and (time.time() - t_s > 10 or len(queue) > 10):
        payload = pickle.dumps(queue, protocol=2)
        queue[:] = []

        header = struct.pack("!L", len(payload))
        message = header + payload
        try:
            conn.send(message)
        except Exception:
            conn = socket.create_connection(('nyx.lan', 2004))
        t_s = time.time()


@route('/ingest')
def consume(request):
    body = io.BytesIO(request.body)
    body.seek(0)
    try:
        e = xml.etree.ElementTree.parse(body).getroot()
    except Exception:
        logger.exception('Fault in decoding {!s}'.format(request.body))
    else:
        for item in e:
            db.hmset('eagle_data|{}'.format(item.tag), {
                    x.tag: x.text for x in item
                })
            if item.tag == 'InstantaneousDemand':
                item = {
                    x.tag: x.text for x in item
                }
                # time_stamp = int(item['TimeStamp'], 16)
                instant_demand = \
                    int(item['Demand'], 16) * \
                    int(item['Multiplier'], 16) / int(item['Divisor'], 16)
                post_metric('power.instant_demand', instant_demand)
    return request.Response(text='')
{'fraction_non_alphanumeric': 0.13439868879328007, 'fraction_numerical': 0.0364679368981766, 'mean_word_length': 2.512230215827338, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 6, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3925196', 'n_tokens_mistral': 1713, 'n_tokens_neox': 1550, 'n_words': 325}