""" Enhanced components with real ARF integration (Streamlit-optional, Hugging Face safe) """ from __future__ import annotations import logging from datetime import datetime from typing import List, Dict, Any import time import plotly.graph_objects as go import plotly.express as px import pandas as pd import numpy as np logger = logging.getLogger(__name__) # --- OPTIONAL Streamlit import (HF-safe) --- try: import streamlit as st # type: ignore except Exception: st = None logger.info("Streamlit not available; ui.components running in headless mode.") # ----------------------------- # Helpers # ----------------------------- def _require_streamlit() -> bool: """Guard to prevent crashes when Streamlit is unavailable.""" return st is not None # ----------------------------- # Mock ARF object (demo-safe) # ----------------------------- class MockHealingIntent: def __init__(self, action, component, confidence, status, rag_similarity_score=None): self.action = action self.component = component self.confidence = confidence self.status = status self.rag_similarity_score = rag_similarity_score self.deterministic_id = f"intent_{int(time.time())}" self.created_at = time.time() def get_execution_summary(self): return { "intent_id": self.deterministic_id, "action": self.action, "component": self.component, "confidence": self.confidence, "status": self.status.value if hasattr(self.status, "value") else self.status, "rag_similarity_score": self.rag_similarity_score, } # ----------------------------- # UI Components (SAFE) # ----------------------------- def create_arf_enhanced_timeline( incident_data: Dict[str, Any], healing_intents: List[Dict[str, Any]] | None = None, ): if not _require_streamlit(): return col1, col2 = st.columns([2, 1]) with col1: st.markdown("### 📈 ARF-Enhanced Incident Timeline") events = [ {"time": "-5m", "event": "📡 Alert Triggered", "phase": "detection", "color": "#FF6B6B"}, {"time": "-4m", "event": "🧠 ARF Analysis Started", "phase": "analysis", "color": "#4ECDC4"}, {"time": "-3.5m", "event": "🔍 RAG Similarity Search", "phase": "rag", "color": "#1E90FF"}, {"time": "-2.5m", "event": "🎯 Pattern Detection", "phase": "pattern", "color": "#9D4EDD"}, {"time": "-1.5m", "event": "💡 HealingIntent Generated", "phase": "intent", "color": "#FFD166"}, {"time": "-1m", "event": "⚡ MCP Execution", "phase": "execution", "color": "#06D6A0"}, {"time": "Now", "event": "✅ Resolution Complete", "phase": "resolution", "color": "#118AB2"}, ] fig = go.Figure() fig.add_trace( go.Scatter( x=list(range(len(events))), y=[0] * len(events), mode="lines+markers", marker=dict(size=16, color=[e["color"] for e in events]), line=dict(width=2), hovertext=[e["event"] for e in events], hoverinfo="text", ) ) fig.update_layout( height=250, showlegend=False, xaxis=dict(visible=False), yaxis=dict(visible=False), margin=dict(l=20, r=20, t=20, b=20), ) st.plotly_chart(fig, use_container_width=True) with col2: st.markdown("### 🎯 ARF Pattern Detection") rag_score = ( healing_intents[0].get("rag_similarity_score", 0.85) if healing_intents else 0.85 ) fig = go.Figure( go.Indicator( mode="gauge+number", value=rag_score * 100, gauge={"axis": {"range": [0, 100]}}, title={"text": "RAG Similarity"}, ) ) fig.update_layout(height=200) st.plotly_chart(fig, use_container_width=True) def create_healing_intent_visualizer(healing_intent: Dict[str, Any]): if not _require_streamlit(): return st.markdown("### 💡 ARF HealingIntent") confidence = healing_intent.get("confidence", 0.85) fig = go.Figure( go.Indicator( mode="gauge+number", value=confidence * 100, gauge={"axis": {"range": [0, 100]}}, title={"text": "Confidence"}, ) ) fig.update_layout(height=180) st.plotly_chart(fig, use_container_width=True) st.json(healing_intent) def create_rag_similarity_panel(query: str, similar_incidents: List[Dict[str, Any]]): if not _require_streamlit(): return st.markdown("### 🔍 RAG Similarity Search") if not similar_incidents: st.info("No similar incidents found") return df = pd.DataFrame(similar_incidents) st.dataframe(df, use_container_width=True) def create_learning_engine_panel(learning_stats: Dict[str, Any]): if not _require_streamlit(): return st.markdown("### 🧠 ARF Learning Engine") st.json(learning_stats) def create_execution_mode_toggle(current_mode: str = "advisory") -> str: if not _require_streamlit(): return current_mode modes = ["advisory", "approval", "autonomous"] return st.selectbox("Execution Mode", modes, index=modes.index(current_mode))