|
|
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}'") |
|
|
|
|
|
|
|
|
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)) |
|
|
|
|
|
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() |
