commit c6ce77b6fcd4855c058c138a363f9ea6fd836666 Author: “Naeel” Date: Tue Jul 21 16:32:08 2026 +0400 Initial commit — IoT Kafka producer diff --git a/Procfile b/Procfile new file mode 100644 index 0000000..c79a8f9 --- /dev/null +++ b/Procfile @@ -0,0 +1 @@ +web: gunicorn -b 0.0.0.0:$PORT app:app diff --git a/app.py b/app.py new file mode 100644 index 0000000..c06f0a1 --- /dev/null +++ b/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) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ba31b98 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +Flask +gunicorn +kafka-python