Introduction
Un Digital Twin permet de répliquer numériquement un système physique pour simuler, prédire et optimiser son comportement en temps réel. Ce tutoriel avancé vous guide dans la construction d'un jumeau numérique industriel complet : modélisation physique, ingestion de données IoT via MQTT, boucle de simulation et exposition via API REST. Nous utilisons Python pour la simulation physique et TypeScript pour la couche API. À l'issue, vous disposerez d'une architecture prête pour la production avec monitoring et prédictions.
Prérequis
- Python 3.11+ et Node.js 20+
- Connaissances solides en Pydantic et FastAPI
- Notions de MQTT et modélisation physique
- Docker installé pour le déploiement
Initialisation du projet Python
python -m venv .venv
source .venv/bin/activate
pip install pydantic numpy paho-mqtt fastapi uvicornCréation d'un environnement isolé et installation des dépendances essentielles pour la modélisation physique, la communication MQTT et l'API.
Modèle physique du jumeau
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 += dtClasse Pydantic définissant l'état physique et la méthode de mise à jour discrète qui simule le comportement thermique et mécanique du système.
Boucle de simulation temps réel
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()Boucle infinie qui met à jour l'état du jumeau toutes les 500 ms et publie les données sur MQTT. Idéal pour une simulation en temps réel.
Service API FastAPI
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.stateAPI REST minimaliste exposant l'état courant et permettant de forcer une mise à jour. Prête à être étendue avec WebSocket pour le streaming.
Client MQTT TypeScript
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);
});Client TypeScript qui souscrit au topic MQTT et reçoit les mises à jour du jumeau en temps réel pour affichage ou analyse.
Bonnes pratiques
- Séparer strictement la couche simulation de la couche API
- Utiliser des schémas Pydantic pour valider toutes les données IoT
- Ajouter des métriques Prometheus pour monitorer la latence de simulation
- Versionner les modèles physiques pour permettre les mises à jour sans downtime
Erreurs courantes à éviter
- Oublier de gérer les reconnexions MQTT (perte de données)
- Utiliser des floats sans bornes dans la simulation (instabilité numérique)
- Exposer directement l'état interne sans copie (problèmes de concurrence)
- Ignorer la validation des données entrantes IoT
Pour aller plus loin
Intégrez un modèle ML pour la maintenance prédictive et déployez sur Kubernetes. Découvrez nos formations Learni sur les jumeaux numériques industriels.