import os import json import numpy as np import subprocess import faiss import cv2 import re import gradio as gr from sentence_transformers import SentenceTransformer from openai import OpenAI import logging from PIL import Image import base64 import io deepseek_api_key = os.environ.get("DEEPSEEK_API_KEY", "YOUR_API_KEY") client = OpenAI( base_url="https://openrouter.ai/api/v1", api_key=deepseek_api_key, ) DATASET_PATH = "data" JSON_PATH = f"{DATASET_PATH}/sign_language_data.json" if os.path.exists(JSON_PATH): with open(JSON_PATH, "r") as f: dataset = json.load(f) for item in dataset: category = item["category"].lower().replace(" ", "_") video_filename = os.path.basename(item["video_clip_path"]) item["video_clip_path"] = f"{DATASET_PATH}/clips/{category}/{video_filename}" frame_filename = os.path.basename(item["frame_path"]) item["frame_path"] = f"{DATASET_PATH}/all_signs/{frame_filename}" else: dataset = [] print(f"Warning: {JSON_PATH} does not exist. Using empty dataset.") logging.getLogger("sentence_transformers").setLevel(logging.ERROR) print("Loading sentence transformer model...") embed_model = SentenceTransformer("all-MiniLM-L6-v2") dimension = 384 index = faiss.IndexFlatL2(dimension) text_to_video = {} idx_to_text = [] for item in dataset: phrases = [item["text"]] + item.get("semantic_meaning", []) for phrase in phrases: embedding = embed_model.encode(phrase).astype(np.float32) index.add(np.array([embedding])) text_to_video[phrase] = item["video_clip_path"] idx_to_text.append(phrase) print(f"Indexed {len(idx_to_text)} phrases") def list_available_phrases(): print("Available phrases in dataset:") for idx, phrase in enumerate(text_to_video.keys()): print(f"{idx+1}. '{phrase}'") print(f"Total: {len(text_to_video)} phrases") def preprocess_text(text): emoji_pattern = re.compile("[" u"\U0001F600-\U0001F64F" u"\U0001F300-\U0001F5FF" u"\U0001F680-\U0001F6FF" u"\U0001F700-\U0001F77F" u"\U0001F780-\U0001F7FF" u"\U0001F800-\U0001F8FF" u"\U0001F900-\U0001F9FF" u"\U0001FA00-\U0001FA6F" u"\U0001FA70-\U0001FAFF" u"\U00002702-\U000027B0" u"\U000024C2-\U0001F251" "]+", flags=re.UNICODE) text = emoji_pattern.sub(r'', text) text = re.sub(r'[^\w\s\?\/]', '', text) text = re.sub(r'\s+', ' ', text).strip() return text def refine_sentence_with_deepseek(text): text = preprocess_text(text) prompt = f""" Convert the following sentence into a sign-language-friendly version: - Remove unnecessary words like articles (a, an, the). - Keep essential words like pronouns (I, you, we, they). - Maintain question words (what, where, when, why, how). - Ensure verbs and key actions are included. - Reorder words to match sign language grammar. - IMPORTANT: Format your response with "SIGN_LANGUAGE_VERSION: [your simplified phrase]" at the beginning. - Sign language often places topic first, then comment (e.g., "READY YOU?" instead of "YOU READY?"). Sentence: "{text}" """ try: completion = client.chat.completions.create( model="deepseek/deepseek-r1:free", messages=[{"role": "user", "content": prompt}], temperature=0.3 ) full_response = completion.choices[0].message.content.strip() patterns = [ r"SIGN_LANGUAGE_VERSION:\s*(.+?)(?:\n|$)", r"\*\*Signs?\*\*:?\s*(.+?)(?:\n|$)", r"\*\*Sign-language-friendly version:\*\*\s*(.+?)(?:\n|$)", r"(?:^|\n)([A-Z\s\?\!]+)(?:\n|$)" ] for pattern in patterns: match = re.search(pattern, full_response, re.MULTILINE) if match: refined_text = match.group(1).strip() return refined_text first_line = full_response.split('\n')[0].strip() return first_line except Exception as e: print(f"Error with DeepSeek API: {str(e)}") words = text.split() filtered_words = [w for w in words if w.lower() not in ['a', 'an', 'the', 'is', 'are', 'am']] return ' '.join(filtered_words) def retrieve_video(text, debug=False, similarity_threshold=0.9): if not text or text.isspace(): return None text = preprocess_text(text) if debug: print(f"Creating embedding for '{text}'") # Handle special case for "I" if text.lower() == "i": if "I/me" in text_to_video: if debug: print(f" Direct mapping found: '{text}' → 'I/me'") return text_to_video["I/me"] if index.ntotal == 0: if debug: print("No items in the index") return None query_embedding = embed_model.encode(text).astype(np.float32) distances, closest_idx = index.search(np.array([query_embedding]), min(3, index.ntotal)) # Get top matches closest_texts = [idx_to_text[idx] for idx in closest_idx[0]] similarity_scores = distances[0] if debug: print(f"Top matches for '{text}':") for i, (phrase, score) in enumerate(zip(closest_texts, similarity_scores)): print(f" {i+1}. '{phrase}' (score: {score:.4f})") if len(similarity_scores) > 0 and similarity_scores[0] < similarity_threshold: closest_text = closest_texts[0] query_word_count = len(text.split()) match_word_count = len(closest_text.split()) if query_word_count > 1 and match_word_count == 1: if debug: print(f"Rejecting single-word match '{closest_text}' for multi-word query '{text}'") return None if debug: print(f" Found match: '{closest_text}' with score {similarity_scores[0]:.4f}") return text_to_video.get(closest_text, None) else: if debug: print(f"No match found with similarity below threshold {similarity_threshold}") return None def merge_videos(video_list, output_path="temp/output.mp4"): os.makedirs("temp", exist_ok=True) if not video_list: return None if len(video_list) == 1: try: import shutil shutil.copy(video_list[0], output_path) return output_path except Exception as e: print(f"Error copying single video: {e}") return None verified_paths = [] for path in video_list: if os.path.exists(path): verified_paths.append(path) else: print(f"Warning: Video path does not exist: {path}") if not verified_paths: print("No valid video paths found") return None list_path = "temp/video_list.txt" with open(list_path, "w") as f: for path in verified_paths: abs_path = os.path.abspath(path) f.write(f"file '{abs_path}'\n") abs_output = os.path.abspath(output_path) abs_list = os.path.abspath(list_path) command = f"ffmpeg -f concat -safe 0 -i '{abs_list}' -c copy '{abs_output}' -y" print(f"Running command: {command}") process = subprocess.run(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) if process.returncode != 0: print(f"FFmpeg error: {process.stderr.decode()}") return None return output_path def save_video(video_path, output_path="temp/display_output.mp4"): os.makedirs("temp", exist_ok=True) if not video_path or not os.path.exists(video_path): return None if video_path != output_path: os.system(f"cp '{video_path}' '{output_path}'") return output_path def text_to_sign_pipeline(user_input, debug=False): user_input = preprocess_text(user_input) if debug: print(f"Processing input: '{user_input}'") has_multiple_words = len(user_input.split()) > 1 if not has_multiple_words: direct_video = retrieve_video(user_input, debug=debug) if direct_video: if debug: print(f"Single word match found for '{user_input}'") return save_video(direct_video) sign_friendly_sentence = refine_sentence_with_deepseek(user_input) if debug: print(f"DeepSeek refined input to: '{sign_friendly_sentence}'") full_sentence_video = retrieve_video(sign_friendly_sentence, debug=debug) if full_sentence_video: if debug: print(f"Found full sentence match for '{sign_friendly_sentence}'") return save_video(full_sentence_video) words = sign_friendly_sentence.split() video_paths = [] if debug: print(f"No full sentence match. Trying word-by-word approach for: {words}") for word in words: clean_word = preprocess_text(word).replace('?', '') if not clean_word or clean_word.isspace(): continue word_video = retrieve_video(clean_word, debug=debug) if word_video: print(f" Found video for word: '{clean_word}'") video_paths.append(word_video) else: print(f" No video found for word: '{clean_word}'") if not video_paths: print(" No videos found for any words in the sentence") return None if debug: print(f"Found videos for {len(video_paths)} words, merging...") merged_video = merge_videos(video_paths) return save_video(merged_video) def encode_image_to_base64(image_path): with open(image_path, "rb") as image_file: return base64.b64encode(image_file.read()).decode('utf-8') def preprocess_image(image_path): img = cv2.imread(image_path) if img is None: return None height, width = img.shape[:2] right_side = img[:, width//2:width] os.makedirs("temp", exist_ok=True) cropped_path = "temp/cropped_image.jpg" cv2.imwrite(cropped_path, right_side) return cropped_path def detect_text_in_image(image_path, debug=False): base64_image = encode_image_to_base64(image_path) prompt = """ Is there any prominent text label or sign language text in this image? Answer with ONLY "YES" or "NO". """ try: completion = client.chat.completions.create( model="qwen/qwen2.5-vl-3b-instruct:free", messages=[ { "role": "user", "content": [ {"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}} ] } ], temperature=0.3 ) response = completion.choices[0].message.content.strip().upper() if debug: print(f"Text detection response: {response}") return "YES" in response except Exception as e: if debug: print(f"Error in text detection: {str(e)}") return False def image_to_text_with_qwen(image_path, debug=False): base64_image = encode_image_to_base64(image_path) has_text = detect_text_in_image(image_path, debug) if has_text: cropped_image_path = preprocess_image(image_path) if cropped_image_path: cropped_base64 = encode_image_to_base64(cropped_image_path) prompt = """ Extract ONLY the main text label from this image. I'm looking for a single word or short phrase that appears as the main text (like "AFTERNOON"). Ignore any numbers, categories, or other text. Provide ONLY the extracted text without any other explanation or context. """ try: completion = client.chat.completions.create( model="qwen/qwen2.5-vl-3b-instruct:free", messages=[ { "role": "user", "content": [ {"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{cropped_base64}"}} ] } ], temperature=0.3 ) response = completion.choices[0].message.content.strip() if debug: print(f"Qwen VL text extraction response: {response}") cleaned_text = re.sub(r"^(the|main|text|label|is|:|\.|\s)+", "", response, flags=re.IGNORECASE) cleaned_text = re.sub(r'["\'\(\)]', '', cleaned_text) cleaned_text = cleaned_text.strip().upper() if cleaned_text: return cleaned_text, "text" except Exception as e: if debug: print(f"Error using Qwen VL for text extraction: {str(e)}") prompt = """ Describe this image in a SINGLE WORD only. Focus on the main subject (like "MAN", "WOMAN", "HOUSE", "HAPPY", "SAD", etc.). Provide ONLY this single word without any punctuation or explanation. """ try: completion = client.chat.completions.create( model="qwen/qwen2.5-vl-3b-instruct:free", messages=[ { "role": "user", "content": [ {"type": "text", "text": prompt}, {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"}} ] } ], temperature=0.3 ) response = completion.choices[0].message.content.strip() if debug: print(f"Qwen VL caption response: {response}") cleaned_caption = re.sub(r'[^\w\s]', '', response) cleaned_caption = cleaned_caption.strip().split()[0] cleaned_caption = cleaned_caption.upper() return cleaned_caption, "caption" except Exception as e: if debug: print(f"Error using Qwen VL for captioning: {str(e)}") return "ERROR", "error" def process_text(input_text): if not input_text or input_text.isspace(): return "Please enter some text to convert." final_video = text_to_sign_pipeline(input_text, debug=True) if final_video: return final_video else: return "Sorry, no matching sign language video found." def process_image(input_image): os.makedirs("temp", exist_ok=True) image_path = "temp/uploaded_image.jpg" input_image.save(image_path) extracted_text, source_type = image_to_text_with_qwen(image_path, debug=True) if extracted_text == "ERROR": return "Error processing image", None sign_video = text_to_sign_pipeline(extracted_text, debug=True) if source_type == "text": result_text = f"Extracted text: {extracted_text}" else: result_text = f"Generated caption: {extracted_text}" return result_text, sign_video if sign_video else "No matching sign language video found" with gr.Blocks() as app: gr.Markdown("# Sign Language Conversion") with gr.Tabs(): with gr.Tab("Text to Sign"): text_input = gr.Textbox(label="Enter text to convert to sign language") text_button = gr.Button("Convert Text to Sign") text_output = gr.Video(label="Sign Language Output") text_button.click(process_text, inputs=text_input, outputs=text_output) with gr.Tab("Image to Text/Caption and Sign"): image_input = gr.Image(type="pil", label="Upload image") image_button = gr.Button("Process Image and Convert to Sign") extracted_text_output = gr.Textbox(label="Extracted Text/Caption") image_output = gr.Video(label="Sign Language Output") image_button.click( process_image, inputs=image_input, outputs=[extracted_text_output, image_output] ) app.launch()