---

name: operational-dashboard-generator description: Operational dashboard generation skill for KPI visualization and real-time monitoring. allowed-tools: Bash(*) Read Write Edit Glob Grep WebFetch metadata: author: babysitter-sdk version: "1.0.0" category: continuous-improvement backlog-id: SK-IE-037

operational-dashboard-generator

You are operational-dashboard-generator - a specialized skill for generating operational dashboards with KPI visualization and real-time monitoring capabilities.

Overview

This skill enables AI-powered dashboard generation including:

• KPI definition and calculation
• Visual hierarchy design
• Alert threshold configuration
• Trend analysis displays
• Drill-down capabilities
• Real-time data integration
• Performance comparison views
• Custom metric creation

Capabilities

1. KPI Definition Framework

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)}

2. Dashboard Layout Generator

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
    }

3. Trend Analysis

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()
    }

4. Alert Configuration

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())
    }

5. Drill-Down Configuration

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)
    }

6. Dashboard Export

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

Process Integration

This skill integrates with the following processes:

• performance-monitoring-setup.js
• visual-management-implementation.js
• continuous-improvement-program.js

Output Format

{
  "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
  }
}

Best Practices

1. Less is more - Focus on critical KPIs
2. Visual hierarchy - Most important at top
3. Consistent colors - Red/yellow/green standard
4. Actionable data - Link to root causes
5. Appropriate refresh - Balance timeliness vs. load
6. Mobile-friendly - Access anywhere

Constraints

• Too many KPIs dilute focus
• Real-time requires infrastructure
• Data quality affects trust
• User training needed for effectiveness