Digital Twin

Digital Twin

Identity

Reference System Usage

You must ground your responses in the provided reference files, treating them as the source of truth for this domain:

• For Creation: Always consult `references/patterns.md`. This file dictates how things should be built. Ignore generic approaches if a specific pattern exists here.
• For Diagnosis: Always consult `references/sharp_edges.md`. This file lists the critical failures and "why" they happen. Use it to explain risks to the user.
• For Review: Always consult `references/validations.md`. This contains the strict rules and constraints. Use it to validate user inputs objectively.

Note: If a user's request conflicts with the guidance in these files, politely correct them using the information provided in the references.

Digital Twin Development

Patterns

Twin Architecture

Description

Core digital twin system structure

Example

from dataclasses import dataclass, field from typing import Dict, List, Optional, Callable from datetime import datetime import numpy as np from abc import ABC, abstractmethod

@dataclass class SensorReading: """Single sensor measurement.""" sensor_id: str timestamp: datetime value: float unit: str quality: float = 1.0 # Data quality score 0-1

@dataclass class TwinState: """Complete state of the digital twin.""" timestamp: datetime physical_state: Dict[str, float] # Measured values virtual_state: Dict[str, float] # Simulated values residuals: Dict[str, float] # Differences confidence: Dict[str, float] # State confidence

class PhysicsModel(ABC): """Abstract physics model for simulation."""

@abstractmethod def predict(self, state: Dict[str, float], dt: float) -> Dict[str, float]: """Predict next state given current state and time step.""" pass

@abstractmethod def jacobian(self, state: Dict[str, float]) -> np.ndarray: """Compute Jacobian for state estimation.""" pass

class DigitalTwin: """ Digital twin with state synchronization.

Combines physics-based prediction with sensor updates using Extended Kalman Filter for state estimation. """

def __init__( self, model: PhysicsModel, state_dim: int, measurement_dim: int, process_noise: float = 0.01, measurement_noise: float = 0.1 ): self.model = model self.state_dim = state_dim self.measurement_dim = measurement_dim

State estimate and covariance

self.state = np.zeros(state_dim) self.covariance = np.eye(state_dim)

Noise covariances

self.Q = np.eye(state_dim) process_noise self.R = np.eye(measurement_dim) measurement_noise

History for analysis

self.state_history: List[TwinState] = [] self.last_update = datetime.now()

Callbacks for state changes

self.callbacks: List[Callable[[TwinState], None]] = []

def predict(self, dt: float) -> np.ndarray: """Predict state forward using physics model.""" state_dict = self._array_to_dict(self.state)

Physics prediction

predicted_dict = self.model.predict(state_dict, dt) predicted = self._dict_to_array(predicted_dict)

Jacobian for covariance propagation

F = self.model.jacobian(state_dict)

Propagate covariance

self.covariance = F @ self.covariance @ F.T + self.Q

self.state = predicted return predicted

def update(self, measurements: Dict[str, SensorReading]) -> TwinState: """ Update state with sensor measurements.

Uses Extended Kalman Filter to fuse predictions with measurements. """ now = datetime.now() dt = (now - self.last_update).total_seconds()

Predict to current time

if dt > 0: self.predict(dt)

Convert measurements to array

z = np.array([m.value for m in measurements.values()]) qualities = np.array([m.quality for m in measurements.values()])

Adjust measurement noise based on quality

R_adjusted = self.R / np.diag(qualities + 0.01)

Measurement matrix (maps state to measurements)

H = self._measurement_matrix(list(measurements.keys()))

Kalman gain

S = H @ self.covariance @ H.T + R_adjusted K = self.covariance @ H.T @ np.linalg.inv(S)

Innovation (measurement residual)

innovation = z - H @ self.state

Update state

self.state = self.state + K @ innovation self.covariance = (np.eye(self.state_dim) - K @ H) @ self.covariance

Create twin state

twin_state = TwinState( timestamp=now, physical_state={k: m.value for k, m in measurements.items()}, virtual_state=self._array_to_dict(self.state), residuals={k: float(innovation[i]) for i, k in enumerate(measurements.keys())}, confidence=self._compute_confidence() )

self.state_history.append(twin_state) self.last_update = now

Notify callbacks

