""" Enhanced visualization engine for ARF Demo with clear boundary indicators FIXED VERSION: Works even when Plotly fails, shows clear real/simulated boundaries """ import logging from typing import Dict, List, Any, Optional, Tuple import random logger = logging.getLogger(__name__) # Try to import Plotly, but have fallbacks try: import plotly.graph_objects as go import plotly.express as px import numpy as np PLOTLY_AVAILABLE = True logger.info("✅ Plotly available for advanced visualizations") except ImportError as e: PLOTLY_AVAILABLE = False logger.warning(f"⚠️ Plotly not available: {e}. Using HTML fallback visualizations.") class EnhancedVisualizationEngine: """Enhanced visualization engine with boundary awareness and fallbacks""" def __init__(self): self.color_palette = { "primary": "#3b82f6", "success": "#10b981", "warning": "#f59e0b", "danger": "#ef4444", "info": "#8b5cf6", "dark": "#1e293b", "light": "#f8fafc", "real_arf": "#10b981", # Green for real ARF "simulated": "#f59e0b", # Amber for simulated "mock": "#64748b", # Gray for mock } def create_executive_dashboard(self, data: Optional[Dict] = None, is_real_arf: bool = True) -> Any: """ Create executive dashboard with clear boundary indicators Args: data: Dashboard data is_real_arf: Whether this is real ARF or simulated/mock Returns: Plotly figure or HTML fallback """ if data is None: data = {"roi_multiplier": 5.2} roi_multiplier = data.get("roi_multiplier", 5.2) if not PLOTLY_AVAILABLE: # HTML fallback return self._create_html_dashboard(roi_multiplier, is_real_arf) try: # Create a multi-panel executive dashboard with boundary indicators fig = go.Figure() # Main ROI gauge with boundary indicator boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF OSS" if is_real_arf else "SIMULATED" fig.add_trace(go.Indicator( mode="number+gauge", value=roi_multiplier, title={ "text": f"ROI Multiplier
" f"" f"💎 {boundary_text}" }, domain={'x': [0.25, 0.75], 'y': [0.6, 1]}, gauge={ 'axis': {'range': [0, 10], 'tickwidth': 1}, 'bar': {'color': boundary_color}, 'steps': [ {'range': [0, 2], 'color': '#e5e7eb'}, {'range': [2, 4], 'color': '#d1d5db'}, {'range': [4, 6], 'color': '#10b981'}, {'range': [6, 10], 'color': '#059669'} ], 'threshold': { 'line': {'color': "black", 'width': 4}, 'thickness': 0.75, 'value': roi_multiplier } } )) # Add boundary indicator in top right fig.add_annotation( x=0.98, y=0.98, xref="paper", yref="paper", text=f"💎 {boundary_text}", showarrow=False, font=dict(size=14, color=boundary_color, family="Arial, sans-serif"), bgcolor="white", bordercolor=boundary_color, borderwidth=2, borderpad=4, opacity=0.9 ) # Add secondary metrics with clear sourcing source_text = "Real ARF OSS" if is_real_arf else "Demo Simulation" fig.add_trace(go.Indicator( mode="number", value=85, title={ "text": f"MTTR Reduction
" f"{source_text}" }, number={'suffix': "%", 'font': {'size': 24}}, domain={'x': [0.1, 0.4], 'y': [0.2, 0.5]} )) fig.add_trace(go.Indicator( mode="number", value=94, title={ "text": f"Detection Accuracy
" f"{source_text}" }, number={'suffix': "%", 'font': {'size': 24}}, domain={'x': [0.6, 0.9], 'y': [0.2, 0.5]} )) fig.update_layout( height=700, paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(0,0,0,0)", font={'family': "Arial, sans-serif"}, margin=dict(t=50, b=50, l=50, r=50), title=f"📊 Executive Dashboard - ARF v3.3.7
" f"" f"Mode: {boundary_text}" ) return fig except Exception as e: logger.error(f"Plotly dashboard creation failed: {e}") return self._create_html_dashboard(roi_multiplier, is_real_arf) def _create_html_dashboard(self, roi_multiplier: float, is_real_arf: bool) -> str: """Create HTML fallback dashboard""" boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF OSS" if is_real_arf else "SIMULATED" return f"""
💎 {boundary_text}

📊 Executive Dashboard - ARF v3.3.7

