Digital Twin

Digital Twin Maturity Levels

Architecture Design

High-Level Architecture Components

A comprehensive digital twin architecture consists of multiple interconnected layers that work together to create, maintain, and utilize the virtual representation.

Physical Layer — The actual assets, equipment, and systems being modeled. This includes sensors, actuators, programmable logic controllers (PLCs), SCADA systems, and other industrial IoT devices that capture and control physical processes.

Connectivity Layer — The infrastructure that transports data between physical and digital domains. This includes industrial protocols (OPC-UA, MQTT, Modbus), network infrastructure, and edge computing devices that preprocess and filter data before transmission.

Data Layer — The storage and management systems that handle the massive volumes of data generated by digital twins. This includes time-series databases, data lakes, data warehouses, and real-time streaming platforms.

Analytics Layer — The computation engines that process data and generate insights. This includes statistical analysis, machine learning models, physics-based simulations, and optimization algorithms.

Application Layer — The user-facing interfaces and applications that interact with the digital twin. This includes dashboards, visualization tools, mobile applications, and integration APIs.

Control Layer — The systems that can affect the physical world based on digital twin insights. This includes automation systems, control algorithms, and human decision support tools.

Reference Architecture Diagram

┌─────────────────────────────────────────────────────────────────┐
│                      APPLICATION LAYER                          │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │Dashboard │  │Mobile App│  │Analytics  │  │Integration│       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
└────────────────────────────┬────────────────────────────────────┘
                             │
┌────────────────────────────▼────────────────────────────────────┐
│                       ANALYTICS LAYER                            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │Real-time │  │Machine   │  │Physics   │  │Optimization│       │
│  │Analytics │  │Learning  │  │Simulation│  │Engine    │       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
└────────────────────────────┬────────────────────────────────────┘
                             │
┌────────────────────────────▼────────────────────────────────────┐
│                         DATA LAYER                               │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │Time-Series│ │Data Lake │ │Asset     │  │Knowledge │       │
│  │Database  │  │          │  │Registry  │  │Graph     │       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
└────────────────────────────┬────────────────────────────────────┘
                             │
┌────────────────────────────▼────────────────────────────────────┐
│                     CONNECTIVITY LAYER                           │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │MQTT     │  │OPC-UA    │  │Edge      │  │Protocol  │       │
│  │Broker   │  │Server    │  │Gateway   │  │Translator│       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
└────────────────────────────┬────────────────────────────────────┘
                             │
┌────────────────────────────▼────────────────────────────────────┐
│                       PHYSICAL LAYER                             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐       │
│  │Sensors  │  │Actuators │  │PLCs     │  │Equipment │       │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘       │
└─────────────────────────────────────────────────────────────────┘

Data Integration

Time-Series Data Management

Digital twins generate massive volumes of time-series data from sensors. Efficient storage and querying of this data is critical for performance.

from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional, Dict, Any
from enum import Enum
import json

class DataQuality(Enum):
    GOOD = "good"
    SUSPECT = "suspect"
    BAD = "bad"
    MISSING = "missing"

@dataclass
class SensorReading:
    asset_id: str
    sensor_id: str
    timestamp: datetime
    value: float
    unit: str
    quality: DataQuality
    metadata: Dict[str, Any]

