Real-Time MQTT Data Pipeline for Digital Twin

IDS 6742 - Real-Time Simulation Modeling for Digital Twins

Dr. Bulent Soykan

---

Overview

This repository contains the implementation of a real-time IoT data pipeline that demonstrates the fundamental architecture of Digital Twin systems. The project simulates a physical temperature sensor using Python, transmits data over the internet using the MQTT (Message Queuing Telemetry Transport) publish/subscribe protocol, and ingests that data in real-time into an AnyLogic Digital Twin model.

Architecture

Physical Layer          →    Communication Layer    →    Digital Twin Layer
[Python Sensor]         →    [MQTT Broker]         →    [AnyLogic Model]
(Publisher)                  (HiveMQ Cloud)             (Subscriber)

The system demonstrates:

• Real-time data streaming from physical sensors to digital models
• Publish/Subscribe messaging pattern for decoupled communication
• Visual state representation that changes based on sensor data
• Time-series visualization of incoming telemetry data
• Automatic reconnection and error handling for production resilience

---

Repository Structure

PhysicalSensor/
│
├── sensor_edge.py           # Basic MQTT publisher (lab version)
├── sensor_edge_robust.py    # Production-ready MQTT publisher with error handling
├── lab.md                   # Lab instructions
├── README.md               # This file
└── .venv/                  # Python virtual environment (auto-created)

---

Prerequisites

System Requirements

• Operating System: Windows 10/11, macOS 10.15+, or Linux (Ubuntu 20.04+)
• Python: Version 3.8 or higher
• AnyLogic: Personal Learning Edition (PLE) or higher
• Internet Connection: Required for MQTT broker communication
• RAM: Minimum 4GB (8GB recommended for AnyLogic)
• Disk Space: 500MB for dependencies

Software Dependencies

1. Python Package Manager: pip or pip3
2. Java Runtime: JRE 8+ (typically bundled with AnyLogic)
3. Text Editor: VS Code, Sublime Text, or any code editor
4. Terminal Access: Command Prompt (Windows), Terminal (macOS/Linux)

---

Installation Guide

Step 1: Python Environment Setup

Option A: Using UV (Recommended for macOS with Homebrew)

# Install UV if not already installed
brew install uv

# Clone or navigate to the project directory
cd ~/Downloads/PhysicalSensor

# Create virtual environment
uv venv

# Activate virtual environment
source .venv/bin/activate  # On macOS/Linux
# OR
.venv\Scripts\activate     # On Windows

# Install MQTT library
uv pip install paho-mqtt

Option B: Using Standard Python venv

# Navigate to project directory
cd ~/Downloads/PhysicalSensor

# Create virtual environment
python3 -m venv .venv      # On macOS/Linux
# OR
python -m venv .venv        # On Windows

# Activate virtual environment
source .venv/bin/activate   # On macOS/Linux
# OR
.venv\Scripts\activate      # On Windows

# Install MQTT library
pip install paho-mqtt

Step 2: Download MQTT Java Client for AnyLogic

1. Navigate to Maven Central Repository:

   https://repo1.maven.org/maven2/org/eclipse/paho/org.eclipse.paho.client.mqttv3/1.2.5/
   

2. Download the JAR file:

   org.eclipse.paho.client.mqttv3-1.2.5.jar
   

3. Save the JAR file in your project directory (same folder as your AnyLogic model)

Step 3: Verify Installation

# Check Python version
python --version

# Verify MQTT library installation
python -c "import paho.mqtt; print('MQTT library installed successfully')"

# Test MQTT connectivity
python -c "import paho.mqtt.publish as publish; publish.single('test/topic', 'test', hostname='broker.hivemq.com'); print('MQTT broker accessible')"

---

Running the Physical Sensor

Using the Basic Sensor (sensor_edge.py)

This version implements the original lab specification:

# Activate virtual environment
source .venv/bin/activate  # macOS/Linux
# OR
.venv\Scripts\activate     # Windows

# Run the sensor
python sensor_edge.py

Note: This version may experience connection timeouts after extended operation due to public broker rate limiting.

Using the Robust Sensor (sensor_edge_robust.py) - RECOMMENDED

This production-ready version includes:

• Automatic reconnection logic
• Connection state management
• Error handling and recovery
• QoS 1 message delivery confirmation
• Clean shutdown handling

# Activate virtual environment
source .venv/bin/activate  # macOS/Linux
# OR
.venv\Scripts\activate     # Windows

