Create interactive KPI dashboards for construction projects. Track schedule, cost, quality, and safety metrics in real-time.
インストールする前に、エージェントに与えられる指示の中身を確認できます。
Project stakeholders struggle with:
Centralized KPI dashboard that aggregates data from multiple sources and presents key metrics with drill-down capabilities.
import pandas as pd
from datetime import datetime, date, timedelta
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
class KPIStatus(Enum):
"""KPI health status."""
ON_TRACK = "on_track"
AT_RISK = "at_risk"
CRITICAL = "critical"
UNKNOWN = "unknown"
class KPICategory(Enum):
"""KPI categories."""
SCHEDULE = "schedule"
COST = "cost"
QUALITY = "quality"
SAFETY = "safety"
PRODUCTIVITY = "productivity"
SUSTAINABILITY = "sustainability"
@dataclass
class KPIMetric:
"""Single KPI metric."""
name: str
category: KPICategory
current_value: float
target_value: float
unit: str
status: KPIStatus
trend: str # up, down, stable
last_updated: datetime
description: str = ""
@property
def variance(self) -> float:
"""Calculate variance from target."""
if self.target_value == 0:
return 0
return ((self.current_value - self.target_value) / self.target_value) * 100
@property
def achievement(self) -> float:
"""Calculate achievement percentage."""
if self.target_value == 0:
return 0
return (self.current_value / self.target_value) * 100
@dataclass
class DashboardConfig:
"""Dashboard configuration."""
project_name: str
project_code: str
start_date: date
end_date: date
budget: float
currency: str = "USD"
refresh_interval_minutes: int = 15
class ProjectKPIDashboard:
"""Construction project KPI dashboard."""
# Standard thresholds for RAG status
THRESHOLDS = {
'schedule': {'green': 0.95, 'amber': 0.85},
'cost': {'green': 1.05, 'amber': 1.15},
'quality': {'green': 0.98, 'amber': 0.95},
'safety': {'green': 0, 'amber': 1} # incident count
}
def __init__(self, config: DashboardConfig):
self.config = config
self.metrics: Dict[str, KPIMetric] = {}
self.history: List[Dict[str, Any]] = []
def add_metric(self, metric: KPIMetric):
"""Add or update a KPI metric."""
self.metrics[metric.name] = metric
self._record_history(metric)
def _record_history(self, metric: KPIMetric):
"""Record metric history for trending."""
self.history.append({
'name': metric.name,
'value': metric.current_value,
'timestamp': metric.last_updated,
'status': metric.status.value
})
def calculate_schedule_kpis(self,
planned_activities: int,
completed_activities: int,
planned_duration_days: int,
actual_duration_days: int) -> List[KPIMetric]:
"""Calculate schedule-related KPIs."""
# Schedule Performance Index (SPI)
spi = completed_activities / planned_activities if planned_activities > 0 else 0
spi_status = self._get_status(spi, 'schedule')
# Schedule Variance
sv = completed_activities - planned_activities
# Percent Complete
pct_complete = (completed_activities / planned_activities * 100) if planned_activities > 0 else 0
metrics = [
KPIMetric(
name="Schedule Performance Index",
category=KPICategory.SCHEDULE,
current_value=round(spi, 2),
target_value=1.0,
unit="ratio",
status=spi_status,
trend=self._calculate_trend("Schedule Performance Index"),
last_updated=datetime.now(),
description="SPI = Earned Value / Planned Value"
),
KPIMetric(
name="Percent Complete",
category=KPICategory.SCHEDULE,
current_value=round(pct_complete, 1),
target_value=100,
unit="%",
status=spi_status,
trend=self._calculate_trend("Percent Complete"),
last_updated=datetime.now()
),
KPIMetric(
name="Schedule Variance",
category=KPICategory.SCHEDULE,
current_value=sv,
target_value=0,
unit="activities",
status=spi_status,
trend=self._calculate_trend("Schedule Variance"),
last_updated=datetime.now()
)
]
for m in metrics:
self.add_metric(m)
return metrics
def calculate_cost_kpis(self,
budgeted_cost: float,
actual_cost: float,
earned_value: float) -> List[KPIMetric]:
"""Calculate cost-related KPIs."""
# Cost Performance Index (CPI)
cpi = earned_value / actual_cost if actual_cost > 0 else 0
cpi_status = self._get_status(cpi, 'cost', inverse=True)
# Cost Variance
cv = earned_value - actual_cost
# Budget utilization
budget_used = (actual_cost / budgeted_cost * 100) if budgeted_cost > 0 else 0
metrics = [
KPIMetric(
name="Cost Performance Index",
category=KPICategory.COST,
current_value=round(cpi, 2),
target_value=1.0,
unit="ratio",
status=cpi_status,
trend=self._calculate_trend("Cost Performance Index"),
last_updated=datetime.now(),
description="CPI = Earned Value / Actual Cost"
),
KPIMetric(
name="Cost Variance",
category=KPICategory.COST,
current_value=round(cv, 2),
target_value=0,
unit=self.config.currency,
status=cpi_status,
trend=self._calculate_trend("Cost Variance"),
last_updated=datetime.now()
),
KPIMetric(
name="Budget Utilization",
category=KPICategory.COST,
current_value=round(budget_used, 1),
target_value=100,
unit="%",
status=cpi_status,
trend=self._calculate_trend("Budget Utilization"),
last_updated=datetime.now()
)
]
for m in metrics:
self.add_metric(m)
return metrics
def calculate_quality_kpis(self,
total_inspections: int,
passed_inspections: int,
rework_items: int,
total_items: int) -> List[KPIMetric]:
"""Calculate quality-related KPIs."""
# First Pass Yield
fpy = passed_inspections / total_inspections if total_inspections > 0 else 0
fpy_status = self._get_status(fpy, 'quality')
# Rework Rate
rework_rate = rework_items / total_items * 100 if total_items > 0 else 0
metrics = [
KPIMetric(
name="First Pass Yield",
category=KPICategory.QUALITY,
current_value=round(fpy * 100, 1),
target_value=98,
unit="%",
status=fpy_status,
trend=self._calculate_trend("First Pass Yield"),
last_updated=datetime.now()
),
KPIMetric(
name="Rework Rate",
category=KPICategory.QUALITY,
current_value=round(rework_rate, 1),
target_value=2,
unit="%",
status=fpy_status,
trend=self._calculate_trend("Rework Rate"),
last_updated=datetime.now()
)
]
for m in metrics:
self.add_metric(m)
return metrics
def calculate_safety_kpis(self,
incidents: int,
near_misses: int,
worked_hours: float,
safety_observations: int) -> List[KPIMetric]:
"""Calculate safety-related KPIs."""
# TRIR (Total Recordable Incident Rate)
trir = (incidents * 200000) / worked_hours if worked_hours > 0 else 0
trir_status = KPIStatus.ON_TRACK if incidents == 0 else (
KPIStatus.AT_RISK if incidents <= 2 else KPIStatus.CRITICAL
)
# LTIR (Lost Time Incident Rate)
ltir = (incidents * 1000000) / worked_hours if worked_hours > 0 else 0
metrics = [
KPIMetric(
name="TRIR",
category=KPICategory.SAFETY,
current_value=round(trir, 2),
target_value=0,
unit="per 200k hrs",
status=trir_status,
trend=self._calculate_trend("TRIR"),
last_updated=datetime.now(),
description="Total Recordable Incident Rate"
),
KPIMetric(
name="Safety Observations",
category=KPICategory.SAFETY,
current_value=safety_observations,
target_value=50,
unit="count",
status=KPIStatus.ON_TRACK if safety_observations >= 50 else KPIStatus.AT_RISK,
trend=self._calculate_trend("Safety Observations"),
last_updated=datetime.now()
),
KPIMetric(
name="Near Miss Reports",
category=KPICategory.SAFETY,
current_value=near_misses,
target_value=10,
unit="count",
status=KPIStatus.ON_TRACK,
trend=self._calculate_trend("Near Miss Reports"),
last_updated=datetime.now()
)
]
for m in metrics:
self.add_metric(m)
return metrics
def _get_status(self, value: float, category: str, inverse: bool = False) -> KPIStatus:
"""Determine RAG status based on thresholds."""
thresholds = self.THRESHOLDS.get(category, {'green': 0.95, 'amber': 0.85})
if inverse:
if value >= thresholds['green']:
return KPIStatus.ON_TRACK
elif value >= thresholds['amber']:
return KPIStatus.AT_RISK
else:
return KPIStatus.CRITICAL
else:
if value >= thresholds['green']:
return KPIStatus.ON_TRACK
elif value >= thresholds['amber']:
return KPIStatus.AT_RISK
else:
return KPIStatus.CRITICAL
def _calculate_trend(self, metric_name: str) -> str:
"""Calculate trend based on historical data."""
history = [h for h in self.history if h['name'] == metric_name]
if len(history) < 2:
return "stable"
recent = history[-1]['value']
previous = history[-2]['value']
if recent > previous * 1.02:
return "up"
elif recent < previous * 0.98:
return "down"
return "stable"
def get_dashboard_summary(self) -> Dict[str, Any]:
"""Generate dashboard summary."""
by_category = {}
for metric in self.metrics.values():
cat = metric.category.value
if cat not in by_category:
by_category[cat] = []
by_category[cat].append({
'name': metric.name,
'value': metric.current_value,
'target': metric.target_value,
'unit': metric.unit,
'status': metric.status.value,
'trend': metric.trend,
'variance': round(metric.variance, 1)
})
# Overall health
statuses = [m.status for m in self.metrics.values()]
critical_count = sum(1 for s in statuses if s == KPIStatus.CRITICAL)
at_risk_count = sum(1 for s in statuses if s == KPIStatus.AT_RISK)
if critical_count > 0:
overall = "CRITICAL"
elif at_risk_count > 2:
overall = "AT_RISK"
else:
overall = "ON_TRACK"
return {
'project': self.config.project_name,
'project_code': self.config.project_code,
'generated_at': datetime.now().isoformat(),
'overall_health': overall,
'metrics_count': len(self.metrics),
'critical_count': critical_count,
'at_risk_count': at_risk_count,
'kpis_by_category': by_category
}
def export_to_dataframe(self) -> pd.DataFrame:
"""Export all KPIs to DataFrame."""
data = []
for metric in self.metrics.values():
data.append({
'KPI': metric.name,
'Category': metric.category.value,
'Current': metric.current_value,
'Target': metric.target_value,
'Unit': metric.unit,
'Variance %': round(metric.variance, 1),
'Status': metric.status.value,
'Trend': metric.trend,
'Last Updated': metric.last_updated
})
return pd.DataFrame(data)
from datetime import date
# Configure dashboard
config = DashboardConfig(
project_name="Office Tower Construction",
project_code="PRJ-2024-001",
start_date=date(2024, 1, 1),
end_date=date(2025, 12, 31),
budget=50000000,
currency="USD"
)
# Initialize dashboard
dashboard = ProjectKPIDashboard(config)
# Calculate schedule KPIs
dashboard.calculate_schedule_kpis(
planned_activities=100,
completed_activities=85,
planned_duration_days=180,
actual_duration_days=195
)
# Calculate cost KPIs
dashboard.calculate_cost_kpis(
budgeted_cost=25000000,
actual_cost=24500000,
earned_value=24000000
)
# Get summary
summary = dashboard.get_dashboard_summary()
print(f"Overall Health: {summary['overall_health']}")
df = dashboard.export_to_dataframe()
critical = df[df['Status'] == 'critical']
print(f"Critical KPIs requiring attention: {len(critical)}")
# Get historical data for a metric
spi_history = [h for h in dashboard.history if h['name'] == 'Schedule Performance Index']
projects = []
for project_config in project_configs:
dash = ProjectKPIDashboard(project_config)
# ... calculate KPIs
projects.append(dash.get_dashboard_summary())
まだレビューはありません。使ってみた感想をお寄せください。
概要と使いどころ
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.
日本語の概要は準備中です。原文の説明を表示しています。