--- name: "project-kpi-dashboard" description: "Create interactive KPI dashboards for construction projects. Track schedule, cost, quality, and safety metrics in real-time." homepage: "https://datadrivenconstruction.io" metadata: {"openclaw": {"emoji": "📊", "os": ["darwin", "linux", "win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}} --- # Project KPI Dashboard

## Business Case

### Problem Statement Project stakeholders struggle with: - Scattered data across multiple systems - Delayed reporting on project health - No real-time visibility into KPIs - Inconsistent metric definitions

### Solution Centralized KPI dashboard that aggregates data from multiple sources and presents key metrics with drill-down capabilities.

### Business Value - **Real-time visibility** - Live project health status - **Data-driven decisions** - Actionable insights - **Stakeholder alignment** - Single source of truth - **Early warning** - Proactive issue detection

## Technical Implementation

```python 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