# Run the robust sensor
python sensor_edge_robust.py

Expected Output

Starting Physical Sensor Simulation...
Client ID: sensor_2547
Publishing to: ucf/ids6742/bulent/machine/temp
Attempting to connect to broker.hivemq.com...
Connected successfully to broker.hivemq.com
Published Data: 102.05 °C
Published Data: 101.09 °C
Published Data: 103.15 °C
...

To stop the sensor: Press Ctrl+C

---

AnyLogic Digital Twin Setup

Step 1: Create New Model

1. Open AnyLogic
2. File → New → Model
3. Model Name: RealTime_Twin_MQTT
4. Model Time Units: seconds
5. Click Finish

Step 2: Add MQTT Dependency

1. In Projects view, click on RealTime_Twin_MQTT (top-level)
2. In Properties → Dependencies section
3. Click green plus (+) under "Jar files and class folders"
4. Browse and select org.eclipse.paho.client.mqttv3-1.2.5.jar
5. Click OK

Step 3: Create State Variable

1. Open Main diagram
2. From Agent palette, drag Variable to canvas
3. Configure:
   • Name: currentTemperature
   • Type: double
   • Initial value: 100.0

Step 4: Build Visual Dashboard

1. Machine Representation:

   • From Presentation palette, drag Rectangle
   • In Properties → Appearance → Fill color
   • Click static/dynamic switch (=) to make dynamic
   • Enter: currentTemperature >= 110.0 ? red : lime

2. Real-time Chart:

   • From Analysis palette, drag Time Plot
   • Configure:
     • Data update: Update automatically
     • Recurrence time: 0.5 seconds
     • Time window: 60 seconds
     • Add data item:
       • Title: Live Temp
       • Value: currentTemperature

Step 5: Implement MQTT Subscription

1. Click empty space in Main diagram

2. Properties → Advanced Java → Imports section:

   import org.eclipse.paho.client.mqttv3.*;

3. Properties → Advanced Java → Additional class code:

   MqttClient client;

4. Properties → Agent actions → On startup:

   try {
       // Connect to HiveMQ public broker
       client = new MqttClient("tcp://broker.hivemq.com:1883", 
                               MqttClient.generateClientId(), null);
       
       // Define message handler
       client.setCallback(new MqttCallback() {
           public void connectionLost(Throwable cause) {
               traceln("Connection lost!");
           }
           
           public void messageArrived(String topic, MqttMessage message) throws Exception {
               double incomingTemp = Double.parseDouble(new String(message.getPayload()));
               currentTemperature = incomingTemp;
               traceln("Received: " + incomingTemp + " on topic: " + topic);
           }
           
           public void deliveryComplete(IMqttDeliveryToken token) {}
       });
       
       // Connect and subscribe
       client.connect();
       client.subscribe("ucf/ids6742/bulent/machine/temp");
       traceln("Digital Twin Connected and Listening...");
       
   } catch (MqttException e) {
       e.printStackTrace();
   }

5. Properties → Agent actions → On destroy:

   try {
       if(client != null && client.isConnected()) {
           client.disconnect();
       }
   } catch (MqttException e) { }

Step 6: Run the Digital Twin

1. Click Run (or press F5)
2. Check console for: "Digital Twin Connected and Listening..."
3. Machine rectangle should be green (temperature < 110°C)
4. Time plot should show flat line at 100.0°C initially

---

Complete System Test

Step 1: Start Digital Twin

1. In AnyLogic, run your model
2. Verify "Digital Twin Connected and Listening..." in console
3. Rectangle should be green

Step 2: Start Physical Sensor

# In terminal
cd ~/Downloads/PhysicalSensor
source .venv/bin/activate
python sensor_edge_robust.py

Step 3: Observe Real-time Synchronization

• Time plot updates every second with new temperature data
• Rectangle turns RED when temperature exceeds 110°C
• Rectangle turns GREEN when temperature drops below 110°C
• Console shows received messages

Step 4: Test Resilience

1. Stop Python script (Ctrl+C) - Digital Twin maintains last state
2. Restart Python script - Digital Twin resumes receiving data
3. Simulate network issues - Robust version auto-reconnects

---

Troubleshooting

Common Issues and Solutions

1. ModuleNotFoundError: No module named 'paho'

# Solution: Install in virtual environment
source .venv/bin/activate
pip install paho-mqtt

2. MQTT Connection Error: Not authorized

