8f05a7c188
- Update ASR, face, OCR, pose processors - Add release pre-flight check script - Add synonym generation, chunk processing scripts - Add face recognition, stamp search utilities
293 lines
9.0 KiB
Python
293 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
直接測試人臉識別功能（不通過 HTTP API）
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import numpy as np
|
||
import cv2
|
||
from pathlib import Path
|
||
|
||
# 添加項目根目錄到 Python 路徑
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
|
||
def test_direct_face_processing():
|
||
"""直接測試人臉處理功能"""
|
||
print("=" * 60)
|
||
print("直接人臉處理測試")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
# 1. 測試人臉註冊
|
||
print("\n1. 測試人臉註冊...")
|
||
from scripts.face_registration import FaceRegistration
|
||
|
||
# 創建測試圖像
|
||
img = np.zeros((480, 640, 3), dtype=np.uint8)
|
||
img.fill(200)
|
||
cv2.circle(img, (320, 240), 100, (255, 200, 150), -1)
|
||
cv2.circle(img, (280, 200), 20, (0, 0, 0), -1)
|
||
cv2.circle(img, (360, 200), 20, (0, 0, 0), -1)
|
||
cv2.ellipse(img, (320, 280), (40, 20), 0, 0, 360, (0, 0, 0), -1)
|
||
|
||
test_image_path = "/tmp/direct_test_face.jpg"
|
||
cv2.imwrite(test_image_path, img)
|
||
print(f"✅ 創建測試圖像: {test_image_path}")
|
||
|
||
# 初始化註冊器
|
||
registration = FaceRegistration()
|
||
registration.load_models(use_mps=False)
|
||
|
||
# 註冊人臉
|
||
result = registration.register_face(
|
||
image_path=test_image_path,
|
||
name="Direct Test Person",
|
||
metadata={"test": True, "method": "direct"},
|
||
)
|
||
|
||
if result["success"]:
|
||
print(f"✅ 人臉註冊成功")
|
||
print(f" - Face ID: {result.get('face_id')}")
|
||
print(f" - 嵌入向量長度: {len(result.get('embedding', []))}")
|
||
|
||
# 保存嵌入向量
|
||
embedding = result.get("embedding", [])
|
||
if embedding:
|
||
np.save("/tmp/test_embedding.npy", embedding)
|
||
else:
|
||
print(f"❌ 人臉註冊失敗: {result.get('message')}")
|
||
|
||
# 2. 測試人臉識別處理器
|
||
print("\n2. 測試人臉識別處理器...")
|
||
from scripts.face_recognition_processor import FaceRecognitionProcessor
|
||
|
||
processor = FaceRecognitionProcessor(
|
||
enable_recognition=True, enable_tracking=True, enable_clustering=True
|
||
)
|
||
processor.load_models(use_mps=False)
|
||
|
||
# 讀取圖像
|
||
image = cv2.imread(test_image_path)
|
||
|
||
# 檢測人臉
|
||
detections = processor.detect_faces(image)
|
||
print(f"✅ 檢測到 {len(detections)} 個人臉")
|
||
|
||
if len(detections) > 0:
|
||
for i, detection in enumerate(detections):
|
||
print(f"\n 人臉 {i + 1}:")
|
||
print(
|
||
f" - 位置: x={detection['x']}, y={detection['y']}, width={detection['width']}, height={detection['height']}"
|
||
)
|
||
print(f" - 置信度: {detection['confidence']:.4f}")
|
||
|
||
if "embedding" in detection and detection["embedding"] is not None:
|
||
embedding = detection["embedding"]
|
||
if hasattr(embedding, "shape"):
|
||
print(f" - 嵌入向量形狀: {embedding.shape}")
|
||
else:
|
||
print(f" - 嵌入向量長度: {len(embedding)}")
|
||
|
||
if "attributes" in detection:
|
||
print(f" - 屬性: {detection['attributes']}")
|
||
|
||
# 3. 測試數據庫操作
|
||
print("\n3. 測試數據庫操作...")
|
||
import psycopg2
|
||
from psycopg2.extras import Json
|
||
|
||
conn = psycopg2.connect(
|
||
host="localhost",
|
||
port=5432,
|
||
database="momentry",
|
||
user="accusys",
|
||
password="accusys",
|
||
)
|
||
|
||
cursor = conn.cursor()
|
||
|
||
# 插入測試數據
|
||
print(" 插入測試檢測記錄...")
|
||
cursor.execute(
|
||
"""
|
||
INSERT INTO face_detections
|
||
(video_uuid, frame_number, timestamp_secs, face_id, x, y, width, height, confidence, attributes)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id;
|
||
""",
|
||
(
|
||
"direct_test_video",
|
||
1,
|
||
0.0,
|
||
"direct_test_face_001",
|
||
100,
|
||
100,
|
||
200,
|
||
200,
|
||
0.95,
|
||
Json({"test": True, "method": "direct", "source": "python_test"}),
|
||
),
|
||
)
|
||
|
||
detection_id = cursor.fetchone()[0]
|
||
print(f" ✅ 插入成功，ID: {detection_id}")
|
||
|
||
# 查詢測試
|
||
print(" 查詢測試數據...")
|
||
cursor.execute("SELECT COUNT(*) as total_faces FROM face_detections")
|
||
total_faces = cursor.fetchone()[0]
|
||
|
||
cursor.execute("SELECT COUNT(*) as total_identities FROM face_identities")
|
||
total_identities = cursor.fetchone()[0]
|
||
|
||
print(f" ✅ 數據庫統計:")
|
||
print(f" - 總人臉檢測記錄: {total_faces}")
|
||
print(f" - 總人臉身份: {total_identities}")
|
||
|
||
# 測試向量搜索
|
||
print(" 測試向量搜索...")
|
||
if embedding:
|
||
cursor.execute(
|
||
"""
|
||
SELECT * FROM find_similar_faces(
|
||
%s::vector,
|
||
0.5, -- similarity_threshold
|
||
5 -- limit_count
|
||
);
|
||
""",
|
||
(embedding,),
|
||
)
|
||
|
||
similar_faces = cursor.fetchall()
|
||
print(f" ✅ 找到 {len(similar_faces)} 個相似人臉")
|
||
|
||
for face in similar_faces:
|
||
print(f" - {face[0]}: {face[1]} (相似度: {face[2]:.4f})")
|
||
|
||
# 清理測試數據
|
||
print("\n4. 清理測試數據...")
|
||
cursor.execute(
|
||
"DELETE FROM face_detections WHERE video_uuid = 'direct_test_video';"
|
||
)
|
||
cursor.execute("DELETE FROM face_identities WHERE face_id LIKE 'test_%';")
|
||
conn.commit()
|
||
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
print("✅ 測試數據清理完成")
|
||
|
||
# 清理文件
|
||
os.remove(test_image_path)
|
||
if os.path.exists("/tmp/test_embedding.npy"):
|
||
os.remove("/tmp/test_embedding.npy")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("✅ 直接測試完成！所有功能正常工作。")
|
||
print("=" * 60)
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 測試失敗: {e}")
|
||
import traceback
|
||
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def test_mps_performance():
|
||
"""測試 MPS 性能"""
|
||
print("\n" + "=" * 60)
|
||
print("MPS 性能測試")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
import time
|
||
from scripts.face_recognition_processor import FaceRecognitionProcessor
|
||
|
||
# 創建測試圖像
|
||
test_image = np.random.randint(0, 255, (640, 480, 3), dtype=np.uint8)
|
||
|
||
# 測試 CPU
|
||
print("測試 CPU 性能...")
|
||
cpu_processor = FaceRecognitionProcessor()
|
||
cpu_processor.load_models(use_mps=False)
|
||
|
||
start_time = time.time()
|
||
cpu_detections = cpu_processor.detect_faces(test_image)
|
||
cpu_time = time.time() - start_time
|
||
|
||
print(f"✅ CPU 檢測時間: {cpu_time:.3f} 秒")
|
||
print(f"✅ CPU 檢測到 {len(cpu_detections)} 個人臉")
|
||
|
||
# 測試 MPS（如果可用）
|
||
print("\n測試 MPS 性能...")
|
||
try:
|
||
mps_processor = FaceRecognitionProcessor()
|
||
mps_processor.load_models(use_mps=True)
|
||
|
||
start_time = time.time()
|
||
mps_detections = mps_processor.detect_faces(test_image)
|
||
mps_time = time.time() - start_time
|
||
|
||
print(f"✅ MPS 檢測時間: {mps_time:.3f} 秒")
|
||
print(f"✅ MPS 檢測到 {len(mps_detections)} 個人臉")
|
||
|
||
if mps_time > 0:
|
||
speedup = cpu_time / mps_time
|
||
print(f"✅ MPS 加速比: {speedup:.2f}x")
|
||
else:
|
||
print("⚠️ MPS 時間測量不準確")
|
||
|
||
except Exception as e:
|
||
print(f"⚠️ MPS 測試失敗: {e}")
|
||
print("⚠️ 回退到 CPU 模式")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("✅ 性能測試完成")
|
||
print("=" * 60)
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 性能測試失敗: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主測試函數"""
|
||
print("人臉識別系統直接測試")
|
||
print("=" * 60)
|
||
|
||
# 測試直接處理
|
||
if not test_direct_face_processing():
|
||
print("❌ 直接處理測試失敗")
|
||
return 1
|
||
|
||
# 測試 MPS 性能
|
||
if not test_mps_performance():
|
||
print("⚠️ 性能測試有問題，但系統仍可工作")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("🎉 所有測試完成！系統功能正常。")
|
||
print("=" * 60)
|
||
print("\n系統狀態:")
|
||
print(" ✅ 人臉註冊功能")
|
||
print(" ✅ 人臉檢測功能")
|
||
print(" ✅ 數據庫操作")
|
||
print(" ✅ MPS 加速支援")
|
||
print("\n下一步:")
|
||
print("1. 配置 API 密鑰進行 HTTP 測試")
|
||
print("2. 使用實際視頻進行測試")
|
||
print("3. 部署到生產環境")
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|