digital-twin-sync

What this skill does

# Digital Twin Synchronization

## Overview

This skill implements digital twin synchronization for construction projects. Connect BIM models with real-time sensor data, progress updates, and field information to create a living digital representation.

**Capabilities:**
- BIM-IoT data binding
- Real-time status updates
- Historical data tracking
- Anomaly detection
- Predictive analytics
- Multi-source data fusion

## Quick Start

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

class ElementStatus(Enum):
    PLANNED = "planned"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    ISSUE = "issue"

@dataclass
class TwinElement:
    element_id: str
    ifc_guid: str
    element_type: str
    status: ElementStatus
    properties: Dict[str, Any] = field(default_factory=dict)
    sensor_bindings: List[str] = field(default_factory=list)
    last_updated: datetime = field(default_factory=datetime.now)

@dataclass
class SensorData:
    sensor_id: str
    value: float
    unit: str
    timestamp: datetime
    quality: float = 1.0

class SimpleTwin:
    """Simple digital twin implementation"""

    def __init__(self, project_id: str):
        self.project_id = project_id
        self.elements: Dict[str, TwinElement] = {}
        self.sensor_data: Dict[str, List[SensorData]] = {}

    def add_element(self, element: TwinElement):
        self.elements[element.element_id] = element

    def bind_sensor(self, element_id: str, sensor_id: str):
        if element_id in self.elements:
            self.elements[element_id].sensor_bindings.append(sensor_id)

    def update_sensor(self, data: SensorData):
        if data.sensor_id not in self.sensor_data:
            self.sensor_data[data.sensor_id] = []
        self.sensor_data[data.sensor_id].append(data)

        # Update linked elements
        for elem in self.elements.values():
            if data.sensor_id in elem.sensor_bindings:
                elem.properties[f'sensor_{data.sensor_id}'] = data.value
                elem.last_updated = data.timestamp

    def get_element_state(self, element_id: str) -> Dict:
        elem = self.elements.get(element_id)
        if not elem:
            return {}

        state = {
            'element_id': elem.element_id,
            'status': elem.status.value,
            'properties': elem.properties,
            'last_updated': elem.last_updated.isoformat()
        }

        # Add latest sensor values
        for sensor_id in elem.sensor_bindings:
            if sensor_id in self.sensor_data and self.sensor_data[sensor_id]:
                latest = self.sensor_data[sensor_id][-1]
                state[f'sensor_{sensor_id}'] = {
                    'value': latest.value,
                    'unit': latest.unit,
                    'timestamp': latest.timestamp.isoformat()
                }

        return state

# Example
twin = SimpleTwin("PROJECT-001")
twin.add_element(TwinElement(
    element_id="WALL-001",
    ifc_guid="2O2Fr$t4X7Zf8NOew3FLOH",
    element_type="IfcWall",
    status=ElementStatus.IN_PROGRESS
))
twin.bind_sensor("WALL-001", "TEMP-001")
twin.update_sensor(SensorData("TEMP-001", 22.5, "°C", datetime.now()))
print(twin.get_element_state("WALL-001"))
```

## Comprehensive Digital Twin System

### Core Twin Model

```python
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Any, Callable
from enum import Enum
import json
import threading
from queue import Queue
import time

class DataSource(Enum):
    BIM = "bim"
    IOT = "iot"
    SCHEDULE = "schedule"
    FIELD = "field"
    DRONE = "drone"
    MANUAL = "manual"

@dataclass
class PropertyValue:
    value: Any
    unit: Optional[str]
    timestamp: datetime
    source: DataSource
    confidence: float = 1.0
    history: List[Dict] = field(default_factory=list)

@dataclass
class DigitalTwinElement:
    element_id: str
    ifc_guid: str
    element_type: str
    name: str
    status: ElementStatus = ElementStatus.PLANNED
    properties: Dict[str, PropertyValue] = field(default_factory=dict)
    sensor_bindings: Dict[str, str] = field(default_factory=dict)  # property_name -> sensor_id
    geometry_ref: Optional[str] = None
    parent_id: Optional[str] = None
    children_ids: List[str] = field(default_factory=list)
    schedule_activity_id: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

    def update_property(self, name: str, value: Any, unit: str = None,
                       source: DataSource = DataSource.MANUAL,
                       confidence: float = 1.0):
        """Update property with history tracking"""
        now = datetime.now()

        if name in self.properties:
            # Store previous value in history
            prev = self.properties[name]
            prev.history.append({
                'value': prev.value,
                'timestamp': prev.timestamp.isoformat(),
                'source': prev.source.value
            })
            # Keep only last 100 values
            prev.history = prev.history[-100:]

            prev.value = value
            prev.unit = unit or prev.unit
            prev.timestamp = now
            prev.source = source
            prev.confidence = confidence
        else:
            self.properties[name] = PropertyValue(
                value=value,
                unit=unit,
                timestamp=now,
                source=source,
                confidence=confidence
            )

        self.updated_at = now

@dataclass
class TwinEvent:
    event_id: str
    event_type: str  # property_update, status_change, alert, etc.
    element_id: str
    timestamp: datetime
    data: Dict
    source: DataSource

class DigitalTwinCore:
    """Core digital twin management system"""

    def __init__(self, project_id: str, project_name: str):
        self.project_id = project_id
        self.project_name = project_name
        self.elements: Dict[str, DigitalTwinElement] = {}
        self.events: List[TwinEvent] = []
        self.event_handlers: Dict[str, List[Callable]] = {}
        self.update_queue: Queue = Queue()
        self._running = False

    def import_from_ifc(self, ifc_data: List[Dict]):
        """Import elements from IFC data"""
        for elem_data in ifc_data:
            element = DigitalTwinElement(
                element_id=elem_data.get('id', f"ELEM-{len(self.elements)}"),
                ifc_guid=elem_data.get('guid', ''),
                element_type=elem_data.get('type', 'IfcBuildingElement'),
                name=elem_data.get('name', 'Unknown'),
                geometry_ref=elem_data.get('geometry_ref')
            )

            # Import properties
            for prop_name, prop_value in elem_data.get('properties', {}).items():
                element.update_property(
                    prop_name,
                    prop_value.get('value'),
                    prop_value.get('unit'),
                    DataSource.BIM
                )

            self.elements[element.element_id] = element

    def bind_sensor_to_property(self, element_id: str, property_name: str,
                               sensor_id: str, transform: Callable = None):
        """Bind IoT sensor to element property"""
        element = self.elements.get(element_id)
        if element:
            element.sensor_bindings[property_name] = sensor_id
            # Store transform function if needed
            if transform:
                element.sensor_bindings[f'{property_name}_transform'] = transform

    def process_sensor_update(self, sensor_id: str, value: float,
                             unit: str, timestamp: datetime = None):
        """Process incoming sensor data"""
        timestamp = timestamp or datetime.now()

        # Find all elements bound to