progress-monitoring-cv

name	progress-monitoring-cv
description	Monitor construction progress using computer vision. Analyze site photos and drone imagery to track work completion, detect safety issues, and compare against BIM models.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"🚀","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}

Progress Monitoring with Computer Vision

Overview

This skill implements computer vision for construction progress monitoring. Analyze site images automatically to track completion, detect hazards, and compare physical progress against planned work.

Applications:

• Progress percentage estimation
• Safety compliance detection (PPE, barriers)
• As-built vs BIM comparison
• Material and equipment tracking
• Quality defect detection

Quick Start

import cv2
import numpy as np
from PIL import Image
import torch
from torchvision import models, transforms

# Load pre-trained model for construction scene analysis
model = models.resnet50(pretrained=True)
model.eval()

# Preprocess image
transform = transforms.Compose([
    transforms.Resize(256),
    transforms.CenterCrop(224),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406],
                        std=[0.229, 0.224, 0.225])
])

# Load site photo
img = Image.open("site_photo.jpg")
input_tensor = transform(img).unsqueeze(0)

# Analyze
with torch.no_grad():
    output = model(input_tensor)

print("Image analyzed successfully")

Progress Detection System

Core Progress Analyzer

import cv2
import numpy as np
from PIL import Image
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import torch
from torchvision import models, transforms
from torchvision.models.detection import fasterrcnn_resnet50_fpn

class ConstructionPhase(Enum):
    EXCAVATION = "excavation"
    FOUNDATION = "foundation"
    STRUCTURE = "structure"
    ENCLOSURE = "enclosure"
    MEP_ROUGH = "mep_rough"
    FINISHES = "finishes"
    COMPLETE = "complete"

@dataclass
class ProgressReport:
    timestamp: str
    image_path: str
    detected_phase: ConstructionPhase
    estimated_progress: float
    detected_elements: List[Dict]
    safety_observations: List[Dict]
    quality_issues: List[Dict]
    comparison_to_plan: Optional[float]

class ConstructionProgressAnalyzer:
    """Analyze construction progress from images"""

    def ():
        .device = torch.device(  use_gpu  torch.cuda.is_available()  )

        
        .detector = fasterrcnn_resnet50_fpn(pretrained=)
        .detector.to(.device)
        .detector.()

        
        .transform = transforms.Compose([
            transforms.ToTensor()
        ])

        
        .construction_labels = {
            : [, , ],
            : [, ],
            : [, , ],
            : [, ],
            : [, ],
            : [, ],
            : [, , ],
            : [, ]
        }

     () -> ProgressReport:
        
        
        img = Image.(image_path).convert()
        img_tensor = .transform(img).to(.device)

        
         torch.no_grad():
            predictions = .detector([img_tensor])

        
        detected_elements = ._process_detections(predictions[])

        
        phase = ._estimate_phase(detected_elements, img)
        progress = ._estimate_progress(phase, detected_elements)

        
        safety_obs = ._analyze_safety(img, detected_elements)

        
        quality_issues = ._check_quality(img)

         ProgressReport(
            timestamp=._get_timestamp(),
            image_path=image_path,
            detected_phase=phase,
            estimated_progress=progress,
            detected_elements=detected_elements,
            safety_observations=safety_obs,
            quality_issues=quality_issues,
            comparison_to_plan=
        )

     () -> []:
        
        elements = []

        boxes = predictions[].cpu().numpy()
        labels = predictions[].cpu().numpy()
        scores = predictions[].cpu().numpy()

         box, label, score  (boxes, labels, scores):
             score > :  
                elements.append({
                    : box.tolist(),
                    : label,
                    : (score),
                    : (box[] - box[]) * (box[] - box[])
                })

         elements

     () -> ConstructionPhase:
        
        
        img_array = np.array(img)

        
        hsv = cv2.cvtColor(img_array, cv2.COLOR_RGB2HSV)

        
        earth_mask = cv2.inRange(hsv, (, , ), (, , ))
        earth_ratio = np.(earth_mask > ) / earth_mask.size

        
        gray_mask = cv2.inRange(hsv, (, , ), (, , ))
        gray_ratio = np.(gray_mask > ) / gray_mask.size

        
        steel_mask = cv2.inRange(hsv, (, , ), (, , ))
        steel_ratio = np.(steel_mask > ) / steel_mask.size

        
         earth_ratio > :
             ConstructionPhase.EXCAVATION
         gray_ratio >   steel_ratio < :
             ConstructionPhase.FOUNDATION
         steel_ratio > :
             ConstructionPhase.STRUCTURE
        :
             ConstructionPhase.ENCLOSURE

     () -> :
        
        phase_base_progress = {
            ConstructionPhase.EXCAVATION: ,
            ConstructionPhase.FOUNDATION: ,
            ConstructionPhase.STRUCTURE: ,
            ConstructionPhase.ENCLOSURE: ,
            ConstructionPhase.MEP_ROUGH: ,
            ConstructionPhase.FINISHES: ,
            ConstructionPhase.COMPLETE: 
        }

        base = phase_base_progress.get(phase, )

        
        element_count = (elements)
        adjustment = (element_count * , )

         (base + adjustment, )

     () -> []:
        
        observations = []
        img_array = np.array(img)

        
        hsv = cv2.cvtColor(img_array, cv2.COLOR_RGB2HSV)

        
        orange_mask = cv2.inRange(hsv, (, , ), (, , ))
        orange_pixels = np.(orange_mask > )

        
        yellow_mask = cv2.inRange(hsv, (, , ), (, , ))
        yellow_pixels = np.(yellow_mask > )

         orange_pixels + yellow_pixels < :  
            observations.append({
                : ,
                : ,
                : 
            })

        
        worker_count = (  e  elements  e.get() == )
         worker_count > :
            observations.append({
                : ,
                : ,
                : 
            })

         observations

     () -> []:
        
        issues = []
        img_array = np.array(img)

        
        gray = cv2.cvtColor(img_array, cv2.COLOR_RGB2GRAY)
        edges = cv2.Canny(gray, , )

        
        edge_density = np.(edges > ) / edges.size

         edge_density > :
            issues.append({
                : ,
                : ,
                : 
            })

         issues

     () -> :
         datetime  datetime
         datetime.now().isoformat()

     () -> :
        
        site_img = cv2.imread(image_path)
        bim_img = cv2.imread(bim_render_path)

        
        target_size = (, )
        site_img = cv2.resize(site_img, target_size)
        bim_img = cv2.resize(bim_img, target_size)

        
        site_gray = cv2.cvtColor(site_img, cv2.COLOR_BGR2GRAY)
        bim_gray = cv2.cvtColor(bim_img, cv2.COLOR_BGR2GRAY)

        
         skimage.metrics  structural_similarity
        similarity, _ = structural_similarity(site_gray, bim_gray, full=)

         similarity

     () -> [ProgressReport]:
        
         [.analyze_image(path)  path  image_paths]