class TimeSeriesIngestService:
    def __init__(self, database_client):
        self.db = database_client
        self.buffer = []
        self.batch_size = 1000
        self.flush_interval_seconds = 5
    
    async def ingest_reading(self, reading: SensorReading):
        """Ingest a single sensor reading"""
        # Validate reading
        if not self._validate_reading(reading):
            await self._handle_invalid_reading(reading)
            return
        
        # Add to buffer
        self.buffer.append(reading)
        
        # Flush if buffer is full
        if len(self.buffer) >= self.batch_size:
            await self._flush_buffer()
    
    async def _flush_buffer(self):
        """Write buffered readings to database"""
        if not self.buffer:
            return
        
        # Convert to database format
        records = [self._to_db_record(r) for r in self.buffer]
        
        # Batch insert
        await self.db.batch_insert("sensor_readings", records)
        
        # Clear buffer
        self.buffer = []
    
    def _validate_reading(self, reading: SensorReading) -> bool:
        """Validate sensor reading"""
        if reading.value is None:
            return False
        
        # Check for reasonable range
        if reading.sensor_id.startswith("temp_"):
            return -50 <= reading.value <= 200
        elif reading.sensor_id.startswith("pressure_"):
            return 0 <= reading.value <= 1000
        
        return True
    
    def _to_db_record(self, reading: SensorReading) -> Dict:
        """Convert reading to database record"""
        return {
            "asset_id": reading.asset_id,
            "sensor_id": reading.sensor_id,
            "timestamp": reading.timestamp.isoformat(),
            "value": reading.value,
            "unit": reading.unit,
            "quality": reading.quality.value,
            "metadata": json.dumps(reading.metadata)
        }
    
    async def _handle_invalid_reading(self, reading: SensorReading):
        """Handle invalid readings"""
        # Log or send to dead letter queue
        pass

Asset Registry

The asset registry maintains the master data for all assets in the digital twin system.

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime

@dataclass
class Asset:
    asset_id: str
    asset_type: str
    name: str
    description: str
    location: Optional[Dict[str, float]]  # lat, lon
    parent_asset_id: Optional[str]
    sensors: List[str] = field(default_factory=list)
    metadata: Dict = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

@dataclass
class Sensor:
    sensor_id: str
    asset_id: str
    sensor_type: str
    name: str
    unit: str
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    sampling_rate_seconds: int = 60

class AssetRegistry:
    def __init__(self, database):
        self.db = database
        self.cache = {}
    
    async def register_asset(self, asset: Asset) -> str:
        """Register a new asset"""
        await self.db.insert("assets", {
            "asset_id": asset.asset_id,
            "asset_type": asset.asset_type,
            "name": asset.name,
            "description": asset.description,
            "location": json.dumps(asset.location) if asset.location else None,
            "parent_asset_id": asset.parent_asset_id,
            "metadata": json.dumps(asset.metadata),
            "created_at": asset.created_at.isoformat(),
            "updated_at": asset.updated_at.isoformat()
        })
        self.cache[asset.asset_id] = asset
        return asset.asset_id
    
    async def register_sensor(self, sensor: Sensor) -> str:
        """Register a sensor for an asset"""
        await self.db.insert("sensors", {
            "sensor_id": sensor.sensor_id,
            "asset_id": sensor.asset_id,
            "sensor_type": sensor.sensor_type,
            "name": sensor.name,
            "unit": sensor.unit,
            "min_value": sensor.min_value,
            "max_value": sensor.max_value,
            "sampling_rate_seconds": sensor.sampling_rate_seconds
        })
        return sensor.sensor_id
    
    async def get_asset_hierarchy(self, root_asset_id: str) -> Dict:
        """Get full asset hierarchy starting from root"""
        # Query all assets with parent relationship
        all_assets = await self.db.query("assets", {})
        
        # Build hierarchy
        assets_by_id = {a["asset_id"]: a for a in all_assets}
        
        def build_tree(asset_id: str) -> Dict:
            asset = assets_by_id.get(asset_id, {})
            children = [
                build_tree(a["asset_id"]) 
                for a in all_assets 
                if a.get("parent_asset_id") == asset_id
            ]
            return {
                **asset,
                "children": children
            }
        
        return build_tree(root_asset_id)

Real-Time Synchronization

Data Pipeline Architecture

import asyncio
from typing import Callable, Dict, List
import json

