File size: 5,467 Bytes
bbe2211 adb7e74 d203412 bbe2211 d203412 cc6b2b9 adb7e74 bbe2211 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 cc6b2b9 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 cc6b2b9 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 dddcb49 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 adb7e74 d203412 cc6b2b9 adb7e74 d203412 adb7e74 d203412 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | """
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))
|