"
return gr.update(visible=True), suggestions_html
except Exception as e:
print(f"Error loading suggestions: {e}")
return gr.update(visible=False), ""
# Set up event handlers
if self.proactive_engine and user_id_input and suggestions_display:
# Suggestions event handlers
suggestions_event = load_suggestions_btn.click(
load_suggestions,
inputs=[user_id_input],
outputs=[suggestions_container, suggestions_display]
)
if refresh_suggestions_btn:
refresh_suggestions_btn.click(
load_suggestions,
inputs=[user_id_input],
outputs=[suggestions_container, suggestions_display]
)
if regenerate_suggestions_btn:
regenerate_suggestions_btn.click(
load_suggestions,
inputs=[user_id_input],
outputs=[suggestions_container, suggestions_display]
)
if cancel_suggestions_btn:
cancel_suggestions_btn.click(fn=None, cancels=suggestions_event)
# Build outputs list for query
outputs_list = [answer_output, footnotes_output]
if self.enhanced_rag_engine:
outputs_list.extend([scenarios_container, scenarios_display])
outputs_list.extend([followup_container, followup_questions_display])
# Query event handlers
query_event = submit_btn.click(
process_query,
inputs=[query_input, user_id_input],
outputs=outputs_list
)
regenerate_answer_btn.click(
process_query,
inputs=[query_input, user_id_input],
outputs=outputs_list
)
if cancel_answer_btn:
cancel_answer_btn.click(fn=None, cancels=query_event)
# Follow-up questions event handlers (regenerate only, cancel is handled by query cancel)
if self.proactive_engine and regenerate_followup_btn:
def regenerate_followup(query, user_id, answer_text):
"""Regenerate follow-up questions based on the current answer"""
if not self.proactive_engine or not user_id or not answer_text:
return gr.update(visible=False), ""
try:
followup_questions = self.proactive_engine.get_follow_up_questions(
user_id, answer_text, max_questions=5
)
if followup_questions:
followup_html = "
"
for i, q_data in enumerate(followup_questions, 1):
question = q_data.get("question", "")
bloom_level = q_data.get("bloom_level", "")
question_escaped = question.replace("'", "\\'").replace('"', '\\"')
followup_html += f"""
{question}
Bloom Level: {bloom_level.title()}
"""
followup_html += "
"
return gr.update(visible=True), followup_html
else:
return gr.update(visible=False), ""
except Exception as e:
print(f"Error regenerating follow-up questions: {e}")
return gr.update(visible=False), ""
followup_event = regenerate_followup_btn.click(
regenerate_followup,
inputs=[query_input, user_id_input, answer_output],
outputs=[followup_container, followup_questions_display]
)
if cancel_followup_btn:
cancel_followup_btn.click(fn=None, cancels=followup_event)
else:
# Build outputs list (must match process_query return values)
outputs_list = [answer_output, footnotes_output]
if self.enhanced_rag_engine:
outputs_list.extend([scenarios_container, scenarios_display])
if self.proactive_engine:
outputs_list.extend([followup_container, followup_questions_display])
query_event = submit_btn.click(
process_query,
inputs=[query_input],
outputs=outputs_list
)
regenerate_answer_btn.click(
process_query,
inputs=[query_input],
outputs=outputs_list
)
if cancel_answer_btn:
cancel_answer_btn.click(fn=None, cancels=query_event)
def _create_knowledge_map_tab(self):
"""Create the knowledge map tab"""
gr.Markdown("## π Car Manual Knowledge Map")
gr.Markdown("This visualization shows how different concepts in the car manual are related.")
knowledge_map_img = gr.Image(
value=str(self.config.output_dir / "knowledge_graph.png"),
label="Knowledge Graph"
)
gr.Markdown("## π₯ Document Similarity Heatmap")
gr.Markdown("This heatmap shows how similar different ADAS features are to each other.")
similarity_heatmap_img = gr.Image(
value=str(self.config.output_dir / "similarity_heatmap.png"),
label="Similarity Heatmap"
)
with gr.Row():
refresh_btn = gr.Button("π Refresh Visualizations", variant="secondary")
def refresh_images():
graph_path, heatmap_path = self.knowledge_graph.generate_visualizations()
return graph_path, heatmap_path
refresh_btn.click(
refresh_images,
inputs=[],
outputs=[knowledge_map_img, similarity_heatmap_img]
)
def _create_test_tab(self):
"""Create the test tab"""
gr.Markdown("## π Test Your Understanding of Mercedes E-class ADAS")
gr.Markdown("Select a topic to test your knowledge with multiple-choice questions based on Bloom's taxonomy levels.")
topic_files = self.rag_engine.get_files_from_vector_store()
with gr.Row():
test_questions = gr.State(None)
current_level_idx = gr.State(0)
selected_topic = gr.State(None)
test_results = gr.State([])
topic_dropdown = gr.Dropdown(
label="Select a Topic",
choices=topic_files,
value=topic_files[0] if topic_files else None,
interactive=True
)
start_test_btn = gr.Button("Start Test", variant="primary")
# Progress indicator
with gr.Column(visible=False) as progress_container:
progress_html = gr.HTML()
# Test container
with gr.Column(visible=False) as test_container:
taxonomy_level = gr.Markdown("Level: Remember")
level_description = gr.Markdown()
question_display = gr.Markdown()
option_radio = gr.Radio(
choices=["A", "B", "C", "D"],
label="Select your answer",
interactive=True
)
submit_answer_btn = gr.Button("Submit Answer", variant="primary")
feedback_display = gr.Markdown(visible=False)
next_question_btn = gr.Button("Next Question", visible=False)
show_summary_btn = gr.Button("Show Summary", visible=False)
# Summary container
with gr.Column(visible=False) as summary_container:
summary_topic = gr.Markdown()
summary_results = gr.HTML()
summary_recommendation = gr.Markdown()
restart_btn = gr.Button("Start Another Test")
# Connect handlers (simplified - full implementation would include all handlers)
# This is a placeholder structure - full implementation would be quite long
def _create_onboarding_tab(self):
"""Create onboarding tab for cold start"""
gr.Markdown("## π― Welcome! Let's Get Started")
gr.Markdown("Complete your profile to get a personalized learning experience.")
if not self.user_profiling:
gr.Markdown("β οΈ Personalized Learning System not initialized.")
return
user_id_input = gr.Textbox(
label="User ID",
placeholder="Enter your user ID",
value="default_user"
)
with gr.Accordion("π Step 1: Background Information", open=True):
background_input = gr.Radio(
label="What's your experience with ADAS systems?",
choices=[
("Beginner - I'm new to ADAS systems", "beginner"),
("Intermediate - I know some basics", "intermediate"),
("Experienced - I have good knowledge", "experienced")
],
value="beginner"
)
with gr.Accordion("π¨ Step 2: Learning Preferences", open=True):
learning_style_input = gr.Radio(
label="How do you prefer to learn?",
choices=[
("Visual - I like diagrams and illustrations", "visual"),
("Textual - I prefer reading and explanations", "textual"),
("Practical - I learn by doing", "practical"),
("Mixed - I like a combination", "mixed")
],
value="mixed"
)
learning_pace_input = gr.Radio(
label="What's your preferred learning pace?",
choices=[
("Slow - I like to take my time", "slow"),
("Medium - Normal pace is fine", "medium"),
("Fast - I want to learn quickly", "fast")
],
value="medium"
)
with gr.Accordion("π― Step 3: Learning Goals", open=True):
learning_goals_input = gr.CheckboxGroup(
label="What are your learning goals?",
choices=[
"Understand basic ADAS functions",
"Learn how to operate ADAS features",
"Master advanced ADAS capabilities",
"Troubleshoot ADAS issues",
"Prepare for certification",
"General knowledge improvement"
],
value=["Understand basic ADAS functions"]
)
with gr.Accordion("π Step 4: Initial Knowledge Assessment", open=True):
gr.Markdown("Rate your familiarity with each topic (0 = No knowledge, 1 = Expert)")
knowledge_sliders = {}
for topic in self.config.available_topics:
display_name = topic.replace("Function of ", "").replace(" Assist", "")
knowledge_sliders[topic] = gr.Slider(
label=display_name,
minimum=0.0,
maximum=1.0,
value=0.0,
step=0.1
)
submit_btn = gr.Button("Complete Setup", variant="primary")
output_result = gr.JSON(label="Setup Result")
def submit_onboarding(user_id, background, learning_style, learning_pace,
learning_goals, *knowledge_values):
"""Submit cold start data"""
if not self.user_profiling:
return {"status": "error", "message": "System not initialized"}
# Convert knowledge_values tuple to dictionary
knowledge_survey = {}
for i, topic in enumerate(self.config.available_topics):
if i < len(knowledge_values):
knowledge_survey[topic] = knowledge_values[i]
else:
knowledge_survey[topic] = 0.0
# Handle tuple values from Radio components
if isinstance(background, tuple):
background = background[1] if len(background) > 1 else background[0]
if isinstance(learning_style, tuple):
learning_style = learning_style[1] if len(learning_style) > 1 else learning_style[0]
if isinstance(learning_pace, tuple):
learning_pace = learning_pace[1] if len(learning_pace) > 1 else learning_pace[0]
onboarding_data = {
'learning_style': learning_style,
'learning_pace': learning_pace,
'background_experience': background,
'learning_goals': learning_goals if learning_goals else [],
'initial_knowledge_survey': knowledge_survey,
'initial_assessment_completed': True
}
try:
profile = self.user_profiling.complete_onboarding(user_id, onboarding_data)
return {
"status": "success",
"message": f"Onboarding completed for {user_id}",
"profile_summary": self.user_profiling.get_profile_summary(user_id)
}
except Exception as e:
import traceback
error_details = traceback.format_exc()
return {"status": "error", "message": f"Error: {str(e)}\nDetails: {error_details}"}
inputs = [user_id_input, background_input, learning_style_input,
learning_pace_input, learning_goals_input] + list(knowledge_sliders.values())
submit_btn.click(submit_onboarding, inputs=inputs, outputs=output_result)
def _create_learning_path_tab(self):
"""Create personalized learning path tab"""
gr.Markdown("## πΊοΈ Your Personalized Learning Journey")
gr.Markdown("Get a customized learning path based on your knowledge profile.")
if not self.adaptive_engine or not self.user_profiling:
gr.Markdown("β οΈ Personalized Learning System not initialized.")
return
# User ID input
with gr.Row():
user_id_input = gr.Textbox(
label="User ID",
placeholder="Enter your user ID",
value="default_user"
)
load_profile_btn = gr.Button("Load My Profile", variant="primary")
# User profile display
with gr.Column(visible=False) as profile_container:
profile_summary = gr.Markdown()
with gr.Row():
with gr.Column():
gr.Markdown("### π Knowledge Profile")
knowledge_level_display = gr.JSON()
with gr.Column():
gr.Markdown("### π Learning Statistics")
learning_stats = gr.JSON()
# Learning path generation
with gr.Row():
focus_areas_input = gr.CheckboxGroup(
label="Focus Areas (Optional)",
choices=self.config.available_topics,
value=[],
interactive=True
)
generate_path_btn = gr.Button("Generate Learning Path", variant="primary")
# Learning path display
with gr.Column(visible=False) as path_container:
gr.Markdown("### πΊοΈ Your Learning Path")
path_progress = gr.HTML()
path_visualization = gr.HTML()
with gr.Row():
with gr.Column():
current_node_info = gr.Markdown()
recommendations_display = gr.JSON()
def check_and_show_onboarding_wrapper(user_id):
"""Check if user needs onboarding"""
if not user_id:
return False
return check_and_show_onboarding(self.user_profiling, user_id)
def load_user_profile(user_id):
"""Load the user profile"""
if not self.user_profiling or not user_id:
return (gr.update(visible=False), "", {}, {}, [])
if check_and_show_onboarding_wrapper(user_id):
return (
gr.update(visible=False),
f"## β οΈ Onboarding Required\n\nPlease complete onboarding first.",
{},
{},
self.config.available_topics
)
profile = self.user_profiling.get_or_create_profile(user_id)
summary = self.user_profiling.get_profile_summary(user_id)
summary_text = f"""
### π€ User Profile: {user_id}
**Learning Style:** {summary['learning_style'].title()}
**Learning Pace:** {summary['learning_pace'].title()}
**Overall Progress:** {summary['overall_progress']:.1%}
**Total Questions:** {summary['total_questions']}
**Total Tests:** {summary['total_tests']}
**Strong Areas:** {', '.join(summary['strong_areas']) if summary['strong_areas'] else 'None'}
**Weak Areas:** {', '.join(summary['weak_areas']) if summary['weak_areas'] else 'None'}
"""
knowledge_data = summary['knowledge_level'] or {"No topics learned yet": 0.0}
stats_data = {
"Total Questions": summary['total_questions'],
"Total Tests": summary['total_tests'],
"Preferred Topics": summary['preferred_topics'][:5] if summary['preferred_topics'] else [],
"Overall Progress": f"{summary['overall_progress']:.1%}"
}
return (
gr.update(visible=True),
summary_text,
knowledge_data,
stats_data,
self.config.available_topics
)
def generate_learning_path(user_id, focus_areas):
"""Generate learning paths"""
if not self.adaptive_engine or not user_id:
return (gr.update(visible=False), "β οΈ System not initialized.", "", "", {})
if check_and_show_onboarding_wrapper(user_id):
return (gr.update(visible=False), "β οΈ Please complete onboarding first.", "", "", {})
path = self.adaptive_engine.create_or_update_path(user_id, focus_areas if focus_areas else None)
progress_html = f"""
{path.completion_percentage*100:.1f}% Complete
Total Nodes: {len(path.nodes)} | Completed: {sum(1 for n in path.nodes if n.status == 'completed')} | Estimated Time: {path.estimated_total_time} minutes
"""
path_html = "
Learning Path:
"
for i, node in enumerate(path.nodes):
status_color = {"completed": "#4CAF50", "in_progress": "#2196F3", "pending": "#9E9E9E", "skipped": "#FF9800"}.get(node.status, "#9E9E9E")
is_current = i == path.current_node_index
highlight = "border: 3px solid #FF5722; padding: 10px;" if is_current else "padding: 10px;"
path_html += f"""
{node.topic} - {node.bloom_level.title()} ({node.content_type}) Difficulty: {node.difficulty:.2f} | Time: {node.estimated_time} min