Build KPI dashboards for construction projects. Track CPI, SPI, quality, safety metrics in real-time.
中文简介正在准备中。当前显示原文说明。
安装前可以查看交给 AI 代理的指令内容。
Project monitoring challenges:
Unified KPI dashboard system for construction projects with automated data collection, visualization, and alerting.
import pandas as pd
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import date, datetime
from enum import Enum
class KPICategory(Enum):
COST = "cost"
SCHEDULE = "schedule"
QUALITY = "quality"
SAFETY = "safety"
PRODUCTIVITY = "productivity"
SUSTAINABILITY = "sustainability"
class KPIStatus(Enum):
ON_TARGET = "on_target"
AT_RISK = "at_risk"
CRITICAL = "critical"
class TrendDirection(Enum):
IMPROVING = "improving"
STABLE = "stable"
DECLINING = "declining"
@dataclass
class KPIDefinition:
kpi_id: str
name: str
category: KPICategory
unit: str
target: float
warning_threshold: float
critical_threshold: float
higher_is_better: bool = True
formula: str = ""
@dataclass
class KPIValue:
kpi_id: str
value: float
date: date
status: KPIStatus
trend: TrendDirection
class KPIDashboard:
"""Build and manage KPI dashboards for construction projects."""
def __init__(self, project_name: str):
self.project_name = project_name
self.kpis: Dict[str, KPIDefinition] = {}
self.history: Dict[str, List[KPIValue]] = {}
self._define_standard_kpis()
def _define_standard_kpis(self):
"""Define standard construction KPIs."""
standard_kpis = [
# Cost KPIs
KPIDefinition("CPI", "Cost Performance Index", KPICategory.COST,
"ratio", 1.0, 0.95, 0.90, True, "BCWP / ACWP"),
KPIDefinition("CV", "Cost Variance", KPICategory.COST,
"$", 0, -50000, -100000, True, "BCWP - ACWP"),
KPIDefinition("BUDGET_USED", "Budget Utilization", KPICategory.COST,
"%", 100, 105, 110, False),
# Schedule KPIs
KPIDefinition("SPI", "Schedule Performance Index", KPICategory.SCHEDULE,
"ratio", 1.0, 0.95, 0.90, True, "BCWP / BCWS"),
KPIDefinition("SV", "Schedule Variance", KPICategory.SCHEDULE,
"days", 0, -7, -14, True),
KPIDefinition("COMPLETION", "Project Completion", KPICategory.SCHEDULE,
"%", 100, 95, 90, True),
# Quality KPIs
KPIDefinition("DEFECT_RATE", "Defect Rate", KPICategory.QUALITY,
"per 1000 units", 0, 5, 10, False),
KPIDefinition("FIRST_PASS", "First Pass Yield", KPICategory.QUALITY,
"%", 95, 90, 85, True),
KPIDefinition("REWORK", "Rework Percentage", KPICategory.QUALITY,
"%", 0, 3, 5, False),
# Safety KPIs
KPIDefinition("TRIR", "Total Recordable Incident Rate", KPICategory.SAFETY,
"per 200k hours", 0, 2, 4, False),
KPIDefinition("LOST_DAYS", "Lost Time Injuries", KPICategory.SAFETY,
"incidents", 0, 1, 3, False),
KPIDefinition("SAFETY_OBSERVATIONS", "Safety Observations", KPICategory.SAFETY,
"count", 50, 30, 20, True),
# Productivity KPIs
KPIDefinition("LABOR_PROD", "Labor Productivity", KPICategory.PRODUCTIVITY,
"%", 100, 90, 80, True),
KPIDefinition("EQUIP_UTIL", "Equipment Utilization", KPICategory.PRODUCTIVITY,
"%", 85, 70, 60, True),
]
for kpi in standard_kpis:
self.kpis[kpi.kpi_id] = kpi
self.history[kpi.kpi_id] = []
def add_custom_kpi(self, kpi: KPIDefinition):
"""Add custom KPI definition."""
self.kpis[kpi.kpi_id] = kpi
self.history[kpi.kpi_id] = []
def record_value(self, kpi_id: str, value: float, record_date: date = None):
"""Record KPI value."""
if kpi_id not in self.kpis:
return
kpi = self.kpis[kpi_id]
record_date = record_date or date.today()
# Calculate status
status = self._calculate_status(kpi, value)
# Calculate trend
trend = self._calculate_trend(kpi_id, value)
kpi_value = KPIValue(
kpi_id=kpi_id,
value=value,
date=record_date,
status=status,
trend=trend
)
self.history[kpi_id].append(kpi_value)
def _calculate_status(self, kpi: KPIDefinition, value: float) -> KPIStatus:
"""Calculate KPI status based on thresholds."""
if kpi.higher_is_better:
if value >= kpi.target:
return KPIStatus.ON_TARGET
elif value >= kpi.warning_threshold:
return KPIStatus.AT_RISK
else:
return KPIStatus.CRITICAL
else:
if value <= kpi.target:
return KPIStatus.ON_TARGET
elif value <= kpi.warning_threshold:
return KPIStatus.AT_RISK
else:
return KPIStatus.CRITICAL
def _calculate_trend(self, kpi_id: str, current_value: float) -> TrendDirection:
"""Calculate trend direction."""
history = self.history.get(kpi_id, [])
if len(history) < 2:
return TrendDirection.STABLE
# Compare with average of last 3 values
recent_values = [h.value for h in history[-3:]]
avg = sum(recent_values) / len(recent_values)
kpi = self.kpis[kpi_id]
diff = current_value - avg
if abs(diff) < avg * 0.05: # Within 5%
return TrendDirection.STABLE
elif (diff > 0 and kpi.higher_is_better) or (diff < 0 and not kpi.higher_is_better):
return TrendDirection.IMPROVING
else:
return TrendDirection.DECLINING
def get_current_values(self) -> Dict[str, KPIValue]:
"""Get most recent value for each KPI."""
current = {}
for kpi_id, history in self.history.items():
if history:
current[kpi_id] = history[-1]
return current
def get_dashboard_summary(self) -> Dict[str, Any]:
"""Get dashboard summary."""
current = self.get_current_values()
summary = {
'project': self.project_name,
'date': date.today().isoformat(),
'total_kpis': len(self.kpis),
'by_status': {s.value: 0 for s in KPIStatus},
'by_category': {},
'alerts': []
}
for kpi_id, value in current.items():
summary['by_status'][value.status.value] += 1
category = self.kpis[kpi_id].category.value
if category not in summary['by_category']:
summary['by_category'][category] = {'on_target': 0, 'at_risk': 0, 'critical': 0}
summary['by_category'][category][value.status.value] += 1
if value.status == KPIStatus.CRITICAL:
summary['alerts'].append({
'kpi': self.kpis[kpi_id].name,
'value': value.value,
'target': self.kpis[kpi_id].target,
'status': 'critical'
})
return summary
def get_kpi_details(self, kpi_id: str) -> Dict[str, Any]:
"""Get detailed KPI information."""
if kpi_id not in self.kpis:
return {}
kpi = self.kpis[kpi_id]
history = self.history.get(kpi_id, [])
return {
'definition': {
'id': kpi.kpi_id,
'name': kpi.name,
'category': kpi.category.value,
'unit': kpi.unit,
'target': kpi.target,
'formula': kpi.formula
},
'current': {
'value': history[-1].value if history else None,
'status': history[-1].status.value if history else None,
'trend': history[-1].trend.value if history else None
},
'history': [
{'date': h.date.isoformat(), 'value': h.value, 'status': h.status.value}
for h in history
]
}
def generate_html_dashboard(self) -> str:
"""Generate HTML dashboard."""
summary = self.get_dashboard_summary()
current = self.get_current_values()
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>KPI Dashboard - {self.project_name}</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.header {{ background: #2196F3; color: white; padding: 20px; margin-bottom: 20px; }}
.kpi-grid {{ display: grid; grid-template-columns: repeat(4, 1fr); gap: 15px; }}
.kpi-card {{ border: 1px solid #ddd; padding: 15px; border-radius: 5px; }}
.on_target {{ border-left: 4px solid #4CAF50; }}
.at_risk {{ border-left: 4px solid #FF9800; }}
.critical {{ border-left: 4px solid #F44336; }}
.kpi-value {{ font-size: 24px; font-weight: bold; }}
.kpi-name {{ color: #666; font-size: 14px; }}
</style>
</head>
<body>
<div class="header">
<h1>{self.project_name} - KPI Dashboard</h1>
<p>Last updated: {summary['date']}</p>
</div>
<div class="kpi-grid">
"""
for kpi_id, value in current.items():
kpi = self.kpis[kpi_id]
html += f"""
<div class="kpi-card {value.status.value}">
<div class="kpi-name">{kpi.name}</div>
<div class="kpi-value">{value.value:.2f} {kpi.unit}</div>
<div>Target: {kpi.target} | Trend: {value.trend.value}</div>
</div>
"""
html += "</div></body></html>"
return html
def export_to_excel(self, output_path: str) -> str:
"""Export dashboard to Excel."""
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Summary
summary = self.get_dashboard_summary()
summary_df = pd.DataFrame([{
'Project': summary['project'],
'Date': summary['date'],
'On Target': summary['by_status']['on_target'],
'At Risk': summary['by_status']['at_risk'],
'Critical': summary['by_status']['critical']
}])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
# Current values
current = self.get_current_values()
current_data = []
for kpi_id, value in current.items():
kpi = self.kpis[kpi_id]
current_data.append({
'KPI': kpi.name,
'Category': kpi.category.value,
'Value': value.value,
'Unit': kpi.unit,
'Target': kpi.target,
'Status': value.status.value,
'Trend': value.trend.value
})
current_df = pd.DataFrame(current_data)
current_df.to_excel(writer, sheet_name='Current KPIs', index=False)
return output_path
# Create dashboard
dashboard = KPIDashboard("Office Building A")
# Record KPI values
dashboard.record_value("CPI", 0.95)
dashboard.record_value("SPI", 1.02)
dashboard.record_value("DEFECT_RATE", 3.5)
dashboard.record_value("TRIR", 1.8)
dashboard.record_value("LABOR_PROD", 92)
# Get summary
summary = dashboard.get_dashboard_summary()
print(f"On Target: {summary['by_status']['on_target']}")
print(f"Critical: {summary['by_status']['critical']}")
html = dashboard.generate_html_dashboard()
with open("dashboard.html", "w") as f:
f.write(html)
details = dashboard.get_kpi_details("CPI")
print(f"Current CPI: {details['current']['value']}")
dashboard.add_custom_kpi(KPIDefinition(
"WASTE_DIVERSION", "Waste Diversion Rate",
KPICategory.SUSTAINABILITY, "%", 75, 60, 50, True
))
还没有评价。欢迎分享你的使用感受。
无需登录即可发表。欢迎分享使用感受、提问，或说明在哪些环境下无法运行。
还没有评论。来写第一条吧。
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.
中文简介正在准备中。当前显示原文说明。