akhfzl
update-vn
4babbcb
import cv2 as cv
import numpy as np
from PIL import Image
import torch
import pandas as pd
from .setConfig import efficientnet_model, face_detector, transform, pca_xgb, faiss, load_db
import os
import requests
def ImgPreprocessing(img):
if len(img.shape) == 2 or img.shape[2] == 1:
img = cv.cvtColor(img, cv.COLOR_GRAY2BGR)
img_yuv = cv.cvtColor(img, cv.COLOR_BGR2YUV)
clahe = cv.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
img_yuv[:, :, 0] = clahe.apply(img_yuv[:, :, 0])
img = cv.cvtColor(img_yuv, cv.COLOR_YUV2BGR)
return img
def YoloFaceDetection(img):
results = face_detector.predict(img, conf=0.7)
keyReturn = {
'message': "",
'status': False,
'coordinate': []
}
if len(results) == 0 or results[0].boxes is None or len(results[0].boxes) == 0:
keyReturn['message'] = "Tidak ada wajah terdeteksi"
return keyReturn
boxes = results[0].boxes.xyxy.cpu().numpy()
confs = results[0].boxes.conf.cpu().numpy()
max_idx = confs.argmax()
x1, y1, x2, y2 = map(int, boxes[max_idx])
keyReturn['message'] = "Face detected"
keyReturn['status'] = True
keyReturn['coordinate'] = [x1, y1, x2, y2]
return keyReturn
def FaceValidationPredict(**face_crop):
x1, y1, x2, y2 = face_crop['coordinate']
face_crop = face_crop['img'][y1:y2, x1:x2]
face_crop = cv.cvtColor(face_crop, cv.COLOR_BGR2GRAY)
face_crop = cv.cvtColor(face_crop, cv.COLOR_GRAY2RGB)
face_pil = Image.fromarray(face_crop)
face_tensor = transform(face_pil).unsqueeze(0)
with torch.no_grad():
features = efficientnet_model(face_tensor).cpu().numpy()
pred = pca_xgb.predict(features)[0]
return pred, features, face_crop
# feature
def FaceValidation(frame: np.ndarray):
if frame is None:
return "No frame captured from webcam"
if isinstance(frame, dict):
frame = frame['image']
pred = ''
img = ImgPreprocessing(frame)
keyReturn = YoloFaceDetection(img)
if keyReturn['status'] is False:
return keyReturn['message']
results = keyReturn['coordinate']
x1, y1, x2, y2 = results
validation, features, face_crop = FaceValidationPredict(coordinate=[x1, y1, x2, y2], img=img)
pred = 'Wajah Valid' if validation == 0 else 'Wajah Tidak Valid'
return f"Predicted class: {pred}"
def FaceRecord(img, name, photo_idx):
if img is None or name is None:
return f"Foto {photo_idx}] Gagal: Tidak ada gambar atau nama"
os.makedirs('users', exist_ok=True)
user_dir = os.path.join("users", name)
os.makedirs(user_dir, exist_ok=True)
img = np.array(img)
img = ImgPreprocessing(img)
keyReturn = YoloFaceDetection(img)
if keyReturn['status'] is False:
return keyReturn['message']
x1, y1, x2, y2 = keyReturn['coordinate']
pred, features, face_crop = FaceValidationPredict(coordinate=[x1, y1, x2, y2], img=img)
if pred == 1:
return f"[Foto {photo_idx}] Gagal: wajah tidak valid"
save_path = os.path.join(user_dir, f"photo_{photo_idx}.jpg")
cv.imwrite(save_path, face_crop)
csv_path = "users/face_features.csv"
row = pd.DataFrame({
"label": [name],
"features": [features.flatten().tolist()]
})
if not os.path.exists(csv_path):
row.to_csv(csv_path, index=False)
else:
row.to_csv(csv_path, mode="a", index=False, header=False)
total_fotos = len([f for f in os.listdir(user_dir) if f.endswith(".jpg")])
if total_fotos < 4:
return f"[Foto {photo_idx}] Berhasil disimpan ke {save_path}. ({total_fotos}/4)"
else:
return f"[Foto {photo_idx}] Berhasil disimpan ke {save_path}. βœ… Semua 4 foto sudah lengkap!"
def Recognize(frame: np.ndarray):
if frame is None:
return "No frame captured from webcam"
if isinstance(frame, dict):
frame = frame['image']
img = ImgPreprocessing(frame)
keyReturn = YoloFaceDetection(img)
if not keyReturn['status']:
return keyReturn['message']
results = keyReturn['coordinate']
x1, y1, x2, y2 = results
pred, features, face_crop = FaceValidationPredict(coordinate=[x1, y1, x2, y2], img=img)
faiss.normalize_L2(features)
if pred == 1:
return 'Wajah tidak terdeteksi'
faiss_index, labels, db = load_db("users/face_features.csv")
if faiss_index is None:
return "Database kosong"
D, I = faiss_index.search(features, k=1)
score = float(D[0][0])
idx = int(I[0][0])
if score < 0.7:
return f"Tidak dikenali"
else:
return f"Terkenali sebagai: {labels[idx]} - (score={score:.2f})"
def UploadVoice(audio_file):
if audio_file is None:
return "❌ Tidak ada file audio. Coba rekam lagi.", "", ""
try:
with open(audio_file, "rb") as f:
response = requests.post(
"https://n8n.smartid.co.id/webhook/voice-upload",
data=f,
headers={"Content-Type": "audio/wav"},
)
if response.status_code != 200:
return f"❌ Gagal upload. Status: {response.status_code}", "", ""
try:
data = response.json()
except Exception:
return f"❌ Server tidak mengirim JSON. Balasan:\n{response}", "", ""
transcription = data.get("transcribe", "(Tidak ada transkripsi)")
summary = data.get("summary", "(Tidak ada ringkasan)")
return "βœ… Rekaman berhasil diproses", f"Transcribe: {transcription}", f"Summary: {summary}"
except Exception as e:
return f"❌ Gagal upload: {e}.", f"Transcribe: ", f"Summary: "