Face_reg_app/FaceFeatureExtractorAPI/examples/api_client_example.py

285 lines
8.5 KiB
Python

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
API客户端调用示例
演示如何使用三个核心接口
"""
import requests
import json
import os
from typing import List, Dict, Optional
# API基础地址
API_BASE = "http://localhost:8000"
class FaceAPIClient:
"""人脸特征API客户端"""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url.rstrip('/')
def extract_feature(self, image_path: str) -> Optional[List[float]]:
"""
接口1: 提取人脸特征
Args:
image_path: 图像文件路径
Returns:
特征向量列表,失败返回None
"""
url = f"{self.base_url}/api/extract_feature"
try:
with open(image_path, "rb") as f:
files = {"image": f}
response = requests.post(url, files=files)
response.raise_for_status()
result = response.json()
if result["success"]:
print(f"✓ 特征提取成功")
print(f" - 特征维度: {result['feature_dim']}")
print(f" - 质量检查: {'通过' if result['quality_passed'] else '未通过'}")
print(f" - 处理时间: {result['processing_time']:.3f}")
return result["feature"]
else:
print(f"✗ 特征提取失败: {result['message']}")
return None
except FileNotFoundError:
print(f"✗ 文件不存在: {image_path}")
return None
except Exception as e:
print(f"✗ 请求失败: {e}")
return None
def compare_features(self, feature1: List[float], feature2: List[float]) -> Optional[Dict]:
"""
接口2: 对比两个特征向量
Args:
feature1: 特征向量1
feature2: 特征向量2
Returns:
对比结果字典,失败返回None
"""
url = f"{self.base_url}/api/compare_features"
try:
data = {
"feature1": feature1,
"feature2": feature2
}
response = requests.post(url, json=data)
response.raise_for_status()
result = response.json()
if result["success"]:
print(f"✓ 特征对比成功")
print(f" - 相似度: {result['similarity']:.4f}")
print(f" - 判断结果: {'同一人' if result['is_same_person'] else '不是同一人'}")
print(f" - 阈值: {result['threshold']}")
return result
else:
print(f"✗ 特征对比失败: {result['message']}")
return None
except Exception as e:
print(f"✗ 请求失败: {e}")
return None
def compare_image_feature(self, image_path: str, feature: List[float]) -> Optional[Dict]:
"""
接口3: 对比图像和特征向量
Args:
image_path: 图像文件路径
feature: 目标特征向量
Returns:
对比结果字典,失败返回None
"""
url = f"{self.base_url}/api/compare_image_feature"
try:
with open(image_path, "rb") as f:
files = {"image": f}
data = {"feature": json.dumps(feature)}
response = requests.post(url, files=files, data=data)
response.raise_for_status()
result = response.json()
if result["success"]:
print(f"✓ 图像特征对比成功")
print(f" - 相似度: {result['similarity']:.4f}")
print(f" - 判断结果: {'同一人' if result['is_same_person'] else '不是同一人'}")
print(f" - 处理时间: {result['processing_time']:.3f}")
return result
else:
print(f"✗ 图像特征对比失败: {result['message']}")
return None
except FileNotFoundError:
print(f"✗ 文件不存在: {image_path}")
return None
except Exception as e:
print(f"✗ 请求失败: {e}")
return None
def health_check(self) -> bool:
"""健康检查"""
url = f"{self.base_url}/health"
try:
response = requests.get(url)
result = response.json()
if result["status"] == "healthy":
print(f"✓ API服务运行正常")
return True
else:
print(f"✗ API服务异常: {result}")
return False
except Exception as e:
print(f"✗ 无法连接到API服务: {e}")
return False
# ==================== 示例场景 ====================
def example1_extract_and_compare():
"""示例1: 提取特征并对比"""
print("\n" + "="*60)
print("示例1: 提取两张图像的特征并对比")
print("="*60)
client = FaceAPIClient()
# 检查服务状态
if not client.health_check():
return
# 图像路径(请替换为实际路径)
image1 = "person1.jpg"
image2 = "person1_verify.jpg"
print(f"\n步骤1: 提取第一张图像特征")
feature1 = client.extract_feature(image1)
if not feature1:
return
print(f"\n步骤2: 提取第二张图像特征")
feature2 = client.extract_feature(image2)
if not feature2:
return
print(f"\n步骤3: 对比两个特征")
result = client.compare_features(feature1, feature2)
def example2_image_feature_compare():
"""示例2: 使用组合接口对比图像和特征"""
print("\n" + "="*60)
print("示例2: 使用组合接口对比图像和特征")
print("="*60)
client = FaceAPIClient()
# 先提取一个已知用户的特征
known_user_image = "person1.jpg"
test_image = "person1_verify.jpg"
print(f"\n步骤1: 提取已知用户的特征")
known_feature = client.extract_feature(known_user_image)
if not known_feature:
return
print(f"\n步骤2: 使用组合接口验证新图像")
result = client.compare_image_feature(test_image, known_feature)
def example3_face_recognition_system():
"""示例3: 简单的人脸识别系统"""
print("\n" + "="*60)
print("示例3: 简单的人脸识别系统")
print("="*60)
client = FaceAPIClient()
# 模拟用户数据库
user_database = {}
# 注册用户
users_to_register = [
("张三", "person1.jpg"),
# ("李四", "person2.jpg"),
# ("王五", "person3.jpg"),
]
print("\n[注册阶段]")
for user_name, image_path in users_to_register:
if not os.path.exists(image_path):
print(f"⚠ 跳过 {user_name}: 文件不存在 {image_path}")
continue
print(f"\n注册用户: {user_name}")
feature = client.extract_feature(image_path)
if feature:
user_database[user_name] = feature
print(f"{user_name} 注册成功")
else:
print(f"{user_name} 注册失败")
print(f"\n当前数据库中有 {len(user_database)} 个用户")
# 识别测试
test_images = [
"test_person1.jpg",
]
print("\n[识别阶段]")
for test_image in test_images:
if not os.path.exists(test_image):
print(f"⚠ 跳过: 文件不存在 {test_image}")
continue
print(f"\n识别图像: {test_image}")
# 提取待识别图像的特征
test_feature = client.extract_feature(test_image)
if not test_feature:
continue
# 与数据库中所有用户对比
best_match = None
best_similarity = 0.0
for user_name, user_feature in user_database.items():
result = client.compare_features(test_feature, user_feature)
if result and result['similarity'] > best_similarity:
best_similarity = result['similarity']
best_match = user_name
# 输出结果
if best_similarity >= 0.7:
print(f"✓ 识别结果: {best_match} (相似度: {best_similarity:.4f})")
else:
print(f"✗ 未识别到已注册用户 (最高相似度: {best_similarity:.4f})")
def main():
"""主函数"""
print("人脸特征API客户端示例")
print("请确保API服务正在运行: python app.py")
# 运行示例
# example1_extract_and_compare()
# example2_image_feature_compare()
example3_face_recognition_system()
if __name__ == "__main__":
main()