Update app.py

app.py

@@ -4,8 +4,11 @@ import mediapipe as mp
 import tempfile
 from micro_gestures import *
 from composite_gestures import detect_composite_gesture
+from transformers import pipeline
+from rapidfuzz import process, fuzz
+import soundfile as sf
 
-# initialize mediapipe modules
+# --- Mediapipe initialization ---
 mp_hands = mp.solutions.hands
 mp_pose = mp.solutions.pose
 mp_drawing = mp.solutions.drawing_utils
@@ -25,15 +28,37 @@ pose = mp_pose.Pose(
     min_tracking_confidence=0.5
 )
 
-def process_video(video_path, target_width=640):
-    # open video file
+# --- ASR initialization (Persian) ---
+asr = pipeline(
+    task="automatic-speech-recognition",
+    model="vhdm/whisper-large-fa-v1",
+    device=-1  # CPU
+)
+
+# --- Custom vocabulary for high accuracy ---
+custom_vocab = ["نرد", "کامپیوتر", "هوش مصنوعی", "ماشین"]
+
+def replace_fuzzy(text, vocab, threshold=70):
+    """
+    Replace similar words in text using fuzzy matching.
+    Uses partial_ratio to catch small variations.
+    """
+    for term in vocab:
+        # find closest substring in text
+        match, score = process.extractOne(term, [text], scorer=fuzz.partial_ratio)
+        if score >= threshold:
+            text = text.replace(match, term)
+    return text
+
+# --- Video processing function ---
+def process_video_with_asr(video_path, target_width=640):
     cap = cv2.VideoCapture(video_path)
     fourcc = cv2.VideoWriter_fourcc(*'mp4v')
     temp_output = tempfile.NamedTemporaryFile(delete=False, suffix=".mp4")
     out = None
 
-    sequence_buffer = []  # buffer for micro-gestures
-    last_label = "None"  # store last detected gesture label
+    sequence_buffer = []
+    last_label = "None"
 
     while cap.isOpened():
         ret, frame = cap.read()
@@ -41,21 +66,18 @@ def process_video(video_path, target_width=640):
             break
 
         h0, w0 = frame.shape[:2]
-        # resize frame keeping aspect ratio
         scale = target_width / float(w0)
         target_height = int(round(h0 * scale))
         frame_small = cv2.resize(frame, (target_width, target_height), interpolation=cv2.INTER_AREA)
-
-        # convert to rgb
         rgb_small = cv2.cvtColor(frame_small, cv2.COLOR_BGR2RGB)
 
-        # hand detection
+        # Hand detection
         hand_results = hands.process(rgb_small)
         micro_label = ""
         if hand_results.multi_hand_landmarks:
             for hand_landmarks in hand_results.multi_hand_landmarks:
-                mp_drawing.draw_landmarks(frame_small, hand_landmarks, mp_hands.HAND_CONNECTIONS)        
-                landmarks = [(lm.x, lm.y, lm.z) for lm in hand_landmarks.landmark]        
+                mp_drawing.draw_landmarks(frame_small, hand_landmarks, mp_hands.HAND_CONNECTIONS)
+                landmarks = [(lm.x, lm.y, lm.z) for lm in hand_landmarks.landmark]
                 if fist_closed(landmarks):
                     micro_label = "fist_closed"
                 elif palm_open(landmarks):
@@ -64,19 +86,19 @@ def process_video(video_path, target_width=640):
                     micro_label = "index_up"
                 elif thumb_up(landmarks):
                     micro_label = "thumb_up"
-        
+
                 if micro_label:
                     sequence_buffer.append(micro_label)
                     if len(sequence_buffer) > 5:
                         sequence_buffer.pop(0)
 
-        # detect composite gesture from micro-gesture sequence
+        # Detect composite gesture
         if sequence_buffer:
             detected = detect_composite_gesture(sequence_buffer)
             if detected:
                 last_label = detected
 
-        # initialize video writer
+        # Initialize video writer
         if out is None:
             fps = cap.get(cv2.CAP_PROP_FPS)
             if fps <= 0 or fps > 120:
@@ -88,29 +110,39 @@ def process_video(video_path, target_width=640):
                 (frame_small.shape[1], frame_small.shape[0])
             )
 
-        # write processed frame
         out.write(frame_small)
 
     cap.release()
     if out:
         out.release()
 
-    # return video path and last detected label
-    return temp_output.name, last_label
+    # --- Extract audio from video for ASR ---
+    temp_audio = tempfile.NamedTemporaryFile(delete=False, suffix=".wav")
+    import moviepy.editor as mpv
+    clip = mpv.VideoFileClip(video_path)
+    clip.audio.write_audiofile(temp_audio.name, fps=16000, nbytes=2, codec='pcm_s16le')
+
+    # ASR
+    result = asr(temp_audio.name, chunk_length_s=30, stride_length_s=[5,5])
+    text = result["text"]
+    final_text = replace_fuzzy(text, custom_vocab, threshold=80)
+
+    return temp_output.name, last_label, final_text
 
-# gradio interface
+# --- Gradio interface ---
 iface = gr.Interface(
-    fn=process_video,
+    fn=process_video_with_asr,
     inputs=[
         gr.Video(label="Upload or Record Video"),
         gr.Slider(minimum=160, maximum=1280, value=640, step=20, label="Processing Width")
     ],
     outputs=[
         gr.Video(label="Processed Video with Landmarks"),
-        gr.Textbox(label="Detected Gesture", interactive=False)
+        gr.Textbox(label="Detected Gesture", interactive=False),
+        gr.Textbox(label="Detected Words (Persian ASR)", interactive=False)
     ],
-    title="Hand & Body Pose Detection",
-    description="Upload or record a video, see MediaPipe detect hand landmarks. Gesture label is shown below the video."
+    title="Hand & Body Pose Detection + Persian ASR",
+    description="Upload or record a video, see MediaPipe detect hand landmarks and Persian speech recognition with custom vocabulary."
 )
 
 if __name__ == "__main__":