digital-twin-sync

name	digital-twin-sync
description	Synchronize construction digital twins with real-time data. Connect BIM models with IoT sensors, progress updates, and field data for live project visualization and monitoring.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

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[, ] = field(default_factory=)  
    geometry_ref: [] = 
    parent_id: [] = 
    children_ids: [] = field(default_factory=)
    schedule_activity_id: [] = 
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

     ():
        
        now = datetime.now()

         name  .properties:
            
            prev = .properties[name]
            prev.history.append({
                : prev.value,
                : prev.timestamp.isoformat(),
                : prev.source.value
            })
            
            prev.history = prev.history[-:]

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

        .updated_at = now


 :
    event_id: 
    event_type:   
    element_id: 
    timestamp: datetime
    data: 
    source: DataSource

 :
    

     ():
        .project_id = project_id
        .project_name = project_name
        .elements: [, DigitalTwinElement] = {}
        .events: [TwinEvent] = []
        .event_handlers: [, []] = {}
        .update_queue: Queue = Queue()
        ._running = 

     ():
        
         elem_data  ifc_data:
            element = DigitalTwinElement(
                element_id=elem_data.get(, ),
                ifc_guid=elem_data.get(, ),
                element_type=elem_data.get(, ),
                name=elem_data.get(, ),
                geometry_ref=elem_data.get()
            )

            
             prop_name, prop_value  elem_data.get(, {}).items():
                element.update_property(
                    prop_name,
                    prop_value.get(),
                    prop_value.get(),
                    DataSource.BIM
                )

            .elements[element.element_id] = element

     ():
        
        element = .elements.get(element_id)
         element:
            element.sensor_bindings[property_name] = sensor_id
            
             transform:
                element.sensor_bindings[] = transform

     ():
        
        timestamp = timestamp  datetime.now()

        
         element  .elements.values():
             prop_name, bound_sensor  element.sensor_bindings.items():
                 bound_sensor == sensor_id   prop_name.endswith():
                    
                    transform_key = 
                     transform_key  element.sensor_bindings:
                        transform = element.sensor_bindings[transform_key]
                        value = transform(value)

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

                    
                    ._emit_event(, element.element_id, {
                        : prop_name,
                        : value,
                        : sensor_id
                    }, DataSource.IOT)

     ():
        
        element = .elements.get(element_id)
          element:
            

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

        ._emit_event(, element_id, {
            : old_status.value,
            : status.value
        }, source)

     ():
        
        event = TwinEvent(
            event_id=,
            event_type=event_type,
            element_id=element_id,
            timestamp=datetime.now(),
            data=data,
            source=source
        )

        .events.append(event)

        
         handler  .event_handlers.get(event_type, []):
            :
                handler(event)
             Exception  e:
                ()

     ():
        
         event_type   .event_handlers:
            .event_handlers[event_type] = []
        .event_handlers[event_type].append(handler)

     () -> :
        
        element = .elements.get(element_id)
          element:
             {}

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

     () -> :
        
        status_counts = {}
         elem  .elements.values():
            status = elem.status.value
            status_counts[status] = status_counts.get(status, ) + 

         {
            : .project_id,
            : .project_name,
            : datetime.now().isoformat(),
            : (.elements),
            : status_counts,
            : [
                {
                    : e.event_id,
                    : e.event_type,
                    : e.element_id,
                    : e.timestamp.isoformat()
                }
                 e  .events[-:]
            ]
        }

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  websocket:
                data = json.loads(message)
                 ._process_message(websocket, data)
         websockets.exceptions.ConnectionClosed:
            
        :
            
             element_id  (.subscribers.keys()):
                 websocket  .subscribers[element_id]:
                    .subscribers[element_id].remove(websocket)

      ():
        
        msg_type = data.get()

         msg_type == :
            element_id = data.get(, )
             element_id   .subscribers:
                .subscribers[element_id] = []
            .subscribers[element_id].append(websocket)

            
             element_id == :
                state = .twin.get_project_snapshot()
            :
                state = .twin.get_element_snapshot(element_id)

             websocket.send(json.dumps({
                : ,
                : state
            }))

         msg_type == :
            
            element_id = data.get()
            updates = data.get(, {})

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

            
             ._broadcast_update(element_id)

         msg_type == :
            element_id = data.get()
            status = ElementStatus(data.get())
            .twin.update_status(element_id, status, DataSource.FIELD)
             ._broadcast_update(element_id)

      ():
        
        state = .twin.get_element_snapshot(element_id)
        message = json.dumps({
            : ,
            : element_id,
            : state
        })

        
         ws  .subscribers.get(element_id, []):
            :
                 ws.send(message)
            :
                

        
         ws  .subscribers.get(, []):
            :
                 ws.send(message)
            :
                

     ():
        
         ():
            :
                data = json.loads(msg.payload.decode())
                .twin.process_sensor_update(
                    sensor_id=data.get(),
                    value=data.get(),
                    unit=data.get(),
                    timestamp=datetime.fromisoformat(data.get())
                )
             Exception  e:
                ()

        mqtt_client.on_message = on_message
        mqtt_client.subscribe()

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  activity.element_ids:
                 elem_id  .twin.elements:
                    .twin.elements[elem_id].schedule_activity_id = activity.activity_id

     ():
        
        activity = .activities.get(activity_id)
          activity:
            

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

        
        status = ._determine_status(percent_complete)
         elem_id  activity.element_ids:
            .twin.update_status(elem_id, status, DataSource.SCHEDULE)

     () -> ElementStatus:
        
         percent == :
             ElementStatus.PLANNED
         percent < :
             ElementStatus.IN_PROGRESS
        :
             ElementStatus.COMPLETED

     () -> :
        
        today = date.today()
        variances = []

         activity  .activities.values():
            planned_duration = (activity.planned_end - activity.planned_start).days
             planned_duration == :
                

             activity.actual_start:
                start_variance = (activity.actual_start - activity.planned_start).days
            :
                start_variance = 

             activity.actual_end:
                end_variance = (activity.actual_end - activity.planned_end).days
            :
                end_variance = 

            
             today >= activity.planned_end:
                expected_progress = 
             today <= activity.planned_start:
                expected_progress = 
            :
                elapsed = (today - activity.planned_start).days
                expected_progress = elapsed / planned_duration * 

            progress_variance = activity.percent_complete - expected_progress

            variances.append({
                : activity.activity_id,
                : activity.name,
                : activity.planned_start.isoformat(),
                : activity.planned_end.isoformat(),
                : activity.percent_complete,
                : expected_progress,
                : progress_variance,
                : start_variance,
                :   progress_variance >     progress_variance < -  
            })

         {
            : today.isoformat(),
            : variances,
            : (  v  variances  v[] == ),
            : (  v  variances  v[] == ),
            : (  v  variances  v[] == )
        }

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) -> :
        
        key = 

        
         key   .value_windows:
            .value_windows[key] = deque(maxlen=.window_size)

        window = .value_windows[key]

        anomaly = {
            : ,
            : ,
            : ,
            : {}
        }

        
         key  .thresholds:
            thresh = .thresholds[key]

             thresh[]     value < thresh[]:
                anomaly[] = 
                anomaly[] = 
                anomaly[] = 
                anomaly[][] = thresh[]
                anomaly[][] = value

             thresh[]     value > thresh[]:
                anomaly[] = 
                anomaly[] = 
                anomaly[] = 
                anomaly[][] = thresh[]
                anomaly[][] = value

        
         (window) >= :
            mean = np.mean(window)
            std = np.std(window)

             std > :
                z_score = (value - mean) / std
                 z_score > :
                    anomaly[] = 
                    anomaly[] = 
                    anomaly[] =   z_score >   
                    anomaly[][] = z_score
                    anomaly[][] = mean
                    anomaly[][] = std

        
        window.append(value)

         anomaly

     () -> []:
        
        element = .twin.elements.get(element_id)
          element:
             []

        anomalies = []
         prop_name, prop  element.properties.items():
             (prop.value, (, )):
                result = .check_value(element_id, prop_name, prop.value)
                 result[]:
                    result[] = element_id
                    result[] = prop_name
                    result[] = prop.timestamp.isoformat()
                    anomalies.append(result)

         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