Cause: Public broker rate limiting or connection limits Solution: Use sensor_edge_robust.py which implements:

• Random client IDs to avoid conflicts
• Automatic reconnection with backoff
• Connection state management

3. AnyLogic: NoClassDefFoundError for MQTT

Solution: Ensure JAR file is properly added:

1. Check JAR file exists in project directory
2. Verify path in Dependencies section
3. Restart AnyLogic if necessary

4. No Data Appearing in AnyLogic

Checklist:

• Topic names must match EXACTLY between Python and AnyLogic
• Check firewall settings (port 1883 must be open)
• Verify internet connectivity
• Check AnyLogic console for error messages

5. Python Script Hangs Without Output

# Add flush=True to print statements for immediate output
print(f"Published Data: {payload} °C", flush=True)

6. Temperature Values Drift Too High/Low

The simulation uses random walk: temp += random.uniform(-1.0, 2.5)

• Average drift: +0.75°C per second
• To stabilize, adjust range to: random.uniform(-1.75, 1.75)

---

Technical Details

MQTT Protocol

• Protocol: MQTT v3.1.1
• Broker: broker.hivemq.com (public)
• Port: 1883 (unencrypted)
• QoS Levels:
  • QoS 0: At most once (fire-and-forget)
  • QoS 1: At least once (acknowledged) - Used in robust version
  • QoS 2: Exactly once (not used here)

Message Format

• Topic Structure: ucf/ids6742/{username}/machine/temp
• Payload: String representation of float (e.g., "102.45")
• Frequency: 1 message per second
• Size: ~6-8 bytes per message

Performance Metrics

• Latency: ~50-200ms (depends on network)
• Throughput: 1 msg/sec (configurable)
• Connection Overhead: ~2KB initial handshake
• Bandwidth: ~10 bytes/second during operation

Security Considerations

⚠️ Warning: This lab uses a public MQTT broker without authentication

• No encryption: Data transmitted in plaintext
• No authentication: Anyone can publish/subscribe
• Public visibility: All topics are accessible

For production systems, implement:

• TLS/SSL encryption (port 8883)
• Username/password authentication
• Access Control Lists (ACLs)
• Private MQTT broker deployment

---

Advanced Modifications

1. Add Multiple Sensors

# In sensor_edge_robust.py
topics = {
    "temperature": "ucf/ids6742/bulent/machine/temp",
    "pressure": "ucf/ids6742/bulent/machine/pressure",
    "vibration": "ucf/ids6742/bulent/machine/vibration"
}

2. Implement Bidirectional Communication

Add control commands from AnyLogic to Python:

// In AnyLogic
client.publish("ucf/ids6742/bulent/machine/control", "STOP");

3. Add Data Persistence

import csv
with open('sensor_log.csv', 'a') as f:
    writer = csv.writer(f)
    writer.writerow([time.time(), temp])

4. Implement Alert System

// In AnyLogic messageArrived method
if(incomingTemp > 115.0) {
    // Trigger alarm
    showMessage("CRITICAL: Temperature exceeds safety threshold!");
}

---

Learning Objectives

This lab demonstrates:

1. IoT Architecture: Understanding publish/subscribe messaging patterns
2. Digital Twin Concepts: Real-time synchronization between physical and digital
3. Protocol Implementation: Working with MQTT for IoT communication
4. System Integration: Connecting heterogeneous systems (Python + Java)
5. Error Handling: Building resilient distributed systems
6. Real-time Visualization: Creating responsive dashboards

---

Additional Resources

Documentation

• MQTT Protocol Specification
• Eclipse Paho Python Client
• HiveMQ Public Broker
• AnyLogic Help

Related Topics

• Industrial IoT (IIoT)
• Industry 4.0
• Cyber-Physical Systems
• Digital Thread
• Predictive Maintenance
• Smart Manufacturing

---

Contact & Support

Course: IDS 6742 - Real-Time Simulation Modeling for Digital Twins
Instructor: Dr. Bulent Soykan
Institution: University of Central Florida

For technical issues with the lab, please ensure you have:

1. Followed all installation steps
2. Checked the troubleshooting section
3. Verified network connectivity
4. Reviewed error messages in both Python and AnyLogic consoles

---

License

This educational material is provided for academic purposes as part of the IDS 6742 course curriculum. Students are encouraged to experiment, modify, and extend the code for learning purposes.

---

Last Updated: March 2026