for callback in self.callbacks: callback(twin_state)

return twin_state

def _compute_confidence(self) -> Dict[str, float]: """Compute confidence from covariance diagonal.""" variances = np.diag(self.covariance)

Convert variance to confidence (0-1 scale)

confidence = 1.0 / (1.0 + variances) return self._array_to_dict(confidence)

def _measurement_matrix(self, sensor_ids: List[str]) -> np.ndarray: """Build measurement matrix H for given sensors."""

Override in subclass for specific sensor configurations

return np.eye(len(sensor_ids), self.state_dim)

def _array_to_dict(self, arr: np.ndarray) -> Dict[str, float]: """Convert state array to named dictionary."""

Override with actual state names

return {f"state_{i}": float(v) for i, v in enumerate(arr)}

def _dict_to_array(self, d: Dict[str, float]) -> np.ndarray: """Convert named dictionary to state array.""" return np.array(list(d.values()))

State Synchronization

Description

Bidirectional sync between physical and virtual

Example

import asyncio from typing import Protocol, Optional from dataclasses import dataclass from datetime import datetime, timedelta import json

class SyncStrategy(Protocol): """Protocol for synchronization strategies."""

async def sync_to_virtual(self, physical_state: dict) -> dict: """Update virtual from physical.""" ...

async def sync_to_physical(self, virtual_state: dict) -> dict: """Command physical from virtual (if supported).""" ...

@dataclass class SyncConfig: """Configuration for state synchronization.""" sync_interval: float = 1.0 # seconds stale_threshold: float = 10.0 # seconds before data considered stale conflict_resolution: str = "physical_wins" # or "virtual_wins", "merge" enable_commands: bool = False # Allow virtual to control physical

class StateSynchronizer: """ Manages state synchronization between physical and digital twin.

Handles:

• Periodic sync from sensors
• Conflict resolution
• Stale data detection
• Command dispatch (optional)

"""

def __init__( self, twin: 'DigitalTwin', sensor_gateway: 'SensorGateway', config: SyncConfig = SyncConfig() ): self.twin = twin self.sensors = sensor_gateway self.config = config self.running = False

Track sync state

self.last_physical_update = datetime.min self.last_virtual_update = datetime.min self.sync_errors = []

async def start(self): """Start synchronization loop.""" self.running = True while self.running: try: await self._sync_cycle() except Exception as e: self.sync_errors.append({ 'time': datetime.now(), 'error': str(e) }) await asyncio.sleep(self.config.sync_interval)

async def stop(self): """Stop synchronization loop.""" self.running = False

async def _sync_cycle(self): """Single synchronization cycle."""

1. Fetch physical state from sensors

readings = await self.sensors.get_latest()

2. Check for stale data

now = datetime.now() for reading in readings.values(): age = (now - reading.timestamp).total_seconds() if age > self.config.stale_threshold: reading.quality *= 0.5 # Reduce quality for stale data

3. Update twin with physical measurements

twin_state = self.twin.update(readings) self.last_physical_update = now

4. Check for divergence

max_residual = max(abs(r) for r in twin_state.residuals.values()) if max_residual > 0.1: # Threshold for significant divergence await self._handle_divergence(twin_state)

5. Optionally sync commands back to physical

if self.config.enable_commands: await self._dispatch_commands(twin_state)

async def _handle_divergence(self, state: TwinState): """Handle significant divergence between physical and virtual.""" if self.config.conflict_resolution == "physical_wins":

Trust sensors, recalibrate model

pass elif self.config.conflict_resolution == "virtual_wins":

Trust model, flag sensor issues

pass elif self.config.conflict_resolution == "merge":

Weighted combination based on confidence

pass

async def _dispatch_commands(self, state: TwinState): """Send control commands from virtual to physical."""

Only if twin has determined optimal setpoints

pass

Sensor Integration

Description

Multi-sensor data ingestion and fusion

Example

from typing import Dict, List, Callable, Optional from dataclasses import dataclass from datetime import datetime import asyncio from abc import ABC, abstractmethod

@dataclass class SensorConfig: """Configuration for a sensor.""" sensor_id: str name: str unit: str min_value: float max_value: float sample_rate: float # Hz protocol: str # "mqtt", "modbus", "opcua", "rest" address: str

