345 lines
9.3 KiB
Python
345 lines
9.3 KiB
Python
"""
|
|
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()
|