Operational dashboard generation skill for KPI visualization and real-time monitoring.
You are operational-dashboard-generator - a specialized skill for generating operational dashboards with KPI visualization and real-time monitoring capabilities.
This skill enables AI-powered dashboard generation including:
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
class KPICategory(Enum):
SAFETY = "safety"
QUALITY = "quality"
DELIVERY = "delivery"
COST = "cost"
PRODUCTIVITY = "productivity"
MORALE = "morale"
@dataclass
class KPIDefinition:
id: str
name: str
category: KPICategory
formula: str
unit: str
target: float
warning_threshold: float
critical_threshold: float
higher_is_better: bool = True
frequency: str = "daily"
def create_kpi_library():
"""Standard industrial KPI definitions"""
return [
KPIDefinition(
id="oee", name="OEE", category=KPICategory.PRODUCTIVITY,
formula="availability * performance * quality",
unit="%", target=85, warning_threshold=75, critical_threshold=65
),
KPIDefinition(
id="fpy", name="First Pass Yield", category=KPICategory.QUALITY,
formula="good_units / total_units * 100",
unit="%", target=98, warning_threshold=95, critical_threshold=90
),
KPIDefinition(
id="otd", name="On-Time Delivery", category=KPICategory.DELIVERY,
formula="on_time_orders / total_orders * 100",
unit="%", target=98, warning_threshold=95, critical_threshold=90
),
KPIDefinition(
id="trir", name="Total Recordable Incident Rate", category=KPICategory.SAFETY,
formula="incidents * 200000 / hours_worked",
unit="per 200k hrs", target=0.5, warning_threshold=1.0, critical_threshold=2.0,
higher_is_better=False
),
KPIDefinition(
id="productivity", name="Labor Productivity", category=KPICategory.PRODUCTIVITY,
formula="units_produced / labor_hours",
unit="units/hr", target=25, warning_threshold=20, critical_threshold=15
),
KPIDefinition(
id="scrap_rate", name="Scrap Rate", category=KPICategory.COST,
formula="scrap_cost / production_cost * 100",
unit="%", target=1.0, warning_threshold=2.0, critical_threshold=3.0,
higher_is_better=False
)
]
def calculate_kpi(kpi: KPIDefinition, data: dict):
"""Calculate KPI value from data"""
# Simple formula evaluation (in production, use safe eval)
try:
value = eval(kpi.formula, {"__builtins__": {}}, data)
# Determine status
if kpi.higher_is_better:
if value >= kpi.target:
status = "green"
elif value >= kpi.warning_threshold:
status = "yellow"
else:
status = "red"
else:
if value <= kpi.target:
status = "green"
elif value <= kpi.warning_threshold:
status = "yellow"
else:
status = "red"
return {
"kpi_id": kpi.id,
"name": kpi.name,
"value": round(value, 2),
"unit": kpi.unit,
"target": kpi.target,
"status": status,
"gap_to_target": round(value - kpi.target, 2)
}
except Exception as e:
return {"kpi_id": kpi.id, "error": str(e)}
def generate_dashboard_layout(kpis: List[KPIDefinition], layout_type: str = "standard"):
"""
Generate dashboard layout configuration
layout_type: 'standard', 'executive', 'operational', 'lean'
"""
if layout_type == "standard":
layout = {
"type": "grid",
"columns": 4,
"rows": 3,
"sections": [
{
"id": "header",
"row": 1, "col_span": 4,
"content": "title_and_period_selector"
},
{
"id": "summary_cards",
"row": 2, "col_span": 4,
"content": "kpi_summary_cards",
"kpis": [k.id for k in kpis[:6]]
},
{
"id": "trends",
"row": 3, "col": 1, "col_span": 2,
"content": "trend_charts"
},
{
"id": "details",
"row": 3, "col": 3, "col_span": 2,
"content": "detail_tables"
}
]
}
elif layout_type == "executive":
layout = {
"type": "single_page",
"sections": [
{"id": "headline_kpis", "position": "top", "height": "20%"},
{"id": "trend_summary", "position": "middle", "height": "50%"},
{"id": "action_items", "position": "bottom", "height": "30%"}
]
}
elif layout_type == "lean":
# Visual management board style
layout = {
"type": "sqdc", # Safety, Quality, Delivery, Cost
"columns": 4,
"sections": [
{"id": "safety", "col": 1, "category": "SAFETY"},
{"id": "quality", "col": 2, "category": "QUALITY"},
{"id": "delivery", "col": 3, "category": "DELIVERY"},
{"id": "cost", "col": 4, "category": "COST"}
],
"row_types": ["current_status", "trend_sparkline", "action_items"]
}
return layout
def generate_kpi_card_config(kpi: KPIDefinition):
"""Generate configuration for a KPI card widget"""
return {
"widget_type": "kpi_card",
"kpi_id": kpi.id,
"title": kpi.name,
"display": {
"value_format": f"{{value:.1f}}{kpi.unit}",
"show_target": True,
"show_trend": True,
"trend_periods": 7
},
"colors": {
"green": kpi.target,
"yellow": kpi.warning_threshold,
"red": kpi.critical_threshold
},
"gauge": kpi.id in ["oee", "fpy", "otd"] # Show gauge for percentage KPIs
}
def analyze_kpi_trends(historical_data: pd.DataFrame, kpi_id: str,
periods: int = 30):
"""
Analyze trends for a KPI
historical_data: DataFrame with ['date', 'kpi_id', 'value']
"""
kpi_data = historical_data[historical_data['kpi_id'] == kpi_id].copy()
kpi_data = kpi_data.sort_values('date').tail(periods)
if len(kpi_data) < 2:
return {"error": "Insufficient data for trend analysis"}
values = kpi_data['value'].values
dates = kpi_data['date'].values
# Calculate statistics
current = values[-1]
previous = values[-2]
change = current - previous
change_pct = (change / previous * 100) if previous != 0 else 0
# Moving averages
ma_7 = np.mean(values[-7:]) if len(values) >= 7 else np.mean(values)
ma_30 = np.mean(values[-30:]) if len(values) >= 30 else np.mean(values)
# Trend direction (linear regression)
x = np.arange(len(values))
slope, intercept = np.polyfit(x, values, 1)
if slope > 0.01:
trend_direction = "improving"
elif slope < -0.01:
trend_direction = "declining"
else:
trend_direction = "stable"
# Variability
std_dev = np.std(values)
cv = (std_dev / np.mean(values) * 100) if np.mean(values) != 0 else 0
return {
"kpi_id": kpi_id,
"current_value": round(current, 2),
"previous_value": round(previous, 2),
"change": round(change, 2),
"change_percent": round(change_pct, 1),
"trend_direction": trend_direction,
"slope": round(slope, 4),
"moving_average_7": round(ma_7, 2),
"moving_average_30": round(ma_30, 2),
"std_deviation": round(std_dev, 2),
"coefficient_of_variation": round(cv, 1),
"sparkline_data": values.tolist()
}
def configure_alerts(kpis: List[KPIDefinition], notification_config: dict):
"""
Configure alert rules for dashboard
notification_config: {'email': [], 'sms': [], 'teams': []}
"""
alerts = []
for kpi in kpis:
# Warning alert
alerts.append({
"alert_id": f"{kpi.id}_warning",
"kpi_id": kpi.id,
"level": "warning",
"condition": f"value {'<' if kpi.higher_is_better else '>'} {kpi.warning_threshold}",
"message": f"{kpi.name} has reached warning level",
"notifications": notification_config.get('email', []),
"frequency": "first_occurrence"
})
# Critical alert
alerts.append({
"alert_id": f"{kpi.id}_critical",
"kpi_id": kpi.id,
"level": "critical",
"condition": f"value {'<' if kpi.higher_is_better else '>'} {kpi.critical_threshold}",
"message": f"{kpi.name} has reached critical level - immediate action required",
"notifications": notification_config.get('email', []) + notification_config.get('sms', []),
"frequency": "every_occurrence",
"escalation_after_minutes": 30
})
# Trend alert
alerts.append({
"alert_id": f"{kpi.id}_trend",
"kpi_id": kpi.id,
"level": "info",
"condition": "consecutive_decline >= 3",
"message": f"{kpi.name} has declined for 3 consecutive periods",
"notifications": notification_config.get('email', []),
"frequency": "daily_digest"
})
return {
"alerts": alerts,
"total_rules": len(alerts),
"notification_channels": list(notification_config.keys())
}
def configure_drilldowns(kpi: KPIDefinition, dimensions: list):
"""
Configure drill-down paths for a KPI
dimensions: ['shift', 'line', 'product', 'operator']
"""
drilldowns = []
for i, dim in enumerate(dimensions):
drilldowns.append({
"level": i + 1,
"dimension": dim,
"aggregation": "sum" if "count" in kpi.formula else "avg",
"chart_type": "bar" if i < 2 else "table",
"filter_enabled": True
})
# Add time-based drilldown
drilldowns.append({
"level": len(dimensions) + 1,
"dimension": "time",
"granularity": ["month", "week", "day", "shift", "hour"],
"chart_type": "line",
"default_granularity": "day"
})
return {
"kpi_id": kpi.id,
"drilldown_path": drilldowns,
"max_levels": len(drilldowns)
}
def export_dashboard_config(layout: dict, kpis: List[dict],
alerts: List[dict], drilldowns: List[dict]):
"""
Export complete dashboard configuration
"""
config = {
"dashboard": {
"name": "Operations Dashboard",
"version": "1.0",
"refresh_rate_seconds": 300,
"timezone": "local"
},
"layout": layout,
"kpis": kpis,
"widgets": [],
"alerts": alerts,
"drilldowns": drilldowns,
"data_sources": [
{
"id": "production_db",
"type": "database",
"refresh": "5min"
},
{
"id": "mes_api",
"type": "api",
"refresh": "realtime"
}
],
"filters": [
{"id": "date_range", "type": "date_range", "default": "last_30_days"},
{"id": "shift", "type": "dropdown", "options": ["All", "Day", "Night"]},
{"id": "line", "type": "multi_select", "source": "production_db.lines"}
]
}
# Generate widgets from KPIs
for kpi in kpis:
config["widgets"].append({
"widget_id": f"card_{kpi['kpi_id']}",
"type": "kpi_card",
"kpi": kpi['kpi_id'],
"position": "auto"
})
return config
This skill integrates with the following processes:
performance-monitoring-setup.jsvisual-management-implementation.jscontinuous-improvement-program.js{
"dashboard_config": {
"name": "Operations Dashboard",
"layout": "standard",
"refresh_rate": 300
},
"kpis": [
{
"id": "oee",
"name": "OEE",
"current_value": 82.5,
"target": 85,
"status": "yellow",
"trend": "improving"
}
],
"alerts": {
"active": 2,
"critical": 0,
"warning": 2
},
"layout_spec": {
"type": "grid",
"columns": 4,
"widgets": 12
}
}