Digital Twins and Simulation Technology: Creating Virtual Replicas for Real-World Insights

# Digital twin architecture concept
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from datetime import datetime
import numpy as np

@dataclass
class SensorData:
    timestamp: datetime
    measurements: Dict[str, float]
    quality: float = 1.0

@dataclass
class DigitalTwinConfig:
    twin_id: str
    entity_type: str
    update_frequency_hz: float
    simulation_precision: str
    historical_data_retention_days: int

class PhysicsSimulation:
    def __init__(self, model_type: str):
        self.model_type = model_type
        self.parameters: Dict[str, float] = {}
        self.state = {}
    
    def initialize(self, initial_conditions: Dict):
        """Set up simulation with initial conditions"""
        self.state = initial_conditions.copy()
    
    def step(self, dt: float, inputs: Dict) -> Dict:
        """Advance simulation by one time step"""
        return self.state
    
    def apply_physics(self, state: Dict, dt: float) -> Dict:
        """Apply physics model"""
        return state
    
    def get_outputs(self) -> Dict:
        """Extract outputs from simulation"""
        return self.state

class DigitalTwin:
    def __init__(self, config: DigitalTwinConfig):
        self.config = config
        self.physical_entity_id: Optional[str] = None
        self.simulation = PhysicsSimulation("default")
        self.sensor_data_buffer: List[SensorData] = []
        self.model_params: Dict[str, Any] = {}
        self.is_running = False
        self.last_sync: Optional[datetime] = None
    
    def connect_physical_entity(self, entity_id: str):
        """Connect to physical entity for data sync"""
        self.physical_entity_id = entity_id
        print(f"Digital twin {self.config.twin_id} connected to {entity_id}")
    
    def ingest_sensor_data(self, data: SensorData):
        """Receive and store sensor data"""
        self.sensor_data_buffer.append(data)
        if len(self.sensor_data_buffer) > 10000:
            self.sensor_data_buffer.pop(0)
    
    def synchronize(self):
        """Sync virtual model with physical entity state"""
        if not self.physical_entity_id:
            return
        
        latest_data = self.sensor_data_buffer[-1] if self.sensor_data_buffer else None
        if latest_data:
            self.simulation.state.update(latest_data.measurements)
            self.last_sync = datetime.now()
    
    def run_simulation(self, duration: float, inputs: Dict) -> Dict:
        """Run simulation for specified duration"""
        results = []
        t = 0.0
        dt = 0.01
        
        while t < duration:
            state = self.simulation.step(dt, inputs)
            results.append(state.copy())
            t += dt
        
        return {
            'trajectory': results,
            'duration': duration,
            'twin_id': self.config.twin_id
        }
    
    def predict_behavior(self, future_inputs: Dict, horizon: float) -> Dict:
        """Predict future behavior based on current state and future inputs"""
        prediction = self.run_simulation(horizon, future_inputs)
        return prediction
    
    def detect_anomalies(self, threshold: float = 3.0) -> List[Dict]:
        """Detect anomalies between physical and virtual states"""
        if len(self.sensor_data_buffer) < 2:
            return []
        
        anomalies = []
        measurements = self.sensor_data_buffer[-1].measurements
        simulated = self.simulation.state
        
        for key in measurements:
            if key in simulated:
                actual = measurements[key]
                expected = simulated[key]
                deviation = abs(actual - expected) / (abs(expected) + 1e-6)
                
                if deviation > threshold:
                    anomalies.append({
                        'parameter': key,
                        'actual': actual,
                        'expected': expected,
                        'deviation': deviation,
                        'timestamp': datetime.now()
                    })
        
        return anomalies
    
    def optimize(self, objective: str, constraints: Dict) -> Dict:
        """Find optimal operating parameters"""
        return {
            'optimal_params': {'param1': 0.5, 'param2': 0.3},
            'predicted_outcome': 0.95,
            'confidence': 0.87
        }

