Plan organizational capacity for construction projects. Forecast resource needs, identify capacity gaps, and support strategic planning for project pursuit and staffing.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
설치 방법 보기설치하기 전에 에이전트에게 전달되는 지시 내용을 확인할 수 있습니다.
Strategic capacity planning for construction organizations. Forecast resource requirements based on project pipeline, identify capacity constraints, optimize staffing levels, and support go/no-go decisions on new project pursuits.
┌─────────────────────────────────────────────────────────────────┐
│ CAPACITY PLANNING │
├─────────────────────────────────────────────────────────────────┤
│ │
│ DEMAND FORECAST CAPACITY ANALYSIS DECISIONS │
│ ─────────────── ───────────────── ───────── │
│ │
│ Current Projects → Available: Pursue new │
│ • Project A (Active) 👷 PM: 5 project? │
│ • Project B (Active) 👷 Supers: 12 ──────── │
│ • Project C (Starting) 📐 Engineers: 8 ✅ Capacity │
│ ⚠️ Stretch │
│ Pipeline: → Required: ❌ Decline │
│ • Bid D (60% win) 👷 PM: 7 │
│ • Bid E (40% win) 👷 Supers: 15 │
│ • Opportunity F 📐 Engineers: 10 │
│ │
│ GAP ANALYSIS: ACTIONS: │
│ • PM: -2 (deficit) • Hire 2 PMs │
│ • Supers: -3 (deficit) • Promote from within │
│ • Engineers: -2 (deficit) • Partner with firm │
│ │
└─────────────────────────────────────────────────────────────────┘
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from datetime import datetime, timedelta
from enum import Enum
import statistics
class ResourceRole(Enum):
PROJECT_MANAGER = "project_manager"
SUPERINTENDENT = "superintendent"
PROJECT_ENGINEER = "project_engineer"
ESTIMATOR = "estimator"
SCHEDULER = "scheduler"
SAFETY_MANAGER = "safety_manager"
QC_MANAGER = "qc_manager"
ADMIN = "admin"
class ProjectPhase(Enum):
PURSUIT = "pursuit"
PRECONSTRUCTION = "preconstruction"
CONSTRUCTION = "construction"
CLOSEOUT = "closeout"
class OpportunityStatus(Enum):
IDENTIFIED = "identified"
PURSUING = "pursuing"
BID_SUBMITTED = "bid_submitted"
NEGOTIATING = "negotiating"
WON = "won"
LOST = "lost"
@dataclass
class StaffMember:
id: str
name: str
role: ResourceRole
capacity: float = 1.0 # FTE
current_assignment: str = ""
availability_date: datetime = None
skills: List[str] = field(default_factory=list)
max_project_value: float = 0 # Max project size they can handle
@dataclass
class ProjectDemand:
project_id: str
project_name: str
value: float
phase: ProjectPhase
start_date: datetime
end_date: datetime
probability: float = 1.0 # 1.0 for active, <1 for pipeline
resource_needs: Dict[ResourceRole, float] = field(default_factory=dict)
@dataclass
class CapacityGap:
role: ResourceRole
period_start: datetime
period_end: datetime
demand: float
capacity: float
gap: float
severity: str
@dataclass
class CapacityForecast:
forecast_date: datetime
horizon_months: int
total_demand_fte: float
total_capacity_fte: float
utilization_pct: float
gaps: List[CapacityGap]
recommendations: List[str]
class CapacityPlanner:
"""Plan organizational capacity for construction projects."""
# Typical staffing ratios by project value
STAFFING_RATIOS = {
ResourceRole.PROJECT_MANAGER: 20000000, # 1 PM per $20M
ResourceRole.SUPERINTENDENT: 10000000, # 1 Super per $10M
ResourceRole.PROJECT_ENGINEER: 15000000, # 1 PE per $15M
ResourceRole.ESTIMATOR: 50000000, # 1 Estimator per $50M (pursuit)
ResourceRole.SCHEDULER: 30000000, # 1 Scheduler per $30M
ResourceRole.SAFETY_MANAGER: 25000000, # 1 Safety per $25M
}
# Phase factors (multiply by role ratio)
PHASE_FACTORS = {
ProjectPhase.PURSUIT: {"estimator": 1.5, "pm": 0.3},
ProjectPhase.PRECONSTRUCTION: {"pm": 0.7, "pe": 0.5, "scheduler": 0.5},
ProjectPhase.CONSTRUCTION: {"pm": 1.0, "super": 1.0, "pe": 1.0, "safety": 1.0},
ProjectPhase.CLOSEOUT: {"pm": 0.5, "pe": 0.3, "admin": 1.0}
}
def __init__(self, organization_name: str):
self.organization_name = organization_name
self.staff: Dict[str, StaffMember] = {}
self.projects: Dict[str, ProjectDemand] = {}
self.pipeline: Dict[str, ProjectDemand] = {}
def add_staff(self, id: str, name: str, role: ResourceRole,
capacity: float = 1.0, current_assignment: str = "",
availability_date: datetime = None,
max_project_value: float = 0) -> StaffMember:
"""Add staff member to capacity pool."""
member = StaffMember(
id=id,
name=name,
role=role,
capacity=capacity,
current_assignment=current_assignment,
availability_date=availability_date or datetime.now(),
max_project_value=max_project_value
)
self.staff[id] = member
return member
def add_active_project(self, id: str, name: str, value: float,
phase: ProjectPhase, start_date: datetime,
end_date: datetime) -> ProjectDemand:
"""Add active project to demand forecast."""
# Calculate resource needs based on value and phase
needs = self._calculate_resource_needs(value, phase)
project = ProjectDemand(
project_id=id,
project_name=name,
value=value,
phase=phase,
start_date=start_date,
end_date=end_date,
probability=1.0,
resource_needs=needs
)
self.projects[id] = project
return project
def add_pipeline_opportunity(self, id: str, name: str, value: float,
win_probability: float,
expected_start: datetime,
duration_months: int) -> ProjectDemand:
"""Add pipeline opportunity to demand forecast."""
needs = self._calculate_resource_needs(value, ProjectPhase.CONSTRUCTION)
opportunity = ProjectDemand(
project_id=id,
project_name=name,
value=value,
phase=ProjectPhase.PURSUIT,
start_date=expected_start,
end_date=expected_start + timedelta(days=duration_months * 30),
probability=win_probability,
resource_needs=needs
)
self.pipeline[id] = opportunity
return opportunity
def _calculate_resource_needs(self, value: float,
phase: ProjectPhase) -> Dict[ResourceRole, float]:
"""Calculate resource needs based on project value and phase."""
needs = {}
for role, ratio in self.STAFFING_RATIOS.items():
base_need = value / ratio
# Apply phase factor
phase_key = role.value.split('_')[0][:3]
factor = 1.0
if phase in self.PHASE_FACTORS:
factor = self.PHASE_FACTORS[phase].get(phase_key, 1.0)
needs[role] = base_need * factor
return needs
def get_current_capacity(self) -> Dict[ResourceRole, float]:
"""Get current capacity by role."""
capacity = {role: 0.0 for role in ResourceRole}
for member in self.staff.values():
if member.availability_date <= datetime.now():
capacity[member.role] += member.capacity
return capacity
def get_capacity_at_date(self, target_date: datetime) -> Dict[ResourceRole, float]:
"""Get projected capacity at future date."""
capacity = {role: 0.0 for role in ResourceRole}
for member in self.staff.values():
if member.availability_date <= target_date:
capacity[member.role] += member.capacity
return capacity
def calculate_demand(self, target_date: datetime,
include_pipeline: bool = True,
pipeline_threshold: float = 0.0) -> Dict[ResourceRole, float]:
"""Calculate resource demand at date."""
demand = {role: 0.0 for role in ResourceRole}
# Active projects
for project in self.projects.values():
if project.start_date <= target_date <= project.end_date:
for role, need in project.resource_needs.items():
demand[role] += need * project.probability
# Pipeline (weighted by probability)
if include_pipeline:
for opp in self.pipeline.values():
if opp.probability >= pipeline_threshold:
if opp.start_date <= target_date <= opp.end_date:
for role, need in opp.resource_needs.items():
demand[role] += need * opp.probability
return demand
def identify_gaps(self, horizon_months: int = 12) -> List[CapacityGap]:
"""Identify capacity gaps over forecast horizon."""
gaps = []
for month in range(horizon_months):
period_start = datetime.now() + timedelta(days=month * 30)
period_end = period_start + timedelta(days=30)
capacity = self.get_capacity_at_date(period_start)
demand = self.calculate_demand(period_start, include_pipeline=True)
for role in ResourceRole:
cap = capacity.get(role, 0)
dem = demand.get(role, 0)
gap = cap - dem
if gap < 0:
severity = "critical" if gap < -1 else "warning"
gaps.append(CapacityGap(
role=role,
period_start=period_start,
period_end=period_end,
demand=dem,
capacity=cap,
gap=gap,
severity=severity
))
return gaps
def can_pursue_project(self, value: float, start_date: datetime,
duration_months: int) -> Dict:
"""Evaluate if organization can pursue new project."""
# Calculate needs for potential project
needs = self._calculate_resource_needs(value, ProjectPhase.CONSTRUCTION)
end_date = start_date + timedelta(days=duration_months * 30)
# Check capacity over project duration
can_staff = True
bottlenecks = []
current_date = start_date
while current_date <= end_date:
capacity = self.get_capacity_at_date(current_date)
demand = self.calculate_demand(current_date)
for role, need in needs.items():
available = capacity.get(role, 0) - demand.get(role, 0)
if need > available:
can_staff = False
bottlenecks.append({
"date": current_date,
"role": role.value,
"needed": need,
"available": available,
"gap": need - available
})
current_date += timedelta(days=30)
# Determine recommendation
if can_staff:
recommendation = "GO - Sufficient capacity"
elif len(bottlenecks) <= 2:
recommendation = "CONDITIONAL - Minor gaps, consider hiring"
else:
recommendation = "CAUTION - Significant capacity constraints"
return {
"can_staff": can_staff,
"recommendation": recommendation,
"resource_needs": {r.value: v for r, v in needs.items()},
"bottlenecks": bottlenecks[:10],
"actions_required": self._suggest_hiring(bottlenecks)
}
def _suggest_hiring(self, bottlenecks: List[Dict]) -> List[str]:
"""Suggest hiring actions based on gaps."""
if not bottlenecks:
return []
# Aggregate gaps by role
role_gaps = {}
for b in bottlenecks:
role = b['role']
if role not in role_gaps:
role_gaps[role] = 0
role_gaps[role] = max(role_gaps[role], b['gap'])
actions = []
for role, gap in sorted(role_gaps.items(), key=lambda x: -x[1]):
hires = int(gap) + 1
actions.append(f"Hire {hires} {role}(s) - Gap: {gap:.1f} FTE")
return actions
def generate_forecast(self, horizon_months: int = 12) -> CapacityForecast:
"""Generate capacity forecast."""
gaps = self.identify_gaps(horizon_months)
# Calculate totals
capacity = self.get_current_capacity()
demand = self.calculate_demand(datetime.now())
total_capacity = sum(capacity.values())
total_demand = sum(demand.values())
utilization = (total_demand / total_capacity * 100) if total_capacity > 0 else 0
# Generate recommendations
recommendations = []
if utilization > 90:
recommendations.append("High utilization - consider hiring")
elif utilization < 60:
recommendations.append("Low utilization - review project pipeline")
# Role-specific recommendations
critical_gaps = [g for g in gaps if g.severity == "critical"]
gap_roles = set(g.role.value for g in critical_gaps)
for role in gap_roles:
recommendations.append(f"Critical gap in {role} - immediate action needed")
return CapacityForecast(
forecast_date=datetime.now(),
horizon_months=horizon_months,
total_demand_fte=total_demand,
total_capacity_fte=total_capacity,
utilization_pct=utilization,
gaps=gaps,
recommendations=recommendations
)
def generate_report(self) -> str:
"""Generate capacity planning report."""
forecast = self.generate_forecast()
lines = [
"# Capacity Planning Report",
"",
f"**Organization:** {self.organization_name}",
f"**Report Date:** {forecast.forecast_date.strftime('%Y-%m-%d')}",
"",
"## Executive Summary",
"",
f"| Metric | Value |",
f"|--------|-------|",
f"| Active Projects | {len(self.projects)} |",
f"| Pipeline Opportunities | {len(self.pipeline)} |",
f"| Total Staff | {len(self.staff)} |",
f"| Current Capacity (FTE) | {forecast.total_capacity_fte:.1f} |",
f"| Current Demand (FTE) | {forecast.total_demand_fte:.1f} |",
f"| Utilization | {forecast.utilization_pct:.0f}% |",
"",
"## Capacity by Role",
"",
"| Role | Capacity | Demand | Gap |",
"|------|----------|--------|-----|"
]
capacity = self.get_current_capacity()
demand = self.calculate_demand(datetime.now())
for role in ResourceRole:
cap = capacity.get(role, 0)
dem = demand.get(role, 0)
gap = cap - dem
gap_icon = "✅" if gap >= 0 else "⚠️" if gap > -1 else "🔴"
lines.append(
f"| {role.value} | {cap:.1f} | {dem:.1f} | {gap:+.1f} {gap_icon} |"
)
# Active projects
lines.extend([
"",
"## Active Projects",
"",
"| Project | Value | Phase | End Date |",
"|---------|-------|-------|----------|"
])
for p in sorted(self.projects.values(), key=lambda x: x.value, reverse=True):
lines.append(
f"| {p.project_name} | ${p.value:,.0f} | {p.phase.value} | "
f"{p.end_date.strftime('%Y-%m-%d')} |"
)
# Pipeline
if self.pipeline:
lines.extend([
"",
"## Pipeline",
"",
"| Opportunity | Value | Probability | Expected Start |",
"|-------------|-------|-------------|----------------|"
])
for p in sorted(self.pipeline.values(), key=lambda x: -x.probability):
lines.append(
f"| {p.project_name} | ${p.value:,.0f} | {p.probability:.0%} | "
f"{p.start_date.strftime('%Y-%m-%d')} |"
)
# Gaps
critical_gaps = [g for g in forecast.gaps if g.severity == "critical"]
if critical_gaps:
lines.extend([
"",
f"## Critical Capacity Gaps ({len(critical_gaps)})",
"",
"| Role | Period | Gap |",
"|------|--------|-----|"
])
for gap in critical_gaps[:10]:
lines.append(
f"| {gap.role.value} | {gap.period_start.strftime('%Y-%m')} | "
f"{gap.gap:.1f} FTE |"
)
# Recommendations
if forecast.recommendations:
lines.extend([
"",
"## Recommendations",
""
])
for rec in forecast.recommendations:
lines.append(f"- {rec}")
return "\n".join(lines)
from datetime import datetime, timedelta
# Initialize planner
planner = CapacityPlanner("ABC Construction")
# Add staff
planner.add_staff("PM-001", "John Smith", ResourceRole.PROJECT_MANAGER)
planner.add_staff("PM-002", "Jane Doe", ResourceRole.PROJECT_MANAGER)
planner.add_staff("SUP-001", "Mike Johnson", ResourceRole.SUPERINTENDENT)
planner.add_staff("SUP-002", "Bob Williams", ResourceRole.SUPERINTENDENT)
planner.add_staff("SUP-003", "Tom Brown", ResourceRole.SUPERINTENDENT)
planner.add_staff("PE-001", "Sarah Davis", ResourceRole.PROJECT_ENGINEER)
planner.add_staff("PE-002", "Chris Wilson", ResourceRole.PROJECT_ENGINEER)
# Add active projects
planner.add_active_project(
"PRJ-001", "Downtown Tower",
value=25000000,
phase=ProjectPhase.CONSTRUCTION,
start_date=datetime(2024, 6, 1),
end_date=datetime(2025, 12, 31)
)
planner.add_active_project(
"PRJ-002", "Hospital Wing",
value=40000000,
phase=ProjectPhase.CONSTRUCTION,
start_date=datetime(2024, 9, 1),
end_date=datetime(2026, 6, 30)
)
# Add pipeline opportunities
planner.add_pipeline_opportunity(
"OPP-001", "Office Complex",
value=30000000,
win_probability=0.6,
expected_start=datetime(2025, 3, 1),
duration_months=18
)
# Check if can pursue new project
evaluation = planner.can_pursue_project(
value=20000000,
start_date=datetime(2025, 6, 1),
duration_months=12
)
print(f"Recommendation: {evaluation['recommendation']}")
for action in evaluation['actions_required']:
print(f" - {action}")
# Generate forecast
forecast = planner.generate_forecast()
print(f"Utilization: {forecast.utilization_pct:.0f}%")
print(f"Critical gaps: {len([g for g in forecast.gaps if g.severity == 'critical'])}")
# Generate report
print(planner.generate_report())
pip install (no external dependencies)
아직 리뷰가 없습니다. 사용해 본 소감을 남겨 주세요.
로그인 없이 쓸 수 있습니다. 사용 후기, 질문, 작동하지 않은 환경 등을 편하게 남겨 주세요.
아직 댓글이 없습니다. 첫 댓글을 남겨 보세요.
Create 4D construction simulations by linking BIM elements to project schedules. Generate time-based visualizations, sequence analysis, and construction phasing with Gantt integration.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Analyze 5000+ IFC and Revit projects at scale for patterns, benchmarks, and insights. Big data analysis for construction.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Check construction AI systems against the EU AI Act (Regulation 2024/1689): classify risk of estimation, scheduling, CV and agent tools, document transparency, keep human oversight. Use when deploying or auditing AI in construction.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Orchestrate multiple AI agents for construction workflows: estimator, scheduler, document, QA and safety agents coordinated by a supervisor agent, with human checkpoints. Use when automating end-to-end project processes with agentic AI.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Create Apache Airflow DAGs for construction data pipelines. Orchestrate ETL, validation, and reporting workflows.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.
Automate as-built documentation and digital handover for construction. Compile project records, generate O&M manuals, create asset databases, and ensure complete project closeout.
한국어 요약은 준비 중입니다. 원문 설명을 표시하고 있습니다.