ROI Multiplier
{roi_multiplier:.1f}×
{boundary_text} Analysis
MTTR Reduction
85%
{boundary_text}
Detection Accuracy
94%
{boundary_text}
📈 Data Source: {boundary_text} v3.3.7 • ⚡ Processing: Real-time analysis • 🎯 Confidence: Deterministic scoring
""" def create_telemetry_plot(self, scenario_name: str, anomaly_detected: bool = True, is_real_arf: bool = True) -> Any: """Create telemetry plot with boundary indicators""" if not PLOTLY_AVAILABLE: return self._create_html_telemetry(scenario_name, anomaly_detected, is_real_arf) try: import numpy as np # Generate realistic telemetry data time_points = np.arange(0, 100, 1) # Different patterns for different scenarios if "Cache" in scenario_name: base_data = 100 + 50 * np.sin(time_points * 0.2) noise = np.random.normal(0, 8, 100) metric_name = "Cache Hit Rate (%)" normal_range = (70, 95) elif "Database" in scenario_name: base_data = 70 + 30 * np.sin(time_points * 0.15) noise = np.random.normal(0, 6, 100) metric_name = "Connection Pool Usage" normal_range = (20, 60) elif "Memory" in scenario_name: base_data = 50 + 40 * np.sin(time_points * 0.1) noise = np.random.normal(0, 10, 100) metric_name = "Memory Usage (%)" normal_range = (40, 80) else: base_data = 80 + 20 * np.sin(time_points * 0.25) noise = np.random.normal(0, 5, 100) metric_name = "System Load" normal_range = (50, 90) data = base_data + noise fig = go.Figure() boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" if anomaly_detected: # Normal operation fig.add_trace(go.Scatter( x=time_points[:70], y=data[:70], mode='lines', name='Normal Operation', line=dict(color=self.color_palette["primary"], width=3), fill='tozeroy', fillcolor='rgba(59, 130, 246, 0.1)' )) # Anomaly period fig.add_trace(go.Scatter( x=time_points[70:], y=data[70:], mode='lines', name='🚨 Anomaly Detected', line=dict(color=self.color_palette["danger"], width=3, dash='dash'), fill='tozeroy', fillcolor='rgba(239, 68, 68, 0.1)' )) # Add detection point fig.add_vline( x=70, line_dash="dash", line_color=self.color_palette["success"], annotation_text=f"ARF Detection ({boundary_text})", annotation_position="top", annotation_font_color=boundary_color ) else: # All normal fig.add_trace(go.Scatter( x=time_points, y=data, mode='lines', name=metric_name, line=dict(color=self.color_palette["primary"], width=3), fill='tozeroy', fillcolor='rgba(59, 130, 246, 0.1)' )) # Add normal range fig.add_hrect( y0=normal_range[0], y1=normal_range[1], fillcolor="rgba(16, 185, 129, 0.1)", opacity=0.2, line_width=0, annotation_text="Normal Range", annotation_position="top left" ) # Add boundary indicator fig.add_annotation( x=0.02, y=0.98, xref="paper", yref="paper", text=f"💎 {boundary_text}", showarrow=False, font=dict(size=12, color=boundary_color), bgcolor="white", bordercolor=boundary_color, borderwidth=2, borderpad=4 ) fig.update_layout( title=f"📈 {metric_name} - Live Telemetry
" f"" f"ARF v3.3.7 • {boundary_text} Analysis", xaxis_title="Time (minutes)", yaxis_title=metric_name, height=350, margin=dict(l=20, r=20, t=70, b=20), plot_bgcolor='rgba(0,0,0,0)', paper_bgcolor='rgba(0,0,0,0)', legend=dict( orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1 ) ) return fig except Exception as e: logger.error(f"Telemetry plot creation failed: {e}") return self._create_html_telemetry(scenario_name, anomaly_detected, is_real_arf) def _create_html_telemetry(self, scenario_name: str, anomaly_detected: bool, is_real_arf: bool) -> str: """HTML fallback for telemetry visualization""" boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" if "Cache" in scenario_name: metric_name = "Cache Hit Rate" normal_range = (70, 95) current_value = 18 if anomaly_detected else 85 elif "Database" in scenario_name: metric_name = "Connection Pool Usage" normal_range = (20, 60) current_value = 92 if anomaly_detected else 45 else: metric_name = "System Load" normal_range = (50, 90) current_value = 185 if anomaly_detected else 65 # Calculate position in range percentage = ((current_value - normal_range[0]) / (normal_range[1] - normal_range[0])) * 100 percentage = max(0, min(100, percentage)) return f"""
💎 {boundary_text}

📈 {metric_name} - Live Telemetry

