Update inspireface to 1.2.0

This commit is contained in:
Jingyu
2025-03-25 00:51:26 +08:00
parent 977ea6795b
commit ca64996b84
388 changed files with 28584 additions and 13036 deletions

View File

@@ -1,4 +1,5 @@
from .modules import *
from .param import *
__version__ = version()

View File

@@ -1,6 +1,8 @@
from .inspire_face import ImageStream, FaceExtended, FaceInformation, SessionCustomParameter, InspireFaceSession, \
from .inspireface import ImageStream, FaceExtended, FaceInformation, SessionCustomParameter, InspireFaceSession, \
launch, FeatureHubConfiguration, feature_hub_enable, feature_hub_disable, feature_comparison, \
FaceIdentity, feature_hub_set_search_threshold, feature_hub_face_insert, SearchResult, \
feature_hub_face_search, feature_hub_face_search_top_k, feature_hub_face_update, feature_hub_face_remove, \
feature_hub_get_face_identity, feature_hub_get_face_count, view_table_in_terminal, version, \
set_logging_level, disable_logging, show_system_resource_statistics
feature_hub_get_face_identity, feature_hub_get_face_count, view_table_in_terminal, version, query_launch_status, reload, set_expansive_pack_path, \
set_logging_level, disable_logging, show_system_resource_statistics, get_recommended_cosine_threshold, cosine_similarity_convert_to_percentage, \
get_similarity_converter_config, set_similarity_converter_config, pull_latest_model, \
HF_PK_AUTO_INCREMENT, HF_PK_MANUAL_INPUT, HF_SEARCH_MODE_EAGER, HF_SEARCH_MODE_EXHAUSTIVE

View File

