From 9ece3d9aeae0c4c0ddc292d90b7f222db54d0dda Mon Sep 17 00:00:00 2001 From: jouel88 Date: Sat, 2 May 2026 01:16:03 +0700 Subject: [PATCH] Initial commit for Flask ML API --- .gitignore | 20 ++ app.py | 181 ++++++++++++++++++ config.py | 15 ++ requirements.txt | 9 + services/__init__.py | 1 + services/extract_service.py | 195 +++++++++++++++++++ services/train_service.py | 186 ++++++++++++++++++ services/verify_service.py | 366 ++++++++++++++++++++++++++++++++++++ utils/__init__.py | 1 + utils/lbp_features.py | 66 +++++++ 10 files changed, 1040 insertions(+) create mode 100644 .gitignore create mode 100644 app.py create mode 100644 config.py create mode 100644 requirements.txt create mode 100644 services/__init__.py create mode 100644 services/extract_service.py create mode 100644 services/train_service.py create mode 100644 services/verify_service.py create mode 100644 utils/__init__.py create mode 100644 utils/lbp_features.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d99230b --- /dev/null +++ b/.gitignore @@ -0,0 +1,20 @@ +# Python +venv/ +__pycache__/ +*.pyc +*.pyo +*.egg-info/ +dist/ +build/ + +# Environment +.env + +# Storage (dataset & model di-generate, bukan disimpan di repo) +storage/face_datasets/ +storage/face_models/ +storage/face_videos/ + +# IDE +.vscode/ +.idea/ diff --git a/app.py b/app.py new file mode 100644 index 0000000..6397738 --- /dev/null +++ b/app.py @@ -0,0 +1,181 @@ +import os +import uuid +import logging +from functools import wraps +from flask import Flask, request, jsonify +from flask_cors import CORS + +from config import DATASETS_DIR, MODELS_DIR, TEMP_DIR, MAX_CONTENT_LENGTH, API_KEY +from services.extract_service import extract_frames +from services.train_service import train_model +from services.verify_service import verify_face + +logging.basicConfig( + level=logging.INFO, + format='[%(asctime)s] %(levelname)s: %(message)s' +) +logger = logging.getLogger(__name__) + +app = Flask(__name__) +app.config['MAX_CONTENT_LENGTH'] = MAX_CONTENT_LENGTH +CORS(app) + + +def require_api_key(f): + @wraps(f) + def decorated(*args, **kwargs): + key = request.headers.get('X-API-Key', '') + if key != API_KEY: + return jsonify({"status": "error", "message": "API key tidak valid."}), 401 + return f(*args, **kwargs) + return decorated + + +@app.route('/health', methods=['GET']) +def health(): + model_exists = os.path.exists(os.path.join(MODELS_DIR, 'face_model.pkl')) + return jsonify({ + "status": "ok", + "model_loaded": model_exists, + "datasets_dir": DATASETS_DIR, + "models_dir": MODELS_DIR, + }) + + +@app.route('/extract-frames', methods=['POST']) +@require_api_key +def api_extract_frames(): + if 'video' not in request.files: + return jsonify({"status": "error", "message": "File video tidak ditemukan."}), 400 + + user_id = request.form.get('user_id') + target_frames = int(request.form.get('target_frames', 200)) + + if not user_id: + return jsonify({"status": "error", "message": "user_id wajib diisi."}), 400 + + video_file = request.files['video'] + temp_video = os.path.join(TEMP_DIR, f"enroll_{user_id}_{uuid.uuid4().hex[:8]}.mp4") + + try: + video_file.save(temp_video) + + output_dir = os.path.join(DATASETS_DIR, str(user_id)) + + result = extract_frames(temp_video, output_dir, target_frames=target_frames) + + logger.info( + f"Extract frames berhasil untuk user {user_id}. " + f"Frames: {result['total_extracted']}" + ) + + return jsonify({ + "status": "success", + "message": f"Berhasil mengekstrak {result['total_extracted']} frame wajah frontal", + **result + }) + + except Exception as e: + logger.error(f"Extract frames GAGAL untuk user {user_id}: {str(e)}") + return jsonify({"status": "error", "message": str(e)}), 500 + + finally: + if os.path.exists(temp_video): + os.remove(temp_video) + + +@app.route('/train-model', methods=['POST']) +@require_api_key +def api_train_model(): + data = request.get_json(silent=True) or {} + approved_user_ids = data.get('approved_user_ids') + + if not approved_user_ids or not isinstance(approved_user_ids, list): + return jsonify({ + "status": "error", + "message": "approved_user_ids wajib diisi sebagai array." + }), 400 + + approved_str = [str(uid) for uid in approved_user_ids] + + try: + logger.info( + f"Memulai training model SVM. " + f"Approved users: {approved_str}" + ) + + result = train_model(DATASETS_DIR, MODELS_DIR, approved_str) + + logger.info( + f"Training selesai. Users: {result['total_users']}, " + f"CV: {result['cv_score']*100:.2f}%, " + f"Test: {result['test_accuracy']*100:.2f}%" + ) + + return jsonify({ + "status": "success", + "message": ( + f"Model SVM berhasil dilatih. " + f"{result['total_users']} user, " + f"CV={result['cv_score']*100:.2f}%, " + f"Test={result['test_accuracy']*100:.2f}%" + ), + **result + }) + + except Exception as e: + logger.error(f"Training model GAGAL: {str(e)}") + return jsonify({"status": "error", "message": str(e)}), 500 + + +@app.route('/verify-face', methods=['POST']) +@require_api_key +def api_verify_face(): + if 'file' not in request.files: + return jsonify({"status": "error", "message": "File tidak ditemukan."}), 400 + + user_id = request.form.get('user_id') + is_video = request.form.get('is_video', 'false').lower() == 'true' + + if not user_id: + return jsonify({"status": "error", "message": "user_id wajib diisi."}), 400 + + uploaded_file = request.files['file'] + ext = 'mp4' if is_video else 'jpg' + temp_file = os.path.join(TEMP_DIR, f"verify_{user_id}_{uuid.uuid4().hex[:8]}.{ext}") + + try: + uploaded_file.save(temp_file) + + result = verify_face(MODELS_DIR, user_id, temp_file, is_video=is_video) + + logger.info( + f"Verifikasi user {user_id}: " + f"status={result.get('verification_status')}, " + f"match={result.get('match')}, " + f"svm_df={result.get('svm_df')}, " + f"confidence={result.get('confidence')}, " + f"predicted={result.get('predicted_user')}, " + f"blur={result.get('blur_score')}, " + f"frames_approved={result.get('frames_approved')}/{result.get('frames_total')}" + ) + + return jsonify(result) + + except Exception as e: + logger.error(f"Verifikasi GAGAL untuk user {user_id}: {str(e)}") + return jsonify({"status": "error", "message": str(e)}), 500 + + finally: + if os.path.exists(temp_file): + os.remove(temp_file) + + +if __name__ == '__main__': + logger.info("=" * 50) + logger.info("MPG HRIS - Flask ML API Server") + logger.info(f"Datasets: {DATASETS_DIR}") + logger.info(f"Models: {MODELS_DIR}") + logger.info("=" * 50) + + app.run(host='0.0.0.0', port=5000, debug=True) diff --git a/config.py b/config.py new file mode 100644 index 0000000..16fb5e7 --- /dev/null +++ b/config.py @@ -0,0 +1,15 @@ +import os + +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) + +STORAGE_DIR = os.path.join(BASE_DIR, 'storage') +DATASETS_DIR = os.path.join(STORAGE_DIR, 'face_datasets') +MODELS_DIR = os.path.join(STORAGE_DIR, 'face_models') +TEMP_DIR = os.path.join(STORAGE_DIR, 'temp') + +MAX_CONTENT_LENGTH = 50 * 1024 * 1024 # 50MB + +API_KEY = os.environ.get('FLASK_ML_API_KEY', 'dev-api-key-mpg-hris') + +for d in [STORAGE_DIR, DATASETS_DIR, MODELS_DIR, TEMP_DIR]: + os.makedirs(d, exist_ok=True) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f344534 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +flask +flask-cors +opencv-contrib-python-headless +numpy +scikit-learn +scikit-image +joblib +Pillow +gunicorn diff --git a/services/__init__.py b/services/__init__.py new file mode 100644 index 0000000..a70b302 --- /dev/null +++ b/services/__init__.py @@ -0,0 +1 @@ +# Services package diff --git a/services/extract_service.py b/services/extract_service.py new file mode 100644 index 0000000..ac3e1ad --- /dev/null +++ b/services/extract_service.py @@ -0,0 +1,195 @@ +import cv2 +import os +import numpy as np + +import sys +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from utils.lbp_features import preprocess_face, FACE_SIZE + +BLUR_THRESHOLD = 30.0 +MIN_FACE_RATIO = 0.15 + +face_cascade = cv2.CascadeClassifier( + cv2.data.haarcascades + 'haarcascade_frontalface_default.xml' +) + + +def get_blur_score(img): + return cv2.Laplacian(img, cv2.CV_64F).var() + + +def detect_frontal_face(gray_frame): + h, w = gray_frame.shape + min_size = int(min(h, w) * MIN_FACE_RATIO) + + faces = face_cascade.detectMultiScale( + gray_frame, scaleFactor=1.1, minNeighbors=5, + minSize=(min_size, min_size) + ) + + if len(faces) == 0: + faces = face_cascade.detectMultiScale( + gray_frame, scaleFactor=1.05, minNeighbors=4, + minSize=(min_size, min_size) + ) + + if len(faces) == 0: + return None + + faces = sorted(faces, key=lambda f: f[2] * f[3], reverse=True) + return faces[0] + + +def estimate_yaw_from_symmetry(gray_face): + h, w = gray_face.shape + mid = w // 2 + + left_half = gray_face[:, :mid].astype(np.float64) + right_half = cv2.flip(gray_face[:, mid:], 1).astype(np.float64) + + min_w = min(left_half.shape[1], right_half.shape[1]) + left_half = left_half[:, :min_w] + right_half = right_half[:, :min_w] + + if left_half.size == 0 or right_half.size == 0: + return 999.0 + + left_mean = np.mean(left_half) + right_mean = np.mean(right_half) + + diff = abs(left_mean - right_mean) + estimated_yaw = diff * 0.8 + + return estimated_yaw + + +def process_frame(frame, gray_frame): + face_rect = detect_frontal_face(gray_frame) + if face_rect is None: + return None, None, "no_face", 0.0, 0.0 + + (x, y, w, h) = face_rect + + padding = int(max(w, h) * 0.1) + x1 = max(0, x - padding) + y1 = max(0, y - padding) + x2 = min(gray_frame.shape[1], x + w + padding) + y2 = min(gray_frame.shape[0], y + h + padding) + + face_crop_gray = gray_frame[y1:y2, x1:x2] + face_crop_color = frame[y1:y2, x1:x2] + + face_resized = cv2.resize(face_crop_gray, FACE_SIZE, interpolation=cv2.INTER_AREA) + + blur_score = get_blur_score(face_resized) + if blur_score < BLUR_THRESHOLD: + return None, None, f"blur ({round(blur_score, 1)})", blur_score, 0.0 + + yaw_estimate = estimate_yaw_from_symmetry(face_resized) + + return face_resized, face_crop_color, "ok", blur_score, yaw_estimate + + +def extract_frames(video_path, output_dir, target_frames=30): + if not os.path.exists(video_path): + raise Exception(f"Video tidak ditemukan: {video_path}") + + if not os.path.exists(output_dir): + os.makedirs(output_dir) + else: + for f in os.listdir(output_dir): + if (f.startswith("frame_") or f.startswith("raw_frame_")) and f.endswith(".jpg"): + os.remove(os.path.join(output_dir, f)) + + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + raise Exception("Gagal membuka file video.") + + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + fps = cap.get(cv2.CAP_PROP_FPS) + + if total_frames <= 0: + raise Exception("Video tidak valid (0 frame).") + + candidates = [] + frame_idx = 0 + skipped_no_face = 0 + skipped_blur = 0 + skipped_side = 0 + + while True: + ret, frame = cap.read() + if not ret: + break + + gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + face_img, face_color, status, blur_score, yaw = process_frame(frame, gray) + + if face_img is None: + if "blur" in status: + skipped_blur += 1 + else: + skipped_no_face += 1 + frame_idx += 1 + continue + + candidates.append({ + 'frame_idx': frame_idx, + 'face_gray': face_img, + 'face_color': face_color, + 'blur_score': blur_score, + 'yaw': yaw + }) + + frame_idx += 1 + + cap.release() + + if len(candidates) == 0: + raise Exception( + "Tidak ada frame wajah frontal berkualitas yang berhasil diekstrak. " + "Pastikan wajah menghadap depan dengan pencahayaan cukup." + ) + + candidates.sort(key=lambda c: c['blur_score'], reverse=True) + + if len(candidates) > target_frames: + candidates_by_idx = sorted(candidates, key=lambda c: c['frame_idx']) + + chunk_size = len(candidates_by_idx) // target_frames + selected = [] + + for i in range(target_frames): + start = i * chunk_size + end = min(start + chunk_size, len(candidates_by_idx)) + chunk = candidates_by_idx[start:end] + + best_in_chunk = max(chunk, key=lambda c: c['blur_score']) + selected.append(best_in_chunk) + + candidates = sorted(selected, key=lambda c: c['frame_idx']) + + saved_count = 0 + for cand in candidates: + filename = f"frame_{saved_count:03d}.jpg" + filename_raw = f"raw_frame_{saved_count:03d}.jpg" + cv2.imwrite( + os.path.join(output_dir, filename), + cand['face_gray'], + [int(cv2.IMWRITE_JPEG_QUALITY), 95] + ) + cv2.imwrite(os.path.join(output_dir, filename_raw), cand['face_color']) + saved_count += 1 + + return { + "total_video_frames": total_frames, + "video_fps": round(fps, 1), + "total_candidates": len(candidates), + "total_extracted": saved_count, + "target_frames": target_frames, + "skipped_no_face": skipped_no_face, + "skipped_blur": skipped_blur, + "skipped_side_face": skipped_side, + "output_dir": output_dir + } diff --git a/services/train_service.py b/services/train_service.py new file mode 100644 index 0000000..a56da23 --- /dev/null +++ b/services/train_service.py @@ -0,0 +1,186 @@ +import cv2 +import os +import sys +import json +import numpy as np +import joblib +from sklearn.svm import SVC +from sklearn.preprocessing import StandardScaler +from sklearn.pipeline import Pipeline +from sklearn.model_selection import train_test_split, GridSearchCV +from skimage.feature import local_binary_pattern + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from utils.lbp_features import ( + extract_lbp_features, extract_lbp_from_augmented, + augmentasi_lbp, preprocess_face, FACE_SIZE, LBP_SCALES +) + + +def load_images(dataset_path): + valid_ext = ('.jpg', '.jpeg', '.png') + images = [] + + if not os.path.exists(dataset_path): + return images + + files = sorted([ + f for f in os.listdir(dataset_path) + if f.lower().endswith(valid_ext) and ( + f.startswith('frame_') or f.startswith('selfie_') + ) + ]) + + for filename in files: + filepath = os.path.join(dataset_path, filename) + img = cv2.imread(filepath, cv2.IMREAD_GRAYSCALE) + if img is not None: + if img.shape != FACE_SIZE: + img = cv2.resize(img, FACE_SIZE, interpolation=cv2.INTER_AREA) + images.append(img) + + return images + + +def train_model(base_datasets_path, model_output_path, approved_user_ids=None): + if not os.path.exists(base_datasets_path): + raise Exception(f"Base datasets path tidak ditemukan: {base_datasets_path}") + + user_images_dict = {} + label_map = {} + user_stats = {} + + for folder_name in sorted(os.listdir(base_datasets_path)): + folder_path = os.path.join(base_datasets_path, folder_name) + if not os.path.isdir(folder_path): + continue + + user_id = folder_name + + if approved_user_ids and user_id not in approved_user_ids: + continue + + images = load_images(folder_path) + if len(images) < 10: + continue + + label_map[user_id] = user_id + user_images_dict[user_id] = images + + if len(label_map) == 0: + raise Exception("Tidak ada user dengan dataset valid (minimal 10 gambar per user).") + + if len(label_map) < 2: + raise Exception( + "Minimal 2 user yang sudah di-approve diperlukan untuk melatih model SVM. " + f"Saat ini baru {len(label_map)} user." + ) + + X_ori = [] + y_ori = [] + X_aug = [] + y_aug = [] + + for user_id, images in user_images_dict.items(): + ori_count = 0 + aug_count = 0 + + for img in images: + proc_img = preprocess_face(img) + features = extract_lbp_features(proc_img) + X_ori.append(features) + y_ori.append(user_id) + ori_count += 1 + + pola = local_binary_pattern(proc_img, P=8, R=1, method='uniform') + lbp_uint8 = (pola * 255.0 / 9).astype(np.uint8) + variasi = augmentasi_lbp(lbp_uint8) + for aug_img in variasi: + aug_features = extract_lbp_from_augmented(aug_img) + X_aug.append(aug_features) + y_aug.append(user_id) + aug_count += 1 + + user_stats[user_id] = ori_count + aug_count + + X = np.array(X_ori + X_aug) + y = np.array(y_ori + y_aug) + + X_train, X_test, y_train, y_test = train_test_split( + X, y, test_size=0.2, random_state=42, stratify=y + ) + + param_grid = { + 'svm__C': [0.1, 1, 10, 20], + 'svm__gamma': ['scale', 'auto', 0.01, 0.001, 0.1] + } + + pipe_search = Pipeline([ + ('scaler', StandardScaler()), + ('svm', SVC(kernel='rbf', probability=False)) + ]) + + grid = GridSearchCV( + pipe_search, param_grid, + cv=5, scoring='accuracy', + refit=False, return_train_score=True, verbose=0 + ) + grid.fit(X_train, y_train) + + best_C = grid.best_params_['svm__C'] + best_gamma = grid.best_params_['svm__gamma'] + cv_score = grid.best_score_ + + model_final = Pipeline([ + ('scaler', StandardScaler()), + ('svm', SVC(kernel='rbf', C=best_C, gamma=best_gamma, probability=True)) + ]) + model_final.fit(X_train, y_train) + test_acc = model_final.score(X_test, y_test) + + if not os.path.exists(model_output_path): + os.makedirs(model_output_path) + + model_file = os.path.join(model_output_path, "face_model.pkl") + scaler_file = os.path.join(model_output_path, "face_scaler.pkl") + labels_file = os.path.join(model_output_path, "face_labels.json") + + temp_model_file = os.path.join(model_output_path, "face_model_temp.pkl") + temp_scaler_file = os.path.join(model_output_path, "face_scaler_temp.pkl") + temp_labels_file = os.path.join(model_output_path, "face_labels_temp.json") + + joblib.dump(model_final.named_steps['svm'], temp_model_file) + joblib.dump(model_final.named_steps['scaler'], temp_scaler_file) + + labels_data = { + "classes": list(model_final.named_steps['svm'].classes_), + "user_ids": list(label_map.keys()), + "best_C": best_C, + "best_gamma": str(best_gamma), + "cv_score": round(cv_score, 4), + "test_accuracy": round(test_acc, 4), + } + with open(temp_labels_file, 'w') as f: + json.dump(labels_data, f, indent=2) + + os.replace(temp_model_file, model_file) + os.replace(temp_scaler_file, scaler_file) + os.replace(temp_labels_file, labels_file) + + return { + "total_users": len(label_map), + "user_stats": user_stats, + "total_samples": len(X), + "train_samples": len(X_train), + "test_samples": len(X_test), + "feature_dimension": int(X.shape[1]), + "best_C": best_C, + "best_gamma": str(best_gamma), + "cv_score": round(cv_score, 4), + "test_accuracy": round(test_acc, 4), + "classes": list(model_final.named_steps['svm'].classes_), + "model_path": model_file, + "scaler_path": scaler_file, + "labels_path": labels_file + } diff --git a/services/verify_service.py b/services/verify_service.py new file mode 100644 index 0000000..93b4f7a --- /dev/null +++ b/services/verify_service.py @@ -0,0 +1,366 @@ +import cv2 +import os +import sys +import json +import logging +import numpy as np +import joblib + +logger = logging.getLogger('flask_ml') + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) + +from utils.lbp_features import extract_lbp_features, preprocess_face, FACE_SIZE + +UNKNOWN_LABEL = "unknown" +BLUR_THRESHOLD = 30.0 +DF_APPROVED = 1.5 +DF_PENDING = 0.5 + +face_cascade = cv2.CascadeClassifier( + cv2.data.haarcascades + 'haarcascade_frontalface_default.xml' +) +profile_cascade = cv2.CascadeClassifier( + cv2.data.haarcascades + 'haarcascade_profileface.xml' +) + + +def get_blur_score(img): + return cv2.Laplacian(img, cv2.CV_64F).var() + + +def fix_exif_rotation(img_path): + try: + from PIL import Image + pil_img = Image.open(img_path) + exif = pil_img.getexif() + orientation = exif.get(274, 1) + + if orientation == 3: + pil_img = pil_img.rotate(180, expand=True) + elif orientation == 6: + pil_img = pil_img.rotate(270, expand=True) + elif orientation == 8: + pil_img = pil_img.rotate(90, expand=True) + + img_array = np.array(pil_img) + if len(img_array.shape) == 3 and img_array.shape[2] == 3: + img_array = cv2.cvtColor(img_array, cv2.COLOR_RGB2BGR) + return img_array + except ImportError: + return None + except Exception: + return None + + +def detect_and_crop_face(gray): + h, w = gray.shape + + if max(h, w) > 640: + scale = 640.0 / max(h, w) + gray = cv2.resize(gray, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA) + h, w = gray.shape + + min_size = int(min(h, w) * 0.1) + + faces = face_cascade.detectMultiScale(gray, 1.1, 5, minSize=(min_size, min_size)) + if len(faces) == 0: + faces = face_cascade.detectMultiScale(gray, 1.05, 3, minSize=(min_size, min_size)) + if len(faces) == 0: + faces = profile_cascade.detectMultiScale(gray, 1.1, 3, minSize=(min_size, min_size)) + if len(faces) == 0: + flipped = cv2.flip(gray, 1) + faces = face_cascade.detectMultiScale(flipped, 1.05, 3, minSize=(min_size, min_size)) + if len(faces) > 0: + gray = flipped + + if len(faces) == 0: + return None + + faces = sorted(faces, key=lambda f: f[2] * f[3], reverse=True) + (x, y, fw, fh) = faces[0] + + padding = int(max(fw, fh) * 0.1) + x1 = max(0, x - padding) + y1 = max(0, y - padding) + x2 = min(w, x + fw + padding) + y2 = min(h, y + fh + padding) + + face_crop = gray[y1:y2, x1:x2] + + if face_crop.shape[0] < 10 or face_crop.shape[1] < 10: + return None + + face_resized = cv2.resize(face_crop, FACE_SIZE, interpolation=cv2.INTER_AREA) + return face_resized + + +def preprocess_image(image_path): + if not os.path.exists(image_path): + raise Exception("File gambar tidak ditemukan.") + + img = fix_exif_rotation(image_path) + if img is None: + img = cv2.imread(image_path) + + if img is None: + raise Exception("Gagal membaca file gambar.") + + gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) if len(img.shape) == 3 else img + + face = detect_and_crop_face(gray) + if face is None: + raise Exception( + "Wajah tidak terdeteksi. Pastikan pencahayaan cukup dan wajah terlihat jelas." + ) + + blur_score = get_blur_score(face) + + if blur_score < BLUR_THRESHOLD: + raise Exception( + f"Foto terlalu buram (Score: {round(blur_score, 1)}). Harap foto ulang." + ) + + face_preprocessed = preprocess_face(face) + return face_preprocessed, blur_score + + +def extract_frames_from_video(video_path, target_frames=10): + cap = cv2.VideoCapture(video_path) + if not cap.isOpened(): + raise Exception("Gagal membuka file video verifikasi.") + + total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + + if total_frames <= 0: + cap.release() + raise Exception("Video tidak valid (0 frame).") + + candidates = [] + frame_idx = 0 + sample_interval = max(1, total_frames // (target_frames * 3)) + + while True: + ret, frame = cap.read() + if not ret: + break + + if frame_idx % sample_interval != 0: + frame_idx += 1 + continue + + gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + face_raw = detect_and_crop_face(gray) + + if face_raw is None: + frame_idx += 1 + continue + + blur_score = get_blur_score(face_raw) + if blur_score < BLUR_THRESHOLD: + frame_idx += 1 + continue + + face_preprocessed = preprocess_face(face_raw) + candidates.append({ + 'face': face_preprocessed, + 'blur_score': blur_score, + 'frame_idx': frame_idx, + }) + + frame_idx += 1 + + cap.release() + + if len(candidates) == 0: + raise Exception( + "Tidak ada frame wajah yang valid dalam video. " + "Pastikan wajah menghadap depan dengan pencahayaan cukup." + ) + + candidates.sort(key=lambda c: c['blur_score'], reverse=True) + selected = candidates[:target_frames] + + return selected + + +def _classify_frame(features, svm, scaler, expected_user): + classes = [str(c) for c in svm.classes_] + features_s = scaler.transform([features]) + df_values = svm.decision_function(features_s)[0] + proba = svm.predict_proba(features_s)[0] + + if expected_user in classes: + idx = classes.index(expected_user) + user_df = float(df_values[idx] if hasattr(df_values, '__len__') else df_values) + user_conf = float(proba[idx]) + else: + user_df = -999.0 + user_conf = 0.0 + + df_sorted = sorted(enumerate(df_values), key=lambda x: -x[1]) + predicted_class = str(classes[df_sorted[0][0]]) + + top_df = float(df_values[df_sorted[0][0]]) + second_df = float(df_values[df_sorted[1][0]]) if len(df_sorted) > 1 else -999.0 + df_margin = top_df - second_df + + # === Verifikasi 1:1 === + # Untuk model LBP+SVM, cukup cek skor OVR user sendiri. + # Gap check dihapus karena terlalu ketat untuk variasi pencahayaan/sudut. + gap = top_df - user_df if predicted_class != expected_user else 0.0 + + if user_df >= DF_APPROVED: + status = "APPROVED" + is_match = True + elif user_df >= DF_PENDING: + status = "PENDING" + is_match = True + else: + status = "REJECTED" + is_match = False + + return { + 'is_match': is_match, + 'status': status, + 'user_df': user_df, + 'confidence': user_conf, + 'predicted_class': predicted_class, + 'df_margin': round(df_margin, 4), + 'gap': round(gap, 4), + } + + +def _aggregate_frame_results(frame_results): + total = len(frame_results) + approved = sum(1 for r in frame_results if r['status'] == 'APPROVED') + pending = sum(1 for r in frame_results if r['status'] == 'PENDING') + rejected = sum(1 for r in frame_results if r['status'] == 'REJECTED') + + approved_ratio = approved / total + pending_ratio = pending / total + + avg_df = float(np.mean([r['user_df'] for r in frame_results])) + avg_conf = float(np.mean([r['confidence'] for r in frame_results])) + + if approved_ratio >= 0.4: + final_status = "APPROVED" + final_match = True + elif (approved_ratio + pending_ratio) >= 0.5: + final_status = "PENDING" + final_match = True + else: + final_status = "REJECTED" + final_match = False + + return { + 'verification_status': final_status, + 'match': final_match, + 'avg_df': round(avg_df, 4), + 'confidence': round(avg_conf, 4), + 'frames_total': total, + 'frames_approved': approved, + 'frames_pending': pending, + 'frames_rejected': rejected, + 'approved_ratio': round(approved_ratio, 3), + } + + +def verify_face(model_dir, user_id, input_path, is_video=False): + model_file = os.path.join(model_dir, "face_model.pkl") + scaler_file = os.path.join(model_dir, "face_scaler.pkl") + labels_file = os.path.join(model_dir, "face_labels.json") + + if not os.path.exists(model_file): + raise Exception("Model wajah belum tersedia. Belum ada data training.") + if not os.path.exists(scaler_file): + raise Exception("File scaler belum tersedia.") + if not os.path.exists(labels_file): + raise Exception("File label belum tersedia.") + + svm = joblib.load(model_file) + scaler = joblib.load(scaler_file) + + expected_user = str(user_id) + + if is_video: + try: + frames = extract_frames_from_video(input_path, target_frames=10) + except Exception as e_vid: + return { + "status": "success", + "match": False, + "confidence": 0, + "verification_status": "PREPROCESSING_FAILED", + "message": str(e_vid), + "blur_score": 0, + "frames_total": 0, + } + + frame_results = [] + blur_scores = [] + + for i, frame_data in enumerate(frames): + face = frame_data['face'] + blur_scores.append(frame_data['blur_score']) + features = extract_lbp_features(face) + result = _classify_frame(features, svm, scaler, expected_user) + frame_results.append(result) + logger.info( + f" Frame {i}: predicted={result['predicted_class']}, " + f"user_df={result['user_df']:.4f}, " + f"gap={result['gap']:.4f}, " + f"conf={result['confidence']:.4f}, " + f"status={result['status']}" + ) + + aggregated = _aggregate_frame_results(frame_results) + avg_blur = float(np.mean(blur_scores)) + + return { + "status": "success", + "match": bool(aggregated['match']), + "verification_status": aggregated['verification_status'], + "confidence": aggregated['confidence'], + "svm_df": aggregated['avg_df'], + "blur_score": round(avg_blur, 1), + "frames_total": aggregated['frames_total'], + "frames_approved": aggregated['frames_approved'], + "frames_pending": aggregated['frames_pending'], + "frames_rejected": aggregated['frames_rejected'], + "approved_ratio": aggregated['approved_ratio'], + "predicted_user": expected_user if aggregated['match'] else "unknown", + "actual_predicted": max(set(r['predicted_class'] for r in frame_results), key=lambda c: sum(1 for r in frame_results if r['predicted_class'] == c)), + "expected_user": expected_user, + "user_id": int(user_id) if str(user_id).isdigit() else str(user_id), + "message": None, + } + + else: + try: + processed_face, blur_score = preprocess_image(input_path) + except Exception as e_proc: + return { + "status": "success", + "match": False, + "confidence": 0, + "verification_status": "PREPROCESSING_FAILED", + "message": str(e_proc), + "blur_score": 0, + } + + features = extract_lbp_features(processed_face) + result = _classify_frame(features, svm, scaler, expected_user) + + return { + "status": "success", + "match": bool(result['is_match']), + "verification_status": result['status'], + "confidence": float(round(result['confidence'], 4)), + "svm_df": float(round(result['user_df'], 4)), + "predicted_user": result['predicted_class'], + "expected_user": expected_user, + "blur_score": float(round(blur_score, 1)), + "user_id": int(user_id) if str(user_id).isdigit() else str(user_id), + "message": None, + } diff --git a/utils/__init__.py b/utils/__init__.py new file mode 100644 index 0000000..dd7ee44 --- /dev/null +++ b/utils/__init__.py @@ -0,0 +1 @@ +# Utils package diff --git a/utils/lbp_features.py b/utils/lbp_features.py new file mode 100644 index 0000000..f47bdf2 --- /dev/null +++ b/utils/lbp_features.py @@ -0,0 +1,66 @@ +import cv2 +import numpy as np +from skimage.feature import local_binary_pattern + +FACE_SIZE = (128, 128) +LBP_SCALES = [(8, 1), (16, 2), (24, 3)] +LBP_TOTAL_BINS = sum(P + 2 for P, R in LBP_SCALES) + + +def apply_clahe(gray_img): + clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)) + return clahe.apply(gray_img) + + +def preprocess_face(image): + if len(image.shape) == 3: + image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) + + if image.shape != FACE_SIZE: + image = cv2.resize(image, FACE_SIZE, interpolation=cv2.INTER_AREA) + + clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)) + image = clahe.apply(image) + return image + + +def extract_lbp_features(image): + if len(image.shape) == 3: + image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY) + + if image.shape != FACE_SIZE: + image = cv2.resize(image, FACE_SIZE, interpolation=cv2.INTER_AREA) + + hists = [] + for P, R in LBP_SCALES: + bins = P + 2 + pola = local_binary_pattern(image, P=P, R=R, method='uniform') + hist, _ = np.histogram(pola, bins=bins, range=(0, bins), density=False) + hists.append(hist) + + return np.concatenate(hists).astype(np.float64) + + +def extract_lbp_from_augmented(lbp_uint8): + hists = [] + for P, R in LBP_SCALES: + bins = P + 2 + hist, _ = np.histogram(lbp_uint8, bins=bins, range=(0, 256), density=False) + hists.append(hist) + return np.concatenate(hists).astype(np.float64) + + +def augmentasi_lbp(lbp_uint8): + hasil = [] + hasil.append(cv2.flip(lbp_uint8, 1)) + hasil.append(cv2.add(lbp_uint8, 30)) + hasil.append(cv2.subtract(lbp_uint8, 30)) + hasil.append(cv2.GaussianBlur(lbp_uint8, (5, 5), 0)) + noise = np.random.normal(0, 10, lbp_uint8.shape).astype(np.int16) + noisy = np.clip(lbp_uint8.astype(np.int16) + noise, 0, 255).astype(np.uint8) + hasil.append(noisy) + M = cv2.getRotationMatrix2D((64, 64), 5, 1.0) + hasil.append(cv2.warpAffine(lbp_uint8, M, (128, 128))) + M2 = cv2.getRotationMatrix2D((64, 64), -5, 1.0) + hasil.append(cv2.warpAffine(lbp_uint8, M2, (128, 128))) + return hasil