"""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("""
âš ī¸ Database Not Connected
Set USE_DATABASE=true in your .env file to enable analytics tracking.
""", unsafe_allow_html=True) st.info("Analytics features require storing analysis results in PostgreSQL for historical tracking.") return st.markdown("") # Time range selector col1, col2, col3 = st.columns([1, 2, 1]) with col1: days = st.selectbox("Time Range", [7, 14, 30, 90], index=0, format_func=lambda x: f"Last {x} days") try: analytics = db_client.get_analytics(days=days) if not analytics: st.info("No analytics data available yet. Run some analyses first!") return st.markdown("") # Summary metrics col1, col2, col3 = st.columns(3) with col1: total = analytics.get("total_messages", 0) st.metric("Total Analyses", total) with col2: companies = len(analytics.get("by_company", {})) st.metric("Companies Analyzed", companies) with col3: types = len(analytics.get("by_type", {})) st.metric("Analysis Types", types) st.markdown("") # Charts col1, col2 = st.columns(2) with col1: by_company = analytics.get("by_company", {}) if by_company: df = pd.DataFrame( [{"Company": k.upper(), "Count": v} for k, v in by_company.items()] ) fig = px.pie( df, values="Count", names="Company", title="Distribution by Company", hole=0.4, color_discrete_sequence=px.colors.sequential.Purp_r, ) fig.update_layout( plot_bgcolor="rgba(0,0,0,0)", paper_bgcolor="rgba(0,0,0,0)", font_color="#94a3b8", ) st.plotly_chart(fig, use_container_width=True) with col2: by_type = analytics.get("by_type", {}) if by_type: df = pd.DataFrame( [{"Type": k, "Count": v} for k, v in by_type.items()] ) fig = px.bar(df, x="Type", y="Count", title="Analysis Types", color_discrete_sequence=["#6366f1"]) fig.update_layout( plot_bgcolor="rgba(0,0,0,0)", paper_bgcolor="rgba(0,0,0,0)", font_color="#94a3b8", 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("") # Recent messages st.markdown('

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"""
{icon}
{title}
{desc}
""", unsafe_allow_html=True) with col2: st.markdown("**Technology Stack**") st.markdown("""
Backend
Python, FastAPI
AI Model
Claude 3.5 Sonnet
Database
PostgreSQL
Dashboard
Streamlit, Plotly
Data Source
SerpAPI Patents
""", unsafe_allow_html=True) st.markdown("") st.markdown("**API Endpoints**") st.code("http://localhost:8000/docs", language=None) st.code("http://localhost:8000/health", language=None) st.markdown("") st.markdown("") # System status st.markdown('

System Status

', unsafe_allow_html=True) col1, col2, col3 = st.columns(3) with col1: db_client = get_db_client() if db_client: st.markdown("""
●
Database
Connected
""", unsafe_allow_html=True) else: st.markdown("""
●
Database
Not Configured
""", unsafe_allow_html=True) with col2: analyzer = get_analyzer() if analyzer: st.markdown("""
●
Analyzer
Ready
""", unsafe_allow_html=True) else: st.markdown("""
●
Analyzer
Not Initialized
""", unsafe_allow_html=True) with col3: st.markdown("""
●
Dashboard
Online
""", unsafe_allow_html=True) def main(): """Main dashboard entry point.""" render_header() tabs = render_navigation() with tabs[0]: render_company_analysis() with tabs[1]: render_batch_analysis() with tabs[2]: render_analytics() with tabs[3]: render_about() if __name__ == "__main__": main()