streamlit_app/utils/ui_helpers.py

#!/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)