<jupyter_start><jupyter_text>Define the value schema<jupyter_code>value_schema_str = """
{
  "namespace": "continuous.analytics",
  "type": "record",
  "name": "TemperatureSensorEvent",
  "fields": [
    {
      "name": "sensor",
      "type": "string"
    },
    {
      "name": "temperature",
      "type": "float"
    },
    {
      "name": "ts",
      "type": "long"
    }
  ]
}
"""
from confluent_kafka import SerializingProducer
from confluent_kafka.serialization import *
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
import time

topic = "TemperatureSensorEvent"
schema_registry_url = "http://schema-registry:8081"

def delivery_report(err, msg):
    if err is not None:
        print("Failed to deliver message: {}".format(err))
    else:
        print("Produced record to topic {} partition [{}] @ offset {}"
              .format(msg.topic(), msg.partition(), msg.offset()))
schema_registry_conf = {'url': schema_registry_url}
schema_registry_client = SchemaRegistryClient(schema_registry_conf)
avro_serializer = AvroSerializer(value_schema_str, schema_registry_client)

producer_conf = {
        'bootstrap.servers': bootstrap_server,
        'key.serializer': StringSerializer('utf_8'),
        'value.serializer': avro_serializer
}

producer = SerializingProducer(producer_conf)<jupyter_output><empty_output><jupyter_text>run the following cell to demostrate that fire is not detected<jupyter_code>from random import gauss

while True:
    key = "S1"
    value = {"sensor": "S1","temperature": gauss(20, 1.0),"ts":int(time.time()*1000.0)}
    producer.produce(topic=topic, value=value, key=key, on_delivery=delivery_report)
    print(value)
    producer.poll(1)
    time.sleep(10)<jupyter_output>{'sensor': 'S1', 'temperature': 50.338292206306306, 'ts': 1605623936394}
{'sensor': 'S1', 'temperature': 50.70999515318572, 'ts': 1605623947413}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 0
{'sensor': 'S1', 'temperature': 51.37361878403081, 'ts': 1605623957422}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 1
{'sensor': 'S1', 'temperature': 48.77852041692449, 'ts': 1605623967432}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 2
{'sensor': 'S1', 'temperature': 50.641243977894305, 'ts': 1605623977442}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 3
{'sensor': 'S1', 'temperature': 50.502307487199815, 'ts': 1605623987447}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 4
{'sensor': 'S1', 'temperature': 53.205666723532524, 'ts': 1605623997458}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 5
{'sensor': 'S1', 'temperature': 50.86978686788763, 'ts[...]<jupyter_text>run the following cell to demostrate to detect fire<jupyter_code>while True:
    key = "S1"
    value = {"sensor": "S1","temperature": gauss(55, 1.0),"ts":int(time.time()*1000.0)}
    producer.produce(topic=topic, value=value, key=key, on_delivery=delivery_report)
    print(value)
    producer.poll(1)
    time.sleep(10)<jupyter_output>{'sensor': 'S1', 'temperature': 57.08832133787175, 'ts': 1605624545856}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 58
{'sensor': 'S1', 'temperature': 55.31187192662198, 'ts': 1605624555859}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 59
{'sensor': 'S1', 'temperature': 54.975452079050946, 'ts': 1605624565870}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 60
{'sensor': 'S1', 'temperature': 56.14230348444072, 'ts': 1605624575880}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 61
{'sensor': 'S1', 'temperature': 52.62000026448661, 'ts': 1605624585890}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 62
{'sensor': 'S1', 'temperature': 54.49939646612739, 'ts': 1605624595901}
Produced record to topic TemperatureSensorEvent partition [0] @ offset 63
{'sensor': 'S1', 'temperature': 54.318403346873446, 'ts': 1605624605908}
Produced record to topic TemperatureSensorEvent pa[...]
{'fraction_non_alphanumeric': 0.1379393355801637, 'fraction_numerical': 0.12181030332209918, 'mean_word_length': 5.767100977198697, 'pattern_counts': {'":': 16, '<': 11, '<?xml version=': 0, '>': 11, '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}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '334100', 'license': 'Apache-2.0', 'n_tokens_mistral': 1710, 'n_tokens_neox': 1367, 'n_words': 337, 'path': 'ksql_firealarm/datagen/temperature_sensor_simulator.ipynb', 'repo_name': 'pavmassimo/continuous-analytics-examples', 'size': '4154'}