#!/usr/bin/env python3 """ UI Helpers Utility functions and components for the Streamlit application UI. Provides reusable UI elements, formatting functions, and visual components. """ import streamlit as st import pandas as pd import plotly.graph_objects as go import plotly.express as px from typing import Dict, Any, List, Optional, Tuple import time from datetime import datetime import json class UIHelpers: """UI helper functions and components""" @staticmethod def create_metric_card(title: str, value: Any, delta: Optional[Any] = None, delta_color: str = "normal", help_text: Optional[str] = None): """Create a styled metric card""" if isinstance(value, float): if title.lower().endswith(('rate', 'ratio', 'percentage', 'percent')): formatted_value = ".1f" else: formatted_value = ".2f" else: formatted_value = str(value) return st.metric( label=title, value=formatted_value, delta=delta, delta_color=delta_color, help=help_text ) @staticmethod def create_progress_bar(progress: float, text: str = "", color: str = "primary"): """Create a styled progress bar with text""" if text: st.write(f"**{text}**") if color == "success": bar_color = "#28a745" elif color == "warning": bar_color = "#ffc107" elif color == "danger": bar_color = "#dc3545" else: bar_color = None st.progress(progress, text=f"{progress:.1%} Complete") @staticmethod def create_info_box(message: str, type: str = "info"): """Create a styled info/warning/success box""" if type == "success": st.success(message) elif type == "warning": st.warning(message) elif type == "error": st.error(message) else: st.info(message) @staticmethod def format_file_size(size_bytes: int) -> str: """Format file size in human-readable format""" for unit in ['B', 'KB', 'MB', 'GB', 'TB']: if size_bytes < 1024.0: return ".1f" size_bytes /= 1024.0 return ".1f" @staticmethod def format_time_duration(seconds: float) -> str: """Format time duration in human-readable format""" if seconds < 60: return ".1f" elif seconds < 3600: minutes = int(seconds // 60) remaining_seconds = seconds % 60 return ".1f" else: hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) return f"{hours}h {minutes}m" @staticmethod def create_performance_chart(data: List[Tuple[float, float]], title: str, y_label: str, color: str = "#1f77b4"): """Create a performance chart using Plotly""" if not data: return None times, values = zip(*data) # Convert timestamps to relative time start_time = min(times) relative_times = [t - start_time for t in times] fig = go.Figure() fig.add_trace(go.Scatter( x=relative_times, y=values, mode='lines+markers', line=dict(color=color, width=2), marker=dict(size=4), name=y_label )) fig.update_layout( title=title, xaxis_title="Time (seconds)", yaxis_title=y_label, template="plotly_white", height=300, margin=dict(l=20, r=20, t=40, b=20) ) return fig @staticmethod def create_comparison_chart(data_dict: Dict[str, List[float]], title: str, x_label: str, y_label: str): """Create a comparison bar chart""" fig = go.Figure() for label, values in data_dict.items(): fig.add_trace(go.Bar( name=label, x=list(range(len(values))), y=values, text=[f"{v:.2f}" for v in values], textposition='auto', )) fig.update_layout( title=title, xaxis_title=x_label, yaxis_title=y_label, template="plotly_white", height=400, margin=dict(l=20, r=20, t=40, b=20) ) return fig @staticmethod def create_analysis_summary(results: List[Dict[str, Any]]) -> Dict[str, Any]: """Create a summary of analysis results""" if not results: return { 'total_analyses': 0, 'total_loopholes': 0, 'avg_confidence': 0, 'total_chunks': 0, 'analysis_types': {} } total_loopholes = sum(len(result.get('loopholes', [])) for result in results) total_confidence = sum(result.get('confidence', 0) for result in results) total_chunks = sum(result.get('chunks_processed', 0) for result in results) # Count analysis types analysis_types = {} for result in results: analysis_type = result.get('analysis_type', 'Unknown') analysis_types[analysis_type] = analysis_types.get(analysis_type, 0) + 1 return { 'total_analyses': len(results), 'total_loopholes': total_loopholes, 'avg_confidence': total_confidence / len(results) if results else 0, 'total_chunks': total_chunks, 'analysis_types': analysis_types } @staticmethod def display_analysis_result(result: Dict[str, Any], index: int = 0): """Display a single analysis result in a formatted way""" with st.expander(f"📋 Analysis {index + 1}: {result.get('title', 'Unknown Title')}", expanded=index == 0): col1, col2 = st.columns([2, 1]) with col1: st.markdown("**Summary:**") st.write(result.get('summary', 'No summary available')) st.markdown("**Key Findings:**") loopholes = result.get('loopholes', []) if loopholes: for i, loophole in enumerate(loopholes, 1): st.markdown(f"{i}. {loophole}") else: st.write("No significant loopholes identified.") if result.get('recommendations'): st.markdown("**Recommendations:**") for rec in result.get('recommendations', []): st.markdown(f"• {rec}") with col2: UIHelpers.create_metric_card( "Confidence", ".2f", help_text="Model confidence in analysis" ) UIHelpers.create_metric_card( "Processing Time", ".2f", help_text="Time taken to analyze this content" ) UIHelpers.create_metric_card( "Chunks Processed", result.get('chunks_processed', 0), help_text="Number of text chunks analyzed" ) st.markdown("**Metadata:**") st.write(f"**Source:** {result.get('source', 'Unknown')}") st.write(f"**Date:** {result.get('date', 'Unknown')}") st.write(f"**Analysis Type:** {result.get('analysis_type', 'Standard')}") @staticmethod def create_export_section(results: List[Dict[str, Any]]): """Create the export section for results""" st.subheader("💾 Export Results") if not results: st.info("No results to export") return col1, col2, col3 = st.columns(3) with col1: if st.button("📄 Export as JSON", use_container_width=True): json_data = json.dumps(results, indent=2, ensure_ascii=False) st.download_button( label="Download JSON", data=json_data, file_name=f"nz_legislation_analysis_{int(time.time())}.json", mime="application/json", use_container_width=True ) with col2: if st.button("📊 Export as CSV", use_container_width=True): df = pd.DataFrame(results) csv_data = df.to_csv(index=False) st.download_button( label="Download CSV", data=csv_data, file_name=f"nz_legislation_analysis_{int(time.time())}.csv", mime="text/csv", use_container_width=True ) with col3: if st.button("📋 Export as Excel", use_container_width=True): df = pd.DataFrame(results) excel_data = df.to_excel(index=False, engine='openpyxl') st.download_button( label="Download Excel", data=excel_data, file_name=f"nz_legislation_analysis_{int(time.time())}.xlsx", mime="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", use_container_width=True ) @staticmethod def create_cache_management_section(cache_manager): """Create cache management section""" st.subheader("🧠 Cache Management") cache_stats = cache_manager.get_stats() col1, col2, col3, col4 = st.columns(4) with col1: UIHelpers.create_metric_card("Cache Hits", cache_stats['hits']) with col2: UIHelpers.create_metric_card("Cache Misses", cache_stats['misses']) with col3: UIHelpers.create_metric_card("Hit Rate", ".1f") with col4: UIHelpers.create_metric_card("Cached Entries", cache_stats['entries']) col1, col2, col3 = st.columns(3) with col1: if st.button("🔄 Clear Cache", type="secondary", use_container_width=True): cache_manager.clear_cache() st.rerun() with col2: if st.button("📤 Export Cache", use_container_width=True): import tempfile with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: success = cache_manager.export_cache(f.name) if success: st.success("Cache exported successfully") else: st.error("Failed to export cache") with col3: uploaded_cache = st.file_uploader("📥 Import Cache", type=['json']) if uploaded_cache: import tempfile with tempfile.NamedTemporaryFile(mode='wb', delete=False) as f: f.write(uploaded_cache.read()) imported_count = cache_manager.import_cache(f.name) st.success(f"Imported {imported_count} cache entries") @staticmethod def create_system_info_section(perf_monitor): """Create system information section""" st.subheader("💻 System Information") sys_info = perf_monitor.get_system_info() col1, col2 = st.columns(2) with col1: st.markdown("**Hardware:**") st.write(f"**CPU Cores:** {sys_info['cpu_count']} physical, {sys_info['cpu_count_logical']} logical") st.write(f"**Total Memory:** {sys_info['total_memory_gb']:.1f} GB") st.write(f"**Available Memory:** {sys_info['available_memory_gb']:.1f} GB") with col2: st.markdown("**Software:**") st.write(f"**Python:** {sys_info['python_version']}") st.write(f"**Platform:** {sys_info['platform']}") st.write(f"**Active Threads:** {st.session_state.performance_monitor.get_stats()['active_threads']}") @staticmethod def create_performance_recommendations(perf_monitor): """Create performance recommendations section""" st.subheader("💡 Performance Recommendations") recommendations = perf_monitor.get_recommendations() if recommendations: for rec in recommendations: if "High" in rec or "Slow" in rec: st.error(rec) elif "Moderate" in rec or "Consider" in rec: st.warning(rec) else: st.info(rec) else: st.success("All performance metrics are within optimal ranges!") @staticmethod def create_loading_spinner(text: str = "Processing..."): """Create a loading spinner""" return st.spinner(text) @staticmethod def create_success_message(message: str): """Create a success message""" st.success(message) @staticmethod def create_error_message(message: str): """Create an error message""" st.error(message) @staticmethod def create_warning_message(message: str): """Create a warning message""" st.warning(message) @staticmethod def create_data_table(data: List[Dict[str, Any]], columns: Optional[List[str]] = None): """Create a formatted data table""" if not data: st.info("No data to display") return df = pd.DataFrame(data) if columns: available_columns = [col for col in columns if col in df.columns] if available_columns: df = df[available_columns] st.dataframe(df, use_container_width=True) @staticmethod def create_json_viewer(data: Dict[str, Any], title: str = "JSON Data"): """Create a JSON viewer""" st.subheader(title) with st.expander("View JSON", expanded=False): st.json(data) @staticmethod def create_file_preview(file_content: str, max_lines: int = 20): """Create a file content preview""" lines = file_content.split('\n') preview_content = '\n'.join(lines[:max_lines]) if len(lines) > max_lines: preview_content += f"\n\n... ({len(lines) - max_lines} more lines)" st.text_area("File Preview", preview_content, height=200, disabled=True)