class DigitalTwinPlatform:
    def __init__(self, name: str):
        self.name = name
        self.twins: Dict[str, DigitalTwin] = {}
        self.data_lake: Dict[str, List[SensorData]] = {}
    
    def create_twin(self, twin_id: str, entity_type: str) -> DigitalTwin:
        """Create a new digital twin"""
        config = DigitalTwinConfig(
            twin_id=twin_id,
            entity_type=entity_type,
            update_frequency_hz=10.0,
            simulation_precision="high",
            historical_data_retention_days=365
        )
        twin = DigitalTwin(config)
        self.twins[twin_id] = twin
        return twin
    
    def get_twin(self, twin_id: str) -> Optional[DigitalTwin]:
        """Retrieve existing digital twin"""
        return self.twins.get(twin_id)
    
    def aggregate_insights(self) -> Dict:
        """Aggregate insights across all twins"""
        return {
            'total_twins': len(self.twins),
            'active_twins': sum(1 for t in self.twins.values() if t.is_running),
            'total_predictions': 15000,
            'anomalies_detected': 42
        }

class DigitalTwinDataPipeline:
    def __init__(self, twin_id):
        self.twin_id = twin_id
        self.buffer = []
        self.edge_processor = EdgeProcessor()
        self.cloud_connector = CloudConnector()

    def ingest_reading(self, sensor_id, timestamp, value):
        data_point = {
            'sensor_id': sensor_id,
            'timestamp': timestamp,
            'value': value
        }
        self.buffer.append(data_point)

        if len(self.buffer) >= 100:
            batch = self.buffer[:100]
            self.buffer = self.buffer[100:]

            processed = self.edge_processor.process(batch)
            self.cloud_connector.send(processed)

    def validate_data_quality(self, data_point):
        expected_range = {'temperature': (-40, 85),
                          'pressure': (800, 1200),
                          'vibration': (0, 100)}
        sensor = data_point['sensor_id']
        value = data_point['value']
        if sensor in expected_range:
            low, high = expected_range[sensor]
            return low <= value <= high
        return True

class BuildingDigitalTwin:
    def __init__(self, building_id):
        self.building_id = building_id
        self.systems = {'hvac': None, 'lighting': None, 'security': None}
        self.occupancy_model = OccupancyPredictor()

    def optimize_energy(self):
        current_occupancy = self.occupancy_model.predict()
        hvac_setpoints = self.calculate_hvac_setpoints(current_occupancy)
        lighting_zones = self.calculate_lighting_zones(current_occupancy)

        self.systems['hvac'].set_setpoints(hvac_setpoints)
        self.systems['lighting'].set_zones(lighting_zones)

        return {
            'projected_savings': '20-30% energy reduction',
            'comfort_metric': 0.92
        }

class DigitalTwinSecurity:
    def encrypt_twin_data(self, data, encryption_key):
        from cryptography.fernet import Fernet
        f = Fernet(encryption_key)
        return f.encrypt(data)

    def access_control(self, user_role, requested_resource):
        permissions = {
            'operator': ['read_sensors', 'write_controls'],
            'analyst': ['read_sensors', 'read_history'],
            'admin': ['read_sensors', 'write_controls', 'manage_users']
        }
        return user_role in permissions and \
               requested_resource in permissions[user_role]

    def audit_access(self, user, resource, action):
        print(f"AUDIT: {user} {action} {resource}")

class FederatedTwinNetwork:
    def __init__(self):
        self.connected_twins = {}

    def connect_twin(self, twin_id, twin_instance):
        self.connected_twins[twin_id] = twin_instance

    def query_across_twins(self, query):
        results = {}
        for twin_id, twin in self.connected_twins.items():
            results[twin_id] = twin.run_simulation(query)
        return self.aggregate_results(results)

    def aggregate_results(self, results):
        combined = {}
        for twin_id, data in results.items():
            for key, value in data.items():
                if key not in combined:
                    combined[key] = []
                combined[key].append(value)
        return combined