class DigitalTwinSynchronizer:
    def __init__(self, mqtt_client, timeseries_db, cache):
        self.mqtt = mqtt_client
        self.tsdb = timeseries_db
        self.cache = cache
        self.subscriptions = {}
        self.handlers = {}
    
    async def start(self):
        """Start the synchronization service"""
        # Subscribe to sensor data topics
        await self.mqtt.subscribe("assets/+/sensors/+/data", self._handle_sensor_data)
        await self.mqtt.subscribe("assets/+/events/+", self._handle_asset_event)
        await self.mqtt.subscribe("assets/+/telemetry", self._handle_telemetry)
    
    async def _handle_sensor_data(self, topic: str, payload: bytes):
        """Handle incoming sensor data"""
        # Parse topic: assets/{asset_id}/sensors/{sensor_id}/data
        parts = topic.split("/")
        asset_id = parts[1]
        sensor_id = parts[3]
        
        # Parse payload
        data = json.loads(payload)
        
        # Create reading
        reading = SensorReading(
            asset_id=asset_id,
            sensor_id=sensor_id,
            timestamp=datetime.fromisoformat(data.get("timestamp", datetime.now().isoformat())),
            value=data["value"],
            unit=data.get("unit", ""),
            quality=DataQuality(data.get("quality", "good")),
            metadata=data.get("metadata", {})
        )
        
        # Ingest to time-series database
        await self.tsdb.ingest_reading(reading)
        
        # Update cache for real-time queries
        cache_key = f"{asset_id}:{sensor_id}:latest"
        await self.cache.set(cache_key, json.dumps({
            "value": reading.value,
            "timestamp": reading.timestamp.isoformat()
        }), ttl=300)
        
        # Invoke registered handlers
        handler_key = f"{asset_id}:{sensor_id}"
        if handler_key in self.handlers:
            await self.handlers[handler_key](reading)
    
    async def _handle_asset_event(self, topic: str, payload: bytes):
        """Handle asset events (state changes, alerts, etc.)"""
        parts = topic.split("/")
        asset_id = parts[1]
        event_type = parts[3]
        
        event = json.loads(payload)
        event["asset_id"] = asset_id
        event["event_type"] = event_type
        event["timestamp"] = datetime.now().isoformat()
        
        # Store event
        await self.tsdb.ingest_event(event)
        
        # Check for alert conditions
        await self._check_alerts(asset_id, event)
    
    def register_handler(self, asset_id: str, sensor_id: str, handler: Callable):
        """Register a handler for specific asset/sensor"""
        key = f"{asset_id}:{sensor_id}"
        self.handlers[key] = handler

Edge-Cloud Integration

import asyncio
from enum import Enum

class ProcessingTier(Enum):
    EDGE = "edge"
    FOG = "fog"
    CLOUD = "cloud"

class DataRouter:
    def __init__(self, edge_client, fog_nodes, cloud_endpoint):
        self.edge = edge_client
        self.fog_nodes = fog_nodes
        self.cloud = cloud_endpoint
        self.routing_rules = {}
    
    def add_routing_rule(
        self, 
        data_type: str, 
        tier: ProcessingTier,
        condition: Callable = None
    ):
        """Add a routing rule for data type"""
        self.routing_rules[data_type] = {
            "tier": tier,
            "condition": condition
        }
    
    async def route_data(self, data: Dict) -> Dict:
        """Route data to appropriate processing tier"""
        data_type = data.get("data_type", "unknown")
        
        rule = self.routing_rules.get(data_type)
        if rule is None:
            # Default to cloud
            return await self._send_to_cloud(data)
        
        # Check condition
        if rule["condition"] and not rule["condition"](data):
            return await self._send_to_cloud(data)
        
        tier = rule["tier"]
        
        if tier == ProcessingTier.EDGE:
            return await self._process_at_edge(data)
        elif tier == ProcessingTier.FOG:
            return await self._send_to_fog(data)
        else:
            return await self._send_to_cloud(data)
    
    async def _process_at_edge(self, data: Dict) -> Dict:
        """Process data at edge device"""
        # Apply edge processing logic
        result = await self.edge.process(data)
        
        # If results need cloud storage, send asynchronously
        if result.get("store_in_cloud"):
            asyncio.create_task(self._send_to_cloud(result))
        
        return result
    
    async def _send_to_fog(self, data: Dict) -> Dict:
        """Send data to nearest fog node"""
        # Find nearest fog node
        fog_node = self._find_nearest_fog(data.get("location"))
        
        # Forward to fog node
        return await self.fog_nodes[fog_node].forward(data)
    
    async def _send_to_cloud(self, data: Dict) -> Dict:
        """Send data to cloud for processing"""
        return await self.cloud.ingest(data)