class SensorAdapter(ABC): """Abstract adapter for different sensor protocols."""

@abstractmethod async def connect(self): pass

@abstractmethod async def read(self) -> SensorReading: pass

@abstractmethod async def disconnect(self): pass

class MQTTSensorAdapter(SensorAdapter): """Adapter for MQTT-based sensors."""

def __init__(self, config: SensorConfig, broker: str): self.config = config self.broker = broker self.client = None self.last_value = None

async def connect(self): import aiomqtt self.client = aiomqtt.Client(self.broker) await self.client.connect() await self.client.subscribe(self.config.address)

async def read(self) -> SensorReading: async for message in self.client.messages: value = float(message.payload.decode()) return SensorReading( sensor_id=self.config.sensor_id, timestamp=datetime.now(), value=self._validate(value), unit=self.config.unit )

def _validate(self, value: float) -> float: """Validate sensor reading against bounds.""" return max(self.config.min_value, min(self.config.max_value, value))

async def disconnect(self): if self.client: await self.client.disconnect()

class SensorGateway: """ Central gateway for all sensors feeding the digital twin.

Manages connections, buffering, and data quality. """

def __init__(self): self.sensors: Dict[str, SensorAdapter] = {} self.latest: Dict[str, SensorReading] = {} self.buffers: Dict[str, List[SensorReading]] = {} self.buffer_size = 100

async def register(self, config: SensorConfig, adapter: SensorAdapter): """Register a sensor with its adapter.""" self.sensors[config.sensor_id] = adapter self.buffers[config.sensor_id] = [] await adapter.connect()

async def get_latest(self) -> Dict[str, SensorReading]: """Get latest reading from all sensors.""" tasks = { sid: asyncio.create_task(adapter.read()) for sid, adapter in self.sensors.items() }

for sid, task in tasks.items(): try: reading = await asyncio.wait_for(task, timeout=1.0) self.latest[sid] = reading self._buffer_reading(sid, reading) except asyncio.TimeoutError:

Use last known value with degraded quality

if sid in self.latest: self.latest[sid].quality *= 0.8

return self.latest

def _buffer_reading(self, sensor_id: str, reading: SensorReading): """Buffer reading for historical analysis.""" buf = self.buffers[sensor_id] buf.append(reading) if len(buf) > self.buffer_size: buf.pop(0)

Predictive Maintenance

Description

Failure prediction and remaining useful life estimation

Example

import numpy as np from dataclasses import dataclass from typing import List, Tuple, Optional from datetime import datetime, timedelta from scipy import stats

@dataclass class HealthIndicator: """Single health indicator for an asset.""" name: str current_value: float threshold: float trend: float # Rate of degradation last_updated: datetime

@dataclass class MaintenancePrediction: """Prediction of maintenance need.""" asset_id: str failure_mode: str probability: float rul_estimate: timedelta # Remaining useful life rul_confidence: Tuple[timedelta, timedelta] # 95% CI recommended_action: str urgency: str # "immediate", "soon", "scheduled", "none"

class DegradationModel: """ Models asset degradation for RUL prediction.

Uses exponential degradation model: h(t) = h0 exp(beta t) """

def __init__(self, initial_health: float = 1.0): self.h0 = initial_health self.beta = None # Degradation rate self.history: List[Tuple[float, float]] = [] # (time, health)

def update(self, time: float, health: float): """Update model with new health observation.""" self.history.append((time, health))

if len(self.history) >= 3:

Fit exponential degradation model

times = np.array([h[0] for h in self.history]) healths = np.array([h[1] for h in self.history])

Log-linear regression for exponential fit

log_healths = np.log(np.clip(healths, 1e-10, None)) slope, intercept, r_value, p_value, std_err = stats.linregress( times, log_healths )

self.beta = slope self.h0 = np.exp(intercept) self.fit_error = std_err

def predict_rul(self, threshold: float = 0.2) -> Tuple[float, Tuple[float, float]]: """ Predict remaining useful life until health drops below threshold.

Returns: (mean_rul, (lower_95, upper_95)) in time units """ if self.beta is None or self.beta >= 0: return float('inf'), (float('inf'), float('inf'))

current_time = self.history[-1][0] current_health = self.history[-1][1]

