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 this sensor
    for element in self.elements.values():
        for prop_name, bound_sensor in element.sensor_bindings.items():
            if bound_sensor == sensor_id and not prop_name.endswith('_transform'):
                # Apply transform if exists
                transform_key = f'{prop_name}_transform'
                if transform_key in element.sensor_bindings:
                    transform = element.sensor_bindings[transform_key]
                    value = transform(value)

                element.update_property(prop_name, value, unit, DataSource.IOT)

                # Create event
                self._emit_event('property_update', element.element_id, {
                    'property': prop_name,
                    'value': value,
                    'sensor_id': sensor_id
                }, DataSource.IOT)

def update_status(self, element_id: str, status: ElementStatus,
                 source: DataSource = DataSource.FIELD):
    """Update element construction status"""
    element = self.elements.get(element_id)
    if not element:
        return

    old_status = element.status
    element.status = status
    element.updated_at = datetime.now()

    self._emit_event('status_change', element_id, {
        'old_status': old_status.value,
        'new_status': status.value
    }, source)

def _emit_event(self, event_type: str, element_id: str,
               data: Dict, source: DataSource):
    """Emit twin event"""
    event = TwinEvent(
        event_id=f"EVT-{len(self.events):06d}",
        event_type=event_type,
        element_id=element_id,
        timestamp=datetime.now(),
        data=data,
        source=source
    )

    self.events.append(event)

    # Notify handlers
    for handler in self.event_handlers.get(event_type, []):
        try:
            handler(event)
        except Exception as e:
            print(f"Event handler error: {e}")

def on_event(self, event_type: str, handler: Callable):
    """Register event handler"""
    if event_type not in self.event_handlers:
        self.event_handlers[event_type] = []
    self.event_handlers[event_type].append(handler)

def get_element_snapshot(self, element_id: str) -> Dict:
    """Get current state snapshot of element"""
    element = self.elements.get(element_id)
    if not element:
        return {}

    return {
        'element_id': element.element_id,
        'ifc_guid': element.ifc_guid,
        'type': element.element_type,
        'name': element.name,
        'status': element.status.value,
        'properties': {
            name: {
                'value': prop.value,
                'unit': prop.unit,
                'timestamp': prop.timestamp.isoformat(),
                'source': prop.source.value,
                'confidence': prop.confidence
            }
            for name, prop in element.properties.items()
        },
        'updated_at': element.updated_at.isoformat()
    }

def get_project_snapshot(self) -> Dict:
    """Get full project state snapshot"""
    status_counts = {}
    for elem in self.elements.values():
        status = elem.status.value
        status_counts[status] = status_counts.get(status, 0) + 1

    return {
        'project_id': self.project_id,
        'project_name': self.project_name,
        'timestamp': datetime.now().isoformat(),
        'element_count': len(self.elements),
        'status_summary': status_counts,
        'recent_events': [
            {
                'event_id': e.event_id,
                'type': e.event_type,
                'element': e.element_id,
                'timestamp': e.timestamp.isoformat()
            }
            for e in self.events[-10:]
        ]
    }

Real-Time Synchronization

import asyncio from typing import Dict, List, Callable import websockets import json

class TwinSynchronizer: """Real-time twin synchronization service"""

def __init__(self, twin: DigitalTwinCore):
    self.twin = twin
    self.subscribers: Dict[str, List[websockets.WebSocketServerProtocol]] = {}
    self.sync_interval = 1.0  # seconds

async def start_server(self, host: str = 'localhost', port: int = 8765):
    """Start WebSocket server for real-time updates"""
    async with websockets.serve(self._handle_connection, host, port):
        print(f"Twin sync server running on ws://{host}:{port}")
        await asyncio.Future()  # Run forever

async def _handle_connection(self, websocket, path):
    """Handle WebSocket connection"""
    try:
        async for message in websocket:
            data = json.loads(message)
            await self._process_message(websocket, data)
    except websockets.exceptions.ConnectionClosed:
        pass
    finally:
        # Remove from all subscriptions
        for element_id in list(self.subscribers.keys()):
            if websocket in self.subscribers[element_id]:
                self.subscribers[element_id].remove(websocket)

