Spaces:
Sleeping
Sleeping
import os | |
import sys | |
import gradio as gr | |
import asyncio | |
import numpy as np | |
import pandas as pd | |
import plotly.express as px | |
import plotly.graph_objects as go | |
from typing import Dict, List, Any, Optional, Tuple | |
import json | |
import base64 | |
import time | |
# Import the integration module | |
try: | |
from gemini_gradio_integration import gemini_integration, connect_to_gemini, send_text_query, send_audio_query | |
GEMINI_AVAILABLE = True | |
except ImportError: | |
print("Warning: Gemini integration not available. Running in demo mode.") | |
GEMINI_AVAILABLE = False | |
# Import repository analysis functions | |
from gradio_repo_analysis import GradioRepoAnalysis, analyze_repository_data | |
from gradio_repo_analysis import create_language_pie_chart, create_commit_history_chart, create_contributors_chart | |
# Custom Gradio theme to match your current UI | |
custom_theme = gr.themes.Monochrome( | |
primary_hue="purple", | |
secondary_hue="blue", | |
neutral_hue="slate", | |
) | |
# Demo mode functions for when Gemini isn't available | |
def demo_process_query(repo_url, query): | |
"""Process a query in demo mode without Gemini""" | |
if not repo_url.startswith("https://github.com/"): | |
return "Please enter a valid GitHub repository URL", None, None, None | |
if not query: | |
return "Please enter a query about the repository", None, None, None | |
# Create simulated response with a delay | |
time.sleep(2) | |
repo_name = repo_url.split("/")[-1] | |
repo_owner = repo_url.split("/")[-2] | |
# Generate simulated data | |
repo_data = { | |
"name": repo_name, | |
"owner": repo_owner, | |
"url": repo_url | |
} | |
# Analyze the repository data | |
analysis_results = analyze_repository_data(repo_data) | |
# Create the visualization charts | |
language_chart = create_language_pie_chart(analysis_results["language_stats"]) | |
commit_chart = create_commit_history_chart(analysis_results["commit_history"]) | |
contributor_chart = create_contributors_chart(analysis_results["contributor_stats"]) | |
# Generate a text response | |
response = f"DEMO MODE - Analysis of {repo_owner}/{repo_name}:\n\n" | |
response += f"This repository is primarily written in {max(analysis_results['language_stats'], key=analysis_results['language_stats'].get)}, " | |
response += f"with {sum(item['commits'] for item in analysis_results['commit_history'])} commits in the last 5 months. " | |
response += f"The most active contributor is {analysis_results['contributor_stats'][0]['name']} with {analysis_results['contributor_stats'][0]['commits']} commits." | |
return response, language_chart, commit_chart, contributor_chart | |
def demo_process_audio(audio_data): | |
"""Process audio in demo mode without Gemini""" | |
if audio_data is None: | |
return "No audio input received", None | |
# Simulate processing delay | |
time.sleep(1.5) | |
# Return placeholder message and no audio (since we can't generate audio in demo mode) | |
return "DEMO MODE - Voice input received. In full mode, Gemini would process this audio and respond.", None | |
# Main application | |
class GradioGeminiApp: | |
def __init__(self): | |
self.gemini_available = GEMINI_AVAILABLE | |
self.connected = False | |
def connect_to_gemini(self): | |
"""Connect to Gemini API""" | |
if not self.gemini_available: | |
return "DEMO MODE - Gemini integration not available" | |
result = connect_to_gemini() | |
self.connected = "Successfully" in result | |
return result | |
def process_text_query(self, repo_url, query): | |
"""Process a text query about a repository""" | |
if not self.gemini_available: | |
return demo_process_query(repo_url, query) | |
# If we have Gemini available, use it | |
if not self.connected: | |
connection_result = self.connect_to_gemini() | |
if "Failed" in connection_result: | |
return f"Could not connect to Gemini: {connection_result}", None, None, None | |
# Send query to Gemini | |
send_result = send_text_query(repo_url, query) | |
# In a real implementation, you would: | |
# 1. Wait for and process Gemini's response | |
# 2. Extract structured data from it | |
# 3. Generate visualizations based on that data | |
# For now, we'll use the demo data as a placeholder | |
return demo_process_query(repo_url, query) | |
def process_audio_query(self, audio_data): | |
"""Process an audio query""" | |
if not self.gemini_available: | |
return demo_process_audio(audio_data) | |
# If we have Gemini available, use it | |
if not self.connected: | |
connection_result = self.connect_to_gemini() | |
if "Failed" in connection_result: | |
return f"Could not connect to Gemini: {connection_result}", None | |
# Send audio to Gemini | |
send_result = send_audio_query(audio_data) | |
# In a real implementation, you would: | |
# 1. Wait for and process Gemini's response | |
# 2. Handle both text and audio responses | |
# For now, return a placeholder message | |
return "Audio sent to Gemini. Awaiting response...", None | |
def build_interface(self): | |
"""Build the Gradio interface""" | |
with gr.Blocks(theme=custom_theme) as interface: | |
gr.Markdown( | |
""" | |
# 🧞♂️ G.E.N.I.E. - GitHub Repository Analysis | |
Analyze GitHub repositories using natural language queries and voice commands. | |
""" | |
) | |
# Connection status | |
with gr.Row(): | |
connection_status = gr.Markdown( | |
"⚠️ **Demo Mode** - Gemini connection not established" | |
if not self.gemini_available else | |
"⚠️ **Ready to Connect** - Click Connect to establish Gemini connection" | |
) | |
connect_btn = gr.Button( | |
"Connect to Gemini", | |
variant="primary", | |
interactive=self.gemini_available | |
) | |
# Main tabs: Text Analysis and Voice Chat | |
with gr.Tabs(): | |
# Text Analysis Tab | |
with gr.Tab("Repository Analysis"): | |
with gr.Row(): | |
with gr.Column(scale=3): | |
repo_url = gr.Textbox( | |
label="GitHub Repository URL", | |
placeholder="https://github.com/owner/repo", | |
info="Enter the URL of a GitHub repository" | |
) | |
query = gr.Textbox( | |
label="Your Query", | |
placeholder="What are the main languages used in this repo?", | |
info="Ask a question about the repository" | |
) | |
with gr.Row(): | |
analyze_btn = gr.Button("Analyze Repository", variant="primary") | |
clear_btn = gr.Button("Clear", variant="secondary") | |
response = gr.Markdown(label="Analysis Results") | |
with gr.Row(): | |
with gr.Column(): | |
language_chart = gr.Plot(label="Repository Language Distribution") | |
with gr.Column(): | |
commit_chart = gr.Plot(label="Commit Activity Over Time") | |
with gr.Row(): | |
contributor_chart = gr.Plot(label="Top Contributors") | |
# Voice Chat Tab | |
with gr.Tab("Voice Chat"): | |
with gr.Row(): | |
with gr.Column(): | |
voice_repo_url = gr.Textbox( | |
label="GitHub Repository URL", | |
placeholder="https://github.com/owner/repo", | |
info="Enter the URL of a GitHub repository" | |
) | |
audio_input = gr.Audio( | |
type="numpy", | |
label="Voice Input" | |
) | |
voice_response_text = gr.Markdown(label="G.E.N.I.E. Text Response") | |
with gr.Column(): | |
audio_output = gr.Audio(label="G.E.N.I.E. Voice Response") | |
# Set up event handlers | |
connect_btn.click( | |
fn=self.connect_to_gemini, | |
inputs=None, | |
outputs=connection_status | |
) | |
analyze_btn.click( | |
fn=self.process_text_query, | |
inputs=[repo_url, query], | |
outputs=[response, language_chart, commit_chart, contributor_chart] | |
) | |
clear_btn.click( | |
fn=lambda: ("", "", None, None, None), | |
inputs=None, | |
outputs=[response, query, language_chart, commit_chart, contributor_chart] | |
) | |
audio_input.change( | |
fn=self.process_audio_query, | |
inputs=[audio_input], | |
outputs=[voice_response_text, audio_output] | |
) | |
return interface | |
# Initialize and launch the application | |
app = GradioGeminiApp() | |
interface = app.build_interface() | |
interface.launch(server_name="0.0.0.0", server_port=7860) | |