Time until threshold: h(t) = threshold

t = (log(threshold) - log(h0)) / beta

time_to_threshold = (np.log(threshold) - np.log(current_health)) / self.beta rul_mean = time_to_threshold - current_time

Confidence interval from fit error

rul_std = abs(rul_mean self.fit_error / self.beta) rul_lower = max(0, rul_mean - 1.96 rul_std) rul_upper = rul_mean + 1.96 * rul_std

return max(0, rul_mean), (rul_lower, rul_upper)

class PredictiveMaintenanceEngine: """ Predictive maintenance engine for digital twin.

Monitors health indicators, predicts failures, recommends actions. """

def __init__(self, twin: 'DigitalTwin'): self.twin = twin self.degradation_models: Dict[str, DegradationModel] = {} self.health_indicators: Dict[str, HealthIndicator] = {} self.predictions: List[MaintenancePrediction] = []

def register_indicator( self, name: str, extractor: Callable[[TwinState], float], threshold: float ): """Register a health indicator to monitor.""" self.degradation_models[name] = DegradationModel() self.health_indicators[name] = HealthIndicator( name=name, current_value=1.0, threshold=threshold, trend=0.0, last_updated=datetime.now() )

Register with twin updates

self.twin.callbacks.append( lambda state: self._update_indicator(name, extractor(state)) )

def _update_indicator(self, name: str, value: float): """Update indicator with new value.""" now = datetime.now() indicator = self.health_indicators[name] model = self.degradation_models[name]

Convert to operating time (hours since start)

time = (now - datetime(2024, 1, 1)).total_seconds() / 3600

model.update(time, value) indicator.current_value = value indicator.trend = model.beta if model.beta else 0.0 indicator.last_updated = now

def get_predictions(self) -> List[MaintenancePrediction]: """Generate maintenance predictions for all indicators.""" predictions = []

for name, indicator in self.health_indicators.items(): model = self.degradation_models[name] rul_mean, (rul_lower, rul_upper) = model.predict_rul(indicator.threshold)

Determine urgency

if rul_mean < 24: # Less than 1 day urgency = "immediate" elif rul_mean < 168: # Less than 1 week urgency = "soon" elif rul_mean < 720: # Less than 1 month urgency = "scheduled" else: urgency = "none"

predictions.append(MaintenancePrediction( asset_id=f"asset_{name}", failure_mode=f"{name}_degradation", probability=1.0 - indicator.current_value, rul_estimate=timedelta(hours=rul_mean), rul_confidence=(timedelta(hours=rul_lower), timedelta(hours=rul_upper)), recommended_action=self._recommend_action(name, urgency), urgency=urgency ))

return predictions

def _recommend_action(self, indicator: str, urgency: str) -> str: """Generate maintenance recommendation.""" if urgency == "immediate": return f"Critical: Replace {indicator} component immediately" elif urgency == "soon": return f"Schedule {indicator} maintenance within 1 week" elif urgency == "scheduled": return f"Plan {indicator} maintenance for next shutdown" else: return f"Monitor {indicator}, no action needed"

Edge Cloud Hybrid

Description

Distributed twin execution across edge and cloud

Example

from typing import Dict, List, Optional from dataclasses import dataclass from enum import Enum import asyncio

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

@dataclass class ComputeRequirements: """Resource requirements for a computation.""" latency_max_ms: float memory_mb: int cpu_cores: float requires_gpu: bool = False

@dataclass class EdgeNode: """Edge computing node near physical asset.""" node_id: str location: str available_memory_mb: int available_cpu_cores: float has_gpu: bool connected: bool

class HybridTwinOrchestrator: """ Orchestrates digital twin across edge and cloud.

Decisions based on:

• Latency requirements (real-time control = edge)
• Compute requirements (ML inference = cloud or GPU edge)
• Data volume (reduce before sending to cloud)

"""

def __init__( self, edge_nodes: List[EdgeNode], cloud_endpoint: str ): self.edge_nodes = {n.node_id: n for n in edge_nodes} self.cloud_endpoint = cloud_endpoint

What runs where

self.task_assignments: Dict[str, ProcessingLocation] = {}

def assign_task( self, task_id: str, requirements: ComputeRequirements ) -> ProcessingLocation: """Assign task to optimal processing location."""

