<commit_before>import asyncio
import zmq
import zmq.asyncio
import msgpack
import time
from decimal import Decimal
import pandas as pd

zmqhost = '127.0.0.1'
numport = 4201
wavport = 4202

influxhost = '127.0.0.1'
influxport = 8089

#<measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]

def run():
    loop = asyncio.get_event_loop()

    numq = asyncio.Queue(loop=loop)
    asyncio.ensure_future(recvFromZmq(loop, numq, zmqhost, numport))
    asyncio.ensure_future(sendToInfluxdb(loop, numq, influxhost, influxport, numEncode))

    wavq = asyncio.Queue(loop=loop)
    asyncio.ensure_future(recvFromZmq(loop, wavq, zmqhost, wavport))
    asyncio.ensure_future(sendToInfluxdb(loop, wavq, influxhost, influxport, wavEncode))

    loop.run_forever()

async def recvFromZmq(loop, queue, host, port):
    ctx = zmq.asyncio.Context()
    sock = ctx.socket(zmq.SUB, io_loop=loop)
    #sock.connect(f'tcp://{host}:{port}')
    sock.bind(f'tcp://{host}:{port}')
    sock.subscribe(b'')
    while loop.is_running():
        msg = await sock.recv()
        await queue.put(msg)

async def sendToInfluxdb(loop, queue, host, port, encoder):
    udpproto = lambda: asyncio.DatagramProtocol()
    transport, proto = await loop.create_datagram_endpoint(udpproto, remote_addr=(host, port))
    while loop.is_running():
        msg = await queue.get()
        print(f'Received {msg}')
        try:
            decoded = decodeMsg(msg)
            line = encoder(decoded)
        except ValueError:
            print(f'Failed to parse: {msg}')
            continue
        print(f'Sending: {line}')
        transport.sendto(line)
    transport.close()

def decodeMsg(msg, wave=False):
    # Parse incoming message
    topic, msgpackdata = msg.split(b' ', 1)
    frame = msgpack.unpackb(msgpackdata, encoding='utf-8')
    topic = topic.decode('ASCII')
    frame['topic'] = topic
    return frame

def numEncode(frame):
    # Encode numerics for InfluxDB
    metadata = frame['meta'].copy()
    metadata['origin'] = frame['topic']
    tags = [f"{tag}={value}" for tag, value in metadata.items()]
    str_tags = ','.join(tags)
    data = frame['data']
    fields = [f"{field}={value}" for field, value in data.items()]
    str_fields = ','.join(fields)
    time = frame['basetime']
    line = f'numerics,{str_tags} {str_fields} {time}'
    return line.encode('ASCII')

def wavEncode(frame):
    # Encode waves for InfluxDB
    metadata = frame['meta'].copy()
    metadata['origin'] = frame['topic']
    tags = [f"{tag}={value}" for tag, value in metadata.items()]<commit_msg>Use only one context for zmq, only one socket for influxdb and only one queue<commit_after>import asyncio
import zmq
import zmq.asyncio
import msgpack
import pandas as pd

zmqhost = '127.0.0.1'
numport = 4201
wavport = 4202

influxhost = '127.0.0.1'
influxport = 8089

#<measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]] <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]

def run():
    loop = asyncio.get_event_loop()
    queue = asyncio.Queue(loop=loop)

    ctx = zmq.asyncio.Context()
    asyncio.ensure_future(recvFromZmq(loop, ctx, queue, zmqhost, numport, numEncode))
    asyncio.ensure_future(recvFromZmq(loop, ctx, queue, zmqhost, wavport, wavEncode))

    asyncio.ensure_future(sendToInfluxdb(loop, queue, influxhost, influxport))
    loop.run_forever()

async def recvFromZmq(loop, ctx, queue, host, port, encoder):
    sock = ctx.socket(zmq.SUB, io_loop=loop)
    #sock.connect(f'tcp://{host}:{port}')
    sock.bind(f'tcp://{host}:{port}')
    sock.subscribe(b'')
    while loop.is_running():
        msg = await sock.recv()
        print(f'Received {msg}')
        try:
            decoded = decodeMsg(msg)
            line = encoder(decoded)
        except ValueError:
            print(f'Failed to parse: {msg}')
            continue
        await queue.put(line)

async def sendToInfluxdb(loop, queue, host, port):
    udpproto = lambda: asyncio.DatagramProtocol()
    transport, proto = await loop.create_datagram_endpoint(udpproto, remote_addr=(host, port))
    while loop.is_running():
        line = await queue.get()
        print(f'Sending: {line}')
        transport.sendto(line)
    transport.close()

def decodeMsg(msg, wave=False):
    # Parse incoming message
    topic, msgpackdata = msg.split(b' ', 1)
    frame = msgpack.unpackb(msgpackdata, encoding='utf-8')
    topic = topic.decode('ASCII')
    frame['topic'] = topic
    return frame

def numEncode(frame):
    # Encode numerics for InfluxDB
    metadata = frame['meta'].copy()
    metadata['origin'] = frame['topic']
    tags = [f"{tag}={value}" for tag, value in metadata.items()]
    str_tags = ','.join(tags)
    data = frame['data']
    fields = [f"{field}={value}" for field, value in data.items()]
    str_fields = ','.join(fields)
    time = frame['basetime']
    line = f'numerics,{str_tags} {str_fields} {time}'
    return line.encode('ASCII')

def wavEncode(frame):
    # Encode waves for InfluxDB
    metadata = frame['meta'].copy()
    metadata['origin'] = frame['topic']
    tags = [f"{tag}={value}" for tag, value in metadata.items()]
{'fraction_non_alphanumeric': 0.14882556796303428, 'fraction_numerical': 0.010011551790527531, 'mean_word_length': 3.7141560798548094, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, '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': '218037', 'n_tokens_mistral': 1837, 'n_tokens_neox': 1776, 'n_words': 388}