"""SPARC Visualization Dashboard. A Streamlit-based dashboard for visualizing patent analysis results. Run with: streamlit run dashboard.py """ import streamlit as st import plotly.express as px import plotly.graph_objects as go import pandas as pd from datetime import datetime, timedelta from SPARC.analyzer import CompanyAnalyzer from SPARC.database import DatabaseClient from SPARC import config st.set_page_config( page_title="SPARC Dashboard", page_icon="âĄ", layout="wide", initial_sidebar_state="collapsed", ) # Modern CSS styling st.markdown(""" """, unsafe_allow_html=True) @st.cache_resource def get_analyzer(): """Get or create the CompanyAnalyzer instance.""" return CompanyAnalyzer() @st.cache_resource def get_db_client(): """Get database client if available.""" if config.use_database: try: client = DatabaseClient() client.connect() return client except Exception: return None return None def render_header(): """Render the modern dashboard header.""" st.markdown("""
""", unsafe_allow_html=True) def render_navigation(): """Render horizontal tab navigation at the top.""" tabs = st.tabs(["đ Company Analysis", "đĻ Batch Analysis", "đ Analytics", "âšī¸ About"]) return tabs def render_company_analysis(): """Render single company analysis page.""" st.markdown('Single Company Analysis
', unsafe_allow_html=True) st.markdown("Analyze a company's patent portfolio using AI-powered insights.") st.markdown("") # Search card with st.container(): col1, col2 = st.columns([3, 1]) with col1: company_name = st.text_input( "Company Name", placeholder="Enter company name (e.g., nvidia, intel, amd)", help="Enter the company name to analyze their patent portfolio", label_visibility="collapsed", ) with col2: analyze_btn = st.button("đ Analyze", type="primary", use_container_width=True) if analyze_btn and company_name: with st.spinner(f"Analyzing {company_name}..."): analyzer = get_analyzer() result = analyzer._analyze_company_safe(company_name) if result.success: st.success(f"â Analysis complete for {company_name.upper()}") st.markdown("") # Metrics row with custom styling col1, col2, col3 = st.columns(3) with col1: st.metric("Patents Found", result.patent_count) with col2: st.metric("Analysis Status", "Complete") with col3: st.metric("Timestamp", result.timestamp.strftime("%H:%M:%S")) st.markdown("") # Analysis content in a styled container st.markdown('AI Analysis Results
', unsafe_allow_html=True) with st.container(): st.markdown(result.analysis) else: st.error(f"Analysis failed: {result.error}") elif not company_name and analyze_btn: st.warning("Please enter a company name to analyze.") def render_batch_analysis(): """Render batch analysis page.""" st.markdown('Batch Company Analysis
', unsafe_allow_html=True) st.markdown("Analyze multiple companies simultaneously for comparative insights.") st.markdown("") # Input section col1, col2 = st.columns([2, 1]) with col1: companies_input = st.text_area( "Company Names", placeholder="Enter company names (one per line or comma-separated):\nnvidia\namd\nintel\nqualcomm", height=150, label_visibility="collapsed", ) with col2: st.markdown("**Configuration**") max_workers = st.slider("Concurrent Workers", 1, 5, 3, help="Number of parallel analysis threads") st.markdown("") analyze_btn = st.button( "đ Run Batch Analysis", type="primary", use_container_width=True ) if analyze_btn and companies_input: # Parse company names companies = [ c.strip() for c in companies_input.replace(",", "\n").split("\n") if c.strip() ] if not companies: st.warning("Please enter at least one company name") return st.info(f"đ Starting analysis of {len(companies)} companies...") # Progress tracking progress_bar = st.progress(0) status_text = st.empty() analyzer = get_analyzer() def update_progress(company: str, completed: int, total: int): progress = completed / total progress_bar.progress(progress) status_text.text(f"Analyzing {company}... ({completed}/{total})") result = analyzer.analyze_companies( companies=companies, max_workers=max_workers, progress_callback=update_progress, ) progress_bar.progress(1.0) status_text.text("â Analysis complete!") st.markdown("") # Summary metrics st.markdown('Results Summary
', unsafe_allow_html=True) col1, col2, col3, col4 = st.columns(4) with col1: st.metric("Total Companies", result.total_companies) with col2: st.metric("Successful", result.successful) with col3: st.metric("Failed", result.failed) with col4: success_rate = ( (result.successful / result.total_companies * 100) if result.total_companies > 0 else 0 ) st.metric("Success Rate", f"{success_rate:.1f}%") # Results chart if result.results: df = pd.DataFrame( [ { "Company": r.company_name.upper(), "Patents": r.patent_count, "Status": "Success" if r.success else "Failed", } for r in result.results ] ) fig = px.bar( df, x="Company", y="Patents", color="Status", color_discrete_map={"Success": "#10b981", "Failed": "#ef4444"}, title="", ) fig.update_layout( plot_bgcolor="rgba(0,0,0,0)", paper_bgcolor="rgba(0,0,0,0)", font_color="#94a3b8", legend=dict( orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1 ), xaxis=dict(showgrid=False), yaxis=dict(showgrid=True, gridcolor="rgba(99, 102, 241, 0.1)"), ) st.plotly_chart(fig, use_container_width=True) st.markdown("") # Individual results st.markdown('Detailed Results
', unsafe_allow_html=True) for r in result.results: status_icon = "â" if r.success else "â" status_class = "status-success" if r.success else "status-error" with st.expander( f"{status_icon} {r.company_name.upper()} â {r.patent_count} patents" ): if r.success: st.markdown(r.analysis) else: st.error(r.error) def render_analytics(): """Render analytics page with database insights.""" st.markdown('Analytics Dashboard
', unsafe_allow_html=True) st.markdown("Track historical analysis data and view insights.") db_client = get_db_client() if not db_client: st.markdown("") st.markdown("""USE_DATABASE=true in your .env file to enable analytics tracking.
Recent Analyses
', unsafe_allow_html=True) messages = db_client.get_messages(limit=10) if messages: for msg in messages: with st.expander( f"đ {msg.get('company_name', 'Unknown').upper()} â {msg.get('analysis_type', 'N/A')} ({msg.get('timestamp', 'N/A')})" ): st.markdown(f"**Model:** `{msg.get('model', 'N/A')}`") if msg.get("response"): st.markdown(msg["response"][:500] + "...") except Exception as e: st.error(f"Error fetching analytics: {e}") def render_about(): """Render about page.""" st.markdown('About SPARC
', unsafe_allow_html=True) col1, col2 = st.columns([2, 1]) with col1: st.markdown(""" **SPARC** (Semiconductor Patent & Analytics Report Core) is an AI-powered patent analysis platform that evaluates company performance by analyzing their patent portfolios with cutting-edge language models. """) st.markdown("") st.markdown("**Key Features**") features = [ ("đ", "Patent Retrieval", "Automated collection via SerpAPI's Google Patents"), ("đ", "Intelligent Parsing", "Extracts key sections from patent documents"), ("đ¤", "AI Analysis", "Deep analysis powered by Claude 3.5 Sonnet"), ("âĄ", "Batch Processing", "Analyze multiple companies concurrently"), ("đ", "REST API", "FastAPI web service for seamless integration"), ("đ", "Analytics", "Track and visualize historical analysis data"), ] for icon, title, desc in features: st.markdown(f"""System Status
', unsafe_allow_html=True) col1, col2, col3 = st.columns(3) with col1: db_client = get_db_client() if db_client: st.markdown("""