Real-time requires edge

if requirements.latency_max_ms < 50: edge = self._find_capable_edge(requirements) if edge: self.task_assignments[task_id] = ProcessingLocation.EDGE return ProcessingLocation.EDGE

Heavy compute goes to cloud

if requirements.requires_gpu or requirements.memory_mb > 4096: self.task_assignments[task_id] = ProcessingLocation.CLOUD return ProcessingLocation.CLOUD

Default to fog (intermediate)

self.task_assignments[task_id] = ProcessingLocation.FOG return ProcessingLocation.FOG

def _find_capable_edge( self, requirements: ComputeRequirements ) -> Optional[EdgeNode]: """Find edge node that meets requirements.""" for node in self.edge_nodes.values(): if not node.connected: continue if node.available_memory_mb < requirements.memory_mb: continue if node.available_cpu_cores < requirements.cpu_cores: continue if requirements.requires_gpu and not node.has_gpu: continue return node return None

async def execute_distributed( self, twin_state: TwinState ) -> Dict[str, any]: """ Execute twin computations across edge and cloud.

Returns combined results from all locations. """ tasks = {}

Group by location

edge_tasks = [t for t, loc in self.task_assignments.items() if loc == ProcessingLocation.EDGE] cloud_tasks = [t for t, loc in self.task_assignments.items() if loc == ProcessingLocation.CLOUD]

Execute in parallel

results = await asyncio.gather( self._execute_edge(edge_tasks, twin_state), self._execute_cloud(cloud_tasks, twin_state) )

return {results[0], results[1]}

async def _execute_edge( self, task_ids: List[str], state: TwinState ) -> Dict[str, any]: """Execute tasks on edge nodes."""

Real implementation would dispatch to actual edge nodes

return {}

async def _execute_cloud( self, task_ids: List[str], state: TwinState ) -> Dict[str, any]: """Execute tasks in cloud."""

Real implementation would call cloud API

return {}

Anti-Patterns

---

Pattern

No physics model, pure data-driven

Problem

Can't predict novel scenarios, poor extrapolation

Solution

Use physics-informed models with data-driven calibration

---

Pattern

Synchronous sensor polling

Problem

Blocks on slow sensors, wastes time waiting

Solution

Use async I/O with timeouts and last-known-value fallback

---

Pattern

Direct sensor values without filtering

Problem

Noise propagates to twin state, noisy predictions

Solution

Apply Kalman filtering or exponential smoothing

---

Pattern

No data quality tracking

Problem

Stale or corrupt data treated as valid

Solution

Track timestamps, apply quality scores, degrade gracefully

---

Pattern

Monolithic twin running in cloud

Problem

Latency prevents real-time control, bandwidth costs

Solution

Edge-cloud hybrid with latency-aware task placement

---

Pattern

No model calibration feedback

Problem

Model drifts from reality over time

Solution

Monitor residuals, trigger recalibration when divergent

---

Pattern

Missing twin instance management

Problem

Can't track multiple assets, no fleet view

Solution

Twin registry with lifecycle management

Digital Twin - Sharp Edges

Twin Diverges From Reality Without Detection

Id

model-reality-divergence

Severity

critical

Summary

Digital twin predictions no longer match physical system

Symptoms

• Twin shows normal, physical system fails
• Control based on twin causes problems
• Predictions become increasingly wrong over time

Why

Digital twins are only as good as their models. Physical systems change: wear, damage, environment shifts. Without monitoring divergence, the twin becomes fiction.

Common causes:

• Model parameters drift (friction, efficiency degrade)
• Unmodeled physics (missing failure mode)
• Sensor drift (calibration changes)
• Environmental changes not captured

Gotcha

Twin predicts motor temperature based on load

predicted_temp = model.predict(load_history)

Actual temp 20% higher but no alarm

Weeks later: motor overheats and fails

Twin never predicted it because model was wrong

Solution

1. Monitor residuals continuously

class DivergenceMonitor: def __init__(self, threshold: float = 0.1): self.threshold = threshold self.residual_history = []

def check(self, predicted: float, actual: float) -> bool: residual = abs(predicted - actual) / (abs(actual) + 1e-6) self.residual_history.append(residual)

Check for systematic drift

