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
368 lines
11 KiB
Python
368 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
測試人臉識別完整集成流程
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import numpy as np
|
||
from pathlib import Path
|
||
|
||
# 添加項目根目錄到 Python 路徑
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
# 測試數據庫連接
|
||
try:
|
||
import psycopg2
|
||
from psycopg2.extras import Json
|
||
|
||
print("✅ psycopg2 已安裝")
|
||
except ImportError:
|
||
print("❌ psycopg2 未安裝，請運行: pip install psycopg2-binary")
|
||
sys.exit(1)
|
||
|
||
# 測試 InsightFace
|
||
try:
|
||
import insightface
|
||
|
||
print("✅ insightface 已安裝")
|
||
except ImportError:
|
||
print("❌ insightface 未安裝，請運行: pip install insightface")
|
||
sys.exit(1)
|
||
|
||
# 測試 ONNX Runtime
|
||
try:
|
||
import onnxruntime as ort
|
||
|
||
print("✅ onnxruntime 已安裝")
|
||
|
||
# 檢查可用的執行提供者
|
||
available_providers = ort.get_available_providers()
|
||
print(f"✅ 可用的執行提供者: {available_providers}")
|
||
|
||
# 檢查 MPS 支援
|
||
if "CoreMLExecutionProvider" in available_providers:
|
||
print("✅ CoreML (MPS) 支援可用")
|
||
elif "CUDAExecutionProvider" in available_providers:
|
||
print("✅ CUDA 支援可用")
|
||
else:
|
||
print("⚠️ 僅 CPU 支援可用")
|
||
|
||
except ImportError:
|
||
print("❌ onnxruntime 未安裝，請運行: pip install onnxruntime")
|
||
sys.exit(1)
|
||
|
||
|
||
def test_database_connection():
|
||
"""測試數據庫連接"""
|
||
print("\n=== 測試數據庫連接 ===")
|
||
try:
|
||
conn = psycopg2.connect(
|
||
host="localhost",
|
||
port=5432,
|
||
database="momentry",
|
||
user="accusys",
|
||
password="accusys",
|
||
)
|
||
|
||
cursor = conn.cursor()
|
||
|
||
# 檢查表是否存在
|
||
cursor.execute("""
|
||
SELECT table_name
|
||
FROM information_schema.tables
|
||
WHERE table_schema = 'public'
|
||
AND table_name LIKE 'face_%'
|
||
ORDER BY table_name;
|
||
""")
|
||
|
||
tables = cursor.fetchall()
|
||
print(f"✅ 找到 {len(tables)} 個人臉相關表:")
|
||
for table in tables:
|
||
print(f" - {table[0]}")
|
||
|
||
# 檢查函數是否存在
|
||
cursor.execute("""
|
||
SELECT proname
|
||
FROM pg_proc
|
||
WHERE proname LIKE '%face%'
|
||
ORDER BY proname;
|
||
""")
|
||
|
||
functions = cursor.fetchall()
|
||
print(f"✅ 找到 {len(functions)} 個人臉相關函數:")
|
||
for func in functions:
|
||
print(f" - {func[0]}")
|
||
|
||
# 測試插入和查詢
|
||
cursor.execute("""
|
||
SELECT find_or_create_face_identity(
|
||
'integration_test_001',
|
||
'Integration Test Person',
|
||
NULL,
|
||
'{"age": 25, "gender": "female", "test": true}'::jsonb,
|
||
'{"source": "integration_test"}'::jsonb
|
||
) AS identity_id;
|
||
""")
|
||
|
||
identity_id = cursor.fetchone()[0]
|
||
print(f"✅ 成功創建人臉身份，ID: {identity_id}")
|
||
|
||
# 檢查插入的數據
|
||
cursor.execute("""
|
||
SELECT id, face_id, name, attributes->>'gender' as gender,
|
||
attributes->>'age' as age, attributes->>'test' as test
|
||
FROM face_identities
|
||
WHERE face_id = 'integration_test_001';
|
||
""")
|
||
|
||
result = cursor.fetchone()
|
||
print(
|
||
f"✅ 查詢結果: ID={result[0]}, FaceID={result[1]}, Name={result[2]}, Gender={result[3]}, Age={result[4]}, Test={result[5]}"
|
||
)
|
||
|
||
# 清理測試數據
|
||
cursor.execute(
|
||
"DELETE FROM face_identities WHERE face_id = 'integration_test_001';"
|
||
)
|
||
conn.commit()
|
||
print("✅ 清理測試數據完成")
|
||
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 數據庫連接測試失敗: {e}")
|
||
return False
|
||
|
||
|
||
def test_insightface_model():
|
||
"""測試 InsightFace 模型"""
|
||
print("\n=== 測試 InsightFace 模型 ===")
|
||
|
||
try:
|
||
# 創建測試圖像（隨機數據）
|
||
test_image = np.random.randint(0, 255, (640, 480, 3), dtype=np.uint8)
|
||
print(f"✅ 創建測試圖像: {test_image.shape}")
|
||
|
||
# 初始化模型
|
||
print("正在初始化 InsightFace 模型...")
|
||
model = insightface.app.FaceAnalysis(name="buffalo_l")
|
||
model.prepare(ctx_id=-1) # -1 表示 CPU
|
||
|
||
print("✅ InsightFace 模型初始化成功")
|
||
|
||
# 測試模型推理（使用隨機圖像）
|
||
print("正在進行模型推理測試...")
|
||
faces = model.get(test_image)
|
||
print(f"✅ 模型推理完成，檢測到 {len(faces)} 個人臉")
|
||
|
||
if len(faces) > 0:
|
||
face = faces[0]
|
||
print(f"✅ 人臉屬性:")
|
||
print(f" - 邊界框: {face.bbox}")
|
||
print(f" - 置信度: {face.det_score:.4f}")
|
||
print(
|
||
f" - 嵌入向量維度: {face.embedding.shape if hasattr(face, 'embedding') else 'N/A'}"
|
||
)
|
||
|
||
if hasattr(face, "age"):
|
||
print(f" - 年齡: {face.age}")
|
||
if hasattr(face, "gender"):
|
||
print(f" - 性別: {face.gender}")
|
||
if hasattr(face, "pose"):
|
||
print(f" - 姿態: {face.pose}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ InsightFace 模型測試失敗: {e}")
|
||
return False
|
||
|
||
|
||
def test_face_processor():
|
||
"""測試人臉處理器"""
|
||
print("\n=== 測試人臉處理器 ===")
|
||
|
||
try:
|
||
# 導入處理器
|
||
from scripts.face_recognition_processor import FaceRecognitionProcessor
|
||
|
||
# 初始化處理器
|
||
print("正在初始化人臉識別處理器...")
|
||
processor = FaceRecognitionProcessor()
|
||
|
||
# 加載模型（不使用 MPS）
|
||
print("正在加載模型...")
|
||
processor.load_models(use_mps=False)
|
||
|
||
print(f"✅ 處理器初始化成功")
|
||
print(f" - 啟用識別: {processor.enable_recognition}")
|
||
print(f" - 啟用追蹤: {processor.enable_tracking}")
|
||
print(f" - 啟用聚類: {processor.enable_clustering}")
|
||
|
||
# 創建測試視頻數據
|
||
test_video_data = {
|
||
"video_path": "/tmp/test_video.mp4", # 虛擬路徑
|
||
"video_uuid": "test_video_001",
|
||
"frame_count": 100,
|
||
"fps": 30.0,
|
||
}
|
||
|
||
print(f"✅ 創建測試視頻數據: {test_video_data['video_uuid']}")
|
||
|
||
# 測試處理器方法
|
||
print("測試處理器方法...")
|
||
|
||
# 測試人臉檢測
|
||
test_image = np.random.randint(0, 255, (640, 480, 3), dtype=np.uint8)
|
||
detections = processor.detect_faces(test_image)
|
||
print(f"✅ 人臉檢測測試完成，檢測到 {len(detections)} 個人臉")
|
||
|
||
if len(detections) > 0:
|
||
detection = detections[0]
|
||
print(f"✅ 檢測結果示例:")
|
||
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)}")
|
||
else:
|
||
print(f" - 嵌入向量: N/A")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ 人臉處理器測試失敗: {e}")
|
||
import traceback
|
||
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def test_api_endpoints():
|
||
"""測試 API 端點配置"""
|
||
print("\n=== 測試 API 端點配置 ===")
|
||
|
||
try:
|
||
# 檢查 Rust 代碼編譯
|
||
print("檢查 Rust 代碼編譯狀態...")
|
||
|
||
# 讀取 API 代碼
|
||
api_file = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"src",
|
||
"api",
|
||
"face_recognition.rs",
|
||
)
|
||
|
||
if os.path.exists(api_file):
|
||
with open(api_file, "r") as f:
|
||
content = f.read()
|
||
|
||
# 檢查關鍵函數是否存在（根據實際代碼）
|
||
endpoints = [
|
||
"register_face_api",
|
||
"recognize_faces",
|
||
"search_faces",
|
||
"get_face_details",
|
||
"list_faces",
|
||
"delete_face",
|
||
"get_recognition_results",
|
||
"store_recognition_results",
|
||
]
|
||
|
||
found_endpoints = []
|
||
for endpoint in endpoints:
|
||
if endpoint in content:
|
||
found_endpoints.append(endpoint)
|
||
|
||
print(f"✅ 找到 {len(found_endpoints)}/{len(endpoints)} 個 API 端點:")
|
||
for endpoint in found_endpoints:
|
||
print(f" - {endpoint}")
|
||
|
||
if len(found_endpoints) == len(endpoints):
|
||
print("✅ 所有 API 端點都已定義")
|
||
else:
|
||
missing = set(endpoints) - set(found_endpoints)
|
||
print(f"⚠️ 缺少端點: {missing}")
|
||
|
||
# 檢查路由配置
|
||
server_file = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"src",
|
||
"api",
|
||
"server.rs",
|
||
)
|
||
|
||
if os.path.exists(server_file):
|
||
with open(server_file, "r") as f:
|
||
content = f.read()
|
||
|
||
if "face_recognition" in content and "merge" in content:
|
||
print("✅ API 路由已正確配置")
|
||
else:
|
||
print("⚠️ API 路由配置可能不完整")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"❌ API 端點測試失敗: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主測試函數"""
|
||
print("=" * 60)
|
||
print("人臉識別集成測試")
|
||
print("=" * 60)
|
||
|
||
tests = [
|
||
("數據庫連接", test_database_connection),
|
||
("InsightFace 模型", test_insightface_model),
|
||
("人臉處理器", test_face_processor),
|
||
("API 端點配置", test_api_endpoints),
|
||
]
|
||
|
||
results = []
|
||
|
||
for test_name, test_func in tests:
|
||
try:
|
||
success = test_func()
|
||
results.append((test_name, success))
|
||
except Exception as e:
|
||
print(f"❌ {test_name} 測試異常: {e}")
|
||
results.append((test_name, False))
|
||
|
||
print("\n" + "=" * 60)
|
||
print("測試結果摘要")
|
||
print("=" * 60)
|
||
|
||
passed = 0
|
||
for test_name, success in results:
|
||
status = "✅ 通過" if success else "❌ 失敗"
|
||
print(f"{test_name}: {status}")
|
||
if success:
|
||
passed += 1
|
||
|
||
print(f"\n總計: {passed}/{len(results)} 個測試通過")
|
||
|
||
if passed == len(results):
|
||
print("\n🎉 所有測試通過！人臉識別集成準備就緒。")
|
||
return 0
|
||
else:
|
||
print(f"\n⚠️ 有 {len(results) - passed} 個測試失敗，請檢查問題。")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|