SKILL.md
Construction Expert
Expert guidance for construction management, project planning, Building Information Modeling (BIM), safety compliance, and modern construction technology solutions.
Core Concepts
Construction Management
- Project planning and scheduling
- Cost estimation and control
- Resource management
- Quality assurance
- Contract management
- Risk management
- Change order management
Technologies
- Building Information Modeling (BIM)
- Construction management software
- Drone surveying and inspection
- 3D printing and modular construction
- IoT sensors for monitoring
- Augmented reality for visualization
- Construction robotics
Standards and Regulations
- OSHA safety regulations
- Building codes (IBC, IRC)
- AIA contracts and standards
- LEED certification
- ISO 19650 (BIM standards)
- CSI MasterFormat
- Environmental regulations
Project Management System
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Optional, Dict
from decimal import Decimal
from enum import Enum
class ProjectPhase(Enum):
PRE_CONSTRUCTION = "pre_construction"
SITE_PREPARATION = "site_preparation"
FOUNDATION = "foundation"
FRAMING = "framing"
MEP = "mep" # Mechanical, Electrical, Plumbing
INTERIOR = "interior"
EXTERIOR = "exterior"
FINAL = "final"
CLOSEOUT = "closeout"
class TaskStatus(Enum):
NOT_STARTED = "not_started"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
DELAYED = "delayed"
ON_HOLD = "on_hold"
@dataclass
class ConstructionProject:
"""Construction project information"""
project_id: str
project_name: str
location: dict
project_type: str # 'residential', 'commercial', 'industrial'
owner: str
general_contractor: str
start_date: datetime
planned_end_date: datetime
actual_end_date: Optional[datetime]
budget: Decimal
current_cost: Decimal
square_footage: float
current_phase: ProjectPhase
@dataclass
class Task:
"""Construction task/activity"""
task_id: str
project_id: str
name: str
description: str
phase: ProjectPhase
status: TaskStatus
assigned_to: str # Subcontractor or crew
planned_start: datetime
planned_end: datetime
actual_start: Optional[datetime]
actual_end: Optional[datetime]
budget: Decimal
actual_cost: Decimal
predecessors: List[str] # Task IDs that must complete first
progress_percent: float
class ConstructionManagementSystem:
"""Construction project management system"""
def __init__(self):
self.projects = {}
self.tasks = {}
self.change_orders = []
self.inspections = []
def create_project_schedule(self, project_id: str, tasks_data: List[dict]) -> dict:
"""Create project schedule using Critical Path Method"""
project = self.projects.get(project_id)
if not project:
return {'error': 'Project not found'}
# Create tasks
tasks = []
for task_data in tasks_data:
task = Task(
task_id=self._generate_task_id(),
project_id=project_id,
name=task_data['name'],
description=task_data.get('description', ''),
phase=ProjectPhase(task_data['phase']),
status=TaskStatus.NOT_STARTED,
assigned_to=task_data['assigned_to'],
planned_start=task_data['planned_start'],
planned_end=task_data['planned_end'],
actual_start=None,
actual_end=None,
budget=Decimal(str(task_data['budget'])),
actual_cost=Decimal('0'),
predecessors=task_data.get('predecessors', []),
progress_percent=0.0
)
tasks.append(task)
self.tasks[task.task_id] = task
# Calculate critical path
critical_path = self._calculate_critical_path(tasks)
# Calculate project duration
if tasks:
project_end = max(t.planned_end for t in tasks)
project_duration = (project_end - project.start_date).days
else:
project_duration = 0
return {
'project_id': project_id,
'total_tasks': len(tasks),
'project_duration_days': project_duration,
'critical_path': [t.task_id for t in critical_path],
'critical_path_duration': sum(
(t.planned_end - t.planned_start).days for t in critical_path
)
}
def _calculate_critical_path(self, tasks: List[Task]) -> List[Task]:
"""Calculate critical path through project network"""
# Simplified critical path calculation
# In production, would use proper CPM algorithm
# Find tasks with no predecessors
start_tasks = [t for t in tasks if not t.predecessors]
# Find longest path through network
critical_path = []
current_tasks = start_tasks
while current_tasks:
# Find task with longest duration
longest_task = max(current_tasks,
key=lambda t: (t.planned_end - t.planned_start).days)
critical_path.append(longest_task)
# Find successors
current_tasks = [
t for t in tasks
if longest_task.task_id in t.predecessors
]
return critical_path
def track_progress(self, project_id: str) -> dict:
"""Track project progress and performance"""
project = self.projects.get(project_id)
if not project:
return {'error': 'Project not found'}
project_tasks = [t for t in self.tasks.values() if t.project_id == project_id]
# Calculate overall progress
if project_tasks:
overall_progress = sum(t.progress_percent for t in project_tasks) / len(project_tasks)
else:
overall_progress = 0.0
# Calculate schedule performance
total_planned_days = (project.planned_end_date - project.start_date).days
elapsed_days = (datetime.now() - project.start_date).days
planned_progress = (elapsed_days / total_planned_days * 100) if total_planned_days > 0 else 0
schedule_variance = overall_progress - planned_progress
# Calculate cost performance
cost_variance = project.budget - project.current_cost
cost_performance_index = float(project.budget / project.current_cost) if project.current_cost > 0 else 1.0
# Calculate estimated completion date
if overall_progress > 0:
estimated_total_days = elapsed_days / (overall_progress / 100)
estimated_completion = project.start_date + timedelta(days=estimated_total_days)
else:
estimated_completion = project.planned_end_date
return {
'project_id': project_id,
'overall_progress_percent': overall_progress,
'schedule_variance_percent': schedule_variance,
'schedule_status': 'ahead' if schedule_variance > 0 else 'behind' if schedule_variance < 0 else 'on_track',
'cost_variance': float(cost_variance),
'cost_performance_index': cost_performance_index,
'budget_status': 'under' if cost_variance > 0 else 'over',
'estimated_completion': estimated_completion.isoformat(),
'days_variance': (estimated_completion - project.planned_end_date).days
}
def manage_change_order(self, project_id: str, change_data: dict) -> dict:
"""Manage construction change orders"""
project = self.projects.get(project_id)
if not project:
return {'error': 'Project not found'}
change_order = {
'co_id': self._generate_co_id(),
'project_id': project_id,
'description': change_data['description'],
'reason': change_data['reason'],
'cost_impact': Decimal(str(change_data['cost_impact'])),
'schedule_impact_days': change_data.get('schedule_impact_days', 0),
'submitted_by': change_data['submitted_by'],
'submitted_date': datetime.now(),
'status': 'pending_approval',
'approved': False
}
self.change_orders.append(change_order)
return {
'change_order_id': change_order['co_id'],
'cost_impact': float(change_order['cost_impact']),
'schedule_impact_days': change_order['schedule_impact_days'],
'new_budget': float(project.budget + change_order['cost_impact']),
'new_completion_date': (
project.planned_end_date + timedelta(days=change_order['schedule_impact_days'])
).isoformat()
}
def estimate_costs(self, project_type: str, square_footage: float, specifications: dict) -> dict:
"""Estimate construction costs"""
# Cost per square foot by project type
base_costs = {
'residential_basic': Decimal('150'),
'residential_luxury': Decimal('300'),
'commercial_office': Decimal('200'),
'industrial_warehouse': Decimal('75')
}
base_cost_per_sf = base_costs.get(project_type, Decimal('150'))
# Calculate base cost
base_cost = base_cost_per_sf * Decimal(str(square_footage))
# Add complexity factors
complexity_factor = Decimal('1.0')
if specifications.get('custom_design', False):
complexity_factor += Decimal('0.15')
if specifications.get('sustainable_materials', False):
complexity_factor += Decimal('0.10')
if specifications.get('complex_site', False):
complexity_factor += Decimal('0.20')
adjusted_cost = base_cost * complexity_factor
# Add contingency (10%)
contingency = adjusted_cost * Decimal('0.10')
# Breakdown by category
breakdown = {
'site_work': float(adjusted_cost * Decimal('0.08')),
'foundation': float(adjusted_cost * Decimal('0.12')),
'structure': float(adjusted_cost * Decimal('0.25')),
'exterior': float(adjusted_cost * Decimal('0.15')),
'interior': float(adjusted_cost * Decimal('0.20')),
'mep': float(adjusted_cost * Decimal('0.20'))
}
total_estimate = adjusted_cost + contingency
return {
'project_type': project_type,
'square_footage': square_footage,
'base_cost_per_sf': float(base_cost_per_sf),
'complexity_factor': float(complexity_factor),
'adjusted_cost': float(adjusted_cost),
'contingency': float(contingency),
'total_estimate': float(total_estimate),
'cost_breakdown': breakdown
}
def _generate_task_id(self) -> str:
import uuid
return f"TASK-{uuid.uuid4().hex[:8].upper()}"
def _generate_co_id(self) -> str:
import uuid
return f"CO-{uuid.uuid4().hex[:6].upper()}"Safety Management System
@dataclass
class SafetyIncident:
"""Safety incident report"""
incident_id: str
project_id: str
incident_type: str # 'injury', 'near_miss', 'property_damage'
severity: str # 'minor', 'moderate', 'severe', 'fatal'
description: str
location: str
occurred_at: datetime
reported_by: str
injured_person: Optional[str]
root_cause: Optional[str]
corrective_actions: List[str]
class SafetyManagementSystem:
"""Construction safety management"""
def __init__(self):
self.incidents = []
self.safety_inspections = []
self.training_records = []
def conduct_safety_inspection(self, project_id: str, inspector: str) -> dict:
"""Conduct safety inspection"""
inspection_items = [
'Personal protective equipment (PPE)',
'Fall protection systems',
'Scaffolding integrity',
'Electrical safety',
'Equipment guarding',
'Housekeeping',
'Fire prevention',
'First aid availability',
'Emergency exits',
'Signage and barriers'
]
violations = []
passed_items = []
# Simulate inspection (in production, would be actual checklist)
for item in inspection_items:
# Random pass/fail for demonstration
import random
if random.random() < 0.85: # 85% pass rate
passed_items.append(item)
else:
violations.append({
'item': item,
'severity': random.choice(['minor', 'major']),
'action_required': 'Correct immediately' if random.random() < 0.3 else 'Correct within 24 hours'
})
inspection = {
'inspection_id': self._generate_inspection_id(),
'project_id': project_id,
'inspector': inspector,
'inspection_date': datetime.now(),
'items_inspected': len(inspection_items),
'items_passed': len(passed_items),
'violations': violations,
'overall_score': (len(passed_items) / len(inspection_items)) * 100,
'status': 'pass' if len(violations) == 0 else 'fail'
}
self.safety_inspections.append(inspection)
return inspection
def report_incident(self, incident_data: dict) -> SafetyIncident:
"""Report safety incident"""
incident = SafetyIncident(
incident_id=self._generate_incident_id(),
project_id=incident_data['project_id'],
incident_type=incident_data['incident_type'],
severity=incident_data['severity'],
description=incident_data['description'],
location=incident_data['location'],
occurred_at=incident_data['occurred_at'],
reported_by=incident_data['reported_by'],
injured_person=incident_data.get('injured_person'),
root_cause=None,
corrective_actions=[]
)
self.incidents.append(incident)
# Notify relevant parties
self._notify_incident(incident)
return incident
def calculate_safety_metrics(self, project_id: str, hours_worked: float) -> dict:
"""Calculate safety performance metrics"""
project_incidents = [
i for i in self.incidents
if i.project_id == project_id
]
# Count recordable incidents
recordable_incidents = [
i for i in project_incidents
if i.incident_type == 'injury' and i.severity in ['moderate', 'severe', 'fatal']
]
# OSHA Incident Rate = (Number of incidents × 200,000) / Total hours worked
if hours_worked > 0:
incident_rate = (len(recordable_incidents) * 200000) / hours_worked
else:
incident_rate = 0
# Days Away, Restricted, or Transferred (DART) Rate
dart_incidents = [
i for i in recordable_incidents
if i.severity in ['severe', 'fatal']
]
dart_rate = (len(dart_incidents) * 200000) / hours_worked if hours_worked > 0 else 0
return {
'project_id': project_id,
'total_hours_worked': hours_worked,
'total_incidents': len(project_incidents),
'recordable_incidents': len(recordable_incidents),
'incident_rate': incident_rate,
'dart_rate': dart_rate,
'safety_rating': 'Excellent' if incident_rate < 1.0 else
'Good' if incident_rate < 3.0 else
'Needs Improvement'
}
def _notify_incident(self, incident: SafetyIncident):
"""Notify stakeholders of incident"""
# Implementation would send notifications
pass
def _generate_inspection_id(self) -> str:
import uuid
return f"INS-{uuid.uuid4().hex[:8].upper()}"
def _generate_incident_id(self) -> str:
import uuid
return f"INC-{uuid.uuid4().hex[:8].upper()}"BIM Integration
class BIMManagement:
"""Building Information Modeling management"""
def __init__(self):
self.models = {}
self.clash_detections = []
def perform_clash_detection(self, model_ids: List[str]) -> dict:
"""Detect clashes between BIM models"""
# Simulate clash detection between disciplines
# In production, would use BIM software APIs (Revit, Navisworks)
clashes = [
{
'clash_id': 'CLASH-001',
'type': 'hard', # 'hard' or 'soft'
'disciplines': ['structural', 'mep'],
'description': 'Steel beam conflicts with HVAC duct',
'location': 'Level 3, Grid B-4',
'severity': 'high',
'status': 'open'
},
{
'clash_id': 'CLASH-002',
'type': 'soft',
'disciplines': ['architectural', 'mep'],
'description': 'Insufficient clearance for plumbing access',
'location': 'Level 2, Grid C-2',
'severity': 'medium',
'status': 'open'
}
]
return {
'models_analyzed': model_ids,
'total_clashes': len(clashes),
'hard_clashes': len([c for c in clashes if c['type'] == 'hard']),
'soft_clashes': len([c for c in clashes if c['type'] == 'soft']),
'clashes': clashes
}
def extract_quantities(self, model_id: str) -> dict:
"""Extract material quantities from BIM model"""
# Simulate quantity takeoff
# In production, would extract from actual BIM model
quantities = {
'concrete': {
'unit': 'cubic_yards',
'quantity': 1250,
'cost_per_unit': 150,
'total_cost': 187500
},
'rebar': {
'unit': 'tons',
'quantity': 85,
'cost_per_unit': 800,
'total_cost': 68000
},
'structural_steel': {
'unit': 'tons',
'quantity': 120,
'cost_per_unit': 1200,
'total_cost': 144000
}
}
total_cost = sum(item['total_cost'] for item in quantities.values())
return {
'model_id': model_id,
'quantities': quantities,
'total_estimated_cost': total_cost
}Best Practices
Project Management
- Use critical path method for scheduling
- Implement regular progress reviews
- Maintain detailed documentation
- Use integrated project delivery (IPD)
- Implement lean construction principles
- Track key performance indicators
- Conduct regular stakeholder meetings
Cost Control
- Develop detailed estimates
- Track costs continuously
- Manage change orders effectively
- Use value engineering
- Implement cost coding systems
- Monitor cash flow
- Conduct regular audits
Safety Management
- Implement comprehensive safety program
- Conduct regular toolbox talks
- Provide proper PPE
- Maintain OSHA compliance
- Investigate all incidents
- Track safety metrics
- Promote safety culture
BIM Implementation
- Use BIM for clash detection
- Implement 4D scheduling
- Extract quantities from model
- Enable collaboration
- Maintain model coordination
- Use BIM for facility management
- Follow ISO 19650 standards
Anti-Patterns
❌ Poor project planning
❌ Inadequate cost tracking
❌ No safety program
❌ Poor communication
❌ Ignoring change orders
❌ No quality control
❌ Inadequate documentation
❌ Poor subcontractor management
❌ No risk management
Resources
- AIA (American Institute of Architects): https://www.aia.org/
- AGC (Associated General Contractors): https://www.agc.org/
- International Building Code: https://www.iccsafe.org/
- buildingSMART (BIM): https://www.buildingsmart.org/
- PMI Construction Extension: https://www.pmi.org/
- LEED Certification: https://www.usgbc.org/leed