""" Tahap 1: Extract Landmark Script ini mengekstrak landmark tangan dan wajah dari video menggunakan MediaPipe dan menyimpannya dalam format .npy Sesuai README: - Input: dataset/raw/ - Output: dataset/landmark/ - Proses: - Membaca seluruh video - Membaca frame menggunakan OpenCV - Deteksi MediaPipe Hands (21 landmark tangan kiri, 21 landmark tangan kanan) - Deteksi MediaPipe Face Mesh (60 landmark) - Membentuk Sliding Window 30 Frame dan 5 Stride - Menyimpan file .npy """ import os import sys import cv2 import numpy as np from pathlib import Path from tqdm import tqdm from typing import List, Tuple, Optional import argparse # Add parent directory to path sys.path.insert(0, str(Path(__file__).parent.parent)) from utils.mediapipe_detector import MediaPipeDetector from utils.config import ( RAW_DIR, LANDMARK_DIR, WINDOW_SIZE, STRIDE, VIDEO_EXTENSIONS, FEATURE_SIZE ) class LandmarkExtractor: """ Extractor untuk mengambil landmark dari video streams. Sesuai README: - Sliding Window 30 Frame - Stride 5 - Output shape: (30, feature_size) """ def __init__( self, window_size: int = WINDOW_SIZE, stride: int = STRIDE, detector: Optional[MediaPipeDetector] = None ): """ Inisialisasi extractor. Args: window_size: Jumlah frame per window (30) stride: Step size untuk sliding window (5) detector: MediaPipe detector instance """ self.window_size = window_size self.stride = stride self.detector = detector or MediaPipeDetector() def extract_from_video(self, video_path: Path) -> Optional[np.ndarray]: """ Extract landmark dari satu video dengan sliding window. Args: video_path: Path ke video file Returns: Array landmark sequence shape (num_windows, 30, 306) atau None jika gagal """ # Open video cap = cv2.VideoCapture(str(video_path)) if not cap.isOpened(): print(f"Error: Cannot open video {video_path}") return None # Collect all frame landmarks all_landmarks = [] frame_count = 0 try: while True: ret, frame = cap.read() if not ret: break # Extract landmarks dari frame left_hand, right_hand, face = self.detector.detect(frame) # Flatten dan concatenate features = np.concatenate([ left_hand.flatten(), # 63 values right_hand.flatten(), # 63 values face.flatten() # 180 values ]) all_landmarks.append(features) frame_count += 1 finally: cap.release() if len(all_landmarks) < self.window_size: print(f"Warning: Video {video_path.name} hanya memiliki {len(all_landmarks)} frames") return None # Create sliding windows windows = [] for i in range(0, len(all_landmarks) - self.window_size + 1, self.stride): window = all_landmarks[i:i + self.window_size] windows.append(np.array(window)) if not windows: # Jika tidak ada window penuh, ambil dari akhir windows.append(np.array(all_landmarks[-self.window_size:])) return np.array(windows) # Shape: (num_windows, 30, 306) def close(self): """Cleanup resources.""" self.detector.close() def get_video_files(directory: Path) -> List[Path]: """ Get semua video files dalam directory dan subdirectories. Args: directory: Root directory Returns: List of video file paths """ video_files = [] for ext in VIDEO_EXTENSIONS: video_files.extend(directory.rglob(f"*{ext}")) return sorted(video_files) def extract_category( raw_dir: Path, landmark_dir: Path, category: str, window_size: int = WINDOW_SIZE, stride: int = STRIDE ) -> Tuple[int, int]: """ Extract landmarks untuk satu kategori. Args: raw_dir: Directory data mentah landmark_dir: Directory output landmark category: Nama kategori window_size: Window size stride: Stride Returns: Tuple (success_count, fail_count) """ category_raw_dir = raw_dir / category if not category_raw_dir.exists(): print(f"Warning: Category directory tidak ditemukan: {category_raw_dir}") return 0, 0 # Get all video files video_files = get_video_files(category_raw_dir) if not video_files: print(f"Warning: Tidak ada video ditemukan untuk kategori {category}") return 0, 0 print(f"\n{'='*60}") print(f"Extracting landmarks untuk kategori: {category}") print(f"Total video files: {len(video_files)}") print(f"Window size: {window_size}, Stride: {stride}") print(f"{'='*60}\n") # Initialize extractor extractor = LandmarkExtractor(window_size=window_size, stride=stride) success_count = 0 fail_count = 0 try: for video_path in tqdm(video_files, desc=f"Extracting {category}"): try: # Get label dari folder structure # Format: raw/{category}/{label}/video.mkv relative_path = video_path.relative_to(category_raw_dir) parts = relative_path.parts if len(parts) >= 2: label = parts[0] # Folder label (huruf/angka/kata) video_name = video_path.stem # Filename tanpa extension # Create output path (sama dengan struktur raw) output_dir = landmark_dir / category / label output_dir.mkdir(parents=True, exist_ok=True) # Extract landmarks windows = extractor.extract_from_video(video_path) if windows is not None and len(windows) > 0: # Save setiap window sebagai file terpisah for i, window in enumerate(windows): output_path = output_dir / f"{video_name}_w{i}.npy" np.save(output_path, window) success_count += 1 else: fail_count += 1 except Exception as e: print(f"Error processing {video_path}: {e}") fail_count += 1 finally: extractor.close() return success_count, fail_count def extract_all_categories( raw_dir: Path = RAW_DIR, landmark_dir: Path = LANDMARK_DIR, categories: List[str] = None, window_size: int = WINDOW_SIZE, stride: int = STRIDE ): """ Extract landmarks untuk semua kategori. Args: raw_dir: Directory data mentah landmark_dir: Directory output landmark categories: List kategori yang akan diproses window_size: Window size (30) stride: Stride (5) """ if categories is None: from utils.config import CATEGORIES categories = CATEGORIES print("="*60) print("TAHAP 1: EXTRACT LANDMARK") print("="*60) print(f"Input: {raw_dir}") print(f"Output: {landmark_dir}") print(f"Categories: {categories}") print(f"Window size: {window_size}, Stride: {stride}") print("="*60) total_success = 0 total_fail = 0 for category in categories: success, fail = extract_category( raw_dir=raw_dir, landmark_dir=landmark_dir, category=category, window_size=window_size, stride=stride ) total_success += success total_fail += fail print("\n" + "="*60) print("EXTRACTION COMPLETE") print("="*60) print(f"Total berhasil: {total_success}") print(f"Total gagal: {total_fail}") if total_success + total_fail > 0: print(f"Success rate: {total_success/(total_success+total_fail)*100:.2f}%") print("="*60) def main(): """Main entry point.""" parser = argparse.ArgumentParser( description="Extract landmarks dari video untuk gesture recognition" ) parser.add_argument( "--input", type=str, default=None, help="Directory data mentah (default: dataset/raw)" ) parser.add_argument( "--output", type=str, default=None, help="Directory output landmark (default: dataset/landmark)" ) parser.add_argument( "--categories", nargs="+", choices=["abjad_bisindo", "abjad_sibi", "angka", "kata"], default=None, help="Kategori yang akan diproses" ) parser.add_argument( "--window-size", type=int, default=WINDOW_SIZE, help=f"Window size (default: {WINDOW_SIZE})" ) parser.add_argument( "--stride", type=int, default=STRIDE, help=f"Stride (default: {STRIDE})" ) args = parser.parse_args() # Setup paths raw_dir = Path(args.input) if args.input else RAW_DIR landmark_dir = Path(args.output) if args.output else LANDMARK_DIR # Extract extract_all_categories( raw_dir=raw_dir, landmark_dir=landmark_dir, categories=args.categories, window_size=args.window_size, stride=args.stride ) if __name__ == "__main__": main()