Digital Twins — Complete Guide: What They Are and How They Work

import asyncio
import json
from datetime import datetime
from typing import Optional

class DigitalTwin:
    """Simple digital twin for an industrial pump with predictive analytics"""

    TEMPERATURE_WARNING_C = 85
    TEMPERATURE_CRITICAL_C = 100
    VIBRATION_WARNING_G = 0.3
    VIBRATION_CRITICAL_G = 0.5

    def __init__(self, asset_id: str, asset_name: str):
        self.asset_id = asset_id
        self.asset_name = asset_name
        self.state = {
            "temperature_c": 0.0,
            "pressure_bar": 0.0,
            "vibration_g": 0.0,
            "rpm": 0,
            "power_kw": 0.0,
            "last_updated": None,
            "health_score": 100,
            "alerts": [],
            "operational_hours": 0,
        }
        self._reading_history = []

    def update_from_sensor(self, sensor_data: dict) -> None:
        """Update twin state with real-time sensor reading"""
        self.state.update(sensor_data)
        self.state["last_updated"] = datetime.utcnow().isoformat() + "Z"
        self._reading_history.append({**sensor_data, "timestamp": self.state["last_updated"]})
        if len(self._reading_history) > 1000:
            self._reading_history.pop(0)  # keep rolling window
        self._evaluate_health()

    def _evaluate_health(self) -> None:
        """Rules-based health assessment — real systems use ML models"""
        alerts = []
        health_deductions = 0

        # Temperature checks
        if self.state["temperature_c"] > self.TEMPERATURE_CRITICAL_C:
            alerts.append({"severity": "CRITICAL", "code": "TEMP_HIGH",
                          "msg": f"Critical temperature {self.state['temperature_c']}°C — shutdown risk"})
            health_deductions += 40
        elif self.state["temperature_c"] > self.TEMPERATURE_WARNING_C:
            alerts.append({"severity": "WARNING", "code": "TEMP_ELEVATED",
                          "msg": f"Elevated temperature {self.state['temperature_c']}°C — check cooling"})
            health_deductions += 15

        # Vibration checks (bearing failure indicator)
        if self.state["vibration_g"] > self.VIBRATION_CRITICAL_G:
            alerts.append({"severity": "CRITICAL", "code": "VIB_HIGH",
                          "msg": "Excessive vibration — imminent bearing failure risk"})
            health_deductions += 50
        elif self.state["vibration_g"] > self.VIBRATION_WARNING_G:
            alerts.append({"severity": "WARNING", "code": "VIB_ELEVATED",
                          "msg": "Elevated vibration — schedule bearing inspection"})
            health_deductions += 20

        self.state["health_score"] = max(0, 100 - health_deductions)
        self.state["alerts"] = alerts

    def predict_failure_days(self) -> Optional[int]:
        """Simplified degradation model — real systems use LSTM/regression"""
        if self.state["vibration_g"] > self.VIBRATION_WARNING_G:
            # Exponential degradation model
            days = int(14 * (self.VIBRATION_CRITICAL_G / self.state["vibration_g"]) ** 2)
            return max(1, days)
        if self.state["temperature_c"] > self.TEMPERATURE_WARNING_C:
            days = int(30 * (self.TEMPERATURE_CRITICAL_C / self.state["temperature_c"]) ** 3)
            return max(1, days)
        return None  # No failure predicted

    def get_status_report(self) -> dict:
        """Generate operator-ready status report"""
        failure_days = self.predict_failure_days()
        return {
            "asset": f"{self.asset_name} ({self.asset_id})",
            "health_score": self.state["health_score"],
            "status": "CRITICAL" if self.state["health_score"] < 50 else
                      "WARNING" if self.state["health_score"] < 80 else "HEALTHY",
            "predicted_failure_days": failure_days,
            "maintenance_urgency": "IMMEDIATE" if failure_days and failure_days <= 7 else
                                   "SOON" if failure_days and failure_days <= 30 else "ROUTINE",
            "alerts": self.state["alerts"],
            "last_updated": self.state["last_updated"],
        }

# Simulate real-time sensor updates
twin = DigitalTwin("PUMP-PROD-001", "Main Circulation Pump")

# Normal operation
twin.update_from_sensor({
    "temperature_c": 72.3, "pressure_bar": 4.2,
    "vibration_g": 0.12, "rpm": 1480, "power_kw": 22.5
})
print("Normal:", twin.get_status_report()["status"])  # HEALTHY

# Degrading bearing
twin.update_from_sensor({
    "temperature_c": 88.1, "pressure_bar": 4.0,
    "vibration_g": 0.38, "rpm": 1475, "power_kw": 23.8
})
report = twin.get_status_report()
print(f"Degraded: {report['status']} — failure in ~{report['predicted_failure_days']} days")