š Session Statistics
Total Queries:
{total_queries}
Average Response Time:
{format_response_time(avg_time)}
Session Started:
{datetime.now().strftime('%H:%M')}
"""
# Document statistics
doc_stats = demo_state.document_stats
if any(doc_stats.values()):
stats_html += """
š Document Statistics
"""
for doc_type, count in doc_stats.items():
if count > 0:
type_info = DOCUMENT_TYPES.get(doc_type, {})
color = type_info.get("color", "#4facfe")
label = type_info.get("label", doc_type.replace("_", " ").title())
stats_html += f"""
{label}:
{count}
"""
stats_html += "
"
if demo_state.recent_queries:
stats_html += """
š Recent Queries
"""
for query_info in demo_state.recent_queries:
stats_html += f"""
{query_info['query']}
{query_info['time']}
"""
stats_html += "
"
return stats_html
# Document processing functions
def process_old_document(file_path: str, query: str = None) -> str:
"""Process old documents using existing old_doc_ingestion"""
try:
# Update stats
demo_state.document_stats["old_documents"] += 1
demo_state.last_uploaded_old_doc = file_path
# Use existing old document processing
status = old_doc_ingestion(file_path)
if query:
answer = old_doc_qa(query)
return f"š Old Document processed: {status}\n\n{answer}"
return f"š Old Document ingested successfully: {status}"
except Exception as e:
logger.error(f"Error processing old document: {e}")
return f"ā Error processing old document: {str(e)}"
def process_user_document(file_path: str, query: str = None) -> str:
"""Process user-specific documents"""
try:
demo_state.document_stats["user_specific"] += 1
result = ingest_file(file_path)
if query:
response = process_query(query)
cleaned_response = clean_pipeline_result(response[0] if isinstance(response, tuple) else str(response))
return f"š¤ User Document processed: {result}\n\n{cleaned_response}"
return f"š¤ User-specific document ingested successfully: {result}"
except Exception as e:
logger.error(f"Error processing user document: {e}")
return f"ā Error processing user document: {str(e)}"
# 5. Create fixed voice chat functions
def start_voice_chat():
"""Start voice chat with proper initialization"""
try:
# Initialize or reset the handler
global voice_handler
voice_handler = GeminiHandler()
return "š¤ Voice chat started... Speak now!"
except Exception as e:
logger.error(f"Error starting voice chat: {e}")
return f"ā Failed to start voice chat: {str(e)}"
def stop_voice_chat():
"""Stop voice chat properly"""
try:
global voice_handler
if 'voice_handler' in globals() and voice_handler:
voice_handler.stop()
return "ā¹ļø Voice chat stopped."
except Exception as e:
logger.error(f"Error stopping voice chat: {e}")
return f"ā Error stopping voice chat: {str(e)}"
def process_domain_document(file_path: str, query: str = None) -> str:
"""Process domain-specific documents"""
try:
demo_state.document_stats["domain_documents"] += 1
result = ingest_file(file_path)
if query:
response = process_query(query)
cleaned_response = clean_pipeline_result(response[0] if isinstance(response, tuple) else str(response))
return f"š„ Domain Document processed: {result}\n\n{cleaned_response}"
return f"š„ Domain document ingested successfully: {result}"
except Exception as e:
logger.error(f"Error processing domain document: {e}")
return f"ā Error processing domain document: {str(e)}"
def process_new_document(file_path: str, query: str = None) -> str:
"""Process new documents"""
try:
demo_state.document_stats["new_documents"] += 1
result = ingest_file(file_path)
if query:
response = process_query(query)
cleaned_response = clean_pipeline_result(response[0] if isinstance(response, tuple) else str(response))
return f"š New Document processed: {result}\n\n{cleaned_response}"
return f"š New document ingested successfully: {result}"
except Exception as e:
logger.error(f"Error processing new document: {e}")
return f"ā Error processing new document: {str(e)}"
# Main processing functions
def process_with_stats(user_input: str, audio_input, chat_history: List) -> Tuple[List, str, str, str, Optional[tuple]]:
"""Process query with timing and statistics"""
start_time = time.time()
try:
# Get input text
query = user_input.strip() if user_input else ""
if audio_input and not query:
query, status = get_transcription_or_text("", audio_input)
if query.startswith("[ERROR]"):
return chat_history, "", f"ā {status}", create_stats_display(), None
if not query:
return chat_history, "", "ā ļø Please provide text or audio input.", create_stats_display(), None
logger.info(f"Processing query: {query[:50]}...")
# Process the query
response = process_query(query)
response_time = time.time() - start_time
# Update statistics
demo_state.update_stats(query, response_time)
# Clean response
if isinstance(response, tuple):
cleaned_response = clean_pipeline_result(response[0] if response[0] else response[1])
else:
cleaned_response = clean_pipeline_result(str(response))
# Update chat history
new_history = chat_history.copy()
new_history.append({"role": "user", "content": query})
new_history.append({"role": "assistant", "content": cleaned_response})
# Create status message
status_msg = f"ā
Response generated in {format_response_time(response_time)}"
# Update stats display
stats_display = create_stats_display()
logger.info(f"Query processed successfully in {response_time:.2f}s")
return new_history, "", status_msg, stats_display, None
except Exception as e:
logger.error(f"Error processing query: {e}")
error_msg = f"ā Error: {str(e)[:100]}..."
return chat_history, "", error_msg, create_stats_display(), None
def process_with_audio(user_input: str, audio_input, voice_dropdown: str, chat_history: List) -> Tuple[List, Optional[tuple], str, str, str, Optional[tuple]]:
"""Fixed audio processing function"""
start_time = time.time()
try:
# Get input text with better audio handling
query = user_input.strip() if user_input else ""
if audio_input and not query:
try:
# Check if audio_utils functions exist and work
query, status = get_transcription_or_text("", audio_input)
if not query or query.startswith("[ERROR]"):
return chat_history, None, "", f"ā Audio transcription failed: {status}", create_stats_display(), None
except Exception as audio_error:
logger.error(f"Audio transcription error: {audio_error}")
return chat_history, None, "", f"ā Audio processing error: {str(audio_error)}", create_stats_display(), None
if not query:
return chat_history, None, "", "ā ļø Please provide text or audio input.", create_stats_display(), None
logger.info(f"Processing query with audio: {query[:50]}...")
# Process the query
try:
response = process_query(query)
except Exception as query_error:
logger.error(f"Query processing error: {query_error}")
return chat_history, None, "", f"ā Query processing failed: {str(query_error)}", create_stats_display(), None
# Clean response
if isinstance(response, tuple):
cleaned_response = clean_pipeline_result(response[0] if response[0] else response[1])
else:
cleaned_response = clean_pipeline_result(str(response))
# Update chat history
new_history = chat_history.copy()
new_history.append({"role": "user", "content": query})
new_history.append({"role": "assistant", "content": cleaned_response})
# Generate audio with better error handling
audio_response = None
tts_status = "Audio generation skipped"
try:
# Check if TTS functions are available
if 'generate_tts_response' in globals():
audio_data, tts_status = generate_tts_response(cleaned_response, voice_dropdown)
audio_response = audio_data if audio_data else None
else:
tts_status = "TTS function not available"
except Exception as audio_error:
logger.error(f"Audio generation error: {audio_error}")
audio_response = None
tts_status = f"Audio generation failed: {str(audio_error)[:50]}..."
response_time = time.time() - start_time
demo_state.update_stats(query, response_time)
status_msg = f"ā
Response generated in {format_response_time(response_time)}"
if audio_response:
status_msg += " | šµ Audio ready"
else:
status_msg += f" | ā ļø {tts_status}"
stats_display = create_stats_display()
return new_history, audio_response, "", status_msg, stats_display, None
except Exception as e:
logger.error(f"Error processing audio query: {e}")
error_msg = f"ā Error: {str(e)[:100]}..."
return chat_history, None, "", error_msg, create_stats_display(), None
# Global handler instance
# 8. Update event handlers with fixed functions
def setup_fixed_event_handlers(components):
"""Setup event handlers with the fixed functions"""
# Fixed upload handler that includes query processing
components['upload_btn'].click(
fn=lambda file, doc_type, query: handle_document_upload(file, doc_type, query),
inputs=[components['doc_file'], components['doc_type'], components['doc_query']],
outputs=[components['upload_status']]
)
# Fixed audio processing
components['both_btn'].click(
fn=process_with_audio,
inputs=[components['user_input'], components['audio_input'], components['voice_dropdown'], components['chat_history']],
outputs=[components['chatbot'], components['audio_output'], components['user_input'], components['status_output'], components['stats_display'], components['audio_input']])
# Fixed voice chat handlers
components['start_voice_btn'].click(
fn=start_voice_chat,
outputs=[components['voice_status']])
components['stop_voice_btn'].click(
fn=stop_voice_chat,
outputs=[components['voice_status']])
global voice_handler
voice_handler = GeminiHandler()
components['webrtc_audio'].stream(
voice_handler,
inputs=[components['webrtc_audio']],
outputs=[components['webrtc_audio']],
time_limit=180 if get_space() else None,
concurrency_limit=2 if get_space() else None,
)
# 8. Audio fallback function if TTS is not working
def safe_audio_processing(text_response: str, voice: str) -> Optional[tuple]:
"""Safe audio processing with fallback"""
try:
# Check if audio_utils is available
if 'generate_tts_response' in globals():
return generate_tts_response(text_response, voice)
else:
logger.warning("TTS function not available, skipping audio generation")
return None
except Exception as e:
logger.error(f"Audio processing failed: {e}")
return None
# 9. Add proper imports check
def check_required_imports():
"""Check if all required modules are available"""
missing_modules = []
try:
from docs_utils import old_doc_ingestion, old_doc_qa, user_doc_ingest, user_doc_qa, rag_dom_ingest, rag_dom_qa
except ImportError as e:
missing_modules.append(f"docs_utils functions: {e}")
try:
from audio_utils import generate_tts_response, get_transcription_or_text
except ImportError as e:
missing_modules.append(f"audio_utils functions: {e}")
try:
from pipeQuery import process_query, clean_pipeline_result
except ImportError as e:
missing_modules.append(f"pipeQuery functions: {e}")
if missing_modules:
logger.warning(f"Missing modules: {missing_modules}")
return False, missing_modules
return True, []
# 8. Audio fallback function if TTS is not working
def safe_audio_processing(text_response: str, voice: str) -> Optional[tuple]:
"""Safe audio processing with fallback"""
try:
# Check if audio_utils is available
if 'generate_tts_response' in globals():
return generate_tts_response(text_response, voice)
else:
logger.warning("TTS function not available, skipping audio generation")
return None
except Exception as e:
logger.error(f"Audio processing failed: {e}")
return None
# 9. Add proper imports check
def handle_document_upload(file, doc_type: str, query: str = "") -> str:
"""Handle document upload with proper function mapping to docs_utils.py"""
if not file:
return "ā ļø Please select a file to upload."
if doc_type == "None" or not doc_type:
return "ā ļø Please select a document type."
try:
logger.info(f"Processing file upload: {file.name} as {doc_type}")
# Map document types to actual functions from docs_utils.py
if doc_type == "old_documents":
from docs_utils import old_doc_ingestion, old_doc_qa
# First ingest the document
ingest_result = old_doc_ingestion(file.name)
demo_state.document_stats["old_documents"] += 1
# If query provided, also get answer
if query and query.strip():
answer = old_doc_qa(query)
return f"š Old Document processed: {ingest_result}\n\nQuery Answer:\n{answer}"
return f"š Old Document ingested: {ingest_result}"
elif doc_type == "user_specific":
from docs_utils import user_doc_ingest, user_doc_qa
# Ingest the document
ingest_result = user_doc_ingest(file.name)
demo_state.document_stats["user_specific"] += 1
# If query provided, also get answer
if query and query.strip():
answer = user_doc_qa(query)
return f"š¤ User Document processed: {ingest_result}\n\nQuery Answer:\n{answer}"
return f"š¤ User-specific document ingested: {ingest_result}"
elif doc_type == "domain_documents":
from docs_utils import rag_dom_ingest, rag_dom_qa
# Ingest the document
ingest_result = rag_dom_ingest(file.name)
demo_state.document_stats["domain_documents"] += 1
# If query provided, also get answer
if query and query.strip():
answer = rag_dom_qa(query)
return f"š„ Domain Document processed: {ingest_result}\n\nQuery Answer:\n{answer}"
return f"š„ Domain document ingested: {ingest_result}"
elif doc_type == "new_documents":
# Use the standard RAG pipeline for new documents
from rag_steps import ingest_file
ingest_result = ingest_file(file.name)
demo_state.document_stats["new_documents"] += 1
# If query provided, process through main pipeline
if query and query.strip():
response = process_query(query)
cleaned_response = clean_pipeline_result(response[0] if isinstance(response, tuple) else str(response))
return f"š New Document processed: {ingest_result}\n\nQuery Answer:\n{cleaned_response}"
return f"š New document ingested: {ingest_result}"
else:
return f"ā Unknown document type: {doc_type}"
except Exception as e:
logger.error(f"File upload error: {e}")
return f"ā File upload failed: {str(e)}"
def process_old_doc_query(query: str, doc_file) -> str:
"""Process query for old documents specifically"""
global demo_state
if doc_file:
try:
upload_dir = os.path.join("assets", "uploaded_docs")
os.makedirs(upload_dir, exist_ok=True)
safe_filename = os.path.basename(doc_file.name)
save_path = os.path.join(upload_dir, safe_filename)
shutil.copy2(doc_file.name, save_path)
demo_state.last_uploaded_old_doc = save_path
return process_old_document(save_path, query)
except Exception as e:
return f"ā Error processing file: {str(e)}"
else:
if demo_state.last_uploaded_old_doc and os.path.exists(demo_state.last_uploaded_old_doc):
try:
answer = old_doc_qa(query)
return f"[Using previously uploaded document: {os.path.basename(demo_state.last_uploaded_old_doc)}]\n\n{answer}"
except Exception as e:
return f"ā Error querying document: {str(e)}"
else:
return "ā No document uploaded. Please upload an old document to proceed."
def clear_chat():
"""Clear chat history and reset stats"""
demo_state.__init__() # Reset stats
return [], None, "š Chat cleared and stats reset!", create_stats_display()
def create_main_chat_tab():
"""Create the main chat interface tab"""
with gr.Row():
with gr.Column(scale=3):
# Main chatbot interface
with gr.Group():
chatbot = gr.Chatbot(
type='messages',
label="š¬ Conversation with Wisal",
height=500,
show_copy_button=True,
elem_classes="chatbot-container",
)
# Status and audio output
with gr.Row():
with gr.Column():
status_output = gr.Textbox(
label="š System Status",
interactive=False,
max_lines=2,
elem_classes="input-container"
)
with gr.Column():
audio_output = gr.Audio(
label="šµ Audio Response",
interactive=False,
show_download_button=True,
elem_classes="input-container"
)
# Input section
with gr.Group():
with gr.Row():
user_input = gr.Textbox(
placeholder="Ask me anything about autism...",
label="š Your Message",
lines=2,
scale=3,
elem_classes="input-container"
)
audio_input = gr.Audio(
sources=["microphone", "upload"],
type="filepath",
label="š¤ Voice Input",
scale=1,
elem_classes="input-container"
)
voice_dropdown = gr.Dropdown(
label="šļø Voice Selection",
choices=["Kore", "Puck", "Zephyr", "Leda", "Fenrir", "Charon", "Orus", "Aoede", "Callirrhoe"],
value="Kore",
elem_classes="input-container"
)
# Action buttons
with gr.Row():
text_btn = gr.Button(
"š¬ Text Response",
variant="secondary",
elem_classes="enhanced-button secondary-button"
)
audio_btn = gr.Button(
"šµ Audio Response",
variant="secondary",
elem_classes="enhanced-button secondary-button"
)
both_btn = gr.Button(
"š Text + Audio",
variant="primary",
elem_classes="enhanced-button primary-button"
)
with gr.Row():
clear_btn = gr.Button(
"šļø Clear Chat",
variant="stop",
elem_classes="enhanced-button danger-button"
)
with gr.Column(scale=1):
# Statistics panel
stats_display = gr.HTML(
create_stats_display(),
label="š Statistics"
)
# Enhanced Document upload section
with gr.Group():
# Fixed Document upload section
gr.Markdown("### š Document Upload")
doc_file = gr.File(
label="Upload Document",
file_types=[".pdf", ".docx", ".txt"],
elem_classes="input-container"
)
# Fixed dropdown with proper choices
doc_type = gr.Dropdown(
label="Document Type",
choices=[
("š Old Documents", "old_documents"),
("š¤ User-Specific Documents", "user_specific"),
("š„ Domain Documents", "domain_documents"),
("š New Documents", "new_documents")
],
value="user_specific",
elem_classes="input-container"
)
# Optional query field for immediate Q&A
doc_query = gr.Textbox(
label="Optional: Ask about this document",
placeholder="What does this document say about...",
lines=2,
elem_classes="input-container"
)
upload_btn = gr.Button(
"š¤ Upload & Process",
elem_classes="enhanced-button primary-button"
)
upload_status = gr.Textbox(
label="Upload Status",
interactive=False,
lines=4,
elem_classes="input-container"
)
# Document type info display
doc_info = gr.HTML(
"""
š Document Types
š Old Documents: Historical documents, legacy materials
š¤ User-Specific: Personal files, user manuals
š„ Domain Documents: Medical papers, research articles
š New Documents: Recent uploads, latest materials
"""
)
upload_btn = gr.Button(
"š¤ Upload",
elem_classes="enhanced-button primary-button"
)
upload_status = gr.Textbox(
label="Upload Status",
interactive=False,
lines=3,
elem_classes="input-container"
)
# Quick actions
with gr.Group():
gr.Markdown("### ā” Quick Actions")
sample_queries = [
"What is autism?",
"Early signs of autism",
"Autism therapy options",
"Supporting autistic children"
]
for query in sample_queries:
btn = gr.Button(
query,
elem_classes="enhanced-button secondary-button",
size="sm"
)
btn.click(
lambda q=query: q,
outputs=[user_input]
)
return {
'chatbot': chatbot,
'user_input': user_input,
'audio_input': audio_input,
'doc_query': doc_query,
'voice_dropdown': voice_dropdown,
'audio_output': audio_output,
'status_output': status_output,
'stats_display': stats_display,
'doc_file': doc_file,
'doc_type': doc_type,
'upload_status': upload_status,
'text_btn': text_btn,
'audio_btn': audio_btn,
'both_btn': both_btn,
'upload_btn': upload_btn,
'clear_btn': clear_btn
}
def create_old_documents_tab():
"""Create the old documents Q&A tab"""
with gr.Row():
with gr.Column(scale=2):
gr.Markdown("## š Old Documents Q&A")
gr.Markdown("Upload historical documents, legacy materials, and archived content for specialized querying.")
query = gr.Textbox(
placeholder="Ask about old documents...",
lines=3,
label="š Your Question about Old Documents",
elem_classes="input-container"
)
doc_file = gr.File(
label="š Upload Old Document (PDF, DOCX, TXT)",
file_types=[".pdf", ".docx", ".txt"],
elem_classes="input-container"
)
with gr.Row():
submit_btn = gr.Button(
"š Submit Query",
variant="primary",
elem_classes="enhanced-button primary-button"
)
clear_old_btn = gr.Button(
"šļø Clear",
variant="secondary",
elem_classes="enhanced-button danger-button"
)
output = gr.Textbox(
label="š Answer from Old Documents",
lines=12,
interactive=False,
elem_classes="input-container"
)
with gr.Column(scale=1):
# Old documents info panel
gr.HTML("""
š” Processing Tips
File Formats: PDF, DOCX, TXT, MD
Size Limit: Up to 50MB per file
Batch Processing: Up to 10 files at once
Processing Time: 10-30 seconds per document
""")
return {
'upload_file': upload_file,
'upload_doc_type': upload_doc_type,
'upload_query': upload_query,
'doc_result': doc_result,
'batch_files': batch_files,
'batch_type': batch_type,
'batch_result': batch_result,
'process_doc_btn': process_doc_btn,
'batch_process_btn': batch_process_btn,
'doc_stats_display': doc_stats_display
}
def process_batch_documents(files, doc_type: str) -> str:
"""Process multiple documents in batch"""
if not files:
return "ā ļø No files selected for batch processing."
if doc_type == "None" or not doc_type:
return "ā ļø Please select a document type for batch processing."
results = []
successful = 0
failed = 0
for file in files:
try:
result = handle_document_upload(file, doc_type)
if result.startswith("ā"):
failed += 1
results.append(f"ā {os.path.basename(file.name)}: Failed")
else:
successful += 1
results.append(f"ā
{os.path.basename(file.name)}: Success")
except Exception as e:
failed += 1
results.append(f"ā {os.path.basename(file.name)}: {str(e)[:50]}...")
summary = f"š¦ Batch Processing Complete:\nā
Successful: {successful}\nā Failed: {failed}\n\n"
summary += "\n".join(results)
return summary
# def create_demo():
# """Create the enhanced integrated Gradio demo"""
# # Custom theme
# theme = gr.themes.Base(
# primary_hue="blue",
# secondary_hue="purple",
# neutral_hue="slate",
# font=gr.themes.GoogleFont("Inter")
# )
# with gr.Blocks(
# title="Wisal - Advanced Autism AI Assistant",
# theme=theme,
# css=ENHANCED_CSS,
# head="""
#
#
š§ Wisal
#
Advanced AI Assistant for Autism Spectrum Disorders
#
#
#
AI Assistant Active - All Services Online
#
#
#
š” How to Use Wisal
#
#
š¬ Main Chat: Primary interface for text and audio conversations with comprehensive features
#
š Old Documents: Specialized processing for historical and legacy documents
#
šļø Voice Chat: Real-time speech-to-speech conversation with advanced voice AI
#
š Document Management: Advanced document upload, processing, and batch operations
#
š¤ Audio Features: Support for voice input, multiple voice options, and audio responses
#
š Statistics: Real-time tracking of usage, performance, and document processing metrics
#
#
š Document Processing Options
š Old Documents
Historical documents, legacy materials, archived content
š¤ User-Specific Documents
Personal files, user manuals, individual documents
š„ Domain Documents
Medical papers, research articles, domain knowledge
š New Documents
Recent uploads, latest materials, standard processing
š” Tip: Each document type uses specialized processing optimized for that content category.
"""
def get_enhanced_css() -> str:
"""Get enhanced CSS from gradio_utils"""
return ENHANCED_CSS
def create_enhanced_file_upload_interface():
"""Create an enhanced file upload interface with all 4 document types"""
with gr.Group():
gr.Markdown("### š Enhanced Document Upload")
doc_file = gr.File(
label="š Upload Document (PDF, DOCX, TXT)",
file_types=[".pdf", ".docx", ".txt"],
elem_classes="input-container"
)
# Fixed dropdown with all 4 options
doc_type = gr.Dropdown(
label="š Document Type",
choices=get_all_document_choices(),
value="user_specific",
elem_classes="input-container"
)
# Optional query field
doc_query = gr.Textbox(
label="š Optional: Ask about this document",
placeholder="What does this document say about...",
lines=2,
elem_classes="input-container"
)
upload_btn = gr.Button(
"š¤ Upload & Process",
variant="primary",
elem_classes="enhanced-button primary-button"
)
upload_status = gr.Textbox(
label="š Upload Status",
interactive=False,
lines=5,
elem_classes="input-container"
)
# Document info panel
doc_info = gr.HTML(create_document_info_panel())
return {
'doc_file': doc_file,
'doc_type': doc_type,
'doc_query': doc_query,
'upload_btn': upload_btn,
'upload_status': upload_status,
'doc_info': doc_info
}
# Export main functions for use in main.py
__all__ = [
'get_all_document_choices',
'enhanced_document_upload_handler',
'enhanced_audio_transcription',
'process_text_with_audio_support',
'process_audio_only_response',
'process_both_text_and_audio_response',
'create_document_info_panel',
'create_enhanced_file_upload_interface',
'get_enhanced_css',
'ENHANCED_CSS',
'demo_state'
]