<commit_before>#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import logging
import datetime

from kafka_connector.avro_loop_producer import AvroLoopProducer

__author__ = u'<NAME>'
__copyright__ = u'2017, <NAME>'
__license__ = u'MIT'

__dirname__ = os.path.dirname(os.path.abspath(__file__))

KAFKA_HOSTS = os.getenv("KAFKA_HOSTS", "kafka:9092")
SCHEMA_REGISTRY_URL = os.getenv("SCHEMA_REGISTRY_URL", "http://schema-registry:8082")
MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto:1883")
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "mqtt/topic")
PRODUCER_TOPIC = os.getenv("PRODUCER_TOPIC", "prod.stcs.???")

LOGGING_LEVEL = os.getenv("LOGGING_LEVEL", "INFO")
logging_format = "%(levelname)8s %(asctime)s %(name)s [%(filename)s:%(lineno)s - %(funcName)s() ] %(message)s"

# TODO schema for every topic
key_schema = __dirname__ + "/config/key.avsc"
value_schema = __dirname__ + "/config/value.avsc"

logging.basicConfig(level=logging.getLevelName(LOGGING_LEVEL), format=logging_format)
logger = logging.getLogger('producer')


producer = AvroLoopProducer(KAFKA_HOSTS, SCHEMA_REGISTRY_URL, PRODUCER_TOPIC, key_schema, value_schema)

# TODO listening to mosquitto and producing to Kafka
while True:

    key = {'timestamp': int(time.time() * 1000)}
    value = {'value': 1}

    producer.produce(key, value)<commit_msg>comitting wip for mqtt kafka bridge<commit_after>#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import logging
import datetime
from glob import glob

import paho.mqtt.client as mqtt

from kafka_connector.avro_loop_producer import AvroLoopProducer

__author__ = u'<NAME>, <NAME>'
__copyright__ = u'2017, <NAME>, <NAME>'
__license__ = u'MIT'

__dirname__ = os.path.dirname(os.path.abspath(__file__))

KAFKA_HOSTS = os.getenv("KAFKA_HOSTS", "kafka:9092")
SCHEMA_REGISTRY_URL = os.getenv("SCHEMA_REGISTRY_URL", "http://schema-registry:8082")
MQTT_HOST = os.getenv("MQTT_HOST", "mosquitto:1883")

# MQTT_TOPIC = os.getenv("MQTT_TOPIC", "mqtt/topic")


# Collecto the name of all folders within the "config" directory, which
# corresponds to the names of all topics that we want to bridge Collect all the topics we want to bridge, and initialize according producers

mqtt_topics = glob("*")

producers = {}

for topic in mqtt_topics:

    PRODUCER_TOPIC = os.getenv("PRODUCER_TOPIC", "prod.stcs." + (topic))

    LOGGING_LEVEL = os.getenv("LOGGING_LEVEL", "INFO")
    logging_format = "%(levelname)8s %(asctime)s %(name)s [%(filename)s:%(lineno)s - %(funcName)s() ] %(message)s"

    # TODO schema for every topic
    key_schema = os.path.join([__dirname__, "config", topic, "key.avsc"])
    value_schema = os.path.join([__dirname__, "config", topic, "value.avsc"])

    logging.basicConfig(level=logging.getLevelName(LOGGING_LEVEL), format=logging_format)
    logger = logging.getLogger('mqtt_kafka_bridge_topic_' + topic)

    producer = AvroLoopProducer(KAFKA_HOSTS, SCHEMA_REGISTRY_URL, PRODUCER_TOPIC, key_schema, value_schema)

    producers[topic] = producer


def on_connect(client, userdata, flags, rc):

    print("Connected with result code "+str(rc))
    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("prod\stcs\+")


def on_message(client, userdata, msg):

    # her comes the publishing

    print(msg.topic+" "+str(msg.payload))



mqtt_client = mqtt.Client()

client.on_connect = on_connect
client.on_message = on_message

client.loop_forever()
{'fraction_non_alphanumeric': 0.13559322033898305, 'fraction_numerical': 0.01235277219189888, 'mean_word_length': 5.545112781954887, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '396238', 'n_tokens_mistral': 1330, 'n_tokens_neox': 1262, 'n_words': 266}