Physics-Based Simulation

Simulation Engine Interface

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, List, Any
import numpy as np

@dataclass
class SimulationState:
    """Current state of a simulated system"""
    time: float
    variables: Dict[str, float]
    derivatives: Dict[str, float]
    metadata: Dict[str, Any]

class SimulationModel(ABC):
    """Base class for physics-based simulation models"""
    
    @abstractmethod
    def get_state(self, state: Dict[str, float]) -> Dict[str, float]:
        """Calculate rate of change (derivatives) for given state"""
        pass
    
    @abstractmethod
    def get_parameters(self) -> Dict[str, Any]:
        """Return model parameters"""
        pass
    
    @abstractmethod
    def validate_state(self, state: Dict[str, float]) -> bool:
        """Check if state is physically valid"""
        pass

class PumpModel(SimulationModel):
    """Physics-based model for a centrifugal pump"""
    
    def __init__(
        self,
        max_flow_rate: float = 100.0,  # L/s
        max_head: float = 50.0,         # meters
        nominal_power: float = 5000.0,  # watts
        efficiency: float = 0.85
    ):
        self.max_flow_rate = max_flow_rate
        self.max_head = max_head
        self.nominal_power = nominal_power
        self.efficiency = efficiency
    
    def get_state(self, state: Dict[str, float]) -> Dict[str, float]:
        """Calculate pump dynamics"""
        flow_rate = state.get("flow_rate", 0.0)
        pressure_in = state.get("pressure_in", 0.0)
        rpm = state.get("rpm", 0.0)
        
        # Pump curve: head decreases with flow
        # H = H_max * (1 - (Q/Q_max)^2)
        normalized_flow = flow_rate / self.max_flow_rate
        head = self.max_head * (1 - normalized_flow ** 2)
        
        # Power consumption
        power = self.nominal_power * (0.3 + 0.7 * (rpm / 3000))
        
        # Efficiency varies with operating point
        efficiency = self._calculate_efficiency(normalized_flow)
        
        # Calculate derivatives (simplified)
        d_flow = (head + pressure_in - state.get("pressure_out", 0)) / 100.0
        d_rpm = (state.get("rpm_setpoint", rpm) - rpm) * 0.5
        
        return {
            "head": head,
            "power": power,
            "efficiency": efficiency,
            "d_flow_rate": d_flow,
            "d_rpm": d_rpm
        }
    
    def _calculate_efficiency(self, normalized_flow: float) -> float:
        """Calculate pump efficiency based on operating point"""
        # Efficiency curve peaks at ~80% of max flow
        peak_flow = 0.8
        efficiency = self.efficiency * (
            1 - ((normalized_flow - peak_flow) / peak_flow) ** 2
        )
        return max(0.1, efficiency)
    
    def get_parameters(self) -> Dict[str, Any]:
        return {
            "max_flow_rate": self.max_flow_rate,
            "max_head": self.max_head,
            "nominal_power": self.nominal_power,
            "efficiency": self.efficiency
        }
    
    def validate_state(self, state: Dict[str, float]) -> bool:
        flow_rate = state.get("flow_rate", 0)
        rpm = state.get("rpm", 0)
        
        return (
            0 <= flow_rate <= self.max_flow_rate * 1.1
            and 0 <= rpm <= 4000
        )

