Initial commit for Flask ML API
This commit is contained in:
commit
9ece3d9aea
|
|
@ -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/
|
||||
|
|
@ -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)
|
||||
|
|
@ -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)
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
flask
|
||||
flask-cors
|
||||
opencv-contrib-python-headless
|
||||
numpy
|
||||
scikit-learn
|
||||
scikit-image
|
||||
joblib
|
||||
Pillow
|
||||
gunicorn
|
||||
|
|
@ -0,0 +1 @@
|
|||
# Services package
|
||||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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,
|
||||
}
|
||||
|
|
@ -0,0 +1 @@
|
|||
# Utils package
|
||||
|
|
@ -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
|
||||
Loading…
Reference in New Issue