""" Reusable UI components for CSH2 Web Dashboard — HUD Theme """ import streamlit as st import pandas as pd from typing import List, Dict, Optional def page_header(title: str, subtitle: str = ""): """Render a HUD-style page header with gradient accent lines""" subtitle_html = f'
{subtitle}
' if subtitle else '' st.markdown(f"""

{title}

{subtitle_html}
""", unsafe_allow_html=True) def metric_row(metrics: List[Dict]): """ Render a row of HUD-style metric cards. metrics: list of dicts with keys: label, value, unit, delta (optional), status (optional) status can be: 'nominal', 'warning', 'alarm' (defaults to no special styling) """ cols = st.columns(len(metrics)) for col, m in zip(cols, metrics): with col: display_value = f"{m['value']}" if m.get('unit'): display_value = f"{m['value']} {m['unit']}" st.metric( label=m['label'], value=display_value, delta=m.get('delta'), ) def analysis_card(title: str, content: str): """Render an LLM analysis result in a HUD terminal-style card""" st.markdown(f"""

// {title.upper()}

{content.replace(chr(10), '
')}
""", unsafe_allow_html=True) def system_status_footer(sensor_count: int, data_start: str, data_end: str, connected: bool = True): """Render a HUD-style system status panel in the sidebar""" conn_color = 'var(--status-nominal)' if connected else 'var(--status-alarm)' conn_text = 'CONNECTED' if connected else 'OFFLINE' st.markdown(f"""
{conn_text}
SENSORS: {sensor_count}
DATA: {data_start} — {data_end}
CSH2 DELPHI v1.0
""", unsafe_allow_html=True) def cycle_selector(cycles: List[Dict], key_prefix: str = "cycle") -> Optional[Dict]: """ Render a cycle selector dropdown. cycles: list of dicts with keys: cycle_id, start_time, end_time, duration_minutes, peak_pressure Returns selected cycle dict or None """ if not cycles: st.info("No testing cycles detected for this period.") return None options = [] for c in cycles: # Use Eastern display time if available, fall back to start_time display_time = c.get('start_time_et') or c['start_time'] start_str = display_time.strftime('%b %d %H:%M') + " ET" duration = f"{c['duration_minutes']:.0f} min" peak = f"{c.get('peak_pressure', 0):.0f} bar" if c.get('peak_pressure') else "N/A" options.append(f"Cycle {c['cycle_id']} | {start_str} | {duration} | Peak (PT130): {peak}") selected_idx = st.selectbox( "Select a testing cycle", range(len(options)), format_func=lambda i: options[i], key=f"{key_prefix}_selector", ) return cycles[selected_idx] if selected_idx is not None else None def tag_multiselect(db_connector, key: str = "tags", default_group: str = None) -> List[str]: """ Render a tag multi-select with group filtering. Selecting a sensor group filters the available options to only show sensors in that group. "All Sensors" shows every tag in the database. Returns list of selected tag names. """ from core.config import SENSOR_GROUPS group_names = ["All Sensors"] + list(SENSOR_GROUPS.keys()) def _clear_selection(): """Reset the multiselect when group changes to avoid stale selections.""" st.session_state.pop(f"{key}_select", None) selected_group = st.selectbox( "Sensor Group", group_names, key=f"{key}_group", on_change=_clear_selection, ) all_tags = db_connector.get_all_tag_names() if selected_group == "All Sensors": available_tags = all_tags default_tags = [] else: group_tags = SENSOR_GROUPS.get(selected_group, []) available_tags = [t for t in group_tags if t in all_tags] default_tags = available_tags # Pre-select all sensors in the chosen group selected_tags = st.multiselect( "Select Sensors", options=available_tags, default=default_tags, key=f"{key}_select", ) return selected_tags def date_range_picker(key: str = "dates"): """Render start/end date and time pickers. Returns (start_datetime, end_datetime).""" from datetime import datetime, time, timedelta col1, col2 = st.columns(2) with col1: start_date = st.date_input( "Start Date", value=datetime(2025, 9, 25), key=f"{key}_start_date", ) start_time = st.time_input("Start Time", value=time(0, 0), key=f"{key}_start_time") with col2: end_date = st.date_input( "End Date", value=datetime(2025, 9, 25), key=f"{key}_end_date", ) end_time = st.time_input("End Time", value=time(23, 59), key=f"{key}_end_time") start_dt = datetime.combine(start_date, start_time) end_dt = datetime.combine(end_date, end_time) return start_dt, end_dt def no_data_message(): """Show a styled no-data message""" st.warning("No data found for the selected sensors and time range. Try adjusting your filters.") def follow_up_section( session_key: str, llm_analyzer, original_prompt: str, original_analysis: str, max_turns: int = 3, ): """Render a follow-up Q&A section below an analysis card. Maintains a multi-turn conversation thread stored in ``st.session_state[session_key]`` as a list of ``{role, content}`` dicts. Args: session_key: Unique session state key for this conversation thread. llm_analyzer: An ``LLMCycleAnalyzer`` instance with ``follow_up()`` method. original_prompt: The original user prompt that generated the analysis. original_analysis: The original assistant analysis text. max_turns: Maximum follow-up exchanges (default 3). """ # Initialize conversation history if needed if session_key not in st.session_state: st.session_state[session_key] = [] followups = st.session_state[session_key] # Display existing follow-up thread if followups: for msg in followups: if msg["role"] == "user": st.markdown(f"""
You: {msg['content']}
""", unsafe_allow_html=True) else: analysis_card("Murphy Follow-up", msg["content"]) # Show input if under max turns turn_count = len([m for m in followups if m["role"] == "user"]) if turn_count >= max_turns: st.caption(f"Maximum {max_turns} follow-up questions reached.") return # Follow-up input input_col, btn_col = st.columns([5, 1]) with input_col: followup_q = st.text_input( "Ask a follow-up question", placeholder="e.g. 'What about the temperature trend?' or 'Explain the plateau at 370 bar'", key=f"{session_key}_input", label_visibility="collapsed", ) with btn_col: followup_btn = st.button( "Ask", type="primary", use_container_width=True, key=f"{session_key}_btn", ) if followup_btn and followup_q: # Build full conversation history for the API call conversation = [ {"role": "user", "content": original_prompt}, {"role": "assistant", "content": original_analysis}, ] + followups with st.spinner("Murphy is thinking..."): response = llm_analyzer.follow_up(conversation, followup_q) # Append to thread followups.append({"role": "user", "content": followup_q}) followups.append({"role": "assistant", "content": response}) st.session_state[session_key] = followups st.rerun()