#!/usr/bin/env python3 """ Real-time Analytics Dashboard Comprehensive real-time monitoring and visualization system for concert events """ import json import math import base64 import hashlib import datetime import re import string import itertools import collections import dataclasses import random from typing import Dict, List, Optional, Any, Union, Tuple from dataclasses import dataclass, field from collections import defaultdict, deque from enum import Enum class AlertSeverity(Enum): LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" class MetricType(Enum): COUNTER = "counter" GAUGE = "gauge" HISTOGRAM = "histogram" TIMELINE = "timeline" @dataclass class Metric: """Real-time metric definition""" metric_id: str name: str description: str metric_type: MetricType unit: str category: str value: Union[float, int, str] = 0.0 timestamp: datetime.datetime = field(default_factory=datetime.datetime.now) tags: Dict[str, str] = field(default_factory=dict) @dataclass class Alert: """Alert definition and configuration""" alert_id: str name: str description: str metric_id: str condition: str # >, <, ==, != threshold: float severity: AlertSeverity enabled: bool = True triggered_count: int = 0 last_triggered: Optional[datetime.datetime] = None @dataclass class DashboardWidget: """Dashboard widget configuration""" widget_id: str title: str widget_type: str # chart, metric, table, gauge metrics: List[str] position: Dict[str, int] # x, y, width, height refresh_interval: int = 5 # seconds chart_config: Dict[str, Any] = field(default_factory=dict) class RealtimeAnalyticsDashboard: """ Real-time analytics dashboard for concert monitoring Provides live metrics, alerts, and visualizations """ def __init__(self): self.metrics = {} self.alerts = {} self.widgets = {} self.metric_history = defaultdict(lambda: deque(maxlen=1000)) self.alert_history = deque(maxlen=500) self.dashboard_state = { "last_update": datetime.datetime.now(), "active_alerts": 0, "total_metrics": 0, "dashboard_health": "healthy" } # Initialize core concert metrics self._initialize_concert_metrics() def _initialize_concert_metrics(self): """Initialize core concert monitoring metrics""" core_metrics = [ Metric("attendance_rate", "Attendance Rate", "Percentage of capacity filled", MetricType.GAUGE, "%", "attendance"), Metric("ticket_sales_rate", "Ticket Sales Rate", "Tickets sold per minute", MetricType.COUNTER, "tickets/min", "sales"), Metric("avg_wait_time", "Average Wait Time", "Average entry wait time", MetricType.GAUGE, "minutes", "operations"), Metric("social_sentiment", "Social Sentiment Score", "Real-time social media sentiment", MetricType.GAUGE, "score", "social"), Metric("audio_quality", "Audio Quality Score", "Real-time audio system performance", MetricType.GAUGE, "score", "technical"), Metric("crowd_density", "Crowd Density", "Current crowd density by zone", MetricType.GAUGE, "people/m²", "safety"), Metric("revenue_per_minute", "Revenue Rate", "Revenue generated per minute", MetricType.COUNTER, "$/min", "financial"), Metric("incident_count", "Incident Count", "Number of security/medical incidents", MetricType.COUNTER, "count", "safety"), ] for metric in core_metrics: self.metrics[metric.metric_id] = metric def update_metric(self, metric_id: str, value: Union[float, int, str], tags: Optional[Dict[str, str]] = None) -> bool: """Update metric value with timestamp""" try: if metric_id not in self.metrics: return False # Update metric self.metrics[metric_id].value = value self.metrics[metric_id].timestamp = datetime.datetime.now() if tags: self.metrics[metric_id].tags.update(tags) # Store in history self.metric_history[metric_id].append({ "value": value, "timestamp": datetime.datetime.now().isoformat(), "tags": tags or {} }) # Check alerts self._check_alerts(metric_id, value) return True except Exception as e: return False def create_alert(self, alert: Alert) -> bool: """Create a new alert rule""" try: self.alerts[alert.alert_id] = alert return True except Exception as e: return False def _check_alerts(self, metric_id: str, value: Union[float, int, str]): """Check if any alerts should be triggered""" for alert in self.alerts.values(): if not alert.enabled or alert.metric_id != metric_id: continue triggered = False try: if alert.condition == ">" and float(value) > alert.threshold: triggered = True elif alert.condition == "<" and float(value) < alert.threshold: triggered = True elif alert.condition == "==" and float(value) == alert.threshold: triggered = True elif alert.condition == "!=" and float(value) != alert.threshold: triggered = True except (ValueError, TypeError): continue if triggered: self._trigger_alert(alert, value) def _trigger_alert(self, alert: Alert, value: Union[float, int, str]): """Trigger an alert""" alert.triggered_count += 1 alert.last_triggered = datetime.datetime.now() alert_entry = { "alert_id": alert.alert_id, "name": alert.name, "severity": alert.severity.value, "value": value, "threshold": alert.threshold, "timestamp": datetime.datetime.now().isoformat() } self.alert_history.append(alert_entry) self.dashboard_state["active_alerts"] += 1 def create_widget(self, widget: DashboardWidget) -> bool: """Create a dashboard widget""" try: self.widgets[widget.widget_id] = widget return True except Exception as e: return False def get_dashboard_data(self, widget_id: Optional[str] = None) -> Dict[str, Any]: """Get dashboard data for rendering""" if widget_id and widget_id in self.widgets: widgets = {widget_id: self.widgets[widget_id]} else: widgets = self.widgets dashboard_data = { "timestamp": datetime.datetime.now().isoformat(), "metrics": {}, "alerts": [], "widgets": {}, "summary": self.dashboard_state } # Get current metric values for metric_id in self.metrics: dashboard_data["metrics"][metric_id] = { "value": self.metrics[metric_id].value, "timestamp": self.metrics[metric_id].timestamp.isoformat(), "unit": self.metrics[metric_id].unit, "type": self.metrics[metric_id].metric_type.value, "category": self.metrics[metric_id].category } # Get recent alerts dashboard_data["alerts"] = list(self.alert_history)[-10:] # Last 10 alerts # Get widget configurations with data for wid, widget in widgets.items(): widget_data = { "id": wid, "title": widget.title, "type": widget.widget_type, "position": widget.position, "data": [] } # Populate widget with metric data for metric_id in widget.metrics: if metric_id in self.metrics: metric_data = { "metric_id": metric_id, "current_value": self.metrics[metric_id].value, "history": list(self.metric_history[metric_id])[-50:] # Last 50 data points } widget_data["data"].append(metric_data) dashboard_data["widgets"][wid] = widget_data return dashboard_data def get_metric_trends(self, metric_id: str, time_range: str = "1h") -> Dict[str, Any]: """Get metric trends over time""" if metric_id not in self.metrics: return {"error": "Metric not found"} history = list(self.metric_history[metric_id]) if not history: return {"error": "No data available"} # Calculate trends values = [entry["value"] for entry in history if isinstance(entry["value"], (int, float))] if not values: return {"error": "No numeric data available"} trends = { "metric_id": metric_id, "time_range": time_range, "current_value": values[-1] if values else 0, "avg_value": sum(values) / len(values) if values else 0, "min_value": min(values) if values else 0, "max_value": max(values) if values else 0, "trend_direction": "stable", "trend_percentage": 0.0, "data_points": len(history) } # Calculate trend direction if len(values) >= 2: recent_avg = sum(values[-10:]) / min(10, len(values)) earlier_avg = sum(values[:10]) / min(10, len(values)) if earlier_avg != 0: change_percent = ((recent_avg - earlier_avg) / earlier_avg) * 100 trends["trend_percentage"] = change_percent if change_percent > 5: trends["trend_direction"] = "increasing" elif change_percent < -5: trends["trend_direction"] = "decreasing" return trends def get_alert_summary(self) -> Dict[str, Any]: """Get comprehensive alert summary""" if not self.alert_history: return {"total_alerts": 0, "by_severity": {}, "recent_alerts": []} # Group alerts by severity severity_counts = defaultdict(int) for alert in self.alert_history: severity_counts[alert["severity"]] += 1 summary = { "total_alerts": len(self.alert_history), "by_severity": dict(severity_counts), "active_alerts": self.dashboard_state["active_alerts"], "recent_alerts": list(self.alert_history)[-5:], # Last 5 alerts "top_triggered": self._get_top_triggered_alerts() } return summary def _get_top_triggered_alerts(self) -> List[Dict[str, Any]]: """Get most frequently triggered alerts""" alert_counts = defaultdict(int) for alert in self.alert_history: alert_counts[alert["alert_id"]] += 1 top_alerts = sorted(alert_counts.items(), key=lambda x: x[1], reverse=True)[:5] result = [] for alert_id, count in top_alerts: if alert_id in self.alerts: result.append({ "alert_id": alert_id, "name": self.alerts[alert_id].name, "triggered_count": count }) return result def simulate_concert_data(self, duration_minutes: int = 60) -> Dict[str, Any]: """Simulate realistic concert data for testing""" simulation_results = { "duration_minutes": duration_minutes, "data_points_generated": 0, "alerts_triggered": 0 } base_metrics = { "attendance_rate": 0.0, "ticket_sales_rate": 50.0, "avg_wait_time": 15.0, "social_sentiment": 0.7, "audio_quality": 0.85, "crowd_density": 2.5, "revenue_per_minute": 1000.0, "incident_count": 0 } for minute in range(duration_minutes): # Simulate metric changes over time progress = minute / duration_minutes # Attendance increases during event base_metrics["attendance_rate"] = min(95.0, progress * 100 + random.random() * 10) # Sales rate peaks then declines base_metrics["ticket_sales_rate"] = max(5.0, 50.0 * (1 - progress) + random.random() * 20) # Wait time varies base_metrics["avg_wait_time"] = 10.0 + random.random() * 20 # Social sentiment fluctuates base_metrics["social_sentiment"] = 0.5 + random.random() * 0.4 # Audio quality generally high with occasional drops base_metrics["audio_quality"] = max(0.6, 0.9 - random.random() * 0.3) # Crowd density peaks mid-event base_metrics["crowd_density"] = 2.0 + abs(math.sin(progress * math.pi)) * 2 + random.random() # Revenue correlates with attendance base_metrics["revenue_per_minute"] = base_metrics["attendance_rate"] * 10 + random.random() * 500 # Random incidents if random.random() < 0.1: # 10% chance per minute base_metrics["incident_count"] += 1 # Update metrics for metric_id, value in base_metrics.items(): self.update_metric(metric_id, value) simulation_results["data_points_generated"] += 1 simulation_results["alerts_triggered"] = self.dashboard_state["active_alerts"] return simulation_results def export_dashboard_config(self) -> Dict[str, Any]: """Export dashboard configuration""" config = { "metrics": {}, "alerts": {}, "widgets": {}, "export_timestamp": datetime.datetime.now().isoformat() } for metric_id, metric in self.metrics.items(): config["metrics"][metric_id] = { "metric_id": metric.metric_id, "name": metric.name, "description": metric.description, "type": metric.metric_type.value, "unit": metric.unit, "category": metric.category } for alert_id, alert in self.alerts.items(): config["alerts"][alert_id] = { "alert_id": alert.alert_id, "name": alert.name, "description": alert.description, "metric_id": alert.metric_id, "condition": alert.condition, "threshold": alert.threshold, "severity": alert.severity.value, "enabled": alert.enabled } for widget_id, widget in self.widgets.items(): config["widgets"][widget_id] = { "widget_id": widget.widget_id, "title": widget.title, "type": widget.widget_type, "metrics": widget.metrics, "position": widget.position, "refresh_interval": widget.refresh_interval } return config def initialize_sample_dashboard() -> RealtimeAnalyticsDashboard: """Initialize sample dashboard with widgets and alerts""" dashboard = RealtimeAnalyticsDashboard() # Create sample alerts alerts = [ Alert("high_wait_time", "High Wait Time", "Entry wait time exceeds threshold", "avg_wait_time", ">", 30, AlertSeverity.HIGH), Alert("low_audio_quality", "Low Audio Quality", "Audio quality below acceptable", "audio_quality", "<", 0.7, AlertSeverity.MEDIUM), Alert("high_crowd_density", "High Crowd Density", "Crowd density exceeds safety limits", "crowd_density", ">", 4.0, AlertSeverity.HIGH), Alert("incident_spike", "Incident Spike", "Multiple incidents detected", "incident_count", ">", 5, AlertSeverity.CRITICAL) ] for alert in alerts: dashboard.create_alert(alert) # Create sample widgets widgets = [ DashboardWidget("attendance_widget", "Live Attendance", "gauge", ["attendance_rate"], {"x": 0, "y": 0, "width": 2, "height": 1}), DashboardWidget("revenue_widget", "Revenue Stream", "chart", ["revenue_per_minute"], {"x": 2, "y": 0, "width": 3, "height": 2}), DashboardWidget("social_widget", "Social Sentiment", "metric", ["social_sentiment"], {"x": 5, "y": 0, "width": 2, "height": 1}), DashboardWidget("safety_widget", "Safety Metrics", "table", ["crowd_density", "incident_count", "avg_wait_time"], {"x": 0, "y": 2, "width": 4, "height": 2}) ] for widget in widgets: dashboard.create_widget(widget) return dashboard if __name__ == "__main__": # Initialize dashboard dashboard = initialize_sample_dashboard() print("=== Real-time Analytics Dashboard ===") # Simulate concert data print("\nSimulating concert data...") simulation = dashboard.simulate_concert_data(30) # 30 minutes print(json.dumps(simulation, indent=2)) # Get dashboard data print("\n=== Dashboard Data ===") dashboard_data = dashboard.get_dashboard_data() print(json.dumps(dashboard_data, indent=2, default=str)) # Get metric trends print("\n=== Metric Trends ===") for metric_id in ["attendance_rate", "revenue_per_minute"]: trends = dashboard.get_metric_trends(metric_id) print(f"\n{metric_id}:") print(json.dumps(trends, indent=2)) # Get alert summary print("\n=== Alert Summary ===") alert_summary = dashboard.get_alert_summary() print(json.dumps(alert_summary, indent=2)) # Export configuration print("\n=== Dashboard Configuration ===") config = dashboard.export_dashboard_config() print(json.dumps(config, indent=2))