class SimulationEngine:
    """Numerical simulation engine using Runge-Kutta integration"""
    
    def __init__(self, model: SimulationModel, dt: float = 0.01):
        self.model = model
        self.dt = dt
    
    def step(self, state: Dict[str, float], duration: float) -> SimulationState:
        """Advance simulation by duration using RK4 integration"""
        current_state = state.copy()
        t = 0.0
        
        while t < duration:
            current_state = self._rk4_step(current_state)
            t += self.dt
        
        derivatives = self.model.get_state(current_state)
        
        return SimulationState(
            time=t,
            variables=current_state,
            derivatives=derivatives,
            metadata={"dt": self.dt}
        )
    
    def _rk4_step(self, state: Dict[str, float]) -> Dict[str, float]:
        """Runge-Kutta 4th order integration step"""
        k1 = self.model.get_state(state)
        
        state_k2 = {k: state[k] + k1.get(f"d_{k}", 0) * self.dt / 2 
                   for k in state}
        k2 = self.model.get_state(state_k2)
        
        state_k3 = {k: state[k] + k2.get(f"d_{k}", 0) * self.dt / 2 
                   for k in state}
        k3 = self.model.get_state(state_k3)
        
        state_k4 = {k: state[k] + k3.get(f"d_{k}", 0) * self.dt 
                   for k in state}
        k4 = self.model.get_state(state_k4)
        
        # Combine
        next_state = {}
        for key in state:
            d_key = f"d_{key}"
            if d_key in k1 or d_key in k2 or d_key in k3 or d_key in k4:
                deriv = (
                    k1.get(d_key, 0) + 
                    2 * k2.get(d_key, 0) + 
                    2 * k3.get(d_key, 0) + 
                    k4.get(d_key, 0)
                ) / 6
                next_state[key] = state[key] + deriv * self.dt
            else:
                next_state[key] = state[key]
        
        return next_state

Predictive Maintenance

Anomaly Detection

from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import numpy as np
from typing import List, Tuple

class AnomalyDetector:
    """Detect anomalies in equipment sensor data"""
    
    def __init__(self, contamination: float = 0.1):
        self.contamination = contamination
        self.scaler = StandardScaler()
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42
        )
        self.is_trained = False
    
    def train(self, historical_data: List[Dict[str, float]]):
        """Train on historical sensor data"""
        # Extract features
        features = self._extract_features(historical_data)
        
        # Scale features
        scaled = self.scaler.fit_transform(features)
        
        # Train model
        self.model.fit(scaled)
        self.is_trained = True
    
    def detect(self, sensor_data: Dict[str, float]) -> Tuple[bool, float]:
        """Detect anomaly in current sensor readings"""
        if not self.is_trained:
            raise ValueError("Model not trained")
        
        # Extract features
        features = self._extract_features([sensor_data])
        scaled = self.scaler.transform(features)
        
        # Predict
        anomaly = self.model.predict(scaled)[0]
        score = self.model.score_samples(scaled)[0]
        
        # Convert to probability-like score
        probability = 1 / (1 + np.exp(-score))
        
        return anomaly == -1, probability
    
    def _extract_features(self, data: List[Dict[str, float]]) -> np.ndarray:
        """Extract numerical features from sensor data"""
        # Extract values for each sensor
        sensors = list(data[0].keys())
        
        features = []
        for record in data:
            row = [record.get(sensor, 0.0) for sensor in sensors]
            features.append(row)
        
        return np.array(features)

class PredictiveMaintenanceEngine:
    """Predict equipment failures and maintenance needs"""
    
    def __init__(
        self, 
        anomaly_detector: AnomalyDetector,
        rul_model: 'RULPredictor'
    ):
        self.anomaly_detector = anomaly_detector
        self.rul_model = rul_model
    
    def assess_health(
        self, 
        asset_id: str,
        sensor_readings: Dict[str, float]
    ) -> Dict:
        """Assess health of equipment"""
        # Detect anomalies
        is_anomaly, anomaly_score = self.anomaly_detector.detect(sensor_readings)
        
        # Predict remaining useful life
        rul = self.rul_model.predict(sensor_readings)
        
        # Determine maintenance recommendation
        if rul < 7:  # Days
            priority = "critical"
            action = "Immediate maintenance required"
        elif rul < 30:
            priority = "high"
            action = "Schedule maintenance within 2 weeks"
        elif is_anomaly:
            priority = "medium"
            action = "Monitor closely, investigate anomaly"
        else:
            priority = "low"
            action = "Continue normal operation"
        
        return {
            "asset_id": asset_id,
            "health_status": "degraded" if is_anomaly else "normal",
            "anomaly_detected": is_anomaly,
            "anomaly_score": anomaly_score,
            "remaining_useful_life_days": rul,
            "priority": priority,
            "recommendation": action,
            "timestamp": datetime.now().isoformat()
        }

