first commit

This commit is contained in:
2025-09-07 21:48:21 +08:00
commit 4e0c898cef
13 changed files with 3116 additions and 0 deletions
+94
View File
@@ -0,0 +1,94 @@
# 腾讯云人脸识别Home Assistant插件
这是一个基于腾讯云人脸识别API的Home Assistant插件,提供了人脸搜索功能。
## 功能特性
- **人脸搜索**:支持通过图片URL或本地文件路径搜索人脸,返回匹配的人员信息
## 安装
1.`custom_components/tencent_face_recognition`目录复制到Home Assistant的`custom_components`目录下
2. 重启Home Assistant
3. 在"配置" -> "集成"中点击"+"添加集成
4. 搜索"腾讯云人脸识别"并点击
5. 输入您的腾讯云凭据和配置信息
## 配置
### 必需配置
- **Secret ID**:腾讯云API的Secret ID
- **Secret Key**:腾讯云API的Secret Key
### 可选配置
- **人员库ID**:默认使用的人员库ID(默认值:Hass)
- **区域**:腾讯云服务区域(默认值:ap-shanghai
- **名称**:集成名称(默认值:腾讯云人脸识别)
## 服务
插件提供以下服务,可以通过开发者工具中的"服务"选项卡调用:
### 人脸搜索
**服务名称**`tencent_face_recognition.face_search`
**描述**:在人员库中搜索人脸
**参数**
- `image_url`(可选):要搜索的图片URL
- `image_path`(可选):要搜索的本地图片路径
- `person_group_id`(可选):要搜索的人员库ID,留空使用默认人员库
- `max_face_num`(可选,默认1):最多处理的人脸数量
- `min_face_size`(可选,默认34):最小人脸尺寸(像素)
- `max_user_num`(可选,默认5):最多返回的匹配人员数量
- `quality_control`(可选,默认1):是否进行质量控制(0:不控制,1:低质量控制,2:高质量控制)
- `need_rotate_check`(可选,默认1):是否进行旋转检查(0:不检查,1:检查)
- `face_match_threshold`(可选,默认60.0):人脸匹配阈值(0-100)
**示例**
``` yaml
action: tencent_face_recognition.face_search
response_variable: face_recognition_result_raw
data:
group_id: Hass
face_match_threshold: 60
min_face_size: 34
max_face_num: 10
max_user_num: 10
image_path: /config/www/camera/face.jpg
```
```yaml
service: tencent_face_recognition.face_search
data:
image_url: "https://example.com/face.jpg"
group_id: "Hass"
```
## 故障排除
### 常见问题
1. **配置失败**:请检查腾讯云凭据是否正确,以及是否有足够的权限
2. **图片处理失败**:请确保图片URL可访问或本地文件路径正确
3. **人脸检测失败**:请确保图片中包含清晰的人脸,且人脸尺寸足够大
4. **API调用失败**:请检查腾讯云账户余额是否充足,以及API调用配额是否足够
### 日志查看
在Home Assistant的"开发者工具" -> "日志"中查看插件日志,可以获取更多错误信息。
## 贡献
欢迎提交Issue和Pull Request来改进这个插件。
## 许可证
MIT许可证
+103
View File
@@ -0,0 +1,103 @@
"""
腾讯云人脸识别Home Assistant插件
"""
import logging
from typing import Dict, Any
from homeassistant.components.http import StaticPathConfig
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
DOMAIN,
PLATFORMS,
CONF_SECRET_ID,
CONF_SECRET_KEY,
CONF_REGION,
CONF_PERSON_GROUP_ID
)
from .tencent_cloud_client import TencentCloudClient
from .face_recognition import FaceRecognition
from .services import async_setup_services, async_unload_services
_LOGGER = logging.getLogger(__name__)
# 添加传感器平台
PLATFORMS.append("sensor")
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""设置配置项"""
_LOGGER.info("设置腾讯云人脸识别插件")
# 获取配置
secret_id = entry.data.get(CONF_SECRET_ID)
secret_key = entry.data.get(CONF_SECRET_KEY)
region = entry.data.get(CONF_REGION, "ap-shanghai")
person_group_id = entry.data.get(CONF_PERSON_GROUP_ID, "Hass")
# 创建腾讯云客户端
try:
client = TencentCloudClient(secret_id, secret_key, region)
except Exception as ex:
_LOGGER.error("创建腾讯云客户端失败: %s", ex)
raise ConfigEntryNotReady(f"创建腾讯云客户端失败: {ex}")
# 创建人脸识别实例
face_recognition = FaceRecognition(hass, client)
# 存储配置和实例
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
"entry": entry,
"client": client,
"face_recognition": face_recognition,
"secret_id": secret_id,
"secret_key": secret_key,
"region": region,
"person_group_id": person_group_id
}
# 设置服务
await async_setup_services(hass)
# 设置平台
# 转发配置项到平台
for platform in PLATFORMS:
hass.async_create_task(
hass.config_entries.async_forward_entry_setups(entry, [platform])
)
# 不再注册前端面板,改为使用配置流管理
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""卸载配置项"""
_LOGGER.info("卸载腾讯云人脸识别插件")
# 注销自定义面板
# 注意:在新版本的 Home Assistant 中,面板注册方式已更改
# 不需要手动注销面板,系统会自动处理
# 卸载平台
unload_ok = True
for platform in PLATFORMS:
result = await hass.config_entries.async_forward_entry_unload(entry, platform)
unload_ok = unload_ok and result
if unload_ok:
# 卸载服务
await async_unload_services(hass)
# 移除数据
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
async def async_remove_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""移除配置项"""
_LOGGER.info("移除腾讯云人脸识别插件配置")
+445
View File
@@ -0,0 +1,445 @@
"""腾讯云人脸识别插件配置流程"""
import logging
import re
from typing import Dict, Any, Optional, List
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.core import callback
from homeassistant.const import CONF_NAME
from homeassistant.helpers.selector import selector
from .const import (
DOMAIN,
CONF_SECRET_ID,
CONF_SECRET_KEY,
CONF_REGION,
CONF_PERSON_GROUP_ID,
DEFAULT_REGION,
ERROR_INVALID_CONFIG,
ERROR_API_ERROR,
)
from .errors import (
validate_config,
InvalidConfigError,
AuthenticationError,
NetworkError,
RateLimitError,
log_and_raise_error
)
from .tencent_cloud_client import TencentCloudClient
_LOGGER = logging.getLogger(__name__)
# 腾讯云区域列表
AVAILABLE_REGIONS = [
"ap-beijing", "ap-shanghai", "ap-guangzhou", "ap-chengdu", "ap-chongqing", "ap-nanjing", "ap-hangzhou"
]
# Secret ID 格式验证
SECRET_ID_REGEX = re.compile(r'^AKID[a-zA-Z0-9\-]{36,}$')
# Secret Key 格式验证
SECRET_KEY_REGEX = re.compile(r'^[a-zA-Z0-9+/]{20,}$')
# 人员库ID格式验证
PERSON_GROUP_ID_REGEX = re.compile(r'^[a-zA-Z0-9\-_]{1,64}$')
class TencentFaceRecognitionConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""腾讯云人脸识别配置流程"""
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
def __init__(self):
"""初始化配置流程"""
self._client = None
self._config_info = {}
async def async_step_user(self, user_input: Optional[Dict[str, Any]] = None):
"""处理用户配置步骤"""
errors = {}
if user_input is not None:
# 验证配置
try:
# 验证输入格式
self._validate_input_format(user_input, errors)
if errors:
return self._show_user_form(errors)
# 基本配置验证
validate_config(user_input)
# 尝试创建腾讯云客户端以验证凭据
validation_result = await self._validate_tencent_cloud_credentials(user_input)
if not validation_result["success"]:
errors = validation_result["errors"]
return self._show_user_form(errors)
# 保存配置并创建配置项
return self.async_create_entry(
title=user_input.get(CONF_NAME, "腾讯云人脸识别"),
data=user_input
)
except InvalidConfigError as ex:
_LOGGER.error("配置验证失败: %s", ex)
errors["base"] = str(ex)
except Exception as ex:
_LOGGER.error("配置验证失败: %s", ex, exc_info=True)
errors["base"] = f"配置验证失败: {str(ex)}"
# 显示配置表单
return self._show_user_form(errors)
def _show_user_form(self, errors: Dict[str, str]) -> dict:
"""显示用户配置表单"""
return self.async_show_form(
step_id="user",
data_schema=vol.Schema({
vol.Required(CONF_SECRET_ID): vol.All(str, vol.Length(min=1)),
vol.Required(CONF_SECRET_KEY): vol.All(str, vol.Length(min=1)),
vol.Optional(CONF_REGION, default=DEFAULT_REGION): selector({
"select": {
"options": [{"label": region, "value": region} for region in AVAILABLE_REGIONS],
"mode": "dropdown"
}
}),
vol.Optional(CONF_PERSON_GROUP_ID, default="Hass"): vol.All(
str,
vol.Length(min=1, max=64)
),
vol.Optional(CONF_NAME, default="腾讯云人脸识别"): vol.All(str, vol.Length(min=1, max=50)),
}),
errors=errors,
description_placeholders={
"secret_id_example": "AKIDxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"secret_key_example": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
"person_group_id_example": "Hass",
"region_example": "ap-shanghai"
}
)
def _validate_input_format(self, user_input: Dict[str, Any], errors: Dict[str, str]) -> None:
"""验证输入格式"""
# 验证Secret ID格式
secret_id = user_input.get(CONF_SECRET_ID, "")
if not SECRET_ID_REGEX.match(secret_id):
errors[CONF_SECRET_ID] = "Secret ID格式不正确,应以AKID开头,后跟36个以上字符"
# 验证Secret Key格式
secret_key = user_input.get(CONF_SECRET_KEY, "")
if not SECRET_KEY_REGEX.match(secret_key):
errors[CONF_SECRET_KEY] = "Secret Key格式不正确,应为20个以上字符"
# 验证区域
region = user_input.get(CONF_REGION, DEFAULT_REGION)
if region not in AVAILABLE_REGIONS:
errors[CONF_REGION] = f"无效的区域,请从列表中选择有效的区域"
# 验证人员库ID格式
person_group_id = user_input.get(CONF_PERSON_GROUP_ID, "Hass")
if not PERSON_GROUP_ID_REGEX.match(person_group_id):
errors[CONF_PERSON_GROUP_ID] = "人员库ID只能包含字母、数字、连字符和下划线,长度1-64个字符"
# 验证名称
name = user_input.get(CONF_NAME, "腾讯云人脸识别")
if len(name) > 50:
errors[CONF_NAME] = "名称长度不能超过50个字符"
async def _validate_tencent_cloud_credentials(self, user_input: Dict[str, Any]) -> Dict[str, Any]:
"""验证腾讯云凭据"""
try:
_LOGGER.info("开始验证腾讯云凭据")
# 创建腾讯云客户端
self._client = TencentCloudClient(
user_input[CONF_SECRET_ID],
user_input[CONF_SECRET_KEY],
user_input.get(CONF_REGION, DEFAULT_REGION)
)
_LOGGER.info("腾讯云客户端创建成功,凭据验证通过")
return {"success": True}
except AuthenticationError as ex:
_LOGGER.error("腾讯云认证失败: %s", ex)
return {
"success": False,
"errors": {
CONF_SECRET_ID: "认证失败,请检查Secret ID",
CONF_SECRET_KEY: "认证失败,请检查Secret Key",
"base": "腾讯云认证失败,请检查Secret ID和Secret Key是否正确"
}
}
except NetworkError as ex:
_LOGGER.error("腾讯云网络错误: %s", ex)
return {
"success": False,
"errors": {
CONF_REGION: "网络错误,请检查区域设置和网络连接",
"base": "连接腾讯云失败,请检查网络连接和区域设置"
}
}
except RateLimitError as ex:
_LOGGER.error("腾讯云速率限制错误: %s", ex)
return {
"success": False,
"errors": {
"base": "腾讯云API调用频率过高,请稍后再试"
}
}
except Exception as ex:
_LOGGER.error("腾讯云凭据验证失败: %s", ex, exc_info=True)
return {
"success": False,
"errors": {
"base": f"腾讯云API验证失败: {str(ex)}"
}
}
@staticmethod
@callback
def async_get_options_flow(config_entry):
"""获取选项流程"""
return TencentFaceRecognitionOptionsFlow(config_entry)
class TencentFaceRecognitionOptionsFlow(config_entries.OptionsFlow):
"""腾讯云人脸识别选项流程"""
def __init__(self, config_entry):
"""初始化选项流程"""
# 存储需要的数据而不是整个 config_entry 实例
self._entry_id = config_entry.entry_id
self._title = config_entry.title
self._data = dict(config_entry.data)
self._options = dict(config_entry.options)
async def async_step_init(self, user_input=None):
"""管理选项流程"""
if user_input is not None:
action = user_input.get("action")
if action == "config":
return await self.async_step_config()
elif action == "test_connection":
return await self.async_step_test_connection()
elif action == "reauth":
return await self.async_step_reauth()
return self.async_create_entry(title="", data=user_input)
# 显示管理菜单
return self.async_show_form(
step_id="init",
data_schema=vol.Schema({
vol.Required("action"): selector({
"select": {
"options": [
{"label": "配置设置", "value": "config"},
{"label": "测试连接", "value": "test_connection"},
{"label": "重新认证", "value": "reauth"}
]
}
})
}),
description_placeholders={
"description": "请选择要管理的项目"
}
)
async def async_step_config(self, user_input=None):
"""配置设置步骤"""
errors = {}
if user_input is not None:
# 验证区域
region = user_input.get(CONF_REGION)
if region not in AVAILABLE_REGIONS:
errors[CONF_REGION] = f"无效的区域,请从列表中选择有效的区域"
if not errors:
# 验证新配置是否有效
try:
# 创建临时客户端进行验证
temp_client = TencentCloudClient(
self._data.get(CONF_SECRET_ID),
self._data.get(CONF_SECRET_KEY),
region
)
_LOGGER.info("配置验证成功")
# 保存配置
return self.async_create_entry(title="", data=user_input)
except AuthenticationError as ex:
_LOGGER.error("腾讯云认证失败: %s", ex)
errors[CONF_REGION] = "认证失败,请检查区域设置"
errors["base"] = "腾讯云认证失败,请检查区域设置"
except NetworkError as ex:
_LOGGER.error("腾讯云网络错误: %s", ex)
errors[CONF_REGION] = "网络错误,请检查区域设置"
errors["base"] = "连接腾讯云失败,请检查网络连接和区域设置"
except Exception as ex:
_LOGGER.error("配置验证失败: %s", ex, exc_info=True)
errors["base"] = f"配置验证失败: {str(ex)}"
options = {
vol.Optional(
CONF_REGION,
default=self._options.get(CONF_REGION, self._data.get(CONF_REGION, DEFAULT_REGION))
): selector({
"select": {
"options": [{"label": region, "value": region} for region in AVAILABLE_REGIONS],
"mode": "dropdown"
}
}),
vol.Optional(
CONF_PERSON_GROUP_ID,
default=self._options.get(CONF_PERSON_GROUP_ID, self._data.get(CONF_PERSON_GROUP_ID, "Hass"))
): str,
}
return self.async_show_form(
step_id="config",
data_schema=vol.Schema(options),
errors=errors,
)
async def async_step_test_connection(self, user_input=None):
"""测试连接步骤"""
errors = {}
connection_status = "未知"
if user_input is None:
# 尝试测试连接
try:
# 使用当前配置创建客户端
client = TencentCloudClient(
self._data.get(CONF_SECRET_ID),
self._data.get(CONF_SECRET_KEY),
self._data.get(CONF_REGION, DEFAULT_REGION)
)
# 使用 GetGroupInfo 测试连接
group_id = self._data.get(CONF_PERSON_GROUP_ID, "Hass")
success = await self.hass.async_add_executor_job(
client._test_api_connection,
group_id
)
if not success:
raise Exception("连接测试失败,请检查人员库ID是否正确")
connection_status = "连接成功"
_LOGGER.info("连接测试成功")
except AuthenticationError as ex:
connection_status = f"认证失败: {str(ex)}"
errors["base"] = "认证失败,请检查Secret ID和Secret Key"
_LOGGER.error("连接测试认证失败: %s", ex)
except NetworkError as ex:
connection_status = f"网络错误: {str(ex)}"
errors["base"] = "网络错误,请检查网络连接和区域设置"
_LOGGER.error("连接测试网络失败: %s", ex)
except Exception as ex:
connection_status = f"测试失败: {str(ex)}"
errors["base"] = f"连接测试失败: {str(ex)}"
_LOGGER.error("连接测试失败: %s", ex, exc_info=True)
return self.async_show_form(
step_id="test_connection",
data_schema=vol.Schema({}),
errors=errors,
description_placeholders={
"connection_status": connection_status
}
)
async def async_step_reauth(self, user_input=None):
"""重新认证步骤"""
errors = {}
if user_input is not None:
# 验证新凭据
try:
# 验证输入格式
temp_errors = {}
self._validate_input_format(user_input, temp_errors)
if temp_errors:
errors = temp_errors
return self.async_show_form(
step_id="reauth",
data_schema=vol.Schema({
vol.Required(CONF_SECRET_ID, default=self._data.get(CONF_SECRET_ID, "")): str,
vol.Required(CONF_SECRET_KEY, default=self._data.get(CONF_SECRET_KEY, "")): str,
}),
errors=errors,
)
# 创建腾讯云客户端验证新凭据
new_client = TencentCloudClient(
user_input[CONF_SECRET_ID],
user_input[CONF_SECRET_KEY],
self._data.get(CONF_REGION, DEFAULT_REGION)
)
_LOGGER.info("重新认证成功")
# 更新配置
new_data = dict(self._data)
new_data.update(user_input)
# 重新创建配置项
self.hass.config_entries.async_update_entry(
self.hass.config_entries.async_get_entry(self._entry_id),
data=new_data
)
return self.async_create_entry(title="", data={})
except AuthenticationError as ex:
_LOGGER.error("重新认证失败: %s", ex)
errors["base"] = "认证失败,请检查Secret ID和Secret Key"
except NetworkError as ex:
_LOGGER.error("重新认证网络错误: %s", ex)
errors["base"] = "网络错误,请检查网络连接和区域设置"
except Exception as ex:
_LOGGER.error("重新认证失败: %s", ex, exc_info=True)
errors["base"] = f"重新认证失败: {str(ex)}"
return self.async_show_form(
step_id="reauth",
data_schema=vol.Schema({
vol.Required(CONF_SECRET_ID, default=self._data.get(CONF_SECRET_ID, "")): str,
vol.Required(CONF_SECRET_KEY, default=self._data.get(CONF_SECRET_KEY, "")): str,
}),
errors=errors,
description_placeholders={
"current_secret_id": self._data.get(CONF_SECRET_ID, ""),
"current_region": self._data.get(CONF_REGION, DEFAULT_REGION)
}
)
def _validate_input_format(self, user_input: Dict[str, Any], errors: Dict[str, str]) -> None:
"""验证输入格式"""
# 验证Secret ID格式
secret_id = user_input.get(CONF_SECRET_ID, "")
if not SECRET_ID_REGEX.match(secret_id):
errors[CONF_SECRET_ID] = "Secret ID格式不正确,应以AKID开头,后跟36个以上字符"
# 验证Secret Key格式
secret_key = user_input.get(CONF_SECRET_KEY, "")
if not SECRET_KEY_REGEX.match(secret_key):
errors[CONF_SECRET_KEY] = "Secret Key格式不正确,应为20个以上字符"
+55
View File
@@ -0,0 +1,55 @@
"""腾讯云人脸识别插件常量定义"""
DOMAIN = "tencent_face_recognition"
PLATFORMS = []
# 配置项键
CONF_SECRET_ID = "secret_id"
CONF_SECRET_KEY = "secret_key"
CONF_REGION = "region"
CONF_PERSON_GROUP_ID = "person_group_id"
# 默认值
DEFAULT_REGION = "ap-shanghai"
DEFAULT_PERSON_GROUP_ID = "Hass"
DEFAULT_MAX_FACE_NUM = 1
DEFAULT_MIN_FACE_SIZE = 34
DEFAULT_MAX_USER_NUM = 5
DEFAULT_QUALITY_CONTROL = 1
DEFAULT_NEED_ROTATE_CHECK = 1
DEFAULT_FACE_MATCH_THRESHOLD = 60.0
# 服务
SERVICE_FACE_SEARCH = "face_search"
SERVICE_DETECT_FACE = "detect_face"
SERVICE_GET_FACE_ATTRIBUTES = "get_face_attributes"
# 服务参数
ATTR_IMAGE_URL = "image_url"
ATTR_IMAGE_PATH = "image_path"
ATTR_IMAGE_FILE = "image_file"
ATTR_MAX_FACE_NUM = "max_face_num"
ATTR_MIN_FACE_SIZE = "min_face_size"
ATTR_MAX_USER_NUM = "max_user_num"
ATTR_QUALITY_CONTROL = "quality_control"
ATTR_NEED_ROTATE_CHECK = "need_rotate_check"
ATTR_FACE_MATCH_THRESHOLD = "face_match_threshold"
# 错误消息
ERROR_INVALID_CONFIG = "无效的配置"
ERROR_API_ERROR = "API调用错误"
ERROR_IMAGE_NOT_FOUND = "图片未找到"
ERROR_FACE_NOT_DETECTED = "未检测到人脸"
# 状态
STATE_CONNECTED = "已连接"
STATE_DISCONNECTED = "未连接"
STATE_CONNECTING = "连接中"
STATE_ERROR = "连接错误"
STATE_PARTIALLY_CONNECTED = "部分连接"
# 日志级别
LOG_LEVEL_DEBUG = "DEBUG"
LOG_LEVEL_INFO = "INFO"
LOG_LEVEL_WARNING = "WARNING"
LOG_LEVEL_ERROR = "ERROR"
+384
View File
@@ -0,0 +1,384 @@
"""错误处理模块"""
import logging
from typing import Dict, Any, Optional
from homeassistant.exceptions import HomeAssistantError
_LOGGER = logging.getLogger(__name__)
# 错误消息本地化
ERROR_MESSAGES = {
"zh_CN": {
# 配置错误
"missing_config": "缺少必需的配置项: {key}",
"invalid_secret_id": "Secret ID 必须是非空字符串,且应以 AKID 开头",
"invalid_secret_key": "Secret Key 必须是非空字符串",
"invalid_region": "区域必须是非空字符串,且为有效的腾讯云区域",
"invalid_person_group_id": "人员库ID必须是非空字符串,只能包含字母、数字、连字符和下划线",
# 图片错误
"image_not_found": "图片未找到: {path}",
"image_too_large": "图片文件过大: {path}, 大小: {size} 字节,最大支持: {max_size}",
"image_format_not_supported": "不支持的图片格式: {format}, 支持的格式: {supported_formats}",
"image_decode_failed": "图片解码失败,请确保图片格式正确",
"image_download_timeout": "下载图片超时: {url}",
"image_download_error": "下载图片失败: {url}, 错误: {error}",
"invalid_image_url": "无效的图片URL: {url}",
"permission_denied": "没有权限读取图片文件: {path}",
# 人脸检测错误
"face_not_detected": "未检测到人脸,请确保图片中包含清晰的人脸",
"face_too_small": "人脸太小,请使用包含更大人脸的图片",
"face_quality_poor": "人脸质量较差,请使用更清晰的图片",
# API错误
"api_error": "API调用失败: {code} - {message}",
"person_not_found": "人员未找到: {person_id}",
"group_not_found": "人员库未找到: {group_id}",
"face_not_found": "人脸未找到: {face_id}",
# 网络错误
"network_error": "网络连接错误,请检查网络连接",
"timeout_error": "请求超时,请稍后重试",
# 认证错误
"auth_failure": "认证失败,请检查Secret ID和Secret Key是否正确",
"auth_expired": "认证已过期,请更新凭证",
# 限制错误
"rate_limit_exceeded": "API调用频率超限,请稍后重试",
"quota_exceeded": "API调用配额已用完,请等待配额重置或升级套餐",
# 通用错误
"unknown_error": "未知错误: {message}",
"invalid_parameter": "无效参数: {parameter}",
},
"en_US": {
# Configuration errors
"missing_config": "Missing required configuration: {key}",
"invalid_secret_id": "Secret ID must be a non-empty string starting with AKID",
"invalid_secret_key": "Secret Key must be a non-empty string",
"invalid_region": "Region must be a non-empty string and a valid Tencent Cloud region",
"invalid_person_group_id": "Person Group ID must be a non-empty string containing only letters, numbers, hyphens, and underscores",
# Image errors
"image_not_found": "Image not found: {path}",
"image_too_large": "Image file too large: {path}, size: {size} bytes, maximum supported: {max_size}",
"image_format_not_supported": "Unsupported image format: {format}, supported formats: {supported_formats}",
"image_decode_failed": "Image decode failed, please ensure the image format is correct",
"image_download_timeout": "Image download timeout: {url}",
"image_download_error": "Image download failed: {url}, error: {error}",
"invalid_image_url": "Invalid image URL: {url}",
"permission_denied": "Permission denied when reading image file: {path}",
# Face detection errors
"face_not_detected": "No face detected, please ensure the image contains clear faces",
"face_too_small": "Face too small, please use an image with larger faces",
"face_quality_poor": "Poor face quality, please use a clearer image",
# API errors
"api_error": "API call failed: {code} - {message}",
"person_not_found": "Person not found: {person_id}",
"group_not_found": "Person group not found: {group_id}",
"face_not_found": "Face not found: {face_id}",
# Network errors
"network_error": "Network connection error, please check your network connection",
"timeout_error": "Request timeout, please try again later",
# Authentication errors
"auth_failure": "Authentication failed, please check if Secret ID and Secret Key are correct",
"auth_expired": "Authentication expired, please update credentials",
# Limit errors
"rate_limit_exceeded": "API call rate limit exceeded, please try again later",
"quota_exceeded": "API call quota used up, please wait for quota reset or upgrade plan",
# General errors
"unknown_error": "Unknown error: {message}",
"invalid_parameter": "Invalid parameter: {parameter}",
}
}
# 错误解决建议
ERROR_SOLUTIONS = {
"zh_CN": {
"missing_config": "请在配置中添加缺少的配置项",
"invalid_secret_id": "请检查Secret ID格式,应以AKID开头",
"invalid_secret_key": "请检查Secret Key格式,确保输入正确",
"invalid_region": "请选择有效的腾讯云区域,如ap-shanghai",
"invalid_person_group_id": "请使用字母、数字、连字符和下划线组合",
"image_not_found": "请检查图片路径是否正确",
"image_too_large": "请使用小于10MB的图片,或对图片进行压缩",
"image_format_not_supported": "请将图片转换为JPG、PNG、BMP或GIF格式",
"image_decode_failed": "请确保图片文件未损坏,并尝试重新保存",
"image_download_timeout": "请检查网络连接,或尝试使用其他图片URL",
"image_download_error": "请检查图片URL是否有效,或稍后重试",
"invalid_image_url": "请提供有效的图片URL",
"permission_denied": "请检查文件权限,或使用其他图片",
"face_not_detected": "请使用包含清晰人脸的图片",
"face_too_small": "请使用人脸更大的图片,或调整拍摄距离",
"face_quality_poor": "请使用光线充足、对焦清晰的图片",
"api_error": "请稍后重试,如果问题持续,请联系技术支持",
"person_not_found": "请检查人员ID是否正确,或确保人员已添加到人员库",
"group_not_found": "请检查人员库ID是否正确,或创建新的人员库",
"face_not_found": "请检查人脸ID是否正确",
"network_error": "请检查网络连接,确保网络通畅",
"timeout_error": "请稍后重试,或在网络状况良好时使用",
"auth_failure": "请检查Secret ID和Secret Key是否正确",
"auth_expired": "请更新腾讯云凭证",
"rate_limit_exceeded": "请降低调用频率,稍后重试",
"quota_exceeded": "请等待配额重置,或升级腾讯云套餐",
"unknown_error": "请记录错误信息并联系技术支持",
"invalid_parameter": "请检查API调用参数是否正确",
},
"en_US": {
"missing_config": "Please add the missing configuration item",
"invalid_secret_id": "Please check the Secret ID format, it should start with AKID",
"invalid_secret_key": "Please check the Secret Key format, ensure it's entered correctly",
"invalid_region": "Please select a valid Tencent Cloud region, such as ap-shanghai",
"invalid_person_group_id": "Please use a combination of letters, numbers, hyphens, and underscores",
"image_not_found": "Please check if the image path is correct",
"image_too_large": "Please use an image smaller than 10MB, or compress the image",
"image_format_not_supported": "Please convert the image to JPG, PNG, BMP, or GIF format",
"image_decode_failed": "Please ensure the image file is not corrupted and try saving it again",
"image_download_timeout": "Please check your network connection, or try using a different image URL",
"image_download_error": "Please check if the image URL is valid, or try again later",
"invalid_image_url": "Please provide a valid image URL",
"permission_denied": "Please check file permissions, or use a different image",
"face_not_detected": "Please use an image that contains clear faces",
"face_too_small": "Please use an image with larger faces, or adjust the shooting distance",
"face_quality_poor": "Please use a well-lit and focused image",
"api_error": "Please try again later, if the problem persists, please contact technical support",
"person_not_found": "Please check if the person ID is correct, or ensure the person has been added to the person group",
"group_not_found": "Please check if the person group ID is correct, or create a new person group",
"face_not_found": "Please check if the face ID is correct",
"network_error": "Please check your network connection, ensure it's working properly",
"timeout_error": "Please try again later, or use when network conditions are good",
"auth_failure": "Please check if Secret ID and Secret Key are correct",
"auth_expired": "Please update Tencent Cloud credentials",
"rate_limit_exceeded": "Please reduce call frequency and try again later",
"quota_exceeded": "Please wait for quota reset, or upgrade Tencent Cloud plan",
"unknown_error": "Please record the error information and contact technical support",
"invalid_parameter": "Please check if the API call parameters are correct",
}
}
class TencentFaceRecognitionError(HomeAssistantError):
"""腾讯云人脸识别基础错误类"""
def __init__(
self,
message: str,
error_key: Optional[str] = None,
error_details: Optional[Dict[str, Any]] = None,
locale: str = "zh_CN"
):
"""初始化错误"""
super().__init__(message)
self.error_key = error_key
self.error_details = error_details or {}
self.locale = locale
# 获取本地化消息
self.localized_message = self._get_localized_message()
# 获取解决建议
self.solution = self._get_solution()
def _get_localized_message(self) -> str:
"""获取本地化错误消息"""
if not self.error_key:
return str(self)
messages = ERROR_MESSAGES.get(self.locale, ERROR_MESSAGES["zh_CN"])
template = messages.get(self.error_key, str(self))
try:
# 使用错误详情填充模板
return template.format(**self.error_details)
except (KeyError, ValueError):
# 如果填充失败,返回原始消息
return str(self)
def _get_solution(self) -> str:
"""获取错误解决建议"""
if not self.error_key:
return ""
solutions = ERROR_SOLUTIONS.get(self.locale, ERROR_SOLUTIONS["zh_CN"])
return solutions.get(self.error_key, "")
class InvalidConfigError(TencentFaceRecognitionError):
"""无效配置错误"""
pass
class ApiError(TencentFaceRecognitionError):
"""API调用错误"""
pass
class ImageNotFoundError(TencentFaceRecognitionError):
"""图片未找到错误"""
pass
class FaceNotDetectedError(TencentFaceRecognitionError):
"""未检测到人脸错误"""
pass
class PersonNotFoundError(TencentFaceRecognitionError):
"""人员未找到错误"""
pass
class PersonGroupNotFoundError(TencentFaceRecognitionError):
"""人员库未找到错误"""
pass
class FaceNotFoundError(TencentFaceRecognitionError):
"""人脸未找到错误"""
pass
class NetworkError(TencentFaceRecognitionError):
"""网络错误"""
pass
class AuthenticationError(TencentFaceRecognitionError):
"""认证错误"""
pass
class RateLimitError(TencentFaceRecognitionError):
"""速率限制错误"""
pass
class QuotaExceededError(TencentFaceRecognitionError):
"""配额超限错误"""
pass
def handle_api_error(error_code: str, error_message: str) -> None:
"""处理API错误"""
error_mapping = {
"FailedOperation.ImageDecodeFailed": (ImageNotFoundError, "image_decode_failed"),
"InvalidParameterValue.NoFaceInPhoto": (FaceNotDetectedError, "face_not_detected"),
"FailedOperation.PersonNotFound": (PersonNotFoundError, "person_not_found"),
"FailedOperation.GroupNotFound": (PersonGroupNotFoundError, "group_not_found"),
"FailedOperation.FaceNotFound": (FaceNotFoundError, "face_not_found"),
"AuthFailure": (AuthenticationError, "auth_failure"),
"RequestLimitExceeded": (RateLimitError, "rate_limit_exceeded"),
"RequestQuotaExceeded": (QuotaExceededError, "quota_exceeded"),
"ResourceUnavailable.NetworkError": (NetworkError, "network_error"),
}
error_info = error_mapping.get(error_code, (ApiError, "api_error"))
error_class = error_info[0]
error_key = error_info[1]
_LOGGER.error("腾讯云API错误: %s - %s", error_code, error_message)
# 创建带有本地化消息的错误
error_details = {
"code": error_code,
"message": error_message
}
raise error_class(
f"{error_code}: {error_message}",
error_key=error_key,
error_details=error_details
)
def log_and_raise_error(error_class, message: str, details: Dict[str, Any] = None, error_key: Optional[str] = None) -> None:
"""记录日志并抛出错误"""
if details:
_LOGGER.error("%s: %s, 详情: %s", error_class.__name__, message, details)
else:
_LOGGER.error("%s: %s", error_class.__name__, message)
# 创建带有本地化消息的错误
raise error_class(
message,
error_key=error_key,
error_details=details or {}
)
def validate_config(config: Dict[str, Any]) -> None:
"""验证配置"""
required_keys = ["secret_id", "secret_key"]
for key in required_keys:
if not config.get(key):
log_and_raise_error(
InvalidConfigError,
f"缺少必需的配置项: {key}",
{"key": key},
"missing_config"
)
# 验证Secret ID格式
secret_id = config.get("secret_id")
if not isinstance(secret_id, str) or len(secret_id.strip()) == 0:
log_and_raise_error(
InvalidConfigError,
"Secret ID必须是非空字符串,且应以 AKID 开头",
{"secret_id": str(secret_id)},
"invalid_secret_id"
)
# 验证Secret Key格式
secret_key = config.get("secret_key")
if not isinstance(secret_key, str) or len(secret_key.strip()) == 0:
log_and_raise_error(
InvalidConfigError,
"Secret Key必须是非空字符串",
{"secret_key": str(secret_key)},
"invalid_secret_key"
)
# 验证区域格式
region = config.get("region", "ap-shanghai")
if not isinstance(region, str) or len(region.strip()) == 0:
log_and_raise_error(
InvalidConfigError,
"区域必须是非空字符串,且为有效的腾讯云区域",
{"region": str(region)},
"invalid_region"
)
# 验证人员库ID格式
person_group_id = config.get("person_group_id", "Hass")
if not isinstance(person_group_id, str) or len(person_group_id.strip()) == 0:
log_and_raise_error(
InvalidConfigError,
"人员库ID必须是非空字符串,只能包含字母、数字、连字符和下划线",
{"person_group_id": str(person_group_id)},
"invalid_person_group_id"
)
_LOGGER.debug("配置验证通过")
+208
View File
@@ -0,0 +1,208 @@
"""人脸识别核心功能"""
import logging
import json
from typing import Dict, Any, List, Optional
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .const import (
ATTR_IMAGE_URL,
ATTR_IMAGE_PATH,
ATTR_MAX_FACE_NUM,
ATTR_MIN_FACE_SIZE,
ATTR_MAX_USER_NUM,
ATTR_QUALITY_CONTROL,
ATTR_NEED_ROTATE_CHECK,
ATTR_FACE_MATCH_THRESHOLD,
ERROR_IMAGE_NOT_FOUND,
ERROR_FACE_NOT_DETECTED,
)
from .tencent_cloud_client import TencentCloudClient
_LOGGER = logging.getLogger(__name__)
class FaceRecognition:
"""人脸识别核心功能"""
def __init__(self, hass: HomeAssistant, client: TencentCloudClient):
"""初始化人脸识别功能"""
self.hass = hass
self.client = client
async def async_search_faces(
self,
image_url: str = None,
image_path: str = None,
image_file: str = None,
group_ids: List[str] = None,
max_face_num: int = 1,
min_face_size: int = 34,
max_user_num: int = 5,
quality_control: int = 1,
need_rotate_check: int = 1,
face_match_threshold: float = 60.0
) -> List[Dict[str, Any]]:
"""搜索人脸"""
try:
# 验证参数
self._validate_image_params(image_url, image_path, image_file)
# 调用腾讯云API搜索人脸
results = await self.hass.async_add_executor_job(
self.client.search_faces,
image_url,
image_path,
image_file,
group_ids,
max_face_num,
min_face_size,
max_user_num,
quality_control,
need_rotate_check,
face_match_threshold
)
# 处理结果
return self._process_search_results(results)
except ValueError as ex:
_LOGGER.error("人脸搜索参数错误: %s", ex)
raise HomeAssistantError(str(ex))
except Exception as ex:
_LOGGER.error("人脸搜索失败: %s", ex)
raise HomeAssistantError(f"人脸搜索失败: {str(ex)}")
def _validate_image_params(self, image_url: str, image_path: str, image_file: str) -> None:
"""验证图片参数"""
if not image_url and not image_path and not image_file:
raise ValueError("必须提供image_url、image_path或image_file")
def _process_search_results(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""处理人脸搜索结果"""
processed_results = []
for result in results:
processed_result = {
"face_id": result.get("face_id"),
"candidates": []
}
for candidate in result.get("candidates", []):
processed_candidate = {
"person_id": candidate.get("person_id"),
"person_name": candidate.get("person_name"),
"score": candidate.get("score"),
"person_tag": candidate.get("person_tag")
}
processed_result["candidates"].append(processed_candidate)
processed_results.append(processed_result)
return processed_results
def _process_detect_results(self, results: List[Dict[str, Any]], include_attributes: bool = False) -> List[Dict[str, Any]]:
"""处理人脸检测结果"""
processed_results = []
for result in results:
face_result = {
"face_id": result.get("face_id", ""),
"x": result.get("x", 0),
"y": result.get("y", 0),
"width": result.get("width", 0),
"height": result.get("height", 0),
"face_rect": result.get("face_rect")
}
# 如果需要包含属性信息
if include_attributes:
face_result.update({
"gender": result.get("gender"),
"age": result.get("age"),
"expression": result.get("expression"),
"beauty": result.get("beauty")
})
processed_results.append(face_result)
return processed_results
def _handle_api_error(self, operation: str, ex: Exception) -> None:
"""处理API错误"""
_LOGGER.error("%s失败: %s", operation, ex)
if isinstance(ex, ValueError):
raise HomeAssistantError(str(ex))
else:
raise HomeAssistantError(f"{operation}失败: {str(ex)}")
def _get_config_entry(self):
"""获取配置项"""
from homeassistant.config_entries import ConfigEntry
from . import DOMAIN
for entry in self.hass.config_entries.async_entries(DOMAIN):
return entry
return None
async def async_get_face_attributes(
self,
image_url: str = None,
image_path: str = None,
image_file: str = None,
max_face_num: int = 1,
need_rotate_check: int = 1
) -> List[Dict[str, Any]]:
"""获取人脸属性"""
try:
# 验证参数
self._validate_image_params(image_url, image_path, image_file)
# 调用腾讯云API获取人脸属性
results = await self.hass.async_add_executor_job(
self.client.get_face_attributes,
image_url,
image_path,
image_file,
max_face_num,
need_rotate_check
)
return results
except Exception as ex:
self._handle_api_error("获取人脸属性", ex)
# 移除 _get_face_attributes_sync 方法,因为现在直接使用客户端的 get_face_attributes 方法
async def async_detect_faces(
self,
image_url: str = None,
image_path: str = None,
image_file: str = None,
max_face_num: int = 1,
min_face_size: int = 34,
need_rotate_check: int = 1
) -> List[Dict[str, Any]]:
"""检测人脸"""
try:
# 验证参数
self._validate_image_params(image_url, image_path, image_file)
# 调用腾讯云API检测人脸
results = await self.hass.async_add_executor_job(
self.client.detect_faces,
image_url,
image_path,
image_file,
max_face_num,
min_face_size,
need_rotate_check
)
return results
except Exception as ex:
self._handle_api_error("人脸检测", ex)
# 移除 _detect_faces_sync 方法,因为现在直接使用客户端的 detect_faces 方法
+16
View File
@@ -0,0 +1,16 @@
{
"domain": "tencent_face_recognition",
"name": "腾讯云人脸识别",
"config_flow": true,
"documentation": "https://code.nextrt.com/Hass/tencent_face_recognition",
"issue_tracker": "https://code.nextrt.com/Hass/tencent_face_recognition/issues",
"requirements": [
"tencentcloud-sdk-python-common",
"tencentcloud-sdk-python-iai"
],
"dependencies": [],
"codeowners": ["@xyzmos"],
"version": "1.0.0",
"iot_class": "cloud_polling",
"resources": []
}
+2
View File
@@ -0,0 +1,2 @@
tencentcloud-sdk-python-common>=3.0.0
tencentcloud-sdk-python-iai>=3.0.0
+206
View File
@@ -0,0 +1,206 @@
"""传感器实体"""
import logging
import base64
from typing import Dict, Any, Optional
from datetime import datetime, timedelta
from homeassistant.components.sensor import SensorEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_time_interval
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""设置传感器实体"""
_LOGGER.debug("设置腾讯云人脸识别传感器实体")
# 获取配置项
name = entry.data.get(CONF_NAME, "腾讯云人脸识别")
# 创建传感器实体
async_add_entities(
[
TencentFaceRecognitionStatusSensor(hass, entry, name),
]
)
# 获取人员列表并创建实体
client = hass.data[DOMAIN][entry.entry_id]["client"]
person_group_id = entry.data.get("person_group_id", "Hass")
try:
persons = await hass.async_add_executor_job(
client.get_person_list, person_group_id
)
for person in persons:
async_add_entities(
[
TencentFaceRecognitionPersonSensor(hass, entry, person),
]
)
except Exception as ex:
_LOGGER.error("获取人员列表失败: %s", ex)
class TencentFaceRecognitionPersonSensor(SensorEntity):
"""腾讯云人脸识别人员传感器"""
_attr_icon = "mdi:account"
def __init__(self, hass: HomeAssistant, entry: ConfigEntry, person: Dict[str, Any]):
"""初始化传感器"""
self.hass = hass
self._entry = entry
self._person = person
self._attr_name = person.get("person_name", "未知人员")
self._attr_unique_id = f"{entry.entry_id}_{person.get('person_id')}"
self._state = "未知"
self._extra_state_attributes = {
"person_id": person.get("person_id"),
"gender": "" if person.get("gender") == 1 else "",
"face_ids": person.get("face_ids"),
}
@property
def state(self) -> str:
"""返回传感器状态"""
return self._state
@property
def extra_state_attributes(self) -> Dict[str, Any]:
"""返回传感器属性"""
return self._extra_state_attributes
class TencentFaceRecognitionStatusSensor(SensorEntity):
"""腾讯云人脸识别状态传感器"""
_attr_icon = "mdi:face-recognition"
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, hass: HomeAssistant, entry: ConfigEntry, name: str):
"""初始化传感器"""
self.hass = hass
self._entry = entry
self._name = name
self._attr_name = f"{name} 状态"
self._attr_unique_id = f"{entry.entry_id}_status"
self._state = "未连接"
self._extra_state_attributes = {
"last_updated": None,
"api_status": "未知",
"client_status": "未知",
"last_test_time": None,
"error_count": 0,
"success_count": 0,
"last_error": None
}
self._client = None
self._last_successful_test = None
@property
def state(self) -> str:
"""返回传感器状态"""
return self._state
@property
def extra_state_attributes(self) -> Dict[str, Any]:
"""返回传感器属性"""
return self._extra_state_attributes
async def async_added_to_hass(self) -> None:
"""当实体添加到Home Assistant时调用"""
# 设置定期更新
async_track_time_interval(
self.hass,
self._async_update,
timedelta(minutes=5) # 每5分钟更新一次
)
# 立即执行一次更新
await self._async_update(None)
async def _async_update(self, _) -> None:
"""更新传感器状态"""
await self.async_update()
async def async_update(self) -> None:
"""更新传感器状态"""
try:
# 获取插件实例
entry_data = self.hass.data[DOMAIN].get(self._entry.entry_id)
if not entry_data:
self._state = "未初始化"
self._extra_state_attributes["api_status"] = "插件未正确初始化"
self._extra_state_attributes["client_status"] = "未初始化"
return
# 获取客户端实例
client = entry_data.get("client")
face_recognition = entry_data.get("face_recognition")
if not client or not face_recognition:
self._state = "未初始化"
self._extra_state_attributes["api_status"] = "客户端未正确初始化"
self._extra_state_attributes["client_status"] = "未初始化"
return
# 更新客户端状态
self._client = client
self._extra_state_attributes["client_status"] = "已初始化"
# 使用内置的测试图片进行连接测试
await self._test_api_connection(face_recognition)
except Exception as ex:
_LOGGER.error("更新腾讯云人脸识别状态传感器失败: %s", ex, exc_info=True)
self._state = "连接错误"
self._extra_state_attributes["api_status"] = f"错误: {str(ex)}"
self._extra_state_attributes["last_error"] = str(ex)
self._extra_state_attributes["error_count"] = self._extra_state_attributes.get("error_count", 0) + 1
# 更新最后更新时间
self._extra_state_attributes["last_updated"] = datetime.now().isoformat()
async def _test_api_connection(self, face_recognition) -> None:
"""测试API连接"""
try:
group_id = self._entry.data.get("person_group_id", "Hass")
# 使用 GetGroupInfo 测试连接
group_info = await self.hass.async_add_executor_job(
self._client.get_group_info,
group_id
)
if group_info:
self._state = "已连接"
self._extra_state_attributes["api_status"] = "正常"
self._extra_state_attributes["success_count"] = self._extra_state_attributes.get("success_count", 0) + 1
self._last_successful_test = datetime.now()
self._extra_state_attributes["last_test_time"] = self._last_successful_test.isoformat()
if "last_error" in self._extra_state_attributes:
self._extra_state_attributes["last_error"] = None
else:
self._state = "连接错误"
self._extra_state_attributes["api_status"] = "获取人员库信息失败"
except Exception as ex:
error_msg = str(ex)
self._state = "连接错误"
self._extra_state_attributes["api_status"] = f"连接失败: {error_msg}"
self._extra_state_attributes["last_error"] = error_msg
self._extra_state_attributes["error_count"] = self._extra_state_attributes.get("error_count", 0) + 1
_LOGGER.debug("API连接测试失败: %s", error_msg)
+258
View File
@@ -0,0 +1,258 @@
"""服务定义"""
import logging
from typing import Dict, Any, List, Optional
import voluptuous as vol
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.helpers import config_validation as cv
from .const import (
DOMAIN,
SERVICE_FACE_SEARCH,
SERVICE_DETECT_FACE,
SERVICE_GET_FACE_ATTRIBUTES,
ATTR_IMAGE_URL,
ATTR_IMAGE_FILE,
ATTR_IMAGE_PATH,
ATTR_MAX_FACE_NUM,
ATTR_MIN_FACE_SIZE,
ATTR_MAX_USER_NUM,
ATTR_QUALITY_CONTROL,
ATTR_NEED_ROTATE_CHECK,
ATTR_FACE_MATCH_THRESHOLD,
)
_LOGGER = logging.getLogger(__name__)
# 人脸搜索服务参数
FACE_SEARCH_SCHEMA = vol.Schema({
vol.Required('group_id'): cv.string,
vol.Optional(ATTR_IMAGE_URL): cv.string,
vol.Optional(ATTR_IMAGE_PATH): cv.string,
vol.Optional(ATTR_IMAGE_FILE): cv.string,
vol.Optional(ATTR_MAX_FACE_NUM, default=1): cv.positive_int,
vol.Optional(ATTR_MIN_FACE_SIZE, default=34): cv.positive_int,
vol.Optional(ATTR_MAX_USER_NUM, default=5): cv.positive_int,
vol.Optional(ATTR_QUALITY_CONTROL, default=1): vol.In([0, 1]),
vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]),
vol.Optional(ATTR_FACE_MATCH_THRESHOLD, default=60.0): vol.Range(min=0, max=100),
})
# 人脸检测服务参数
DETECT_FACE_SCHEMA = vol.Schema({
vol.Optional(ATTR_IMAGE_URL): cv.string,
vol.Optional(ATTR_IMAGE_PATH): cv.string,
vol.Optional(ATTR_IMAGE_FILE): cv.string,
vol.Optional(ATTR_MAX_FACE_NUM, default=1): cv.positive_int,
vol.Optional(ATTR_MIN_FACE_SIZE, default=34): cv.positive_int,
vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]),
})
# 获取人脸属性服务参数
GET_FACE_ATTRIBUTES_SCHEMA = vol.Schema({
vol.Optional(ATTR_IMAGE_URL): cv.string,
vol.Optional(ATTR_IMAGE_PATH): cv.string,
vol.Optional(ATTR_IMAGE_FILE): cv.string,
vol.Optional(ATTR_MAX_FACE_NUM, default=1): cv.positive_int,
vol.Optional(ATTR_NEED_ROTATE_CHECK, default=1): vol.In([0, 1]),
})
async def async_setup_services(hass: HomeAssistant) -> None:
"""设置服务"""
_LOGGER.info("设置腾讯云人脸识别服务")
# 注册人脸搜索服务
hass.services.async_register(
DOMAIN,
SERVICE_FACE_SEARCH,
async_face_search_service,
schema=FACE_SEARCH_SCHEMA,
supports_response=True
)
# 注册人脸检测服务
hass.services.async_register(
DOMAIN,
SERVICE_DETECT_FACE,
async_detect_face_service,
schema=DETECT_FACE_SCHEMA
)
# 注册获取人脸属性服务
hass.services.async_register(
DOMAIN,
SERVICE_GET_FACE_ATTRIBUTES,
async_get_face_attributes_service,
schema=GET_FACE_ATTRIBUTES_SCHEMA
)
async def async_unload_services(hass: HomeAssistant) -> None:
"""卸载服务"""
_LOGGER.info("卸载腾讯云人脸识别服务")
# 注销所有服务
hass.services.async_remove(DOMAIN, SERVICE_FACE_SEARCH)
hass.services.async_remove(DOMAIN, SERVICE_DETECT_FACE)
hass.services.async_remove(DOMAIN, SERVICE_GET_FACE_ATTRIBUTES)
async def async_face_search_service(call: ServiceCall) -> Dict[str, Any]:
"""人脸搜索服务"""
from homeassistant.exceptions import HomeAssistantError
hass = call.hass
data = call.data
# 获取配置项
config_entry = _get_config_entry(hass)
if not config_entry:
raise HomeAssistantError("未找到腾讯云人脸识别配置")
# 获取插件实例
entry_data = _get_entry_data(hass)
if not entry_data:
raise HomeAssistantError("腾讯云人脸识别插件未正确初始化")
face_recognition = entry_data.get("face_recognition")
if not face_recognition:
raise HomeAssistantError("腾讯云人脸识别插件未正确初始化")
# 调用人脸搜索功能
try:
group_id = data.get('group_id')
image_url = data.get(ATTR_IMAGE_URL)
image_path = data.get(ATTR_IMAGE_PATH)
image_file = data.get(ATTR_IMAGE_FILE)
max_face_num = data.get(ATTR_MAX_FACE_NUM, 1)
min_face_size = data.get(ATTR_MIN_FACE_SIZE, 34)
max_user_num = data.get(ATTR_MAX_USER_NUM, 5)
quality_control = data.get(ATTR_QUALITY_CONTROL, 1)
need_rotate_check = data.get(ATTR_NEED_ROTATE_CHECK, 1)
face_match_threshold = data.get(ATTR_FACE_MATCH_THRESHOLD, 60.0)
results = await face_recognition.async_search_faces(
image_url=image_url,
image_path=image_path,
image_file=image_file,
group_ids=[group_id],
max_face_num=max_face_num,
min_face_size=min_face_size,
max_user_num=max_user_num,
quality_control=quality_control,
need_rotate_check=need_rotate_check,
face_match_threshold=face_match_threshold
)
return {"results": results}
except Exception as ex:
_LOGGER.error("人脸搜索服务调用失败: %s", ex)
raise HomeAssistantError(f"人脸搜索服务调用失败: {str(ex)}")
async def async_detect_face_service(call: ServiceCall) -> Dict[str, Any]:
"""人脸检测服务"""
from homeassistant.exceptions import HomeAssistantError
hass = call.hass
data = call.data
# 获取配置项
config_entry = _get_config_entry(hass)
if not config_entry:
raise HomeAssistantError("未找到腾讯云人脸识别配置")
# 获取插件实例
entry_data = _get_entry_data(hass)
if not entry_data:
raise HomeAssistantError("腾讯云人脸识别插件未正确初始化")
face_recognition = entry_data.get("face_recognition")
if not face_recognition:
raise HomeAssistantError("腾讯云人脸识别插件未正确初始化")
# 调用人脸检测功能
try:
image_url = data.get(ATTR_IMAGE_URL)
image_path = data.get(ATTR_IMAGE_PATH)
image_file = data.get(ATTR_IMAGE_FILE)
max_face_num = data.get(ATTR_MAX_FACE_NUM, 1)
min_face_size = data.get(ATTR_MIN_FACE_SIZE, 34)
need_rotate_check = data.get(ATTR_NEED_ROTATE_CHECK, 1)
results = await face_recognition.async_detect_faces(
image_url=image_url,
image_path=image_path,
image_file=image_file,
max_face_num=max_face_num,
min_face_size=min_face_size,
need_rotate_check=need_rotate_check
)
return {"results": results}
except Exception as ex:
_LOGGER.error("人脸检测服务调用失败: %s", ex)
raise HomeAssistantError(f"人脸检测服务调用失败: {str(ex)}")
async def async_get_face_attributes_service(call: ServiceCall) -> Dict[str, Any]:
"""获取人脸属性服务"""
from homeassistant.exceptions import HomeAssistantError
hass = call.hass
data = call.data
# 获取配置项
config_entry = _get_config_entry(hass)
if not config_entry:
raise HomeAssistantError("未找到腾讯云人脸识别配置")
# 获取插件实例
entry_data = _get_entry_data(hass)
if not entry_data:
raise HomeAssistantError("腾讯云人脸识别插件未正确初始化")
face_recognition = entry_data.get("face_recognition")
if not face_recognition:
raise HomeAssistantError("腾讯云人脸识别插件未正确初始化")
# 调用获取人脸属性功能
try:
image_url = data.get(ATTR_IMAGE_URL)
image_path = data.get(ATTR_IMAGE_PATH)
image_file = data.get(ATTR_IMAGE_FILE)
max_face_num = data.get(ATTR_MAX_FACE_NUM, 1)
need_rotate_check = data.get(ATTR_NEED_ROTATE_CHECK, 1)
results = await face_recognition.async_get_face_attributes(
image_url=image_url,
image_path=image_path,
image_file=image_file,
max_face_num=max_face_num,
need_rotate_check=need_rotate_check
)
return {"results": results}
except Exception as ex:
_LOGGER.error("获取人脸属性服务调用失败: %s", ex)
raise HomeAssistantError(f"获取人脸属性服务调用失败: {str(ex)}")
def _get_config_entry(hass: HomeAssistant):
"""获取配置项"""
for entry in hass.config_entries.async_entries(DOMAIN):
return entry
return None
def _get_entry_data(hass: HomeAssistant):
"""获取配置项数据"""
for entry_id, data in hass.data.get(DOMAIN, {}).items():
if isinstance(data, dict) and "face_recognition" in data:
return data
return None
+185
View File
@@ -0,0 +1,185 @@
# 腾讯云人脸识别服务定义
face_search:
name: 人脸搜索
description: 在指定的人员库中搜索人脸。
fields:
group_id:
name: 人员库ID
description: 人员库 ID。
required: true
example: "Hass"
selector:
text:
image_url:
name: 图片URL
description: 要搜索的人脸图片的 URL。
example: "https://example.com/image.jpg"
selector:
text:
image_path:
name: 图片路径
description: Home Assistant 本地可访问的人脸图片路径。
example: "/config/www/image.jpg"
selector:
text:
image_file:
name: 图片文件
description: 通过文件上传的人脸图片。
selector:
text:
max_face_num:
name: 最大人脸数量
description: 最多处理的人脸数量。
default: 1
example: 1
selector:
number:
min: 1
max: 10
step: 1
mode: box
min_face_size:
name: 最小人脸尺寸
description: 最小人脸尺寸(像素)。
default: 34
example: 34
selector:
number:
min: 1
max: 200
step: 1
mode: box
max_user_num:
name: 最大返回人员数量
description: 最多返回的匹配人员数量。
default: 5
example: 5
selector:
number:
min: 1
max: 20
step: 1
mode: box
quality_control:
name: 质量控制
description: 是否进行质量控制。
default: true
example: true
selector:
boolean:
need_rotate_check:
name: 旋转检查
description: 是否进行旋转检查。
default: true
example: true
selector:
boolean:
face_match_threshold:
name: 人脸匹配阈值
description: 人脸匹配阈值(0-100)。
default: 60
example: 60
selector:
number:
min: 0
max: 100
step: 1
mode: slider
response:
description: "人脸搜索结果"
fields:
results:
name: "结果"
description: "包含人脸ID和候选人列表的数组"
example: '[{"face_id": "3440118425101275818", "candidates": [{"person_id": "person_1", "person_name": "张三", "score": 99.9, "person_tag": "员工"}]}]'
detect_face:
name: 人脸检测
description: 检测图片中的人脸。
fields:
image_url:
name: 图片URL
description: 要检测的图片URL。
example: "https://example.com/image.jpg"
selector:
text:
image_path:
name: 图片路径
description: 要检测的本地图片路径。
example: "/config/www/image.jpg"
selector:
text:
image_file:
name: 图片文件
description: 要检测的图片文件。
selector:
text:
max_face_num:
name: 最大人脸数量
description: 最多检测的人脸数量。
default: 1
example: 1
selector:
number:
min: 1
max: 10
step: 1
mode: box
min_face_size:
name: 最小人脸尺寸
description: 最小人脸尺寸(像素)。
default: 34
example: 34
selector:
number:
min: 1
max: 200
step: 1
mode: box
need_rotate_check:
name: 旋转检查
description: 是否进行旋转检查。
default: true
example: true
selector:
boolean:
get_face_attributes:
name: 获取人脸属性
description: 获取图片中人脸的属性信息。
fields:
image_url:
name: 图片URL
description: 要分析的图片URL。
example: "https://example.com/image.jpg"
selector:
text:
image_path:
name: 图片路径
description: 要分析的本地图片路径。
example: "/config/www/image.jpg"
selector:
text:
image_file:
name: 图片文件
description: 要分析的图片文件。
selector:
text:
max_face_num:
name: 最大人脸数量
description: 最多分析的人脸数量。
default: 1
example: 1
selector:
number:
min: 1
max: 10
step: 1
mode: box
need_rotate_check:
name: 旋转检查
description: 是否进行旋转检查。
default: true
example: true
selector:
boolean:
+144
View File
@@ -0,0 +1,144 @@
{
"config": {
"step": {
"user": {
"title": "腾讯云人脸识别",
"description": "请输入您的腾讯云人脸识别服务凭据",
"data": {
"secret_id": "Secret ID",
"secret_key": "Secret Key",
"region": "区域",
"person_group_id": "人员组ID",
"name": "名称"
},
"data_description": {
"secret_id": "腾讯云API的Secret ID",
"secret_key": "腾讯云API的Secret Key",
"region": "腾讯云服务区域,例如:ap-shanghai",
"person_group_id": "人脸库人员组ID,用于存储和搜索人脸",
"name": "集成名称"
}
}
},
"error": {
"invalid_config": "无效的配置"
},
"abort": {
"already_configured": "已经配置了腾讯云人脸识别服务"
}
},
"options": {
"step": {
"init": {
"title": "腾讯云人脸识别选项",
"description": "配置腾讯云人脸识别服务选项",
"data": {
"region": "区域"
},
"data_description": {
"region": "腾讯云服务区域,例如:ap-shanghai"
}
}
}
},
"services": {
"face_search": {
"name": "人脸搜索",
"description": "搜索人脸",
"fields": {
"image_url": {
"name": "图片URL",
"description": "要搜索的图片URL"
},
"image_path": {
"name": "图片路径",
"description": "要搜索的本地图片路径"
},
"image_file": {
"name": "图片文件",
"description": "要搜索的图片文件(Base64编码)"
},
"max_face_num": {
"name": "最大人脸数量",
"description": "最多处理的人脸数量"
},
"min_face_size": {
"name": "最小人脸尺寸",
"description": "最小人脸尺寸(像素)"
},
"max_user_num": {
"name": "最大返回人员数量",
"description": "最多返回的匹配人员数量"
},
"quality_control": {
"name": "质量控制",
"description": "是否进行质量控制"
},
"need_rotate_check": {
"name": "旋转检查",
"description": "是否进行旋转检查"
},
"face_match_threshold": {
"name": "人脸匹配阈值",
"description": "人脸匹配阈值(0-100"
}
}
},
"detect_face": {
"name": "人脸检测",
"description": "检测图片中的人脸",
"fields": {
"image_url": {
"name": "图片URL",
"description": "要检测的图片URL"
},
"image_path": {
"name": "图片路径",
"description": "要检测的本地图片路径"
},
"image_file": {
"name": "图片文件",
"description": "要检测的图片文件(Base64编码)"
},
"max_face_num": {
"name": "最大人脸数量",
"description": "最多检测的人脸数量"
},
"min_face_size": {
"name": "最小人脸尺寸",
"description": "最小人脸尺寸(像素)"
},
"need_rotate_check": {
"name": "旋转检查",
"description": "是否进行旋转检查"
}
}
},
"get_face_attributes": {
"name": "获取人脸属性",
"description": "获取图片中人脸的属性信息",
"fields": {
"image_url": {
"name": "图片URL",
"description": "要分析的图片URL"
},
"image_path": {
"name": "图片路径",
"description": "要分析的本地图片路径"
},
"image_file": {
"name": "图片文件",
"description": "要分析的图片文件(Base64编码)"
},
"max_face_num": {
"name": "最大人脸数量",
"description": "最多分析的人脸数量"
},
"need_rotate_check": {
"name": "旋转检查",
"description": "是否进行旋转检查"
}
}
}
}
}
File diff suppressed because it is too large Load Diff