{current_value}
Normal: {normal_range[0]}
Warning: {normal_range[1]}
Current: {current_value}
Status
{'🚨 Anomaly Detected' if anomaly_detected else '✅ Normal'}
ARF Mode
{boundary_text}
""" def create_impact_gauge(self, scenario_name: str, is_real_arf: bool = True) -> Any: """Create business impact gauge with boundary indicators""" impact_map = { "Cache Miss Storm": {"revenue": 8500, "severity": "critical", "users": 45000}, "Database Connection Pool Exhaustion": {"revenue": 4200, "severity": "high", "users": 25000}, "Kubernetes Memory Leak": {"revenue": 5500, "severity": "high", "users": 35000}, "API Rate Limit Storm": {"revenue": 3800, "severity": "medium", "users": 20000}, "Network Partition": {"revenue": 12000, "severity": "critical", "users": 75000}, "Storage I/O Saturation": {"revenue": 6800, "severity": "high", "users": 30000} } impact = impact_map.get(scenario_name, {"revenue": 5000, "severity": "medium", "users": 30000}) if not PLOTLY_AVAILABLE: return self._create_html_impact_gauge(impact, is_real_arf) try: boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" severity_color = self._get_severity_color(impact["severity"]) fig = go.Figure(go.Indicator( mode="gauge+number", value=impact["revenue"], title={ "text": f"💰 Hourly Revenue Risk
" f"" f"{boundary_text} Analysis" }, number={'prefix': "$", 'font': {'size': 28}}, gauge={ 'axis': {'range': [0, 15000], 'tickwidth': 1}, 'bar': {'color': severity_color}, 'steps': [ {'range': [0, 3000], 'color': '#10b981'}, {'range': [3000, 7000], 'color': '#f59e0b'}, {'range': [7000, 15000], 'color': '#ef4444'} ], 'threshold': { 'line': {'color': "black", 'width': 4}, 'thickness': 0.75, 'value': impact["revenue"] } } )) # Add users affected annotation fig.add_annotation( x=0.5, y=0.3, xref="paper", yref="paper", text=f"👥 {impact['users']:,} users affected", showarrow=False, font=dict(size=14, color="#64748b") ) fig.update_layout( height=350, margin=dict(l=20, r=20, t=80, b=20), paper_bgcolor='rgba(0,0,0,0)', plot_bgcolor='rgba(0,0,0,0)' ) return fig except Exception as e: logger.error(f"Impact gauge creation failed: {e}") return self._create_html_impact_gauge(impact, is_real_arf) def _create_html_impact_gauge(self, impact: Dict, is_real_arf: bool) -> str: """HTML fallback for impact gauge""" boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" severity_color = self._get_severity_color(impact["severity"]) # Calculate percentage for gauge revenue = impact["revenue"] percentage = min(100, (revenue / 15000) * 100) return f"""
💎 {boundary_text}

💰 Business Impact

${revenue:,}
per hour
{impact['severity']} SEVERITY
👥 Users Affected
{impact['users']:,}
Analysis: {boundary_text} v3.3.7
""" def create_timeline_comparison(self, is_real_arf: bool = True) -> Any: """Create timeline comparison chart""" if not PLOTLY_AVAILABLE: return self._create_html_timeline(is_real_arf) try: import numpy as np phases = ["Detection", "Analysis", "Decision", "Execution", "Recovery"] manual_times = [300, 1800, 1200, 1800, 3600] # seconds arf_times = [45, 30, 60, 720, 0] # Convert to minutes for readability manual_times_min = [t/60 for t in manual_times] arf_times_min = [t/60 for t in arf_times] fig = go.Figure() boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" fig.add_trace(go.Bar( name='Manual Process', x=phases, y=manual_times_min, marker_color=self.color_palette["danger"], text=[f"{t:.0f}m" for t in manual_times_min], textposition='auto' )) fig.add_trace(go.Bar( name=f'ARF Autonomous ({boundary_text})', x=phases, y=arf_times_min, marker_color=boundary_color, text=[f"{t:.0f}m" for t in arf_times_min], textposition='auto' )) total_manual = sum(manual_times_min) total_arf = sum(arf_times_min) fig.update_layout( title=f"⏰ Incident Timeline Comparison
" f"" f"Total: {total_manual:.0f}m manual vs {total_arf:.0f}m ARF " f"({((total_manual - total_arf) / total_manual * 100):.0f}% faster)", barmode='group', height=400, plot_bgcolor='rgba(0,0,0,0)', paper_bgcolor='rgba(0,0,0,0)', legend=dict( orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1 ), yaxis_title="Time (minutes)" ) return fig except Exception as e: logger.error(f"Timeline comparison creation failed: {e}") return self._create_html_timeline(is_real_arf) def _create_html_timeline(self, is_real_arf: bool) -> str: """HTML fallback for timeline comparison""" boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" phases = ["Detection", "Analysis", "Decision", "Execution", "Recovery"] manual_times = [5, 30, 20, 30, 60] # minutes arf_times = [0.75, 0.5, 1, 12, 0] # minutes # Calculate totals total_manual = sum(manual_times) total_arf = sum(arf_times) percent_faster = ((total_manual - total_arf) / total_manual) * 100 return f"""
⏰ {boundary_text}

