Skip to content
Learni
View all tutorials
Industrie 4.0

How to Implement an Advanced Digital Twin in 2026

18 minADVANCED
IoTDigital TwinSimulation
Lire en français

Introduction

A digital twin digitally replicates a physical system to simulate, predict, and optimize its behavior in real time. This advanced tutorial guides you through building a complete industrial digital twin: physical modeling, MQTT-based IoT data ingestion, simulation loops, and REST API exposure. We use Python for the physics simulation and TypeScript for the API layer. By the end, you will have a production-ready architecture with monitoring and predictive capabilities.

Prerequisites

  • Python 3.11+ and Node.js 20+
  • Solid knowledge of Pydantic and FastAPI
  • Basic understanding of MQTT and physical modeling
  • Docker installed for deployment

Python Project Initialization

terminal
python -m venv .venv
source .venv/bin/activate
pip install pydantic numpy paho-mqtt fastapi uvicorn

Create an isolated environment and install the core dependencies for physical modeling, MQTT communication, and the API.

Digital Twin Physics Model

digital_twin/models.py
from pydantic import BaseModel
class MachineState(BaseModel):
    temperature: float
    vibration: float
    rpm: int
    timestamp: float
class DigitalTwin:
    def __init__(self):
        self.state = MachineState(temperature=65.0, vibration=2.1, rpm=1500, timestamp=0.0)
    def update(self, dt: float, ambient_temp: float):
        self.state.temperature += (ambient_temp - self.state.temperature) * 0.05 * dt
        self.state.vibration = max(0.5, self.state.vibration + (self.state.rpm - 1500) * 0.001)
        self.state.rpm = int(self.state.rpm * 0.999 + 1500 * 0.001)
        self.state.timestamp += dt

Pydantic class defining the physical state and the discrete update method that simulates the system's thermal and mechanical behavior.

Real-Time Simulation Loop

digital_twin/simulation.py
import time
import paho.mqtt.client as mqtt
from models import DigitalTwin

twin = DigitalTwin()
client = mqtt.Client()
client.connect('localhost', 1883)

def run_simulation():
    while True:
        twin.update(0.5, 22.0)
        client.publish('twin/state', twin.state.model_dump_json())
        time.sleep(0.5)

if __name__ == '__main__':
    run_simulation()

Infinite loop that updates the twin state every 500 ms and publishes data to MQTT. Ideal for real-time simulation.

FastAPI Service

api/main.py
from fastapi import FastAPI
from models import DigitalTwin

app = FastAPI(title='Digital Twin API')
twin = DigitalTwin()

@app.get('/state')
async def get_state():
    return twin.state

@app.post('/update')
async def update_state(ambient: float):
    twin.update(1.0, ambient)
    return twin.state

Minimal REST API exposing the current state and allowing forced updates. Ready to be extended with WebSockets for streaming.

TypeScript MQTT Client

client/mqtt-client.ts
import mqtt from 'mqtt';
const client = mqtt.connect('mqtt://localhost:1883');
client.on('connect', () => {
  client.subscribe('twin/state');
});
client.on('message', (topic, message) => {
  const state = JSON.parse(message.toString());
  console.log('Twin update:', state);
});

TypeScript client that subscribes to the MQTT topic and receives real-time twin updates for display or analysis.

Best Practices

  • Strictly separate the simulation layer from the API layer
  • Use Pydantic schemas to validate all IoT data
  • Add Prometheus metrics to monitor simulation latency
  • Version physical models to enable updates without downtime

Common Mistakes to Avoid

  • Forgetting to handle MQTT reconnections (data loss)
  • Using unbounded floats in the simulation (numerical instability)
  • Exposing internal state directly without copying (concurrency issues)
  • Ignoring validation of incoming IoT data

Going Further

Integrate an ML model for predictive maintenance and deploy on Kubernetes. Discover our Learni trainings on industrial digital twins.