Time-Lapse Analysis

class TimeLapseAnalyzer:
    """Analyze construction progress over time from image series"""

    def __init__(self, analyzer: ConstructionProgressAnalyzer):
        self.analyzer = analyzer
        self.reports: List[ProgressReport] = []

    def add_image(self, image_path: str, date: str):
        """Add image to time series"""
        report = self.analyzer.analyze_image(image_path)
        report.timestamp = date
        self.reports.append(report)

    def get_progress_curve(self) -> pd.DataFrame:
        """Generate progress curve from analyzed images"""
        data = [{
            'date': r.timestamp,
            'phase': r.detected_phase.value,
            'progress': r.estimated_progress,
            'element_count': len(r.detected_elements)
        } for r in self.reports]

        return pd.DataFrame(data).sort_values('date')

    def detect_delays(self, planned_progress: pd.DataFrame) -> List[Dict]:
        """Compare actual vs planned progress"""
        actual = self.get_progress_curve()
        delays = []

        for _, row in actual.iterrows():
            planned_row = planned_progress[
                planned_progress[] == row[]
            ]
              planned_row.empty:
                planned_pct = planned_row.iloc[][]
                actual_pct = row[]

                 actual_pct < planned_pct - :  
                    delays.append({
                        : row[],
                        : planned_pct,
                        : actual_pct,
                        : planned_pct - actual_pct
                    })

         delays

     ():
        
        progress_df = .get_progress_curve()

         pd.ExcelWriter(output_path, engine=)  writer:
            progress_df.to_excel(writer, sheet_name=, index=)

            
            safety_data = []
             r  .reports:
                 obs  r.safety_observations:
                    safety_data.append({
                        : r.timestamp,
                        : obs[],
                        : obs[],
                        : obs[]
                    })
             safety_data:
                pd.DataFrame(safety_data).to_excel(
                    writer, sheet_name=, index=
                )

         output_path

Quick Reference

Analysis Type	Method	Output
Phase Detection	Color analysis + Object detection	Construction phase
Progress %	Element counting + Phase base	Completion percentage
Safety Check	Color detection (PPE) + Worker count	Safety observations
Quality Check	Edge detection + Anomaly detection	Quality issues
BIM Comparison	Structural similarity	Similarity score

Resources

• OpenCV: https://opencv.org
• PyTorch: https://pytorch.org
• YOLO: https://ultralytics.com/yolov8
• DDC Website: https://datadrivenconstruction.io

Next Steps

• See 4d-simulation for schedule comparison
• See data-visualization for progress dashboards
• See risk-assessment-ml for delay prediction