Spaces:
Paused
Paused
File size: 6,051 Bytes
bbae929 | 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 | """Reference events panel + prediction details."""
from __future__ import annotations
import html
import streamlit as st
from app.styles import TOKENS, domain_color, severity_color
from src.models.schemas import PredictionResult
def _render_one_reference(entry: dict) -> str:
"""Render one reference row as HTML.
``entry`` follows PredictionResult.reference_info shape:
``{event_id, similarity[, country, date, num_cascade_nodes, top_domains]}``.
Optional fields are omitted from the rendered secondary line when missing
(fallback mode uses only event_id + similarity).
"""
event_id = entry.get("event_id", "?")
similarity = entry.get("similarity") or 0
pct = int(round(similarity * 100))
# Secondary metadata line: country · date · num_cascade_nodes
meta_bits: list[str] = []
country = entry.get("country")
date_ = entry.get("date")
if country and date_:
meta_bits.append(f"{html.escape(str(country))} · {html.escape(str(date_))}")
elif country:
meta_bits.append(html.escape(str(country)))
elif date_:
meta_bits.append(html.escape(str(date_)))
n_nodes = entry.get("num_cascade_nodes")
if isinstance(n_nodes, int):
meta_bits.append(f"{n_nodes} cascade nodes")
meta_line = (
f'<div class="node-meta" style="margin:2px 0 6px 0">{ " · ".join(meta_bits) }</div>'
if meta_bits else ""
)
# top_domains pills (reuse the DAG legend style)
top_domains = entry.get("top_domains") or []
pills = ""
if top_domains:
pill_html = "".join(
f'<span class="domain-pill" style="font-size:10px;padding:2px 8px">'
f'<span class="swatch" style="background:{domain_color(d)}"></span>'
f'{html.escape(str(d))}</span>'
for d in top_domains
)
pills = (
f'<div style="display:flex;flex-wrap:wrap;gap:6px;'
f'margin:0 0 6px 0">{pill_html}</div>'
)
return (
f'<div class="ref-row" style="flex-direction:column;'
f'align-items:stretch;gap:4px">'
f' <div style="display:flex;align-items:center;gap:14px">'
f' <span class="ref-id">{html.escape(event_id)}</span>'
f' <span class="ref-meter">'
f' <span class="ref-meter-fill" style="width:{pct}%"></span>'
f' </span>'
f' <span class="ref-pct">{pct}%</span>'
f' </div>'
f' {meta_line}'
f' {pills}'
f'</div>'
)
def render_reference_events(result: PredictionResult) -> None:
"""Historical analogues with similarity meters + enriched provenance.
Consumes ``result.reference_info`` when populated (post-v2 schema). Falls
back to event_id + similarity pairs when the chain index was unavailable
at predict time, so older callers/artifacts still render correctly.
"""
if not result.reference_event_ids and not result.reference_info:
st.markdown(
'<div class="node-meta">No historical analogues retrieved.</div>',
unsafe_allow_html=True,
)
return
# Prefer enriched info when present; otherwise synthesize minimal entries
# from the parallel lists so rendering stays uniform.
if result.reference_info:
entries = result.reference_info
else:
entries = [
{"event_id": eid,
"similarity": result.similarity_scores[i]
if i < len(result.similarity_scores) else 0}
for i, eid in enumerate(result.reference_event_ids)
]
rows = "".join(_render_one_reference(e) for e in entries)
st.markdown(
f'<div class="panel" style="padding:6px 16px 4px 16px">{rows}</div>',
unsafe_allow_html=True,
)
st.markdown(
'<div class="node-meta" style="margin-top:8px">'
'RAG · MiniLM-L6 · cosine similarity'
'</div>',
unsafe_allow_html=True,
)
def render_prediction_details(result: PredictionResult) -> None:
"""Per-node detail cards."""
chain = result.predicted_chain
if not chain.cascade_events:
return
for node in chain.cascade_events:
confidence = result.confidence_scores.get(node.id, 0)
sev_col = severity_color(node.severity)
dcol = domain_color(node.domain)
# Plain-text label (Streamlit doesn't render HTML here)
label = (
f"{node.id} · {node.description[:72]}"
f"{'…' if len(node.description) > 72 else ''}"
f" [{node.severity.upper()}]"
)
with st.expander(label, expanded=False):
meta = (
f'<div class="node-meta">'
f'<span style="color:{dcol}">● {node.domain}</span>'
f' · <span style="color:{sev_col}">'
f'SEVERITY {node.severity.upper()}</span>'
f' · +{node.time_offset_hours}h'
)
if confidence:
meta += f' · CONF {confidence:.0%}'
if node.parent_ids:
joint_label = (
f' <span style="color:{TOKENS["amber"]}">⋈ JOINT</span>'
if len(node.parent_ids) >= 2 else ""
)
meta += (
f' · CAUSED BY {", ".join(node.parent_ids)}'
f'{joint_label}'
)
meta += "</div>"
st.markdown(meta, unsafe_allow_html=True)
st.markdown(
f'<div style="font-family:Figtree;font-size:14.5px;'
f'color:{TOKENS["text"]};line-height:1.6;margin-top:10px">'
f'<span style="color:{TOKENS["text_muted"]};font-family:JetBrains Mono;'
f'font-size:10px;letter-spacing:0.18em;text-transform:uppercase;'
f'display:block;margin-bottom:4px">Mechanism</span>'
f'{node.mechanism}'
f'</div>',
unsafe_allow_html=True,
)
|