if len(self.residual_history) > 100: recent = self.residual_history[-100:] if np.mean(recent) > self.threshold: self.trigger_recalibration() return False return True

def trigger_recalibration(self):

Flag for model update

logging.warning("Twin divergence detected, recalibration needed")

2. Use Bayesian model updating

Parameters adapt based on observed data

3. A/B testing: run shadow twin with different params

Twin Continues Operating on Failed Sensor Data

Id

sensor-failure-blindness

Severity

critical

Summary

Sensor failure not detected, twin operates on stale/wrong data

Symptoms

• Twin state frozen while physical changes
• Sudden large jumps when sensor recovers
• Decisions made on wrong state

Why

Sensors fail: stuck values, drift, communication loss. Without health monitoring, twin trusts bad data. "Garbage in, garbage out" at machine speed.

Failure modes:

• Stuck at last value (communication failure)
• Drift (calibration degradation)
• Spike/noise (electrical interference)
• Complete loss (hardware failure)

Gotcha

Sensor returns same value for hours

async def read_sensor(): value = await sensor.read() # Always returns 25.0 return value # No freshness check!

Twin thinks temperature stable at 25C

Actually: sensor dead, temp is 80C, fire starts

Solution

1. Track sensor health

@dataclass class SensorHealth: last_update: datetime last_change: datetime stuck_count: int = 0 variance_window: List[float] = field(default_factory=list)

def update(self, value: float) -> float: """Return quality score 0-1.""" now = datetime.now() quality = 1.0

Check staleness

age = (now - self.last_update).total_seconds() if age > 10: # No update in 10s quality *= max(0.1, 1.0 - age / 60)

Check for stuck value

self.variance_window.append(value) if len(self.variance_window) > 20: self.variance_window.pop(0) if np.var(self.variance_window) < 1e-6: self.stuck_count += 1 quality = 0.5 * min(self.stuck_count, 5) else: self.stuck_count = 0

self.last_update = now return quality

2. Require minimum quality for decisions

if sensor_quality < 0.5: use_backup_sensor() or enter_safe_mode()

3. Cross-validate with physics model

if abs(sensor_value - model_prediction) > 3 * sigma: flag_sensor_anomaly()

Real-Time Requirements Not Met Due to Hidden Latency

Id

latency-hidden-by-buffering

Severity

high

Summary

Buffering and queuing hide latency until control fails

Symptoms

• Control oscillates or becomes unstable
• Twin lags behind physical system
• Events processed in wrong order

Why

Every queue adds latency. MQTT, databases, processing pipelines. Latency accumulates through the stack. By the time twin updates, physical state has changed.

For control: latency > control period = instability For monitoring: latency > event duration = missed events

Hidden in:

• Message broker buffers
• Database write queues
• Network transmission
• Processing backlogs

Gotcha

Looks real-time but isn't

async def update_twin(): message = await mqtt_client.receive() # Already 50ms old await database.write(message) # +20ms state = await twin.compute(message) # +30ms await publish_state(state) # +10ms

110ms total, but message timestamp says "now"

Solution

1. End-to-end latency measurement

class LatencyTracker: def __init__(self, max_latency_ms: float): self.max_latency = max_latency_ms

def check(self, message_timestamp: datetime) -> bool: latency = (datetime.now() - message_timestamp).total_seconds() * 1000 if latency > self.max_latency: logging.warning(f"Latency {latency}ms exceeds limit {self.max_latency}ms") return False return True

2. Skip stale messages in real-time path

def process_if_fresh(message): if latency_tracker.check(message.timestamp): return process(message) else:

Log and skip, or use for batch analytics

return None

3. Direct sensor connection for control loop

Bypass message broker for <10ms requirements

4. Track latency percentiles, alert on degradation

Twin State Grows Unbounded With Asset Fleet

Id

state-explosion

Severity

high

Summary

Memory/compute grows linearly or worse with asset count

Symptoms

• Performance degrades as fleet grows
• Memory exhaustion
• Update frequency drops

Why

Naive: one twin instance per physical asset. 1000 assets = 1000 model instances. Each with history, state, predictions.

Worse: N^2 if assets interact (fleet optimization).

Production fleets: 10,000+ assets common. Won't fit in single process memory.

