petter2025's picture
Update ui/components.py
d203412 verified
raw
history blame
5.47 kB
"""
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))