Warren
e75c4d6f07
- Remove session-ses_2f27.md (161KB raw session log) - Remove 49 ROOT_* duplicate files across REFERENCE/ - Remove 14 duplicate files between REFERENCE/ root and history/ - Remove asr_legacy.rs (dead code, replaced by asr.rs) - Remove src/core/worker/ (duplicate JobWorker) - Remove src/core/layers/ (empty directory) - Remove 4 .bak files in src/ - Remove 7 dead private methods in worker/processor.rs - Remove backup directory from git tracking
291 lines
9.0 KiB
Python
291 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
直接測試人臉識別功能（不通過 HTTP API）
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import numpy as np
|
|
import cv2
|
|
|
|
# 添加項目根目錄到 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("✅ 人臉註冊成功")
|
|
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(" ✅ 數據庫統計:")
|
|
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())
|