13. Dashboard Updates

Static dashboards quickly become stale. Implement automated refresh mechanisms that update visualizations when underlying data changes or when AI detects significant trends.

Webhook-Based Updates

Grafana and similar tools support webhooks for dashboard updates.

# dashboard_updater.py
import json
import requests
from datetime import datetime
from pathlib import Path
import ollama

class DashboardUpdater:
    def __init__(self, config_path: str = "dashboard_config.json"):
        self.config = self._load_config(config_path)
        self.update_log = Path("update_log.jsonl")
    
    def _load_config(self, path: str) -> dict:
        with open(path) as f:
            return json.load(f)
    
    def check_for_updates(self) -> list[dict]:
        """Check all configured dashboards for required updates."""
        updates_needed = []
        
        for dashboard in self.config["dashboards"]:
            if self._needs_update(dashboard):
                updates_needed.append(self._prepare_update(dashboard))
        
        return updates_needed
    
    def _needs_update(self, dashboard: dict) -> bool:
        """Determine if dashboard requires refresh."""
        last_update = dashboard.get("last_update")
        refresh_interval = dashboard.get("refresh_minutes", 60)
        
        if not last_update:
            return True
        
        last = datetime.fromisoformat(last_update)
        age_minutes = (datetime.now() - last).total_seconds() / 60
        
        return age_minutes >= refresh_interval
    
    def _prepare_update(self, dashboard: dict) -> dict:
        """Generate new data for dashboard."""
        dashboard_type = dashboard["type"]
        
        if dashboard_type == "anomaly_detection":
            return self._prepare_anomaly_dashboard(dashboard)
        elif dashboard_type == "trend_analysis":
            return self._prepare_trend_dashboard(dashboard)
        elif dashboard_type == "forecast":
            return self._prepare_forecast_dashboard(dashboard)
        else:
            return {"error": f"Unknown dashboard type: {dashboard_type}"}
    
    def _prepare_anomaly_dashboard(self, dashboard: dict) -> dict:
        """Update anomaly detection dashboard."""
        # Fetch latest metrics
        metrics = self._fetch_metrics(dashboard["metrics"])
        
        # Run AI analysis
        prompt = f"""Analyze these metrics for anomalies:
{json.dumps(metrics, indent=2)}

Return JSON with:
- "anomalies": list of detected anomalies
- "severity": "low" | "medium" | "high"
- "recommendations": list of suggested actions
- "chart_data": object for visualization update
"""
        
        response = ollama.chat(
            model=self.config.get("model", "llama3"),
            messages=[{"role": "user", "content": prompt}],
            options={"temperature": 0.3}
        )
        
        try:
            analysis = json.loads(response["message"]["content"])
            return {
                "dashboard_id": dashboard["id"],
                "panel_updates": analysis.get("chart_data", {}),
                "alerts": analysis.get("anomalies", []),
                "severity": analysis.get("severity", "low"),
                "timestamp": datetime.now().isoformat()
            }
        except json.JSONDecodeError:
            return {"dashboard_id": dashboard["id"], "error": "Parse failed"}
    
    def _fetch_metrics(self, metric_names: list[str]) -> dict:
        """Fetch current values for specified metrics."""
        # Implementation depends on metrics source
        return {name: self._fetch_single_metric(name) for name in metric_names}
    
    def _fetch_single_metric(self, metric_name: str) -> dict:
        """Fetch a single metric value."""
        # Placeholder - implement based on your metrics system
        return {"value": 0, "timestamp": datetime.now().isoformat()}
    
    def _prepare_trend_dashboard(self, dashboard: dict) -> dict:
        """Update trend analysis dashboard."""
        historical_data = self._fetch_historical(dashboard["time_range"])
        
        prompt = f"""Analyze these historical trends and provide forecast:
{json.dumps(historical_data)}

Return JSON with:
- "trend_direction": "up" | "down" | "stable"
- "growth_rate": percentage
- "forecast": array of predicted values for next period
- "insights": key observations
"""
        
        response = ollama.chat(
            model=self.config.get("model", "llama3"),
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            analysis = json.loads(response["message"]["content"])
            return {
                "dashboard_id": dashboard["id"],
                "trend_data": analysis,
                "timestamp": datetime.now().isoformat()
            }
        except json.JSONDecodeError:
            return {"dashboard_id": dashboard["id"], "error": "Parse failed"}
    
    def _prepare_forecast_dashboard(self, dashboard: dict) -> dict:
        """Update forecast dashboard."""
        features = self._prepare_forecast_features(dashboard)
        
        prompt = f"""Generate forecast based on these features:
{json.dumps(features)}

Return JSON with:
- "predictions": list of predicted values with confidence intervals
- "model_used": description of forecasting approach
- "accuracy_estimate": percentage
"""
        
        response = ollama.chat(
            model=self.config.get("model", "llama3"),
            messages=[{"role": "user", "content": prompt}]
        )
        
        try:
            forecast = json.loads(response["message"]["content"])
            return {
                "dashboard_id": dashboard["id"],
                "forecast": forecast,
                "timestamp": datetime.now().isoformat()
            }
        except json.JSONDecodeError:
            return {"dashboard_id": dashboard["id"], "error": "Parse failed"}
    
    def push_updates(self, updates: list[dict]):
        """Push dashboard updates to visualization platform."""
        for update in updates:
            if "error" in update:
                continue
            
            dashboard = self._find_dashboard(update["dashboard_id"])
            if not dashboard:
                continue
            
            self._update_grafana(dashboard, update) if dashboard["platform"] == "grafana" else None
            self._update_grafana(dashboard, update) if dashboard["platform"] == "grafana" else None
    
    def _update_grafana(self, dashboard: dict, update: dict):
        """Update Grafana panel with new data."""
        panel_url = dashboard["panel_url"]
        
        # Update specific panel via Grafana API
        headers = {"Authorization": f"Bearer {dashboard['api_key']}"}
        
        payload = self._format_grafana_payload(update)
        response = requests.patch(panel_url, json=payload, headers=headers)
        
        self._log_update(dashboard["id"], response.status_code)
    
    def _log_update(self, dashboard_id: str, status_code: int):
        """Record update attempt."""
        with open(self.update_log, "a") as f:
            f.write(json.dumps({
                "dashboard_id": dashboard_id,
                "status_code": status_code,
                "timestamp": datetime.now().isoformat()
            }) + "\n")

Automated Refresh Schedule

# Run as a continuous service
if __name__ == "__main__":
    import time
    
    updater = DashboardUpdater()
    
    while True:
        try:
            updates = updater.check_for_updates()
            if updates:
                updater.push_updates(updates)
                print(f"Pushed {len(updates)} dashboard updates")
        except Exception as e:
            print(f"Update cycle failed: {e}")
        
        time.sleep(60)  # Check every minute

// dashboard_config.json
{
  "model": "llama3",
  "dashboards": [
    {
      "id": "sales_anomalies",
      "type": "anomaly_detection",
      "metrics": ["revenue", "orders", "conversion_rate"],
      "refresh_minutes": 15,
      "platform": "grafana",
      "panel_url": "https://grafana.example.com/api/panels/123",
      "api_key": "your-api-key",
      "last_update": null
    },
    {
      "id": "inventory_trends",
      "type": "trend_analysis",
      "time_range": "7d",
      "refresh_minutes": 60,
      "platform": "grafana",
      "panel_url": "https://grafana.example.com/api/panels/456",
      "api_key": "your-api-key"
    },
    {
      "id": "demand_forecast",
      "type": "forecast",
      "time_range": "30d",
      "refresh_minutes": 240,
      "platform": "grafana",
      "panel_url": "https://grafana.example.com/api/panels/789",
      "api_key": "your-api-key"
    }
  ]
}