"""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'
{ " · ".join(meta_bits) }
' 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'' f'' f'{html.escape(str(d))}' for d in top_domains ) pills = ( f'
{pill_html}
' ) return ( f'
' f'
' f' {html.escape(event_id)}' f' ' f' ' f' ' f' {pct}%' f'
' f' {meta_line}' f' {pills}' f'
' ) 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( '
No historical analogues retrieved.
', 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'
{rows}
', unsafe_allow_html=True, ) st.markdown( '
' 'RAG · MiniLM-L6 · cosine similarity' '
', 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'
' f'● {node.domain}' f'  ·  ' f'SEVERITY {node.severity.upper()}' f'  ·  +{node.time_offset_hours}h' ) if confidence: meta += f'  ·  CONF {confidence:.0%}' if node.parent_ids: joint_label = ( f' ⋈ JOINT' if len(node.parent_ids) >= 2 else "" ) meta += ( f'  ·  CAUSED BY {", ".join(node.parent_ids)}' f'{joint_label}' ) meta += "
" st.markdown(meta, unsafe_allow_html=True) st.markdown( f'
' f'Mechanism' f'{node.mechanism}' f'
', unsafe_allow_html=True, )