@@ -1,12 +1,11 @@
import ctypes
import cv2
import numpy as np
from .core import *
from typing import Tuple, List
from dataclasses import dataclass
from loguru import logger
from .utils import ResourceManager
class ImageStream(object):
"""
@@ -268,6 +267,12 @@ class InspireFaceSession(object):
Raises:
Exception: If session creation fails.
"""
# If InspireFace is not initialized, run launch() use Pikachu model
if not query_launch_status():
ret = launch()
if not ret:
raise Exception("Launch InspireFace failure")
self.multiple_faces = None
self._sess = HFSession()
self.param = param
@@ -330,6 +335,20 @@ class InspireFaceSession(object):
return infos
else:
return []
def get_face_five_key_points(self, single_face: FaceInformation):
num_landmarks = 5
landmarks_array = (HPoint2f * num_landmarks)()
ret = HFGetFaceFiveKeyPointsFromFaceToken(single_face._token, landmarks_array, num_landmarks)
if ret != 0:
logger.error(f"An error occurred obtaining a dense landmark for a single face: {ret}")
landmark = []
for point in landmarks_array:
landmark.append(point.x)
landmark.append(point.y)
return np.asarray(landmark).reshape(-1, 2)
def get_face_dense_landmark(self, single_face: FaceInformation):
num_landmarks = HInt32()
@@ -379,6 +398,21 @@ class InspireFaceSession(object):
if ret != 0:
logger.error(f"Set filter minimum face pixel size error: {ret}")
def set_track_mode_smooth_ratio(self, ratio=0.025):
ret = HFSessionSetTrackModeSmoothRatio(self._sess, ratio)
if ret != 0:
logger.error(f"Set track mode smooth ratio error: {ret}")
def set_track_mode_num_smooth_cache_frame(self, num=15):
ret = HFSessionSetTrackModeNumSmoothCacheFrame(self._sess, num)
if ret != 0:
logger.error(f"Set track mode num smooth cache frame error: {ret}")
def set_track_model_detect_interval(self, num=20):
ret = HFSessionSetTrackModeDetectInterval(self._sess, num)
if ret != 0:
logger.error(f"Set track model detect interval error: {ret}")
def face_pipeline(self, image, faces: List[FaceInformation], exec_param) -> List[FaceExtended]:
"""
Processes detected faces to extract additional attributes based on the provided execution parameters.
@@ -478,22 +512,22 @@ class InspireFaceSession(object):
def _update_face_interact_confidence(self, exec_param, flag, extends):
if (flag == "object" and exec_param.enable_interaction_liveness) or (
flag == "bitmask" and exec_param & HF_ENABLE_INTERACTION):
results = HFFaceIntereactionState()
ret = HFGetFaceIntereactionStateResult(self._sess, PHFFaceIntereactionState(results))
results = HFFaceInteractionState()
ret = HFGetFaceInteractionStateResult(self._sess, PHFFaceInteractionState(results))
if ret == 0:
for i in range(results.num):
extends[i].left_eye_status_confidence = results.leftEyeStatusConfidence[i]
extends[i].right_eye_status_confidence = results.rightEyeStatusConfidence[i]
else:
logger.error(f"Get face interact result error: {ret}")
actions = HFFaceIntereactionsActions()
ret = HFGetFaceIntereactionActionsResult(self._sess, PHFFaceIntereactionsActions(actions))
actions = HFFaceInteractionsActions()
ret = HFGetFaceInteractionActionsResult(self._sess, PHFFaceInteractionsActions(actions))
if ret == 0:
for i in range(results.num):
extends[i].action_normal = actions.normal[i]
extends[i].action_shake = actions.shake[i]
extends[i].action_jaw_open = actions.jawOpen[i]
extends[i].action_head_raise = actions.headRiase[i]
extends[i].action_head_raise = actions.headRaise[i]
extends[i].action_blink = actions.blink[i]
else:
logger.error(f"Get face action result error: {ret}")
@@ -571,12 +605,13 @@ class InspireFaceSession(object):
# == Global API ==
def launch(resource_path: str) -> bool:
def launch(model_name: str = "Pikachu", resource_path: str = None) -> bool:
"""
Launches the InspireFace system with the specified resource directory.
Args:
resource_path (str): The file path to the resource directory necessary for operation.
model_name (str): the name of the model to use.
resource_path (str): if None, use the default model path.
Returns:
bool: True if the system was successfully launched, False otherwise.
@@ -584,6 +619,9 @@ def launch(resource_path: str) -> bool:
Notes:
A specific error is logged if duplicate loading is detected or if there is any other launch failure.
"""
if resource_path is None:
sm = ResourceManager()
resource_path = sm.get_model(model_name)
path_c = String(bytes(resource_path, encoding="utf8"))
ret = HFLaunchInspireFace(path_c)
if ret != 0:
@@ -595,6 +633,48 @@ def launch(resource_path: str) -> bool:
return False
return True
def set_expansive_pack_path(path: str):
path_c = String(bytes(path, encoding="utf8"))
ret = HFSetExpansiveHardwareAppleCoreMLModelPath(path_c)
if ret != 0:
logger.error(f"Set expansive pack path error: {ret}")
return False
return True
def pull_latest_model(model_name: str = "Pikachu") -> str:
sm = ResourceManager()
resource_path = sm.get_model(model_name, re_download=True)
return resource_path
def reload(model_name: str = "Pikachu", resource_path: str = None) -> bool:
if resource_path is None:
sm = ResourceManager()
resource_path = sm.get_model(model_name)
path_c = String(bytes(resource_path, encoding="utf8"))
ret = HFReloadInspireFace(path_c)
if ret != 0:
if ret == 1363:
logger.warning("Duplicate loading was found")
return True
else:
logger.error(f"Launch InspireFace failure: {ret}")
return False
return True
def query_launch_status() -> bool:
"""
Queries the launch status of the InspireFace SDK.
Returns:
bool: True if InspireFace is launched, False otherwise.
"""
status = HInt32()
ret = HFQueryInspireFaceLaunchStatus(byref(status))
if ret != 0:
logger.error(f"Query launch status error: {ret}")
return False
return status.value == 1
@dataclass
class FeatureHubConfiguration:
@@ -608,9 +688,9 @@ class FeatureHubConfiguration:
search_threshold (float): The threshold value for considering a match.
search_mode (int): The mode of searching in the database.
"""
feature_block_num: int
enable_use_db: bool
db_path: str
primary_key_mode: int
enable_persistence: bool
persistence_db_path: str
search_threshold: float
search_mode: int
@@ -622,9 +702,9 @@ class FeatureHubConfiguration:
HFFeatureHubConfiguration: A C-structure for feature hub configuration.
"""
return HFFeatureHubConfiguration(
enablePersistence=int(self.enable_use_db),
dbPath=String(bytes(self.db_path, encoding="utf8")),
featureBlockNum=self.feature_block_num,
primaryKeyMode=self.primary_key_mode,
enablePersistence=int(self.enable_persistence),
persistenceDbPath=String(bytes(self.persistence_db_path, encoding="utf8")),
searchThreshold=self.search_threshold,
searchMode=self.search_mode
)
@@ -714,7 +794,7 @@ class FaceIdentity(object):
_c_struct: Converts the instance back to a compatible C structure.
"""
def __init__(self, data: np.ndarray, custom_id: int, tag: str):
def __init__(self, data: np.ndarray, id: int):
"""
Initializes a new FaceIdentity instance with facial feature data, a custom identifier, and a tag.
@@ -723,9 +803,12 @@ class FaceIdentity(object):
custom_id (int): A custom identifier for tracking or referencing the face identity.
tag (str): A descriptive tag or label for the face identity.
"""
if data.dtype != np.float32:
logger.error("The input data must be in float32 format")
raise ValueError("The input data must be in float32 format")
self.feature = data
self.custom_id = custom_id
self.tag = tag
self.id = id
@staticmethod
def from_ctypes(raw_identity: HFFaceFeatureIdentity):
@@ -741,10 +824,9 @@ class FaceIdentity(object):
feature_size = raw_identity.feature.contents.size
feature_data_ptr = raw_identity.feature.contents.data
feature_data = np.ctypeslib.as_array(cast(feature_data_ptr, HPFloat), (feature_size,))
custom_id = raw_identity.customId
tag = raw_identity.tag.data.decode('utf-8')
id_ = raw_identity.id
return FaceIdentity(data=feature_data, custom_id=custom_id, tag=tag)
return FaceIdentity(data=feature_data, id=id_)
def _c_struct(self):
"""
@@ -758,8 +840,7 @@ class FaceIdentity(object):
feature.size = HInt32(self.feature.size)
feature.data = data_ptr
return HFFaceFeatureIdentity(
customId=self.custom_id,
tag=String(bytes(self.tag, encoding="utf8")),
customId=HFaceId(self.id),
feature=PHFFaceFeature(feature)
)
@@ -774,7 +855,7 @@ def feature_hub_set_search_threshold(threshold: float):
HFFeatureHubFaceSearchThresholdSetting(threshold)
def feature_hub_face_insert(face_identity: FaceIdentity) -> bool:
def feature_hub_face_insert(face_identity: FaceIdentity) -> Tuple[bool, int]:
"""
Inserts a face identity into the FeatureHub database.
@@ -787,11 +868,12 @@ def feature_hub_face_insert(face_identity: FaceIdentity) -> bool:
Notes:
Logs an error if the insertion process fails.
"""
ret = HFFeatureHubInsertFeature(face_identity._c_struct())
alloc_id = HFaceId()
ret = HFFeatureHubInsertFeature(face_identity._c_struct(), HPFaceId(alloc_id))
if ret != 0:
logger.error(f"Failed to insert face feature data into FeatureHub: {ret}")
return False
return True
return False, -1
return True, int(alloc_id.value)
@dataclass
@@ -820,18 +902,21 @@ def feature_hub_face_search(data: np.ndarray) -> SearchResult:
Notes:
If the search operation fails, logs an error and returns a SearchResult with a confidence of -1.
"""
if data.dtype != np.float32:
logger.error("The input data must be in float32 format")
raise ValueError("The input data must be in float32 format")
feature = HFFaceFeature(size=HInt32(data.size), data=data.ctypes.data_as(HPFloat))
confidence = HFloat()
most_similar = HFFaceFeatureIdentity()
ret = HFFeatureHubFaceSearch(feature, HPFloat(confidence), PHFFaceFeatureIdentity(most_similar))
if ret != 0:
logger.error(f"Failed to search face: {ret}")
return SearchResult(confidence=-1, similar_identity=FaceIdentity(np.zeros(0), most_similar.customId, "None"))
if most_similar.customId != -1:
return SearchResult(confidence=-1, similar_identity=FaceIdentity(np.zeros(0), most_similar.id))
if most_similar.id != -1:
search_identity = FaceIdentity.from_ctypes(most_similar)
return SearchResult(confidence=confidence.value, similar_identity=search_identity)
else:
none = FaceIdentity(np.zeros(0), most_similar.customId, "None")
none = FaceIdentity(np.zeros(0, dtype=np.float32), most_similar.id)
return SearchResult(confidence=confidence.value, similar_identity=none)
@@ -849,6 +934,10 @@ def feature_hub_face_search_top_k(data: np.ndarray, top_k: int) -> List[Tuple]:
Notes:
If the search operation fails, an empty list is returned.
"""
if data.dtype != np.float32:
logger.error("The input data must be in float32 format")
raise ValueError("The input data must be in float32 format")
feature = HFFaceFeature(size=HInt32(data.size), data=data.ctypes.data_as(HPFloat))
results = HFSearchTopKResults()
ret = HFFeatureHubFaceSearchTopK(feature, top_k, PHFSearchTopKResults(results))
@@ -856,8 +945,8 @@ def feature_hub_face_search_top_k(data: np.ndarray, top_k: int) -> List[Tuple]:
if ret == 0:
for idx in range(results.size):
confidence = results.confidence[idx]
customId = results.customIds[idx]
outputs.append((confidence, customId))
id_ = results.ids[idx]
outputs.append((confidence, id_))
return outputs
@@ -894,7 +983,7 @@ def feature_hub_face_remove(custom_id: int) -> bool:
Notes:
Logs an error if the removal operation fails.
"""
ret = HFFeatureHubFaceRemove(custom_id)
ret = HFFeatureHubFaceRemove(HFaceId(custom_id))
if ret != 0:
logger.error(f"Failed to remove face feature data from FeatureHub: {ret}")
return False
@@ -915,7 +1004,7 @@ def feature_hub_get_face_identity(custom_id: int):
Logs an error if retrieving the face identity fails.
"""
identify = HFFaceFeatureIdentity()
ret = HFFeatureHubGetFaceIdentity(custom_id, PHFFaceFeatureIdentity(identify))
ret = HFFeatureHubGetFaceIdentity(HFaceId(custom_id), PHFFaceFeatureIdentity(identify))
if ret != 0:
logger.error("Get face identity errors from FeatureHub")
return None
@@ -952,6 +1041,52 @@ def view_table_in_terminal():
if ret != 0:
logger.error(f"Failed to view DB: {ret}")
def get_recommended_cosine_threshold() -> float:
"""
Retrieves the recommended cosine threshold.
"""
threshold = HFloat()
HFGetRecommendedCosineThreshold(threshold)
return float(threshold.value)
def get_similarity_converter_config() -> dict:
"""
Retrieves the similarity converter configuration.
"""
config = HFSimilarityConverterConfig()
ret = HFGetCosineSimilarityConverter(PHFSimilarityConverterConfig(config))
if ret != 0:
logger.error(f"Failed to get cosine similarity converter config: {ret}")
cfg = {
"threshold": config.threshold,
"middleScore": config.middleScore,
"steepness": config.steepness,
"outputMin": config.outputMin,
"outputMax": config.outputMax
}
return cfg
def set_similarity_converter_config(cfg: dict):
"""
Sets the similarity converter configuration.
"""
config = HFSimilarityConverterConfig()
config.threshold = cfg["threshold"]
config.middleScore = cfg["middleScore"]
config.steepness = cfg["steepness"]
config.outputMin = cfg["outputMin"]
config.outputMax = cfg["outputMax"]
HFUpdateCosineSimilarityConverter(config)
def cosine_similarity_convert_to_percentage(similarity: float) -> float:
"""
Converts a cosine similarity score to a percentage similarity score.
"""
result = HFloat()
ret = HFCosineSimilarityConvertToPercentage(HFloat(similarity), HPFloat(result))
if ret != 0:
logger.error(f"Failed to convert cosine similarity to percentage: {ret}")
return float(result.value)
def version() -> str:
"""
@@ -986,3 +1121,4 @@ def show_system_resource_statistics():
Displays the system resource information.
"""
HFDeBugShowResourceStatistics()

View File

@@ -0,0 +1 @@
from .resource import *

View File

@@ -0,0 +1,116 @@
import os
import sys
from pathlib import Path
import urllib.request
import ssl
import hashlib
def get_file_hash_sha256(file_path):
sha256 = hashlib.sha256()
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b''):
sha256.update(chunk)
return sha256.hexdigest()
class ResourceManager:
def __init__(self):
"""Initialize resource manager and create necessary directories"""
self.user_home = Path.home()
self.base_dir = self.user_home / '.inspireface'
self.models_dir = self.base_dir / 'models'
# Create directories
self.base_dir.mkdir(exist_ok=True)
self.models_dir.mkdir(exist_ok=True)
# Model URLs
self._MODEL_LIST = {
"Pikachu": {
"url": "https://github.com/HyperInspire/InspireFace/releases/download/v1.x/Pikachu",
"filename": "Pikachu",
"md5": "f2983a2d884902229c1443fdc921b8e5f49cf2daba8a4f103cd127910dc9e7cd"
},
"Megatron": {
"url": "https://github.com/HyperInspire/InspireFace/releases/download/v1.x/Megatron",
"filename": "Megatron",
"md5": "28f2284c5e7cf53b0e152ff524a416c966ab21e724002643b1304aedc4af6b06"
}
}
def get_model(self, name: str, re_download: bool = False) -> str:
"""
Get model path. Download if not exists or re_download is True.
Args:
name: Model name
re_download: Force re-download if True
Returns:
str: Full path to model file
"""
if name not in self._MODEL_LIST:
raise ValueError(f"Model '{name}' not found. Available models: {list(self._MODEL_LIST.keys())}")
model_info = self._MODEL_LIST[name]
model_file = self.models_dir / model_info["filename"]
downloading_flag = model_file.with_suffix('.downloading')
# Check if model exists and is complete
if model_file.exists() and not downloading_flag.exists() and not re_download:
current_hash = get_file_hash_sha256(model_file)
if current_hash == model_info["md5"]:
return str(model_file)
else:
print(f"Model file hash mismatch for '{name}'. Re-downloading...")
# Start download
try:
print(f"Downloading model '{name}'...")
downloading_flag.touch()
# Create SSL context and headers
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
req = urllib.request.Request(model_info["url"], headers=headers)
with urllib.request.urlopen(req, context=ssl_context) as response:
total_size = int(response.headers.get('content-length', 0))
block_size = 8192
downloaded_size = 0
with open(model_file, 'wb') as f:
while True:
buffer = response.read(block_size)
if not buffer:
break
downloaded_size += len(buffer)
f.write(buffer)
if total_size > 0:
percent = (downloaded_size / total_size) * 100
sys.stdout.write(f"\rDownloading {name}: {percent:.1f}%")
sys.stdout.flush()
print("\nDownload completed")
downloading_flag.unlink() # Remove the downloading flag
return str(model_file)
except Exception as e:
if model_file.exists():
model_file.unlink()
if downloading_flag.exists():
downloading_flag.unlink()
raise RuntimeError(f"Failed to download model: {e}")
# Usage example
if __name__ == "__main__":
try:
rm = ResourceManager()
model_path = rm.get_model("Pikachu")
print(f"Model path: {model_path}")
except Exception as e:
print(f"Error: {e}")

View File

@@ -2,7 +2,7 @@
# Session option
from inspireface.modules.core.native import HF_ENABLE_NONE, HF_ENABLE_FACE_RECOGNITION, HF_ENABLE_LIVENESS, HF_ENABLE_IR_LIVENESS, \
HF_ENABLE_MASK_DETECT, HF_ENABLE_FACE_ATTRIBUTE, HF_ENABLE_QUALITY, HF_ENABLE_INTERACTION
HF_ENABLE_MASK_DETECT, HF_ENABLE_FACE_ATTRIBUTE, HF_ENABLE_QUALITY, HF_ENABLE_INTERACTION, HF_PK_AUTO_INCREMENT, HF_PK_MANUAL_INPUT
# Face track mode
from inspireface.modules.core.native import HF_DETECT_MODE_ALWAYS_DETECT, HF_DETECT_MODE_LIGHT_TRACK, HF_DETECT_MODE_TRACK_BY_DETECTION