⏰ Timeline Comparison

Manual: {total_manual:.0f} minutes • ARF ({boundary_text}): {total_arf:.0f} minutes • Faster: {percent_faster:.0f}%
{self._create_timeline_segments(phases, manual_times, arf_times, boundary_color)}
Manual Process
ARF ({boundary_text})
🎯 ARF Value: Reduces MTTR from {total_manual:.0f} to {total_arf:.0f} minutes
💎 Mode: {boundary_text} autonomous execution
📈 Impact: {percent_faster:.0f}% faster resolution
""" def _create_timeline_segments(self, phases: List[str], manual_times: List[float], arf_times: List[float], boundary_color: str) -> str: """Create timeline segments HTML""" segments_html = "" total_width = 0 for i, (phase, manual_time, arf_time) in enumerate(zip(phases, manual_times, arf_times)): # Calculate widths as percentages manual_width = (manual_time / 125) * 100 # 125 = sum of all times arf_width = (arf_time / 125) * 100 segments_html += f"""
{phase}
{manual_time:.0f}m
{arf_time:.0f}m
""" total_width += manual_width return segments_html def _get_severity_color(self, severity: str) -> str: """Get color for severity level""" color_map = { "critical": self.color_palette["danger"], "high": self.color_palette["warning"], "medium": self.color_palette["info"], "low": self.color_palette["success"] } return color_map.get(severity.lower(), self.color_palette["info"]) # Keep other methods from original file but add boundary parameters def create_agent_performance_chart(self, is_real_arf: bool = True) -> Any: """Create agent performance comparison chart""" if not PLOTLY_AVAILABLE: # HTML fallback boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" agents = ["Detection", "Recall", "Decision"] accuracy = [98.7, 92.0, 94.0] speed = [45, 30, 60] rows = "" for agent, acc, spd in zip(agents, accuracy, speed): rows += f"""
{agent}
{acc}%
{spd}s
""" return f"""
🤖 {boundary_text}

🤖 Agent Performance

{rows}
Agent Accuracy Speed
Mode: {boundary_text} v3.3.7 • Avg Accuracy: {(sum(accuracy)/len(accuracy)):.1f}% • Avg Speed: {(sum(speed)/len(speed)):.0f}s
""" # Original Plotly implementation with boundary additions try: agents = ["Detection", "Recall", "Decision"] accuracy = [98.7, 92.0, 94.0] speed = [45, 30, 60] # seconds confidence = [99.8, 92.0, 94.0] boundary_color = self.color_palette["real_arf"] if is_real_arf else self.color_palette["simulated"] boundary_text = "REAL ARF" if is_real_arf else "SIMULATED" fig = go.Figure(data=[ go.Bar(name='Accuracy (%)', x=agents, y=accuracy, marker_color=self.color_palette["primary"]), go.Bar(name='Speed (seconds)', x=agents, y=speed, marker_color=boundary_color), go.Bar(name='Confidence (%)', x=agents, y=confidence, marker_color=self.color_palette["info"]) ]) # Add boundary annotation fig.add_annotation( x=0.98, y=0.98, xref="paper", yref="paper", text=f"🤖 {boundary_text}", showarrow=False, font=dict(size=12, color=boundary_color), bgcolor="white", bordercolor=boundary_color, borderwidth=2, borderpad=4 ) fig.update_layout( title=f"🤖 Agent Performance Metrics
" f"{boundary_text} v3.3.7", barmode='group', height=400, plot_bgcolor='rgba(0,0,0,0)', paper_bgcolor='rgba(0,0,0,0)', legend=dict( orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1 ) ) return fig except Exception as e: logger.error(f"Agent performance chart creation failed: {e}") # Return HTML fallback return self.create_agent_performance_chart.__wrapped__(self, is_real_arf)