| """ |
| 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__) |
|
|
| |
| try: |
| import streamlit as st |
| except Exception: |
| st = None |
| logger.info("Streamlit not available; ui.components running in headless mode.") |
|
|
|
|
| |
| |
| |
| def _require_streamlit() -> bool: |
| """Guard to prevent crashes when Streamlit is unavailable.""" |
| return st is not None |
|
|
|
|
| |
| |
| |
| 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, |
| } |
|
|
|
|
| |
| |
| |
| 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)) |
|
|