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))