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 __init__(self, use_gpu: bool = True):
        self.device = torch.device('cuda' if use_gpu and torch.cuda.is_available() else 'cpu')

        # Load detection model
        self.detector = fasterrcnn_resnet50_fpn(pretrained=True)
        self.detector.to(self.device)
        self.detector.eval()

        # Image transform
        self.transform = transforms.Compose([
            transforms.ToTensor()
        ])

        # Construction element labels (would need fine-tuned model in production)
        self.construction_labels = {
            'column': ['column', 'pillar', 'post'],
            'beam': ['beam', 'girder'],
            'slab': ['floor', 'slab', 'deck'],
            'wall': ['wall', 'partition'],
            'scaffold': ['scaffold', 'scaffolding'],
            'crane': ['crane', 'tower crane'],
            'equipment': ['excavator', 'loader', 'truck'],
            'worker': ['person', 'worker']
        }

    def analyze_image(self, image_path: str) -> ProgressReport:
        """Analyze a single construction site image"""
        # Load image
        img = Image.open(image_path).convert('RGB')
        img_tensor = self.transform(img).to(self.device)

        # Run detection
        with torch.no_grad():
            predictions = self.detector([img_tensor])

        # Process detections
        detected_elements = self._process_detections(predictions[0])

        # Estimate phase and progress
        phase = self._estimate_phase(detected_elements, img)
        progress = self._estimate_progress(phase, detected_elements)

        # Safety analysis
        safety_obs = self._analyze_safety(img, detected_elements)

        # Quality check (simplified)
        quality_issues = self._check_quality(img)

        return ProgressReport(
            timestamp=self._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=None
        )

    def _process_detections(self, predictions: Dict) -> List[Dict]:
        """Process model predictions into detected elements"""
        elements = []

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

        for box, label, score in zip(boxes, labels, scores):
            if score > 0.5:  # Confidence threshold
                elements.append({
                    'box': box.tolist(),
                    'label': label,
                    'score': float(score),
                    'area': (box[2] - box[0]) * (box[3] - box[1])
                })

        return elements

    def _estimate_phase(self, elements: List[Dict], img: Image) -> ConstructionPhase:
        """Estimate construction phase from detected elements"""
        # Convert to numpy for color analysis
        img_array = np.array(img)

        # Color-based phase estimation (simplified)
        hsv = cv2.cvtColor(img_array, cv2.COLOR_RGB2HSV)

        # Brown/earth tones indicate excavation
        earth_mask = cv2.inRange(hsv, (10, 50, 50), (30, 255, 255))
        earth_ratio = np.sum(earth_mask > 0) / earth_mask.size

        # Gray tones indicate concrete
        gray_mask = cv2.inRange(hsv, (0, 0, 50), (180, 50, 200))
        gray_ratio = np.sum(gray_mask > 0) / gray_mask.size

        # Steel colors
        steel_mask = cv2.inRange(hsv, (0, 0, 100), (180, 30, 255))
        steel_ratio = np.sum(steel_mask > 0) / steel_mask.size

        # Simple phase logic
        if earth_ratio > 0.3:
            return ConstructionPhase.EXCAVATION
        elif gray_ratio > 0.2 and steel_ratio < 0.1:
            return ConstructionPhase.FOUNDATION
        elif steel_ratio > 0.1:
            return ConstructionPhase.STRUCTURE
        else:
            return ConstructionPhase.ENCLOSURE

    def _estimate_progress(self, phase: ConstructionPhase,
                          elements: List[Dict]) -> float:
        """Estimate progress percentage within phase"""
        phase_base_progress = {
            ConstructionPhase.EXCAVATION: 5,
            ConstructionPhase.FOUNDATION: 15,
            ConstructionPhase.STRUCTURE: 35,
            ConstructionPhase.ENCLOSURE: 60,
            ConstructionPhase.MEP_ROUGH: 75,
            ConstructionPhase.FINISHES: 90,
            ConstructionPhase.COMPLETE: 100
        }

        base = phase_base_progress.get(phase, 0)

        # Adjust based on detected elements
        element_count = len(elements)
        adjustment = min(element_count * 0.5, 10)

        return min(base + adjustment, 100)

    def _analyze_safety(self, img: Image, elements: List[Dict]) -> List[Dict]:
        """Analyze safety compliance"""
        observations = []
        img_array = np.array(img)

        # Check for safety vest colors (orange, yellow, green)
        hsv = cv2.cvtColor(img_array, cv2.COLOR_RGB2HSV)

        # Orange vest detection
        orange_mask = cv2.inRange(hsv, (10, 100, 100), (25, 255, 255))
        orange_pixels = np.sum(orange_mask > 0)

        # Yellow vest detection
        yellow_mask = cv2.inRange(hsv, (25, 100, 100), (35, 255, 255))
        yellow_pixels = np.sum(yellow_mask > 0)

        if orange_pixels + yellow_pixels < 100:  # Threshold
            observations.append({
                'type': 'PPE_VISIBILITY',
                'severity': 'Medium',
                'message': 'Limited high-visibility clothing detected'
            })

        # Check for workers detected
        worker_count = sum(1 for e in elements if e.get('label') == 1)
        if worker_count > 0:
            observations.append({
                'type': 'WORKER_COUNT',
                'severity': 'Info',
                'message': f'{worker_count} wo