import streamlit as st import os import numpy as np import pandas as pd import matplotlib.pyplot as plt import base64 import time import sqlite3 import datetime import sys import re import queue from concurrent.futures import ThreadPoolExecutor from PIL import Image from io import BytesIO from gamification import GamificationSystem import random from collections import defaultdict import io from vocam_ui import apply_custom_css from streamlit.components.v1 import components import hashlib from functools import lru_cache from example_sentences import ExampleSentenceGenerator import tensorflow as tf import tensorflow_hub as hub import requests from deep_translator import GoogleTranslator from transformers import pipeline, AutoTokenizer, AutoModelForSeq2SeqLM # First, display Python version for st.set_page_config( page_title="Vocam", page_icon="๐", layout="wide", initial_sidebar_state="expanded" ) # Import the UI enhancement module from vocam_ui import ( apply_custom_css, success_message, info_message, warning_message, error_message, show_loading_spinner, vocam_card, word_card, add_result_separator, add_scroll_indicator, style_title, style_section_title, add_footer ) apply_custom_css() try: from pronunciation_practice import create_pronunciation_practice has_pronunciation_practice = True print("โ Enhanced pronunciation practice with AI feedback loaded") except ImportError as e: has_pronunciation_practice = False print(f"โ Pronunciation practice not available: {e}") # Try importing OCR with fallback try: import pytesseract has_tesseract = True except ImportError as e: has_tesseract = False # Dummy implementation class DummyTesseract: def image_to_string(self, *args, **kwargs): return "OCR requires pytesseract. Install with: pip install pytesseract" pytesseract = DummyTesseract() # Try importing OpenCV with robust fallback mechanism try: import cv2 except ImportError as e: # Create dummy CV2 class to prevent crashes class DummyCV2: def __init__(self): pass def __getattr__(self, name): def dummy_method(*args, **kwargs): return None return dummy_method def cvtColor(self, *args, **kwargs): return args[0] # Return the input image unchanged @staticmethod def imread(path): try: from PIL import Image import numpy as np img = Image.open(path) return np.array(img) except Exception: return None @staticmethod def imwrite(path, img): try: from PIL import Image import numpy as np Image.fromarray(img).save(path) return True except False: return False # Replace cv2 with our dummy implementation cv2 = DummyCV2() # Import other dependencies with careful error handling try: import torch except ImportError as e: # Dummy torch for fallback class DummyTorch: def __init__(self): self.hub = type('obj', (object,), { 'load': lambda *args, **kwargs: DummyModel() }) class DummyModel: def __call__(self, *args, **kwargs): return type('obj', (object,), { 'xyxy': [[]], 'render': lambda: [[np.zeros((300, 300, 3), dtype=np.uint8)]], 'names': {0: 'unknown'} }) def eval(self): return self torch = DummyTorch() # Try importing gTTS try: from gtts import gTTS except ImportError as e: # Create a dummy gTTS class class DummyGTTS: def __init__(self, text="", lang="en", slow=False): self.text = text self.lang = lang def write_to_fp(self, fp): fp.write(b'dummy audio data') gTTS = DummyGTTS # Import database module with error handling try: from database import LanguageLearningDB except ImportError as e: # Define a basic LanguageLearningDB class for fallback class LanguageLearningDB: def __init__(self, db_path): self.db_path = db_path def start_session(self): return None def end_session(self, session_id, words_studied, words_learned): return True # Import custom audio recorder try: from custom_audio_recorder import audio_recorder has_custom_recorder = True print("Custom audio recorder imported successfully") except ImportError as e: has_custom_recorder = False print(f"Custom audio recorder not available: {e}") def check_pronunciation_dependencies(): """Check and report pronunciation practice dependencies""" dependencies = { 'streamlit_webrtc': False, 'speech_recognition': False, 'librosa': False, 'Levenshtein': False, 'av': False } try: import streamlit_webrtc dependencies['streamlit_webrtc'] = True except ImportError: pass try: import speech_recognition dependencies['speech_recognition'] = True except ImportError: pass try: import librosa dependencies['librosa'] = True except ImportError: pass try: import Levenshtein dependencies['Levenshtein'] = True except ImportError: pass try: import av dependencies['av'] = True except ImportError: pass return dependencies def draw_detections(image_np, detections): """Draw bounding boxes and labels on the image.""" result_image = image_np.copy() for detection in detections: bbox = detection['bbox'] left, top, right, bottom = [int(x) for x in bbox] label = detection['label'] confidence = detection['confidence'] # Use different colors for different object types color = get_detection_color(label) # Draw bounding box cv2.rectangle(result_image, (left, top), (right, bottom), color, 3) # Prepare label text label_text = f"{label} {confidence:.2f}" label_size, _ = cv2.getTextSize(label_text, cv2.FONT_HERSHEY_SIMPLEX, 0.7, 2) # Draw background for text cv2.rectangle(result_image, (left, top - label_size[1] - 10), (left + label_size[0], top), color, -1) # Draw text (white or black depending on background) text_color = (255, 255, 255) if sum(color) < 400 else (0, 0, 0) cv2.putText(result_image, label_text, (left, top - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.7, text_color, 2) return result_image def apply_nms(boxes, classes, scores, image_shape, iou_threshold=0.45): """Apply Non-Maximum Suppression to remove duplicate detections.""" final_detections = [] unique_classes = np.unique(classes) height, width = image_shape[:2] for class_id in unique_classes: # Get all detections for this class class_mask = classes == class_id class_boxes = boxes[class_mask] class_scores = scores[class_mask] if len(class_boxes) == 0: continue # Convert normalized coordinates to pixel coordinates pixel_boxes = [] for box in class_boxes: ymin, xmin, ymax, xmax = box pixel_boxes.append([ int(xmin * width), # left int(ymin * height), # top int(xmax * width), # right int(ymax * height) # bottom ]) pixel_boxes = np.array(pixel_boxes) # Apply simple NMS (since OpenCV might cause issues) keep_indices = simple_nms(pixel_boxes, class_scores, iou_threshold) # Add kept detections to final list for idx in keep_indices: class_name = COCO_CLASS_NAMES.get(class_id, f"unknown_{class_id}") bbox = pixel_boxes[idx] final_detections.append({ 'label': class_name.lower(), 'confidence': float(class_scores[idx]), 'bbox': [float(bbox[0]), float(bbox[1]), float(bbox[2]), float(bbox[3])], 'class_id': int(class_id) }) # Sort detections by confidence (highest first) final_detections.sort(key=lambda x: x['confidence'], reverse=True) return final_detections # Alternative simplified NMS function if OpenCV NMS doesn't work def simple_nms(boxes, scores, iou_threshold=0.5): """Simple Non-Maximum Suppression implementation.""" if len(boxes) == 0: return [] # Sort by confidence score (highest first) sorted_indices = np.argsort(scores)[::-1] keep = [] while len(sorted_indices) > 0: # Take the detection with highest confidence current = sorted_indices[0] keep.append(current) if len(sorted_indices) == 1: break # Calculate IoU with all other boxes current_box = boxes[current] remaining_indices = sorted_indices[1:] # Calculate IoU with remaining boxes ious = [] for idx in remaining_indices: iou = calculate_iou(current_box, boxes[idx]) ious.append(iou) # Keep only boxes with IoU below threshold ious = np.array(ious) keep_mask = ious < iou_threshold sorted_indices = remaining_indices[keep_mask] return keep def calculate_iou(box1, box2): """Calculate Intersection over Union (IoU) of two bounding boxes.""" # box format: [left, top, right, bottom] x1 = max(box1[0], box2[0]) y1 = max(box1[1], box2[1]) x2 = min(box1[2], box2[2]) y2 = min(box1[3], box2[3]) if x2 <= x1 or y2 <= y1: return 0.0 intersection = (x2 - x1) * (y2 - y1) area1 = (box1[2] - box1[0]) * (box1[3] - box1[1]) area2 = (box2[2] - box2[0]) * (box2[3] - box2[1]) union = area1 + area2 - intersection return intersection / union if union > 0 else 0.0 def get_detection_color(label): """Get a consistent color for each object type.""" # Color mapping for different object categories color_map = { # Electronics - Blue shades 'cell phone': (255, 100, 100), 'laptop': (255, 150, 100), 'tv': (255, 200, 100), 'mouse': (200, 255, 100), 'keyboard': (150, 255, 100), 'remote': (100, 255, 100), # People - Green shades 'person': (100, 255, 150), # Furniture - Purple shades 'chair': (150, 100, 255), 'couch': (200, 100, 255), 'bed': (255, 100, 255), # Food - Orange/Red shades 'bottle': (100, 150, 255), 'cup': (100, 200, 255), 'bowl': (100, 255, 255), # Default color 'default': (0, 255, 0) } return color_map.get(label, color_map['default']) def show_detection_settings(): """Show detection settings in the sidebar.""" with st.sidebar.expander("๐๏ธ Detection Settings"): st.markdown("**Non-Maximum Suppression (NMS)**") st.markdown("โ Enabled - Removes duplicate detections") # Allow user to adjust IOU threshold iou_threshold = st.slider( "Overlap Threshold", min_value=0.1, max_value=0.9, value=0.45, step=0.05, help="Lower values = fewer duplicates, Higher values = more detections" ) st.markdown(f"**Current Settings:**") st.markdown(f"- Overlap: {iou_threshold:.2f}") st.markdown("- Confidence: Set below โฌ๏ธ") return iou_threshold # Helper function to convert AttrDict to a regular dict recursively def convert_to_dict(obj): if isinstance(obj, dict): return {key: convert_to_dict(value) for key, value in obj.items()} elif isinstance(obj, list): return [convert_to_dict(item) for item in obj] else: return obj # Define object categories for better organization OBJECT_CATEGORIES = { "food": ["banana", "apple", "sandwich", "orange", "broccoli", "carrot", "hot dog", "pizza", "donut", "cake", "bottle", "wine glass", "cup", "fork", "knife", "spoon", "bowl"], "animals": ["bird", "cat", "dog", "horse", "sheep", "cow", "elephant", "bear", "zebra", "giraffe"], "vehicles": ["bicycle", "car", "motorcycle", "airplane", "bus", "train", "truck", "boat"], "electronics": ["tv", "laptop", "mouse", "remote", "keyboard", "cell phone", "microwave", "oven", "toaster", "refrigerator"], "furniture": ["chair", "couch", "potted plant", "bed", "dining table", "toilet", "bench"], "personal": ["backpack", "umbrella", "handbag", "tie", "suitcase"], "sports": ["frisbee", "skis", "snowboard", "sports ball", "kite", "baseball bat", "baseball glove", "skateboard", "surfboard", "tennis racket"], "household": ["bottle", "wine glass", "cup", "fork", "knife", "spoon", "bowl", "book", "clock", "vase", "scissors", "teddy bear", "hair drier", "toothbrush", "sink"] } COCO_CLASS_NAMES = { 1: 'person', 2: 'bicycle', 3: 'car', 4: 'motorcycle', 5: 'airplane', 6: 'bus', 7: 'train', 8: 'truck', 9: 'boat', 10: 'traffic light', 11: 'fire hydrant', 13: 'stop sign', 14: 'parking meter', 15: 'bench', 16: 'bird', 17: 'cat', 18: 'dog', 19: 'horse', 20: 'sheep', 21: 'cow', 22: 'elephant', 23: 'bear', 24: 'zebra', 25: 'giraffe', 27: 'backpack', 28: 'umbrella', 31: 'handbag', 32: 'tie', 33: 'suitcase', 34: 'frisbee', 35: 'skis', 36: 'snowboard', 37: 'sports ball', 38: 'kite', 39: 'baseball bat', 40: 'baseball glove', 41: 'skateboard', 42: 'surfboard', 43: 'tennis racket', 44: 'bottle', 46: 'wine glass', 47: 'cup', 48: 'fork', 49: 'knife', 50: 'spoon', 51: 'bowl', 52: 'banana', 53: 'apple', 54: 'sandwich', 55: 'orange', 56: 'broccoli', 57: 'carrot', 58: 'hot dog', 59: 'pizza', 60: 'donut', 61: 'cake', 62: 'chair', 63: 'couch', 64: 'potted plant', 65: 'bed', 67: 'dining table', 70: 'toilet', 72: 'tv', 73: 'laptop', 74: 'mouse', 75: 'remote', 76: 'keyboard', 77: 'cell phone', 78: 'microwave', 79: 'oven', 80: 'toaster', 81: 'sink', 82: 'refrigerator', 84: 'book', 85: 'clock', 86: 'vase', 87: 'scissors', 88: 'teddy bear', 89: 'hair drier', 90: 'toothbrush' } # Define question types QUESTION_TYPES = [ "translation_to_target", # English โ Target language "translation_to_english", # Target language โ English "image_recognition", # Show image, select correct word "category_match", # Match word to correct category "sentence_completion", # Fill in blank in a sentence "multiple_choice_category", # Choose words from same category "audio_recognition" # Hear word, select correct option ] def get_object_category(label): """Get the category for a detected object label.""" label = label.lower() for category, items in OBJECT_CATEGORIES.items(): if label in items: return category return "other" # Add this to optimize API usage and reduce costs @lru_cache(maxsize=100) def cached_vision_detection(image_hash, confidence_threshold): """Cache detection results based on image hash to avoid redundant API calls.""" # This is a placeholder - the actual implementation would be tied to your caching mechanism # Return None to indicate cache miss return None def get_image_hash(image): """Create a hash of an image for caching purposes.""" img_byte_arr = io.BytesIO() image.save(img_byte_arr, format='JPEG', quality=70) # Lower quality for hash stability return hashlib.md5(img_byte_arr.getvalue()).hexdigest() # Add rate limiting to avoid excessive API calls last_api_call = 0 MIN_API_CALL_INTERVAL = 0.5 # seconds def rate_limited_detection(image, confidence_threshold=0.5, iou_threshold=0.45): """Rate-limited version of detect_objects to avoid excessive API calls.""" global last_api_call # Check cache first image_hash = get_image_hash(image) cached_result = cached_vision_detection(image_hash, confidence_threshold) if cached_result: return cached_result # Rate limiting current_time = time.time() time_since_last_call = current_time - last_api_call if time_since_last_call < MIN_API_CALL_INTERVAL: time.sleep(MIN_API_CALL_INTERVAL - time_since_last_call) # Make the API call result = detect_objects(image, confidence_threshold, iou_threshold) last_api_call = time.time() return result # Function to detect objects in image def detect_objects(image, confidence_threshold=0.5, iou_threshold=0.45): """Detect objects using Faster R-CNN with Non-Maximum Suppression to remove duplicates.""" try: # Load the Faster R-CNN model detector = load_faster_rcnn_model() if detector is None: error_message("Failed to load Faster R-CNN model") return [], np.array(image) # Convert PIL image to numpy array if needed if hasattr(image, 'convert'): image_np = np.array(image.convert('RGB')) else: image_np = np.array(image) # Convert to tensor image_tensor = tf.convert_to_tensor(image_np) image_tensor = image_tensor[tf.newaxis, ...] # Run object detection results = detector(image_tensor) # Extract results boxes = results['detection_boxes'][0].numpy() classes = results['detection_classes'][0].numpy().astype(int) scores = results['detection_scores'][0].numpy() # Filter by confidence threshold first valid_indices = scores >= confidence_threshold filtered_boxes = boxes[valid_indices] filtered_classes = classes[valid_indices] filtered_scores = scores[valid_indices] if len(filtered_boxes) == 0: return [], image_np # Apply Non-Maximum Suppression to remove duplicate detections final_detections = apply_nms(filtered_boxes, filtered_classes, filtered_scores, image_np.shape, iou_threshold) # Draw bounding boxes on result image result_image = draw_detections(image_np, final_detections) print(f"โ Faster R-CNN detected {len(final_detections)} unique objects (after NMS)") return final_detections, result_image except Exception as e: error_message(f"Faster R-CNN detection error: {str(e)}") # Return empty result on error dummy_image = np.array(image) if hasattr(image, 'convert') else image return [], dummy_image # Function to enhance image quality def enhance_image(image, enhance_type="auto"): """Enhance the image to improve object detection.""" try: # Convert PIL image to numpy array img_array = np.array(image) if enhance_type == "auto" or enhance_type == "brightness": # Auto-adjust brightness gray = cv2.cvtColor(img_array, cv2.COLOR_RGB2GRAY) mean_brightness = np.mean(gray) if mean_brightness < 100: # Image is too dark # Increase brightness hsv = cv2.cvtColor(img_array, cv2.COLOR_RGB2HSV) h, s, v = cv2.split(hsv) # Calculate how much to increase brightness (more for darker images) brightness_factor = max(1.0, (130 - mean_brightness) / 80) v = cv2.add(v, np.array([brightness_factor * 30.0], dtype=np.uint8)) final_hsv = cv2.merge((h, s, v)) img_array = cv2.cvtColor(final_hsv, cv2.COLOR_HSV2RGB) elif mean_brightness > 200: # Image is too bright # Decrease brightness hsv = cv2.cvtColor(img_array, cv2.COLOR_RGB2HSV) h, s, v = cv2.split(hsv) # Reduce brightness v = cv2.subtract(v, np.array([30], dtype=np.uint8)) final_hsv = cv2.merge((h, s, v)) img_array = cv2.cvtColor(final_hsv, cv2.COLOR_HSV2RGB) if enhance_type == "auto" or enhance_type == "contrast": # Enhance contrast lab = cv2.cvtColor(img_array, cv2.COLOR_RGB2LAB) l, a, b = cv2.split(lab) # Apply CLAHE (Contrast Limited Adaptive Histogram Equalization) clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8)) cl = clahe.apply(l) # Merge the CLAHE enhanced L-channel with the a and b channels enhanced_lab = cv2.merge((cl, a, b)) img_array = cv2.cvtColor(enhanced_lab, cv2.COLOR_LAB2RGB) # Convert back to PIL image enhanced_image = Image.fromarray(img_array) return enhanced_image except Exception as e: error_message(f"Image enhancement error: {e}") return image # Return original image on error # Function to detect text in image (OCR) def detect_text_in_image(image): """Detect text in image using OCR.""" try: if not has_tesseract: return "OCR functionality requires installing pytesseract." # Convert PIL image to numpy array img_array = np.array(image) # Convert to grayscale gray = cv2.cvtColor(img_array, cv2.COLOR_RGB2GRAY) # Apply threshold to get image with only black and white _, binary = cv2.threshold(gray, 150, 255, cv2.THRESH_BINARY_INV) # Apply dilation and erosion to remove noise kernel = np.ones((1, 1), np.uint8) processed = cv2.dilate(binary, kernel, iterations=1) processed = cv2.erode(processed, kernel, iterations=1) # Invert back processed = cv2.bitwise_not(processed) # Detect text detected_text = pytesseract.image_to_string(processed) # Clean and process the text detected_text = detected_text.strip() return detected_text except Exception as e: return f"Text detection error: {e}" # Function to get example sentence def get_example_sentence(word, target_language): """Generate an example sentence using the word via the example generator.""" # Try to determine category from OBJECT_CATEGORIES category = None for cat_name, items in OBJECT_CATEGORIES.items(): if word.lower() in [item.lower() for item in items]: category = cat_name break # Call the generator with the category hint return example_generator.get_example_sentence(word, target_language, category) # Function to get pronunciation guide def get_pronunciation_guide(word, language_code): """Generate a simple pronunciation guide for the word.""" try: # Map of common sounds in different languages pronunciation_maps = { "es": { # Spanish 'j': 'h', 'll': 'y', 'รฑ': 'ny', 'rr': 'rolled r' }, "fr": { # French 'eau': 'oh', 'au': 'oh', 'ai': 'eh', 'ou': 'oo', 'u': 'รผ', 'r': 'guttural r' }, "de": { # German 'sch': 'sh', 'ch': 'kh/sh', 'ei': 'eye', 'ie': 'ee', 'รค': 'eh', 'รถ': 'er', 'รผ': 'รผ' }, "it": { # Italian 'gn': 'ny', 'gli': 'ly', 'ch': 'k', 'c+e/i': 'ch', 'c+a/o/u': 'k' } } # Get pronunciation map for this language sound_map = pronunciation_maps.get(language_code, {}) # Build pronunciation guide notes = [] for sound, pronunciation in sound_map.items(): if sound in word.lower(): notes.append(f"'{sound}' sounds like '{pronunciation}'") return notes except Exception as e: return [f"Pronunciation guide unavailable: {str(e)}"] # Function to create a database session def create_session_direct(): """Create a session directly using SQLite.""" try: # Connect to the database conn = sqlite3.connect("language_learning.db") cursor = conn.cursor() # Insert a new session with the current time current_time = datetime.datetime.now() cursor.execute( "INSERT INTO sessions (start_time, words_studied, words_learned) VALUES (?, 0, 0)", (current_time,) ) conn.commit() # Get the last inserted ID session_id = cursor.lastrowid conn.close() return session_id except Exception as e: error_message(f"Direct session creation error: {str(e)}") return None # Function to add vocabulary to the database def add_vocabulary_direct(word_original, word_translated, language_translated, category=None, image_path=None): """Add vocabulary directly using SQLite with improved error handling for duplicates and locks.""" try: # Original function code here... # Connect to the database with timeout to handle locks conn = sqlite3.connect("language_learning.db", timeout=10.0) cursor = conn.cursor() # Check if this word already exists in this language cursor.execute( "SELECT id FROM vocabulary WHERE word_original = ? AND language_translated = ?", (word_original, language_translated) ) existing_word = cursor.fetchone() # If word exists, update it rather than inserting a new one if existing_word: vocab_id = existing_word[0] # Update the existing word with new translation and image if provided cursor.execute( "UPDATE vocabulary SET word_translated = ?, category = ?, image_path = ? WHERE id = ?", (word_translated, category, image_path, vocab_id) ) # Let the user know we're updating info_message(f"Word '{word_original}' already exists in {language_translated}. Updating with new information.") else: # Current time for timestamps current_time = datetime.datetime.now() # Insert a new word try: # Try with source column cursor.execute(''' INSERT INTO vocabulary (word_original, word_translated, language_translated, category, image_path, date_added, source) VALUES (?, ?, ?, ?, ?, ?, 'manual') ''', (word_original, word_translated, language_translated, category, image_path, current_time)) except sqlite3.OperationalError as e: if 'no column named source' in str(e): # Try without source column cursor.execute(''' INSERT INTO vocabulary (word_original, word_translated, language_translated, category, image_path, date_added) VALUES (?, ?, ?, ?, ?, ?) ''', (word_original, word_translated, language_translated, category, image_path, current_time)) else: raise e # Get the last inserted ID vocab_id = cursor.lastrowid # Check if we need to add user progress cursor.execute("SELECT id FROM user_progress WHERE vocabulary_id = ?", (vocab_id,)) if not cursor.fetchone(): # Initialize user progress for this vocabulary cursor.execute(''' INSERT INTO user_progress (vocabulary_id, last_reviewed, proficiency_level) VALUES (?, ?, 0) ''', (vocab_id, current_time)) # Commit changes and close conn.commit() conn.close() # NEW CODE: Integration with gamification system - with error handling if vocab_id: try: # Check for gamification achievements gamification.check_achievements( "word_learned", word=word_original, category=category, language=language_translated ) # Check for daily challenges gamification.check_challenge_progress( word_original=word_original, word_translated=word_translated, language=language_translated ) except Exception as e: print(f"Gamification error in add_vocabulary_direct: {e}") return vocab_id except sqlite3.OperationalError as e: # Handle database locks with specific advice if 'database is locked' in str(e): error_message("Database is currently locked. Please wait a moment and try again.") # Add a small delay to allow the database to unlock time.sleep(1.5) else: error_message(f"Database error: {str(e)}") return None except Exception as e: error_message(f"Direct vocabulary save error: {str(e)}") return None # Function to get all vocabulary items from the database def get_all_vocabulary_direct(): """Get all vocabulary items directly from SQLite.""" try: # Connect to the database conn = sqlite3.connect("language_learning.db") # Use dictionary cursor for easier access conn.row_factory = sqlite3.Row cursor = conn.cursor() # Get all vocabulary with user progress info cursor.execute(''' SELECT v.id, v.word_original, v.word_translated, v.language_translated, v.category, v.image_path, v.date_added, up.proficiency_level, up.review_count, up.correct_count, up.last_reviewed FROM vocabulary v LEFT JOIN user_progress up ON v.id = up.vocabulary_id ORDER BY v.date_added DESC ''') # Fetch all results results = cursor.fetchall() # Convert to list of dictionaries vocabulary = [] for row in results: # Convert row to dictionary word = dict(row) vocabulary.append(word) conn.close() return vocabulary except Exception as e: error_message(f"Error retrieving vocabulary: {str(e)}") return [] # Function to get session statistics def get_session_stats_direct(days=30): """Get session statistics directly from SQLite.""" try: # Connect to the database conn = sqlite3.connect("language_learning.db") cursor = conn.cursor() # Calculate date for filtering current_time = datetime.datetime.now() start_date = current_time - datetime.timedelta(days=days) # Convert to string format start_date_str = start_date.strftime("%Y-%m-%d") # Get total sessions cursor.execute( "SELECT COUNT(*) FROM sessions WHERE start_time >= ?", (start_date_str,) ) total_sessions = cursor.fetchone()[0] # Get words studied and learned cursor.execute( "SELECT SUM(words_studied), SUM(words_learned) FROM sessions WHERE start_time >= ?", (start_date_str,) ) result = cursor.fetchone() total_words_studied = result[0] if result[0] else 0 total_words_learned = result[1] if result[1] else 0 # Calculate averages avg_words_per_session = total_words_studied / total_sessions if total_sessions > 0 else 0 # Get session durations cursor.execute( """ SELECT start_time, end_time FROM sessions WHERE start_time >= ? AND end_time IS NOT NULL """, (start_date_str,) ) # Calculate average session length total_minutes = 0 session_count = 0 for start_time_str, end_time_str in cursor.fetchall(): try: # Parse the datetime strings start_time = datetime.datetime.fromisoformat(start_time_str.replace(' ', 'T')) end_time = datetime.datetime.fromisoformat(end_time_str.replace(' ', 'T')) # Calculate duration in minutes duration = (end_time - start_time).total_seconds() / 60 total_minutes += duration session_count += 1 except: pass avg_session_minutes = total_minutes / session_count if session_count > 0 else 0 conn.close() # Return stats dictionary return { 'total_sessions': total_sessions, 'total_words_studied': total_words_studied, 'total_words_learned': total_words_learned, 'avg_words_per_session': avg_words_per_session, 'avg_session_minutes': avg_session_minutes } except Exception as e: error_message(f"Error retrieving session stats: {str(e)}") return {} # Function to check if database is properly set up def check_database_setup(): """Check if the database is properly set up and try to fix if needed.""" try: conn = sqlite3.connect("language_learning.db") cursor = conn.cursor() # Check if tables exist cursor.execute("SELECT name FROM sqlite_master WHERE type='table';") tables = [table[0] for table in cursor.fetchall()] required_tables = ['vocabulary', 'user_progress', 'sessions', 'camera_translations'] missing_tables = [table for table in required_tables if table not in tables] if missing_tables: # Create missing tables if 'vocabulary' in missing_tables: cursor.execute(''' CREATE TABLE IF NOT EXISTS vocabulary ( id INTEGER PRIMARY KEY, word_original TEXT NOT NULL, word_translated TEXT NOT NULL, language_translated TEXT NOT NULL, category TEXT, image_path TEXT, date_added TIMESTAMP DEFAULT CURRENT_TIMESTAMP, source TEXT DEFAULT 'manual' ); ''') if 'user_progress' in missing_tables: cursor.execute(''' CREATE TABLE IF NOT EXISTS user_progress ( id INTEGER PRIMARY KEY, vocabulary_id INTEGER, review_count INTEGER DEFAULT 0, correct_count INTEGER DEFAULT 0, last_reviewed TIMESTAMP, proficiency_level INTEGER DEFAULT 0, FOREIGN KEY (vocabulary_id) REFERENCES vocabulary (id) ); ''') if 'sessions' in missing_tables: cursor.execute(''' CREATE TABLE IF NOT EXISTS sessions ( id INTEGER PRIMARY KEY, start_time TIMESTAMP, end_time TIMESTAMP, words_studied INTEGER DEFAULT 0, words_learned INTEGER DEFAULT 0 ); ''') if 'camera_translations' in missing_tables: cursor.execute(''' CREATE TABLE IF NOT EXISTS camera_translations ( id INTEGER PRIMARY KEY, image_path TEXT, detected_text TEXT, translated_text TEXT, source_language TEXT, target_language TEXT, date_captured TIMESTAMP DEFAULT CURRENT_TIMESTAMP, is_saved_to_vocabulary BOOLEAN DEFAULT 0 ); ''') conn.commit() conn.close() return True except Exception as e: error_message(f"Database error: {e}") return False def prepare_vocabulary_for_diverse_questions(vocabulary, languages): """Enhance vocabulary data to support diverse question types.""" total_words = len(vocabulary) words_with_categories = 0 words_with_images = 0 words_with_examples = 0 # Count and prepare vocabulary for diverse questions for word in vocabulary: # Check/count category if word.get('category') and word['category'] not in ['other', 'manual', '']: words_with_categories += 1 # Check/count image if word.get('image_path') and os.path.exists(word.get('image_path', '')): words_with_images += 1 # Test for example sentence try: example = get_example_sentence(word.get('word_original', ''), word.get('language_translated', 'en')) if example and example.get('translated'): words_with_examples += 1 except: pass if st.session_state.debug_quiz: st.sidebar.markdown("### Vocabulary Stats") st.sidebar.markdown(f"Total words: {total_words}") st.sidebar.markdown(f"With categories: {words_with_categories}") st.sidebar.markdown(f"With images: {words_with_images}") st.sidebar.markdown(f"With examples: {words_with_examples}") return vocabulary if 'db_checked' not in st.session_state: st.session_state.db_checked = check_database_setup() def debug_button(label, **kwargs): """Debug wrapper that shows what parameters are being passed to a button and ensures uniqueness""" import inspect import time # Get the caller info caller = inspect.getframeinfo(inspect.currentframe().f_back) # Create a unique key based on the calling file, line number, and timestamp if 'key' not in kwargs: caller_id = f"{caller.filename.split('/')[-1]}_{caller.lineno}" timestamp = int(time.time() * 1000) % 10000 # Use last 4 digits of timestamp for readability unique_key = f"{label.replace(' ', '_')}_{caller_id}_{timestamp}" kwargs['key'] = unique_key # For debugging, uncomment this line to see what keys are being generated # print(f"Button: {label}, Key: {kwargs['key']}") # Remove any problematic parameters if present if 'use_column_width' in kwargs: del kwargs['use_column_width'] if 'type' in kwargs and kwargs['type'] == 'primary': del kwargs['type'] # Use the cleaned kwargs return st.button(label, **kwargs) def safe_button(label, **kwargs): """Safe wrapper for st.button that ensures uniqueness and removes problematic parameters""" import time # Generate a unique key if none provided if 'key' not in kwargs: # Create unique key based on label and timestamp timestamp = int(time.time() * 1000) % 10000 # Use last 4 digits of timestamp for readability unique_key = f"{label.replace(' ', '_')}_{timestamp}" kwargs['key'] = unique_key # Remove problematic parameters if present if 'use_column_width' in kwargs: del kwargs['use_column_width'] if 'type' in kwargs and kwargs['type'] == 'primary': del kwargs['type'] # Use the cleaned kwargs return st.button(label, **kwargs) # Initialize database @st.cache_resource def get_database(): return LanguageLearningDB("language_learning.db") db = get_database() # Initialize processing queue in session state for background tasks if 'processing_queue' not in st.session_state: st.session_state.processing_queue = queue.Queue() if 'processing_results' not in st.session_state: st.session_state.processing_results = {} if 'processing_complete' not in st.session_state: st.session_state.processing_complete = False # Initialize session state for manual mode if 'manual_mode' not in st.session_state: st.session_state.manual_mode = False if 'manual_label' not in st.session_state: st.session_state.manual_label = "" # Initialize session state variables if 'target_language' not in st.session_state: st.session_state.target_language = "es" # Default to Spanish if 'session_id' not in st.session_state: st.session_state.session_id = None if 'words_studied' not in st.session_state: st.session_state.words_studied = 0 if 'words_learned' not in st.session_state: st.session_state.words_learned = 0 if 'quiz_score' not in st.session_state: st.session_state.quiz_score = 0 if 'quiz_total' not in st.session_state: st.session_state.quiz_total = 0 if 'current_quiz_word' not in st.session_state: st.session_state.current_quiz_word = None if 'quiz_options' not in st.session_state: st.session_state.quiz_options = [] if 'answered' not in st.session_state: st.session_state.answered = False if 'detection_checkboxes' not in st.session_state: st.session_state.detection_checkboxes = {} # Ensure session state variables are initialized first if 'level' not in st.session_state: st.session_state.level = 1 if 'points' not in st.session_state: st.session_state.points = 0 if 'streak_days' not in st.session_state: st.session_state.streak_days = 0 if 'daily_challenges' not in st.session_state: st.session_state.daily_challenges = [] if 'word_of_the_day' not in st.session_state: st.session_state.word_of_the_day = None # For debugging question type selection if 'debug_quiz' not in st.session_state: st.session_state.debug_quiz = False # Add these initializations with your other session state initializations # Add these initializations with your other session state initializations if 'audio_data' not in st.session_state: st.session_state.audio_data = None if 'audio_data_received' not in st.session_state: st.session_state.audio_data_received = False if 'current_recording_word' not in st.session_state: st.session_state.current_recording_word = None if 'use_vision_api' not in st.session_state: st.session_state.use_vision_api = True # Always force it to be True st.session_state.use_vision_api = True if 'app_mode' not in st.session_state: st.session_state.app_mode = "Camera Mode" # Add flag to track save button state if 'save_button_clicked' not in st.session_state: st.session_state.save_button_clicked = False if 'words_just_saved' not in st.session_state: st.session_state.words_just_saved = False if 'saved_count' not in st.session_state: st.session_state.saved_count = 0 if 'saved_items' not in st.session_state: st.session_state.saved_items = [] if 'faster_rcnn_model_loaded' not in st.session_state: st.session_state.faster_rcnn_model_loaded = False def get_gamification(): # Initialize GamificationSystem without the translate function for now return GamificationSystem() # Initialize gamification gamification = get_gamification() # Make sure state is explicitly initialized gamification.initialize_state() # Function to translate text class FreeTranslationService: def __init__(self): self.translation_cache = {} self.last_request_time = 0 self.rate_limit_delay = 1.0 # seconds between requests def translate_text(self, text, target_language, source_language='en'): """ Translate text using multiple free services with fallbacks """ # Check cache first cache_key = f"{text}_{source_language}_{target_language}" if cache_key in self.translation_cache: return self.translation_cache[cache_key] # Rate limiting current_time = time.time() if current_time - self.last_request_time < self.rate_limit_delay: time.sleep(self.rate_limit_delay - (current_time - self.last_request_time)) translation = None # Method 1: Deep Translator (Free Google Translate web interface) try: translator = GoogleTranslator(source=source_language, target=target_language) translation = translator.translate(text) if translation and translation != text: self.translation_cache[cache_key] = translation self.last_request_time = time.time() return translation except Exception as e: print(f"Deep Translator failed: {e}") # Method 2: MyMemory Translation API (Free tier: 10,000 chars/day) try: translation = self._translate_with_mymemory(text, source_language, target_language) if translation: self.translation_cache[cache_key] = translation self.last_request_time = time.time() return translation except Exception as e: print(f"MyMemory failed: {e}") # Method 3: LibreTranslate (if you have a server) try: translation = self._translate_with_libretranslate(text, source_language, target_language) if translation: self.translation_cache[cache_key] = translation self.last_request_time = time.time() return translation except Exception as e: print(f"LibreTranslate failed: {e}") # Method 4: Hugging Face models (for specific language pairs) try: translation = self._translate_with_huggingface(text, source_language, target_language) if translation: self.translation_cache[cache_key] = translation self.last_request_time = time.time() return translation except Exception as e: print(f"Hugging Face translation failed: {e}") # Fallback: Return formatted message return f"[Translation to {target_language} unavailable]" def _translate_with_mymemory(self, text, source_lang, target_lang): """MyMemory Translation API - Free tier""" url = "https://api.mymemory.translated.net/get" params = { 'q': text, 'langpair': f"{source_lang}|{target_lang}" } response = requests.get(url, params=params, timeout=10) if response.status_code == 200: data = response.json() if data.get('responseStatus') == 200: return data['responseData']['translatedText'] return None def _translate_with_libretranslate(self, text, source_lang, target_lang): """LibreTranslate - Free self-hosted option""" # You can use the free public instance (limited) or host your own url = "https://libretranslate.de/translate" # Public instance data = { 'q': text, 'source': source_lang, 'target': target_lang, 'format': 'text' } response = requests.post(url, data=data, timeout=10) if response.status_code == 200: result = response.json() return result.get('translatedText') return None def _translate_with_huggingface(self, text, source_lang, target_lang): """Hugging Face translation models - Completely free""" try: # Map language codes to model names (add more as needed) model_map = { ('en', 'es'): 'Helsinki-NLP/opus-mt-en-es', ('en', 'fr'): 'Helsinki-NLP/opus-mt-en-fr', ('en', 'de'): 'Helsinki-NLP/opus-mt-en-de', ('en', 'it'): 'Helsinki-NLP/opus-mt-en-it', ('en', 'pt'): 'Helsinki-NLP/opus-mt-en-pt', ('en', 'ru'): 'Helsinki-NLP/opus-mt-en-ru', # Add reverse translations ('es', 'en'): 'Helsinki-NLP/opus-mt-es-en', ('fr', 'en'): 'Helsinki-NLP/opus-mt-fr-en', ('de', 'en'): 'Helsinki-NLP/opus-mt-de-en', } model_name = model_map.get((source_lang, target_lang)) if not model_name: return None # Load model and tokenizer translator = pipeline( "translation", model=model_name, return_all_scores=False, max_length=512 ) result = translator(text) return result[0]['translation_text'] except Exception as e: print(f"Hugging Face model error: {e}") return None # Initialize the service free_translator = FreeTranslationService() # Replace your translate_text function with this: def translate_text(text, target_language, source_language='en'): return free_translator.translate_text(text, target_language, source_language) # Connect the translation function to gamification system after both are initialized gamification.set_translate_func(translate_text) @st.cache_resource def get_example_generator(): """Initialize and cache the example sentence generator.""" return ExampleSentenceGenerator(translate_func=translate_text, debug=True) example_generator = get_example_generator() # Function for text-to-speech def text_to_speech(text, lang): try: tts = gTTS(text=text, lang=lang, slow=False) mp3_fp = BytesIO() tts.write_to_fp(mp3_fp) mp3_fp.seek(0) audio_bytes = mp3_fp.read() return audio_bytes except Exception as e: error_message(f"Text-to-speech error: {e}") return None # Function to generate HTML for audio playback def get_audio_html(audio_bytes): """Generate HTML for audio playback without autoplay.""" audio_base64 = base64.b64encode(audio_bytes).decode() # Remove the autoplay attribute - only keep controls audio_tag = f'' return audio_tag # Function to load RCNN Model @st.cache_resource def load_faster_rcnn_model(): """Load and cache the Faster R-CNN model from TensorFlow Hub.""" try: print("Loading Faster R-CNN model...") model_url = "https://tfhub.dev/tensorflow/faster_rcnn/resnet50_v1_640x640/1" detector = hub.load(model_url) print("โ Faster R-CNN model loaded successfully!") return detector except Exception as e: print(f"โ Error loading Faster R-CNN model: {e}") return None # Background worker function for object detection def detect_objects_worker(image, confidence_threshold, iou_threshold, task_id): """Worker function to run detection in background.""" try: # Run detection detections, rendered_image = detect_objects(image, confidence_threshold, iou_threshold) # Store results st.session_state.processing_results[task_id] = { 'detections': detections, 'result_image': rendered_image } # Mark task as complete st.session_state.processing_complete = True except Exception as e: # Store error st.session_state.processing_results[task_id] = { 'error': str(e) } st.session_state.processing_complete = True # Function to start or end a learning session def manage_session(action): """Start or end learning session with improved error handling.""" if action == "start": try: # Try to use the direct method instead of the database object session_id = create_session_direct() if session_id: st.session_state.session_id = session_id st.session_state.words_studied = 0 st.session_state.words_learned = 0 success_message(f"Started new learning session!") return True else: error_message("Failed to create a session directly. Check database permissions.") return False except Exception as e: error_message(f"Error starting session: {str(e)}") return False elif action == "end" and st.session_state.session_id: try: # Connect directly to the database conn = sqlite3.connect("language_learning.db") cursor = conn.cursor() # Update the session with end time and stats current_time = datetime.datetime.now() cursor.execute( "UPDATE sessions SET end_time = ?, words_studied = ?, words_learned = ? WHERE id = ?", (current_time, st.session_state.words_studied, st.session_state.words_learned, st.session_state.session_id) ) conn.commit() conn.close() success_message(f"Session completed! Words studied: {st.session_state.words_studied}, Words learned: {st.session_state.words_learned}") # Clear session state st.session_state.session_id = None st.session_state.words_studied = 0 st.session_state.words_learned = 0 return True except Exception as e: error_message(f"Error ending session: {str(e)}") return False return False # Function to save image def save_image(image, label): try: # Convert PIL Image to OpenCV format img_array = np.array(image) img_cv = cv2.cvtColor(img_array, cv2.COLOR_RGB2BGR) # Create directory if it doesn't exist os.makedirs("object_images", exist_ok=True) # Save image filename = f"object_images/{label}_{int(time.time())}.jpg" cv2.imwrite(filename, img_cv) return filename except Exception as e: error_message(f"Error saving image: {e}") return None # Function to start a new quiz def start_new_quiz(vocabulary, num_questions=5): # Reset quiz state st.session_state.quiz_score = 0 st.session_state.quiz_total = 0 st.session_state.answered = False if not vocabulary or len(vocabulary) < 4: warning_message("Not enough vocabulary words for a quiz (need at least 4).") return False # Start a new session if needed if not st.session_state.session_id: st.session_state.session_id = create_session_direct() st.session_state.words_studied = 0 st.session_state.words_learned = 0 # Set up first question setup_new_question(vocabulary) return True # Function to set up a new quiz question def setup_new_question(vocabulary): if not vocabulary: return False # Select a random word as the question st.session_state.current_quiz_word = np.random.choice(vocabulary) # Create options (3 wrong + 1 correct) options = [st.session_state.current_quiz_word] while len(options) < 4: wrong_option = np.random.choice(vocabulary) if wrong_option['id'] != st.session_state.current_quiz_word['id'] and not any(o['id'] == wrong_option['id'] for o in options): options.append(wrong_option) # Shuffle options np.random.shuffle(options) st.session_state.quiz_options = options st.session_state.answered = False return True # Function to update word progress in the database def update_word_progress_direct(vocab_id, is_correct): """Update word progress directly using SQLite.""" try: # Connect to the database conn = sqlite3.connect("language_learning.db") cursor = conn.cursor() # Current time for timestamp current_time = datetime.datetime.now() # Get current progress cursor.execute( """ SELECT review_count, correct_count, proficiency_level FROM user_progress WHERE vocabulary_id = ? """, (vocab_id,) ) result = cursor.fetchone() if result: review_count, correct_count, proficiency_level = result # Increment counts review_count = review_count + 1 if review_count else 1 correct_count = correct_count + 1 if correct_count and is_correct else (1 if is_correct else 0) # Calculate proficiency (0-5 scale) if review_count > 0: accuracy = correct_count / review_count if accuracy >= 0.9 and review_count >= 5: proficiency_level = 5 elif accuracy >= 0.8 and review_count >= 4: proficiency_level = 4 elif accuracy >= 0.6 and review_count >= 3: proficiency_level = 3 elif accuracy >= 0.4 and review_count >= 2: proficiency_level = 2 elif accuracy >= 0.2: proficiency_level = 1 else: proficiency_level = 0 # Update progress cursor.execute( """ UPDATE user_progress SET review_count = ?, correct_count = ?, proficiency_level = ?, last_reviewed = ? WHERE vocabulary_id = ? """, (review_count, correct_count, proficiency_level, current_time, vocab_id) ) else: # Create new progress entry cursor.execute( """ INSERT INTO user_progress (vocabulary_id, review_count, correct_count, proficiency_level, last_reviewed) VALUES (?, ?, ?, ?, ?) """, (vocab_id, 1, 1 if is_correct else 0, 1 if is_correct else 0, current_time) ) conn.commit() conn.close() return True except Exception as e: error_message(f"Error updating word progress: {str(e)}") return False # Function to check quiz answer def check_answer(selected_index): """Check if selected quiz answer is correct and update progress.""" if st.session_state.answered: return selected_word = st.session_state.quiz_options[selected_index] is_correct = selected_word['id'] == st.session_state.current_quiz_word['id'] # Update database using direct method instead of db class update_word_progress_direct(st.session_state.current_quiz_word['id'], is_correct) # Update session stats st.session_state.words_studied += 1 if is_correct: st.session_state.words_learned += 1 st.session_state.quiz_score += 1 st.session_state.quiz_total += 1 st.session_state.answered = True # Check if any challenges are completed - with error handling try: gamification.check_challenge_progress( quiz_score=st.session_state.quiz_score, quiz_total=st.session_state.quiz_total ) # Check for quiz-related achievements if st.session_state.quiz_total >= 5: # Only check if quiz is substantial gamification.check_achievements( "quiz_completed", score=st.session_state.quiz_score, total=st.session_state.quiz_total ) except Exception as e: print(f"Gamification error in check_answer: {e}") return is_correct # Global counter for truly unique widget IDs if 'widget_counter' not in st.session_state: st.session_state.widget_counter = 0 def truly_safe_button(label, **kwargs): """Button helper that guarantees unique keys even with rapid clicks""" # Increment the global counter st.session_state.widget_counter += 1 # Generate a unique key if none provided if 'key' not in kwargs: # Create unique key based on counter + millisecond timestamp import time timestamp = int(time.time() * 1000000) % 1000000 # Microsecond part only counter = st.session_state.widget_counter unique_key = f"{label.replace(' ', '_').lower()}_{counter}_{timestamp}" kwargs['key'] = unique_key # Remove problematic parameters if present if 'use_column_width' in kwargs: del kwargs['use_column_width'] if 'type' in kwargs and kwargs['type'] == 'primary': kwargs['type'] = None # Set to None instead of deleting # Use the cleaned kwargs return st.button(label, **kwargs) def safe_button(label, **kwargs): """Alias for truly_safe_button for backward compatibility""" return truly_safe_button(label, **kwargs) # Main sidebar for navigation st.sidebar.title("๐ Vocam") app_mode_options = ["Camera Mode", "My Vocabulary", "Quiz Mode", "Statistics", "My Progress", "Pronunciation Practice"] if 'app_mode' in st.session_state: # Use the session state value as the default index for the selectbox default_index = app_mode_options.index(st.session_state.app_mode) if st.session_state.app_mode in app_mode_options else 0 else: default_index = 0 app_mode = st.sidebar.selectbox( "Choose a mode", app_mode_options, index=default_index ) # Update session state with the current selection (might be from selectbox or previous setting) st.session_state.app_mode = app_mode # Add gamification info to the sidebar try: gamification.update_sidebar() except Exception as e: st.sidebar.markdown('
System initializing...
' '