Digital Twin Testing for Industrial Systems: From Concept to CI
A digital twin is a real-time virtual model of a physical system that receives the same sensor data as the physical asset and can predict behavior, simulate failures, and serve as a test environment for control logic changes. For testing purposes, digital twins remove the most dangerous constraint in OT testing: you no longer need physical access to a running process to inject failures.
What Digital Twins Enable for Testing
The primary testing value of a digital twin:
Risk-free failure injection: Inject pump cavitation, motor over-temperature, valve seizure, communication loss — without affecting the physical plant or requiring a planned outage.
Regression testing before deployment: Test new PLC logic or SCADA configurations against the twin before applying changes to production hardware.
Edge case simulation: Physical systems rarely enter certain failure states under normal operation. A twin can be driven into those states deterministically.
Continuous integration: Attach the twin to a CI pipeline. Every code commit runs against a realistic simulation of the physical system.
The key caveat: twin fidelity matters. A low-fidelity twin that doesn't model fluid dynamics, thermal behavior, or mechanical coupling will give you false confidence. Understanding your twin's fidelity limits is itself a testing discipline.
Building a Digital Twin Test Environment
Architecture Overview
┌─────────────────────────────────────────────────────┐
│ Physical Plant │
│ Sensors → PLC → Historian → SCADA │
└──────────────────┬──────────────────────────────────┘
│ Real-time data (OPC-UA / MQTT)
▼
┌─────────────────────────────────────────────────────┐
│ Digital Twin Platform (Eclipse Ditto / Azure DT) │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Twin Model │ │ Simulation │ │
│ │ (state sync) │ │ Engine │ │
│ └──────────────┘ └──────────────┘ │
└──────────────────┬──────────────────────────────────┘
│ Twin API
▼
┌─────────────────────────────────────────────────────┐
│ Test Framework │
│ - Fault injection │
│ - Regression suite │
│ - CI pipeline integration │
└─────────────────────────────────────────────────────┘Eclipse Ditto Setup
Eclipse Ditto is the leading open-source digital twin platform. It manages thing state and provides a REST/WebSocket API.
# Launch Ditto via Docker Compose
git clone https://github.com/eclipse-ditto/ditto.git
cd ditto/deployment/docker
# Start Ditto (nginx, gateway, things, policies, connectivity services)
docker-compose up -d
# Verify health
curl -s http://localhost:8080/health | python3 -m json.toolDefine a twin model for a pump station:
// PUT http://localhost:8080/api/2/things/industrial:pump-station-01
{
"attributes": {
"manufacturer": "GroundFos",
"model": "CM10-2",
"installation_date": "2021-03-15"
},
"features": {
"pump1": {
"properties": {
"status": {
"running": false,
"speed_rpm": 0,
"current_amps": 0.0,
"temperature_c": 22.0
},
"alarms": {
"over_temperature": false,
"low_flow": false,
"vibration_fault": false
}
}
},
"flow_sensor": {
"properties": {
"flow_lpm": 0.0,
"pressure_bar": 0.0,
"quality": "good"
}
}
}
}Synchronization from Physical to Twin
import asyncio
import aiohttp
from asyncua import Client as OpcClient
DITTO_BASE = "http://localhost:8080/api/2"
DITTO_AUTH = ("ditto", "ditto")
OPC_SERVER = "opc.tcp://plc-server:4840"
THING_ID = "industrial:pump-station-01"
# OPC-UA node mapping to Ditto feature paths
NODE_MAPPING = {
"ns=2;i=1001": "pump1/properties/status/running",
"ns=2;i=1002": "pump1/properties/status/speed_rpm",
"ns=2;i=1003": "pump1/properties/status/temperature_c",
"ns=2;i=2001": "flow_sensor/properties/flow_lpm",
"ns=2;i=2002": "flow_sensor/properties/pressure_bar",
}
async def sync_twin():
async with aiohttp.ClientSession(auth=aiohttp.BasicAuth(*DITTO_AUTH)) as http:
opc_client = OpcClient(OPC_SERVER)
await opc_client.connect()
while True:
updates = {}
for node_id, ditto_path in NODE_MAPPING.items():
node = opc_client.get_node(node_id)
value = await node.read_value()
# Build nested update dict from path
parts = ditto_path.split("/")
d = updates
for part in parts[:-1]:
d = d.setdefault(part, {})
d[parts[-1]] = value
# Merge feature updates into Ditto
for feature_name, feature_data in updates.items():
await http.patch(
f"{DITTO_BASE}/things/{THING_ID}/features/{feature_name}",
json=feature_data
)
await asyncio.sleep(1) # 1Hz sync rate
asyncio.run(sync_twin())Synchronization Testing: Detecting Twin Drift
Twin drift occurs when the digital twin's state diverges from the physical system due to missed updates, communication failures, or simulation inaccuracies. Detecting drift is a testing discipline.
import requests
import time
from dataclasses import dataclass
from typing import Dict, Any
@dataclass
class DriftReport:
timestamp: float
feature: str
property_path: str
twin_value: Any
physical_value: Any
deviation_pct: float
def get_twin_state(thing_id) -> Dict:
resp = requests.get(
f"{DITTO_BASE}/things/{thing_id}/features",
auth=DITTO_AUTH
)
return resp.json()
def get_physical_state(opc_nodes: Dict[str, str]) -> Dict:
"""Read current values from OPC-UA server."""
from opcua import Client
client = Client(OPC_SERVER)
client.connect()
values = {}
for node_id, key in opc_nodes.items():
values[key] = client.get_node(node_id).get_value()
client.disconnect()
return values
def detect_drift(thing_id, opc_nodes, tolerance_pct=5.0) -> list[DriftReport]:
"""Compare twin state to physical state and report deviations."""
twin = get_twin_state(thing_id)
physical = get_physical_state(opc_nodes)
reports = []
for key, physical_val in physical.items():
# Navigate twin nested dict using key as path
parts = key.split(".")
twin_val = twin
try:
for part in parts:
twin_val = twin_val[part]
except KeyError:
reports.append(DriftReport(
timestamp=time.time(), feature=parts[0],
property_path=key, twin_value=None,
physical_value=physical_val, deviation_pct=100.0
))
continue
if isinstance(physical_val, (int, float)) and physical_val != 0:
deviation = abs(twin_val - physical_val) / abs(physical_val) * 100
if deviation > tolerance_pct:
reports.append(DriftReport(
timestamp=time.time(), feature=parts[0],
property_path=key, twin_value=twin_val,
physical_value=physical_val, deviation_pct=deviation
))
return reports
# Run drift detection continuously
while True:
drift = detect_drift(THING_ID, NODE_MAPPING)
if drift:
for d in drift:
print(f"DRIFT DETECTED: {d.property_path} "
f"twin={d.twin_value} physical={d.physical_value} "
f"({d.deviation_pct:.1f}% deviation)")
time.sleep(10)Fault Injection Testing
Digital twins allow injecting fault conditions that would be dangerous or impossible to create in the physical system.
import requests
def inject_fault(thing_id, fault_type, parameters=None):
"""Inject a fault condition into the digital twin."""
fault_configs = {
"pump_cavitation": {
"feature": "pump1",
"updates": {
"properties/status/speed_rpm": 0,
"properties/alarms/vibration_fault": True,
"properties/status/temperature_c": 85.0
}
},
"sensor_stuck": {
"feature": "flow_sensor",
"updates": {
"properties/quality": "bad",
"properties/flow_lpm": parameters.get("stuck_value", 0)
}
},
"communication_loss": {
"feature": "pump1",
"updates": {
"properties/status/running": None, # Unknown state
"properties/status/speed_rpm": None
}
}
}
config = fault_configs[fault_type]
feature = config["feature"]
for path, value in config["updates"].items():
requests.patch(
f"{DITTO_BASE}/things/{thing_id}/features/{feature}/{path}",
json=value, auth=DITTO_AUTH
)
print(f"Fault injected: {fault_type} on {thing_id}")
def test_alarm_response_to_over_temperature():
"""Verify SCADA alarm fires when pump temperature exceeds threshold."""
# Record baseline alarm count
baseline_alarms = scada_api.get_active_alarms()
baseline_count = len(baseline_alarms)
# Inject fault
inject_fault(THING_ID, "pump_cavitation")
# Wait for SCADA to poll twin and process alarm
import time
time.sleep(5)
active_alarms = scada_api.get_active_alarms()
new_alarms = [a for a in active_alarms if a not in baseline_alarms]
temp_alarms = [a for a in new_alarms if "temperature" in a["message"].lower()]
assert len(temp_alarms) >= 1, \
f"Expected temperature alarm, got {len(new_alarms)} new alarms: {new_alarms}"
print(f"PASS: Temperature alarm fired within 5s of fault injection")
# Clear fault
requests.patch(
f"{DITTO_BASE}/things/{THING_ID}/features/pump1/properties/alarms",
json={"over_temperature": False, "vibration_fault": False},
auth=DITTO_AUTH
)CI Pipeline Integration
# .github/workflows/scada-integration-tests.yml
name: SCADA Integration Tests (Digital Twin)
on:
push:
paths:
- 'plc-programs/**'
- 'scada-configs/**'
jobs:
twin-tests:
runs-on: self-hosted
services:
ditto:
image: eclipse/ditto:latest
ports:
- 8080:8080
options: --health-cmd "curl -f http://localhost:8080/health" --health-interval 10s
steps:
- uses: actions/checkout@v3
- name: Initialize digital twin
run: |
python3 tests/twin/setup_twin.py \
--config tests/twin/pump-station-01.json \
--ditto-url http://localhost:8080
- name: Load new PLC program to SoftPLC
run: |
codesys --deploy plc-programs/ConveyorControl.project \
--target softplc://localhost:4840
- name: Run fault injection test suite
run: |
pytest tests/twin/ -v \
--twin-url http://localhost:8080 \
--opc-url opc.tcp://localhost:4840 \
--junitxml=test-results/twin-tests.xml
- name: Publish results
uses: EnricoMi/publish-unit-test-result-action@v2
with:
files: "test-results/*.xml"Physics Simulation Fidelity Validation
A twin that ignores physics will miss real failure modes. Validate simulation fidelity by comparing twin predictions against historical physical data.
import pandas as pd
import numpy as np
def validate_pump_model_fidelity(historian_export_csv, twin_simulation_csv):
"""
Compare twin simulation predictions to actual historian data.
Both CSVs: timestamp, flow_lpm, pressure_bar, temperature_c
"""
physical = pd.read_csv(historian_export_csv, parse_dates=["timestamp"])
twin = pd.read_csv(twin_simulation_csv, parse_dates=["timestamp"])
# Align on timestamps (nearest match)
merged = pd.merge_asof(
physical.sort_values("timestamp"),
twin.sort_values("timestamp"),
on="timestamp",
tolerance=pd.Timedelta("5s"),
suffixes=("_physical", "_twin")
)
metrics = {}
for column in ["flow_lpm", "pressure_bar", "temperature_c"]:
physical_col = f"{column}_physical"
twin_col = f"{column}_twin"
valid = merged[[physical_col, twin_col]].dropna()
rmse = np.sqrt(((valid[physical_col] - valid[twin_col]) ** 2).mean())
mae = (valid[physical_col] - valid[twin_col]).abs().mean()
metrics[column] = {"rmse": rmse, "mae": mae, "n": len(valid)}
print(f"{column}: RMSE={rmse:.3f}, MAE={mae:.3f} (n={len(valid)})")
# Acceptance criteria for twin fidelity
assert metrics["flow_lpm"]["rmse"] < 5.0, \
f"Flow model RMSE {metrics['flow_lpm']['rmse']:.2f} exceeds 5 LPM threshold"
assert metrics["pressure_bar"]["rmse"] < 0.2, \
f"Pressure model RMSE {metrics['pressure_bar']['rmse']:.3f} exceeds 0.2 bar threshold"
return metricsKey Takeaways
- Digital twins remove the physical risk constraint from OT testing — fault injection that would cause process disruption becomes safe and repeatable
- Twin drift detection must be built in from day one; a twin that diverges from reality provides false confidence
- Fidelity validation using historical data quantifies how much to trust simulation results — set RMSE thresholds appropriate to your system
- Integrate the twin into CI so PLC code changes are validated against simulation before physical deployment
- Eclipse Ditto and Azure Digital Twins provide production-grade twin platforms with REST APIs suitable for test automation integration