*What happens when equipment starts behaving differently?
Let's build a simple event-driven pipeline to find out.*
Why I Wanted to Explore This
Imagine you're monitoring a transformer at an energy facility.
Everything looks normal until the temperature starts climbing. A few minutes later, the load increases, and the vibration readings begin changing too.
The data is coming in, but someone still needs to notice the pattern, understand what it means, and decide whether it needs attention.
That's the problem I wanted to explore.
During my recent technical presentation at the IEEE Houston Section on AI-driven digital transformation in energy, I discussed how organizations can move beyond simply collecting operational data and start using it to support better decisions.
One of the ideas I kept coming back to was this: What if we could build a small system that reacts to equipment data as it arrives, instead of waiting for someone to review a dashboard?
For this example, we'll use Python and Apache Kafka to build a simple monitoring pipeline.
Nothing here connects to actual energy equipment. We'll generate synthetic transformer readings, stream them through Kafka, and create alerts when the measurements cross demonstration thresholds.
It's a small project, but it gives us a useful foundation for exploring event-driven architecture and, eventually, AI-powered monitoring.
What We're Building
The idea is straightforward.
A Python script generates transformer measurements every few seconds. Kafka receives those measurements and makes them available to another service. A second Python script reads the events and checks whether anything needs attention.
When a measurement crosses one of our sample thresholds, the detector publishes an alert.
Here's the flow:
Simulated Transformer
|
v
Python Telemetry Generator
|
v
Kafka: transformer-telemetry
|
v
Python Anomaly Detector
|
v
Kafka: transformer-alerts
|
v
Monitoring / Dashboard
(Next Step)
I like this approach because each component has one responsibility.
The generator produces data. Kafka transports it. The detector evaluates it. A future dashboard or notification service could consume the alerts.
We can change one part without having to rebuild everything else.
That's one of the reasons event-driven architecture works well for systems where multiple applications need to react to the same information.
Step 1: Get Kafka Running Locally
Before writing the Python code, we need a Kafka broker.
I'll use Docker Compose because it keeps the setup relatively simple.
You'll need Python 3.10 or later, Docker, and Docker Compose.
First, install the Python Kafka client:
pip install kafka-python
Next, create a file named docker-compose.yml:
services:
kafka:
image: apache/kafka:3.9.0
container_name: energy-kafka
ports:
- "9092:9092"
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_NUM_PARTITIONS: 1
This is intended for a local learning environment. It isn't a secure or production-ready Kafka configuration.
Start the broker:
bash
docker compose up -d
Once Kafka is running, we can start sending events.
Step 2: Simulate Transformer Readings
Since we don't have access to actual equipment telemetry, we'll generate our own.
Each event will contain a transformer ID, a timestamp, and three measurements: temperature, load, and vibration.
Create producer.py:
python
import json
import random
import time
import uuid
from datetime import datetime, timezone
from kafka import KafkaProducer
producer = KafkaProducer(
bootstrap_servers="localhost:9092",
value_serializer=lambda value: json.dumps(value).encode("utf-8")
)
asset_id = "TR-104"
while True:
event = {
"eventId": str(uuid.uuid4()),
"assetId": asset_id,
"timestamp": datetime.now(timezone.utc).isoformat(),
"temperatureC": round(random.uniform(65, 105), 2),
"loadPercent": round(random.uniform(65, 95), 2),
"vibrationMmS": round(random.uniform(1.5, 7.5), 2),
"source": "synthetic-telemetry"
}
producer.send(
"transformer-telemetry",
key=asset_id.encode("utf-8"),
value=event
)
producer.flush()
print("Published:", event)
time.sleep(2)
Every two seconds, this script creates a new event and publishes it to the transformer-telemetry topic.
A typical message might look like this:
json
{
"eventId": "evt-example-001",
"assetId": "TR-104",
"timestamp": "2026-10-29T18:04:00Z",
"temperatureC": 96.0,
"loadPercent": 91.0,
"vibrationMmS": 6.8,
"source": "synthetic-telemetry"
}
The values are randomly generated, so they're not meant to represent realistic transformer behavior.
At this point, we're mainly interested in getting events from one application to another.
Step 3: Detect Readings That Need Attention
Now let's create the service that consumes the events.
We'll start with a few simple rules:
- Temperature above 90°C
- Load above 85%
- Vibration above 5 mm/s
These numbers are arbitrary demonstration thresholds, not recommended operating limits for actual transformers.
Create detector.py:
import json
import uuid
from datetime import datetime, timezone
from kafka import KafkaConsumer, KafkaProducer
consumer = KafkaConsumer(
"transformer-telemetry",
bootstrap_servers="localhost:9092",
auto_offset_reset="earliest",
group_id="energy-anomaly-detector",
value_deserializer=lambda value: json.loads(value.decode("utf-8"))
)
producer = KafkaProducer(
bootstrap_servers="localhost:9092",
value_serializer=lambda value: json.dumps(value).encode("utf-8")
)
for message in consumer:
event = message.value
reasons = []
if event["temperatureC"] > 90:
reasons.append("Temperature exceeded demo threshold")
if event["loadPercent"] > 85:
reasons.append("Load exceeded demo threshold")
if event["vibrationMmS"] > 5:
reasons.append("Vibration exceeded demo threshold")
if reasons:
alert = {
"alertId": str(uuid.uuid4()),
"sourceEventId": event["eventId"],
"assetId": event["assetId"],
"detectedAt": datetime.now(timezone.utc).isoformat(),
"severity": "REVIEW_REQUIRED",
"reasons": reasons,
"status": "PENDING_HUMAN_REVIEW"
}
producer.send(
"transformer-alerts",
key=event["assetId"].encode("utf-8"),
value=alert
)
producer.flush()
print("ALERT:", alert)
There isn't any machine learning here yet.
We're simply checking each incoming event against a few conditions.
That's intentional.
When building a new data pipeline, I prefer starting with something easy to understand and troubleshoot. Once the event flow is working reliably, we can make the detection logic more sophisticated.
Step 4: See the Events Flow
Open two terminals.
Start the detector in the first terminal:
python detector.py
Then start the producer in the second:
python producer.py
You should see telemetry messages being published.
When a reading crosses one of the sample thresholds, the detector should publish an alert.
To inspect the alert topic, open another terminal and run:
docker exec -it energy-kafka \
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic transformer-alerts \
--from-beginning
An alert might look like this:
{
"alertId": "example-alert-001",
"sourceEventId": "example-event-001",
"assetId": "TR-104",
"detectedAt": "2026-10-29T18:04:05+00:00",
"severity": "REVIEW_REQUIRED",
"reasons": [
"Temperature exceeded demo threshold",
"Vibration exceeded demo threshold"
],
"status": "PENDING_HUMAN_REVIEW"
}
This is an illustrative output, not a captured execution result.
We now have the basic structure of an event-driven monitoring pipeline: one service publishes telemetry, another evaluates it, and alerts are available for other applications to consume.
Step 5: Where AI Could Make This More Useful
So far, our detector only checks whether individual measurements exceed fixed thresholds.
That's useful for demonstrating the architecture, but real equipment behavior is more complicated.
A temperature reading that looks unusual under one operating condition might be expected under another.
This is where machine learning could add value.
Instead of evaluating each measurement independently, a model could examine relationships between temperature, load, vibration, and historical operating patterns.
For example, we could experiment with an anomaly detection algorithm such as Isolation Forest.
The model could flag combinations of measurements that differ from the patterns observed during training.
But we'd still need to validate whether those anomalies correspond to meaningful operational conditions.
A model detecting an unusual pattern is not the same as a model diagnosing equipment failure.
That distinction matters.
Step 6: Could Generative AI Explain the Alert?
This is another interesting extension.
Imagine an engineer receives an alert containing several measurements and technical details.
Rather than making the engineer interpret raw JSON, we could use Generative AI to prepare a readable summary.
For example:
Transformer TR-104 has reported measurements above the demonstration monitoring thresholds. Temperature and vibration readings should be reviewed alongside recent operating conditions. Verify the sensor data before determining whether additional investigation is required.
That kind of explanation could make alerts easier to understand.
A more advanced version could use Retrieval-Augmented Generation (RAG) to retrieve relevant maintenance procedures or approved technical documents.
However, I wouldn't let a language model invent a diagnosis or independently issue operational commands.
Its role would be to help organize and explain the available evidence, with an engineer remaining responsible for the decision.
Step 7: What This Prototype Doesn't Solve
It's important to be clear about the limitations.
This is a small learning project. It doesn't include actual SCADA integration, trained predictive models, validated equipment thresholds, or a production dashboard.
There are also several engineering problems we haven't addressed.
What happens when Kafka becomes unavailable? What if a message arrives twice? How should the detector handle missing fields? What if a sensor sends outdated readings?
In a real implementation, we'd need to address those questions.
We'd also need authentication, encryption, observability, schema management, alert deduplication, model monitoring, and appropriate security boundaries between operational and enterprise systems.
For critical energy infrastructure, these aren't optional improvements. They're part of making the system safe and dependable.
What I'd Build Next
If I were extending this project, I'd focus on four improvements.
First, I'd add a small dashboard to display incoming measurements and alerts.
Next, I'd introduce an anomaly detection model and compare its results against our simple threshold rules.
Then, I'd explore a Generative AI explanation service that retrieves approved documentation and produces evidence-based summaries.
Finally, I'd add a human-review workflow so an engineer could acknowledge an alert, record findings, and decide whether further action is needed.
That would bring the prototype closer to the bigger idea: connecting operational data to useful, accountable decisions.
Final Thoughts
One thing I appreciate about event-driven architecture is that we don't need to build the entire solution at once.
We can start with a simple producer, a message broker, and a consumer.
Once the data is moving reliably, we can add analytics, AI capabilities, dashboards, and workflows without turning everything into one tightly coupled application.
For energy monitoring, the opportunity isn't simply to generate more alerts or introduce another AI model.
It's to give people the information they need to recognize problems, understand the evidence, and make better decisions.
And sometimes, the best way to explore that opportunity is to start with a small, understandable prototype.
This article was inspired by my presentation, “AI-Driven Digital Transformation in Energy: From Operational Data to Intelligent Decision Systems,” delivered at the IEEE Houston Section on October 29, 2026.
All telemetry values and examples are synthetic. The code is intended for local experimentation and has not been verified here through an end-to-end execution.
About the Author
Supraja Chinthala is an IEEE Senior Member and Technical Product Management professional with over 13 years of experience in enterprise integration, event-driven architecture, digital transformation, and AI-enabled platforms. Her interests include building practical, responsible AI solutions that connect technology capabilities with real business needs.
Top comments (0)