async def _process_message(self, websocket, data: Dict):
    """Process incoming message"""
    msg_type = data.get('type')

    if msg_type == 'subscribe':
        element_id = data.get('element_id', '*')
        if element_id not in self.subscribers:
            self.subscribers[element_id] = []
        self.subscribers[element_id].append(websocket)

        # Send current state
        if element_id == '*':
            state = self.twin.get_project_snapshot()
        else:
            state = self.twin.get_element_snapshot(element_id)

        await websocket.send(json.dumps({
            'type': 'state',
            'data': state
        }))

    elif msg_type == 'update':
        # Handle incoming update from field
        element_id = data.get('element_id')
        updates = data.get('updates', {})

        for prop_name, value in updates.items():
            element = self.twin.elements.get(element_id)
            if element:
                element.update_property(prop_name, value, source=DataSource.FIELD)

        # Broadcast update
        await self._broadcast_update(element_id)

    elif msg_type == 'status_update':
        element_id = data.get('element_id')
        status = ElementStatus(data.get('status'))
        self.twin.update_status(element_id, status, DataSource.FIELD)
        await self._broadcast_update(element_id)

async def _broadcast_update(self, element_id: str):
    """Broadcast element update to subscribers"""
    state = self.twin.get_element_snapshot(element_id)
    message = json.dumps({
        'type': 'update',
        'element_id': element_id,
        'data': state
    })

    # Send to specific element subscribers
    for ws in self.subscribers.get(element_id, []):
        try:
            await ws.send(message)
        except:
            pass

    # Send to wildcard subscribers
    for ws in self.subscribers.get('*', []):
        try:
            await ws.send(message)
        except:
            pass

def setup_sensor_integration(self, mqtt_client):
    """Setup MQTT integration for IoT sensors"""
    def on_message(client, userdata, msg):
        try:
            data = json.loads(msg.payload.decode())
            self.twin.process_sensor_update(
                sensor_id=data.get('sensor_id'),
                value=data.get('value'),
                unit=data.get('unit'),
                timestamp=datetime.fromisoformat(data.get('timestamp'))
            )
        except Exception as e:
            print(f"Sensor message error: {e}")

    mqtt_client.on_message = on_message
    mqtt_client.subscribe("sensors/#")

Schedule Integration

from datetime import date, datetime

@dataclass class ScheduleActivity: activity_id: str name: str planned_start: date planned_end: date actual_start: Optional[date] = None actual_end: Optional[date] = None percent_complete: float = 0 element_ids: List[str] = field(default_factory=list)

class ScheduleTwinIntegrator: """Integrate schedule with digital twin"""

def __init__(self, twin: DigitalTwinCore):
    self.twin = twin
    self.activities: Dict[str, ScheduleActivity] = {}

def import_schedule(self, schedule_data: List[Dict]):
    """Import schedule activities"""
    for act_data in schedule_data:
        activity = ScheduleActivity(
            activity_id=act_data['id'],
            name=act_data['name'],
            planned_start=date.fromisoformat(act_data['start']),
            planned_end=date.fromisoformat(act_data['end']),
            element_ids=act_data.get('elements', [])
        )
        self.activities[activity.activity_id] = activity

        # Link elements to activity
        for elem_id in activity.element_ids:
            if elem_id in self.twin.elements:
                self.twin.elements[elem_id].schedule_activity_id = activity.activity_id

def update_activity_progress(self, activity_id: str, percent_complete: float,
                            actual_start: date = None, actual_end: date = None):
    """Update activity progress"""
    activity = self.activities.get(activity_id)
    if not activity:
        return

    activity.percent_complete = percent_complete
    if actual_start:
        activity.actual_start = actual_start
    if actual_end:
        activity.actual_end = actual_end

    # Update linked elements
    status = self._determine_status(percent_complete)
    for elem_id in activity.element_ids:
        self.twin.update_status(elem_id, status, DataSource.SCHEDULE)

def _determine_status(self, percent: float) -> ElementStatus:
    """Determine element status from progress percentage"""
    if percent == 0:
        return ElementStatus.PLANNED
    elif percent < 100:
        return ElementStatus.IN_PROGRESS
    else:
        return ElementStatus.COMPLETED