Industry Applications

Manufacturing Digital Twin

# Example: Manufacturing line digital twin
class ManufacturingLineTwin:
    """Digital twin for manufacturing production line"""
    
    def __init__(self, line_id: str):
        self.line_id = line_id
        self.stations = []
        self.buffer_zones = []
        self.downtime_tracker = DowntimeTracker()
        self.oee_calculator = OEECalculator()
    
    async def add_station(self, station: 'WorkStation'):
        """Add a workstation to the line"""
        self.stations.append(station)
    
    async def get_current_state(self) -> Dict:
        """Get current state of entire production line"""
        states = []
        for station in self.stations:
            states.append(await station.get_state())
        
        # Calculate overall line metrics
        total_cycle_time = sum(s["cycle_time"] for s in states)
        available_stations = sum(1 for s in states if s["status"] == "running")
        
        return {
            "line_id": self.line_id,
            "stations": states,
            "overall_status": self._determine_line_status(states),
            "total_cycle_time": total_cycle_time,
            "utilization": available_stations / len(stations) if states else 0,
            "oee": self.oee_calculator.calculate(states),
            "downtime": await self.downtime_tracker.get_current_downtime(self.line_id)
        }
    
    async def simulate_ bottleneck(self) -> Dict:
        """Identify bottlenecks through simulation"""
        # Simulate with different station speeds
        results = []
        for station in self.stations:
            original_speed = station.processing_speed
            station.processing_speed *= 1.2
            
            # Simulate production
            output = await self._simulate_production(3600)  # 1 hour
            results.append({
                "station": station.station_id,
                "original_output": output,
                "improved_output": output * 1.2
            })
            
            station.processing_speed = original_speed
        
        # Find biggest impact
        max_improvement = max(results, key=lambda x: x["improved_output"] - x["original_output"])
        
        return {
            "bottleneck": max_improvement["station"],
            "potential_improvement": max_improvement["improved_output"] - max_improvement["original_output"],
            "analysis": results
        }

Smart Building Digital Twin

class BuildingTwin:
    """Digital twin for smart building management"""
    
    def __init__(self, building_id: str):
        self.building_id = building_id
        self.zones = {}
        self.hvac_model = HVACModel()
        self.energy_optimizer = EnergyOptimizer()
        self.comfort_calculator = ComfortCalculator()
    
    async def optimize_energy(
        self, 
        occupancy: Dict[str, int],
        electricity_price: float,
        outdoor_temp: float
    ) -> Dict:
        """Optimize HVAC energy consumption"""
        # Get current states
        current_states = {}
        for zone_id, zone in self.zones.items():
            current_states[zone_id] = await zone.get_state()
        
        # Optimize setpoints
        optimized_setpoints = self.energy_optimizer.optimize(
            zones=current_states,
            occupancy=occupancy,
            price=electricity_price,
            outdoor_temp=outdoor_temp
        )
        
        return {
            "building_id": self.building_id,
            "optimization_type": "energy",
            "setpoints": optimized_setpoints,
            "estimated_savings_percent": 15.5,
            "comfort_impact": self.comfort_calculator.assess(
                optimized_setpoints, occupancy
            )
        }

Best Practices

1. Start with Clear Objectives — Define specific use cases and success metrics before building.

2. Invest in Data Quality — Garbage in, garbage out. Sensor quality and data integration are critical.

3. Plan for Scale — Design architecture to handle growth in assets, data volume, and users.

4. Prioritize Security — Digital twins are high-value targets. Implement robust security from the start.

5. Iterate and Improve — Start with a minimum viable twin and add capabilities over time.

6. Bridge IT and OT — Successful digital twins require collaboration between IT and operations technology teams.

7. Choose the Right Level of Fidelity — More detail isn't always better. Match simulation fidelity to use case requirements.

8. Maintain the Digital Twin — Plan for ongoing maintenance as physical assets change.