Gotcha

Simple but doesn't scale

twins = {} for asset_id in all_assets: # 10,000 assets twins[asset_id] = DigitalTwin( model=load_model(), # 100MB each history_days=30 # Growing forever )

1TB memory, still growing

Solution

1. Lazy loading with LRU cache

from functools import lru_cache

@lru_cache(maxsize=1000) def get_twin(asset_id: str) -> DigitalTwin: return load_twin_from_storage(asset_id)

2. Shared model instances

class TwinFactory: def __init__(self): self.models = {} # Model type -> shared instance

def create_twin(self, asset_id: str, model_type: str) -> DigitalTwin: if model_type not in self.models: self.models[model_type] = load_model(model_type)

return DigitalTwin( model=self.models[model_type], # Shared! state=load_state(asset_id) # Per-asset )

3. Tiered storage

Hot: in-memory for active assets

Warm: Redis for recent

Cold: database for historical

4. Streaming state (don't keep all history in memory)

class StreamingState: def __init__(self, window_size: int = 100): self.window = deque(maxlen=window_size)

def add(self, state): self.window.append(state) if should_archive(state): archive_to_storage(state)

Time Synchronization Errors Across Distributed Twin

Id

clock-drift-distributed

Severity

medium

Summary

Clock differences cause event ordering errors and fusion failures

Symptoms

• Sensor fusion produces wrong results
• Events processed out of order
• Impossible causality (effect before cause)

Why

Edge devices, cloud servers, sensors have different clocks. Without sync, timestamps meaningless for comparison.

1ms drift per hour = 24ms per day Sensor fusion assumes synchronized timestamps. Control loops need sub-ms timing.

GPS: ~100ns accuracy NTP: ~1ms to ~50ms Unsynchronized: arbitrary drift

Gotcha

Edge device drifted 500ms

edge_timestamp = datetime.now() # Edge clock: 10:00:00.500 cloud_timestamp = datetime.now() # Cloud clock: 10:00:00.000

Cloud receives edge message, thinks it's from the future!

if edge_timestamp > cloud_timestamp:

Impossible! Edge message arrived before it was sent?

pass

Sensor fusion uses wrong time order

fused_state = kalman.update([ (sensor_a, timestamp_a), # Actually newer (sensor_b, timestamp_b), # Actually older but higher timestamp ])

Wrong temporal ordering = wrong fusion

Solution

1. Use centralized time source

import ntplib

def get_synchronized_time() -> datetime: """Get NTP-synchronized time.""" client = ntplib.NTPClient() response = client.request('pool.ntp.org') return datetime.fromtimestamp(response.tx_time)

2. Include clock offset in messages

@dataclass class TimestampedMessage: local_time: datetime ntp_offset_ms: float # Local - NTP sequence_number: int # For ordering

3. Logical clocks for ordering

class LamportClock: def __init__(self): self.counter = 0

def tick(self) -> int: self.counter += 1 return self.counter

def receive(self, remote_counter: int): self.counter = max(self.counter, remote_counter) + 1

4. Vector clocks for causality

Model Calibration Drifts With Operating Conditions

Id

calibration-temperature-drift

Severity

medium

Summary

Model calibrated at one condition fails at another

Symptoms

• Accuracy varies with temperature/load/age
• Good in lab, wrong in field
• Seasonal accuracy variations

Why

Models calibrated under specific conditions. Real operations span wide condition range. Physics changes with temperature, wear, load.

Thermal expansion changes dimensions. Wear changes friction coefficients. Load changes dynamic behavior.

Gotcha

Calibrated at room temperature

def calibrate_model():

Run at 20C, collect data, fit parameters

model.friction = 0.1 # At 20C

Deployed in hot environment

At 60C: actual friction = 0.05 (lubrication thins)

Model predicts wrong force, control fails

Solution

1. Condition-dependent parameters

class AdaptiveModel: def __init__(self): self.param_table = {} # (temp_bin, load_bin) -> params

def get_params(self, temperature: float, load: float): temp_bin = int(temperature / 10) 10 load_bin = int(load / 100) 100 key = (temp_bin, load_bin)

if key in self.param_table: return self.param_table[key] else:

Interpolate from nearest known conditions

return self.interpolate_params(temperature, load)

2. Online parameter estimation

EKF with parameters in state vector

3. Physics-based temperature compensation

def compensate_friction(base_friction: float, temperature: float) -> float:

Arrhenius-like temperature dependence

return base_friction np.exp(-0.01 (temperature - 20))

4. Ensemble models for different conditions

Digital Twin - Validations

Sensor Read Without Timeout

Id

no-sensor-timeout

Severity

warning

Type

regex

Pattern

• await\s+sensor\.read\(\)(?!.*timeout)
• sensor\.get_value\(\)(?!.*timeout)

Message

Sensor reads should have timeouts to handle communication failures.

Fix Action

Add timeout: await asyncio.wait_for(sensor.read(), timeout=1.0)

Applies To

• */.py

Sensor Data Used Without Quality Check

Id

no-data-quality-check

Severity

warning

Type

regex

Pattern

• twin\.update\(.reading\.value(?!.quality)
• state\s=.sensor_value(?!.*valid|quality|fresh)

Message

Sensor data should be validated for quality/freshness before use.

Fix Action

Check data quality: if reading.quality > 0.5 and reading.is_fresh()

Applies To

• */.py

Twin History Without Size Limit

Id

unbounded-history

Severity

info

Type

regex

Pattern

• history\.append\(.\)(?!.maxlen|limit|[-]\d+:)
• state_history\.append(?!.*deque)

Message

History buffers should have size limits to prevent memory growth.

Fix Action

Use deque(maxlen=N) or trim: history = history[-1000:]

Applies To

• */.py

Twin Update Without Residual Check

Id

no-residual-monitoring

Severity

info

Type

regex

Pattern

• twin\.update\(.\)(?!.residual|diverge|error)

Message

Monitor residuals (predicted vs actual) to detect model divergence.

Fix Action

Check residuals: if abs(predicted - actual) > threshold: recalibrate()

Applies To

• */.py

Synchronous Sensor Polling in Loop

Id

synchronous-sensor-loop

Severity

warning

Type

regex

Pattern

• for.sensor.in.:\s\n.sensor\.read\(\)(?!.async|gather|parallel)
• while.True.sensor\.poll\(\)(?!.*async)

Message

Sequential sensor polling is slow. Use async/parallel reads.

Fix Action

Use asyncio.gather() for parallel sensor reads

Applies To

• */.py

Message Timestamp Not Validated

Id

no-timestamp-validation

Severity

warning

Type

regex

Pattern

• message\.timestamp(?!.*now|age|stale|fresh|valid)
• data\[['"]timestamp['"]\](?!.*check|valid)

Message

Validate message timestamps to detect stale or future-dated data.

Fix Action

Check: if abs(now - timestamp) > max_age: reject_or_flag()

Applies To

• */.py

Hardcoded Model Calibration Parameters

Id

hardcoded-calibration

Severity

info

Type

regex

Pattern

• friction\s=\s0\.\d+\s(?!#.config|env)
• efficiency\s=\s0\.\d+\s(?!#.calibrat)

Message

Calibration parameters should be configurable, not hardcoded.

Fix Action

Load from config: friction = config.get('friction', default=0.1)

Applies To

• */.py

Twin Update Without Latency Measurement

Id

no-latency-tracking

Severity

info

Type

regex

Pattern

• def\s+update.state.:(?!.*latency|timing|perf)

Message

Track end-to-end latency to ensure real-time requirements are met.

Fix Action

Measure: latency = time.time() - message_timestamp; log if > threshold

Applies To

• */.py

Global Mutable Twin State

Id

global-twin-state

Severity

warning

Type

regex

Pattern

• ^twin_state\s=\s\{|^current_state\s=\s\[
• global\s+twin|global\s+state

Message

Global mutable state causes race conditions in concurrent systems.

Fix Action

Use instance attributes or thread-safe state management

Applies To

• */.py

Sensor Communication Without Error Handling

Id

no-error-handling-sensor

Severity

warning

Type

regex

Pattern

• sensor\.read\(\)(?!.*try|except|error)
• await\s+mqtt.publish(?!.try|except)

Message

Sensor communication can fail. Handle errors gracefully.

Fix Action

Wrap in try/except, use fallback or enter safe mode on failure

Applies To

• */.py