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
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
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