def calculate_schedule_variance(self) -> Dict:
    """Calculate schedule performance"""
    today = date.today()
    variances = []

    for activity in self.activities.values():
        planned_duration = (activity.planned_end - activity.planned_start).days
        if planned_duration == 0:
            continue

        if activity.actual_start:
            start_variance = (activity.actual_start - activity.planned_start).days
        else:
            start_variance = None

        if activity.actual_end:
            end_variance = (activity.actual_end - activity.planned_end).days
        else:
            end_variance = None

        # Calculate expected progress
        if today >= activity.planned_end:
            expected_progress = 100
        elif today <= activity.planned_start:
            expected_progress = 0
        else:
            elapsed = (today - activity.planned_start).days
            expected_progress = elapsed / planned_duration * 100

        progress_variance = activity.percent_complete - expected_progress

        variances.append({
            'activity_id': activity.activity_id,
            'name': activity.name,
            'planned_start': activity.planned_start.isoformat(),
            'planned_end': activity.planned_end.isoformat(),
            'actual_progress': activity.percent_complete,
            'expected_progress': expected_progress,
            'progress_variance': progress_variance,
            'start_variance_days': start_variance,
            'status': 'ahead' if progress_variance > 0 else 'behind' if progress_variance < -5 else 'on_track'
        })

    return {
        'date': today.isoformat(),
        'activities': variances,
        'on_track_count': sum(1 for v in variances if v['status'] == 'on_track'),
        'ahead_count': sum(1 for v in variances if v['status'] == 'ahead'),
        'behind_count': sum(1 for v in variances if v['status'] == 'behind')
    }

Anomaly Detection

import numpy as np from collections import deque

class TwinAnomalyDetector: """Detect anomalies in digital twin data"""

def __init__(self, twin: DigitalTwinCore, window_size: int = 100):
    self.twin = twin
    self.window_size = window_size
    self.value_windows: Dict[str, deque] = {}  # key: element_id:property
    self.thresholds: Dict[str, Dict] = {}

def set_threshold(self, element_id: str, property_name: str,
                 min_value: float = None, max_value: float = None,
                 std_multiplier: float = 3.0):
    """Set threshold for property monitoring"""
    key = f"{element_id}:{property_name}"
    self.thresholds[key] = {
        'min': min_value,
        'max': max_value,
        'std_multiplier': std_multiplier
    }

def check_value(self, element_id: str, property_name: str, value: float) -> Dict:
    """Check value for anomalies"""
    key = f"{element_id}:{property_name}"

    # Initialize window if needed
    if key not in self.value_windows:
        self.value_windows[key] = deque(maxlen=self.window_size)

    window = self.value_windows[key]

    anomaly = {
        'is_anomaly': False,
        'type': None,
        'severity': 'normal',
        'details': {}
    }

    # Check against thresholds
    if key in self.thresholds:
        thresh = self.thresholds[key]

        if thresh['min'] is not None and value < thresh['min']:
            anomaly['is_anomaly'] = True
            anomaly['type'] = 'below_minimum'
            anomaly['severity'] = 'warning'
            anomaly['details']['threshold'] = thresh['min']
            anomaly['details']['value'] = value

        if thresh['max'] is not None and value > thresh['max']:
            anomaly['is_anomaly'] = True
            anomaly['type'] = 'above_maximum'
            anomaly['severity'] = 'warning'
            anomaly['details']['threshold'] = thresh['max']
            anomaly['details']['value'] = value

    # Statistical anomaly detection
    if len(window) >= 10:
        mean = np.mean(window)
        std = np.std(window)

        if std > 0:
            z_score = abs(value - mean) / std
            if z_score > 3:
                anomaly['is_anomaly'] = True
                anomaly['type'] = 'statistical_outlier'
                anomaly['severity'] = 'critical' if z_score > 5 else 'warning'
                anomaly['details']['z_score'] = z_score
                anomaly['details']['mean'] = mean
                anomaly['details']['std'] = std

    # Add to window
    window.append(value)

    return anomaly

def monitor_element(self, element_id: str) -> List[Dict]:
    """Monitor all properties of an element for anomalies"""
    element = self.twin.elements.get(element_id)
    if not element:
        return []

    anomalies = []
    for prop_name, prop in element.properties.items():
        if isinstance(prop.value, (int, float)):
            result = self.check_value(element_id, prop_name, prop.value)
            if result['is_anomaly']:
                result['element_id'] = element_id
                result['property'] = prop_name
                result['timestamp'] = prop.timestamp.isoformat()
                anomalies.append(result)

    return anomalies

Quick Reference

Data Source Update Frequency Reliability

BIM Model On change High

IoT Sensors Real-time Variable

Schedule Daily High

Field Updates Event-driven Medium

Drone Surveys Periodic High

Resources

• Digital Twin Consortium: https://www.digitaltwinconsortium.org

• IFC/BIM Standards: https://www.buildingsmart.org

• DDC Website: https://datadrivenconstruction.io

Next Steps

• See material-tracking-iot for IoT integration

• See 4d-simulation for schedule visualization

• See bim-validation-pipeline for model validation