python-kafka-sample.py

from kafka import KafkaProducer

from functools import partial
from vivint_generated.constants import Fields

utf8_encode = partial(str.encode, encoding="UTF-8")

config = {
    'bootstrap_servers': 'localhost',
    'key_serializer': utf8_encode,
    'value_serializer': utf8_encode,
    'acks': 0,
}
k = KafkaProducer(**config)

sample_ap_msg = {
    "_id": "234149061247496|1",
    "da": {"stuff": "here"},
    "t": "account_partition",
    "op": "u",
    "panid": 234149061247496,
    "parid": 1
}

sample_as_msg = {
    "da": {"stuff": "here"},
    "_id": 234149060930506,
    "t": "account_system",
    "op": "u"
}

def get_id(msg):
    msg.get(Fields.PanelId, msg.get(Fields.Id, "No panel id found... strange"))

def send_msg(msg):
    k.send(msg[Fields.Type], get_id(msg), msg)

# for i in [sample_ap_msg, sample_as_msg]:
#     send_msg(i)