Initial commit — IoT Kafka producer

Procfile

@@ -0,0 +1 @@
+web: gunicorn -b 0.0.0.0:$PORT app:app

app.py

@@ -0,0 +1,117 @@
+"""
+IoT Producer — Flask app that generates fake IoT sensor events
+and sends them to Kafka.
+
+Env vars (set by Terraform):
+  KAFKA_BROKERS     — comma-separated Kafka bootstrap servers
+  KAFKA_TOPIC       — topic name
+  KAFKA_USERNAME    — SASL username
+  KAFKA_PASSWORD    — SASL password
+  PRODUCE_INTERVAL  — seconds between events (default 3)
+"""
+
+import json
+import os
+import random
+import time
+import threading
+import uuid
+from datetime import datetime, timezone
+
+from flask import Flask, jsonify
+from kafka import KafkaProducer
+from kafka.errors import NoBrokersAvailable
+
+app = Flask(__name__)
+
+# --- Config from env ---
+KAFKA_BROKERS = os.environ["KAFKA_BROKERS"].split(",")
+KAFKA_TOPIC = os.environ["KAFKA_TOPIC"]
+KAFKA_USERNAME = os.environ["KAFKA_USERNAME"]
+KAFKA_PASSWORD = os.environ["KAFKA_PASSWORD"]
+PRODUCE_INTERVAL = float(os.environ.get("PRODUCE_INTERVAL", 3))
+
+# --- IoT simulation config ---
+SENSORS = [
+    {"id": "temp_living",   "type": "temperature", "location": "living_room", "unit": "°C",   "min": 18.0, "max": 28.0},
+    {"id": "temp_bedroom",  "type": "temperature", "location": "bedroom",     "unit": "°C",   "min": 16.0, "max": 26.0},
+    {"id": "temp_office",   "type": "temperature", "location": "office",      "unit": "°C",   "min": 19.0, "max": 27.0},
+    {"id": "humidity_living","type": "humidity",    "location": "living_room", "unit": "%",    "min": 30.0, "max": 70.0},
+    {"id": "humidity_bed",  "type": "humidity",    "location": "bedroom",     "unit": "%",    "min": 35.0, "max": 65.0},
+    {"id": "power_kitchen", "type": "power_meter", "location": "kitchen",     "unit": "kW",   "min": 0.1,  "max": 3.5},
+    {"id": "power_living",  "type": "power_meter", "location": "living_room", "unit": "kW",   "min": 0.2,  "max": 2.0},
+    {"id": "power_office",  "type": "power_meter", "location": "office",      "unit": "kW",   "min": 0.3,  "max": 4.0},
+]
+
+producer = None
+events_sent = 0
+errors_count = 0
+
+
+def get_producer():
+    global producer
+    if producer is None:
+        producer = KafkaProducer(
+            bootstrap_servers=KAFKA_BROKERS,
+            security_protocol="SASL_PLAINTEXT",
+            sasl_mechanism="PLAIN",
+            sasl_plain_username=KAFKA_USERNAME,
+            sasl_plain_password=KAFKA_PASSWORD,
+            value_serializer=lambda v: json.dumps(v).encode("utf-8"),
+            key_serializer=lambda k: k.encode("utf-8") if k else None,
+        )
+    return producer
+
+
+def generate_event():
+    sensor = random.choice(SENSORS)
+    value = round(random.uniform(sensor["min"], sensor["max"]), 2)
+    return {
+        "event_id": str(uuid.uuid4()),
+        "sensor_id": sensor["id"],
+        "device_type": sensor["type"],
+        "location": sensor["location"],
+        "value": value,
+        "unit": sensor["unit"],
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    }
+
+
+def produce_loop():
+    global events_sent, errors_count
+    while True:
+        try:
+            event = generate_event()
+            p = get_producer()
+            future = p.send(KAFKA_TOPIC, key=event["sensor_id"], value=event)
+            future.get(timeout=5)
+            events_sent += 1
+            print(f"[PRODUCER] Sent: {event['sensor_id']} = {event['value']}{event['unit']} ({event['location']})")
+        except Exception as e:
+            errors_count += 1
+            print(f"[PRODUCER] Error: {e}")
+        time.sleep(PRODUCE_INTERVAL)
+
+
+@app.route("/")
+def index():
+    return jsonify({
+        "service": "iot-producer",
+        "status": "running",
+        "kafka_topic": KAFKA_TOPIC,
+        "events_sent": events_sent,
+        "errors": errors_count,
+        "interval": PRODUCE_INTERVAL,
+        "sensors": len(SENSORS),
+    })
+
+
+@app.route("/health")
+def health():
+    return jsonify({"status": "ok", "events_sent": events_sent})
+
+
+if __name__ == "__main__":
+    threading.Thread(target=produce_loop, daemon=True).start()
+    port = int(os.environ.get("PORT", 5000))
+    app.run(host="0.0.0.0", port=port)

requirements.txt

@@ -0,0 +1,3 @@
+Flask
+gunicorn
+kafka-python