feat(worker): 接口1 四庭七眼测量真实实现（替换 Mock）
worker 侧从 Mock 替换为真实算法： - face_analysis 包：detector(MediaPipe 478点) / pose(solvePnP 姿态) / calibration(虹膜直径法) / hair_segmenter+bisenet_model(方案B 头发分割) / measure(方案A兜底+B/A决策+七眼+换算) / annotation(numpy渐变线+中文标注) - app.py：/api/v1/face/measure 接真实实现，返回 annotated_image_base64 （不落盘不拼URL，落盘由网关做）；加 X-Internal-Token 鉴权、/health 就绪态、 可配置分辨率门槛、异常兜底 - 部署：start.sh/run_worker.sh/hair-worker.service 监听 8187；worker_config 示例 - 测试 tests/：Tier1合成真值<1e-6 + Tier2缩放不变 + Tier3叠加 + 错误码集成 + 数值回归，pytest 24 项全绿 - 文档补实测基线表 + RTX5090/torch 说明 注：worker 为 RTX 5090(sm_120)，pinned torch 2.2.2(cu121) 只到 sm_90， BiSeNet 已自动回退 CPU（方案B 正常）；要用 GPU 需换 torch cu128(≥2.7)。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,162 @@
|
||||
"""标注图层生成（透明底 RGBA PNG，仅标注、不含人物）。
|
||||
|
||||
规格（技术方案 §6）：线/字色 #FFFFFF、字体 10pt、线宽 1pt、透明底。
|
||||
- 四庭水平分界线：numpy 向量化渐变消失（中间亮、两侧渐隐）。
|
||||
- 四庭 cm 数值：图片左侧。
|
||||
- 七眼标注：眼宽/两眼间距/脸宽，虚线带箭头，标签上下穿插。
|
||||
中文字体用打包的思源黑体绝对路径加载，缺字体直接抛错（不静默降级成方块）。
|
||||
"""
|
||||
import os
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
FONT_PATH = os.path.join(os.path.dirname(__file__), "fonts", "NotoSansCJKsc-Regular.otf")
|
||||
FONT_SIZE = 10
|
||||
LINE_COLOR = (255, 255, 255, 255) # #FFFFFF 100%
|
||||
LINE_WIDTH = 1
|
||||
|
||||
|
||||
def _load_font():
|
||||
if not os.path.isfile(FONT_PATH):
|
||||
raise FileNotFoundError(f"中文字体缺失：{FONT_PATH}（请按 OFFLINE_ASSETS.md 放置）")
|
||||
return ImageFont.truetype(FONT_PATH, FONT_SIZE)
|
||||
|
||||
|
||||
def draw_gradient_horizontal_line(buf, cx, cy, color=LINE_COLOR, half_length=None):
|
||||
"""在 RGBA numpy 缓冲 buf 上，以 (cx,cy) 为中心画向两侧渐变消失的水平线。
|
||||
|
||||
numpy 向量化：一次性算整行 alpha，避免逐像素 draw.point。
|
||||
"""
|
||||
h, w = buf.shape[:2]
|
||||
cy = int(round(cy)); cx = int(round(cx))
|
||||
if not (0 <= cy < h):
|
||||
return
|
||||
half = half_length or (w // 3)
|
||||
xs = np.arange(w)
|
||||
dist = np.abs(xs - cx)
|
||||
alpha = np.clip(1.0 - dist / half, 0.0, 1.0) * color[3]
|
||||
mask = alpha > 0
|
||||
row = buf[cy]
|
||||
row[mask, 0] = color[0]
|
||||
row[mask, 1] = color[1]
|
||||
row[mask, 2] = color[2]
|
||||
row[mask, 3] = np.maximum(row[mask, 3], alpha[mask].astype(np.uint8))
|
||||
|
||||
|
||||
def draw_dashed_line_with_arrows(draw, x1, y1, x2, y2, color=LINE_COLOR,
|
||||
dash_len=6, gap_len=4, arrow_size=5):
|
||||
"""两点间画虚线，两端带箭头（等腰三角）。"""
|
||||
total = ((x2 - x1) ** 2 + (y2 - y1) ** 2) ** 0.5
|
||||
if total == 0:
|
||||
return
|
||||
dx = (x2 - x1) / total
|
||||
dy = (y2 - y1) / total
|
||||
pos = 0.0
|
||||
while pos < total:
|
||||
seg_end = min(pos + dash_len, total)
|
||||
draw.line([(x1 + dx * pos, y1 + dy * pos),
|
||||
(x1 + dx * seg_end, y1 + dy * seg_end)], fill=color, width=LINE_WIDTH)
|
||||
pos += dash_len + gap_len
|
||||
# 法向量（用于箭头两翼张开）
|
||||
nx, ny = -dy, dx
|
||||
for (ex, ey, sdx, sdy) in [(x1, y1, dx, dy), (x2, y2, -dx, -dy)]:
|
||||
p1 = (ex + sdx * arrow_size + nx * arrow_size * 0.6,
|
||||
ey + sdy * arrow_size + ny * arrow_size * 0.6)
|
||||
p2 = (ex + sdx * arrow_size - nx * arrow_size * 0.6,
|
||||
ey + sdy * arrow_size - ny * arrow_size * 0.6)
|
||||
draw.line([p1, (ex, ey)], fill=color, width=LINE_WIDTH)
|
||||
draw.line([p2, (ex, ey)], fill=color, width=LINE_WIDTH)
|
||||
|
||||
|
||||
def create_annotated_image(image_bgr, measure_result):
|
||||
"""生成标注图层 PNG（透明底 RGBA，尺寸同原图）。返回 PIL.Image。"""
|
||||
h, w = image_bgr.shape[:2]
|
||||
v = measure_result.vertical
|
||||
|
||||
# --- 1. 四庭水平分界线（numpy 渐变） ---
|
||||
buf = np.zeros((h, w, 4), dtype=np.uint8)
|
||||
order = ["hair_top", "hairline", "brow_center", "nose_bottom", "chin_tip"]
|
||||
ys = [v[name][1] for name in order]
|
||||
cx_line = v["brow_center"][0]
|
||||
for cy in ys:
|
||||
draw_gradient_horizontal_line(buf, cx_line, cy)
|
||||
|
||||
canvas = Image.fromarray(buf, mode="RGBA")
|
||||
draw = ImageDraw.Draw(canvas)
|
||||
font = _load_font()
|
||||
|
||||
# --- 2. 四庭 cm 数值（左侧） ---
|
||||
court_labels = [
|
||||
("顶庭", measure_result.top_cm),
|
||||
("上庭", measure_result.upper_cm),
|
||||
("中庭", measure_result.middle_cm),
|
||||
("下庭", measure_result.lower_cm),
|
||||
]
|
||||
left_margin = 16
|
||||
for i, (label, cm_val) in enumerate(court_labels):
|
||||
y_mid = (ys[i] + ys[i + 1]) / 2 - FONT_SIZE / 2
|
||||
draw.text((left_margin, y_mid), f"{label} {cm_val:.2f}cm",
|
||||
fill=LINE_COLOR, font=font)
|
||||
|
||||
# --- 3. 七眼标注（虚线箭头 + 上下穿插标签） ---
|
||||
pts = measure_result.eyes["points"]
|
||||
pc = measure_result.px_per_cm
|
||||
eye_y = (pts["left_inner"][1] + pts["right_inner"][1]) / 2
|
||||
|
||||
def hline(p_left, p_right, label, cm_val, above):
|
||||
y = eye_y
|
||||
draw_dashed_line_with_arrows(draw, p_left[0], y, p_right[0], y)
|
||||
text = f"{label} {cm_val:.2f}cm"
|
||||
tx = (p_left[0] + p_right[0]) / 2
|
||||
ty = y - FONT_SIZE - 4 if above else y + 4
|
||||
bbox = draw.textbbox((0, 0), text, font=font)
|
||||
tw = bbox[2] - bbox[0]
|
||||
draw.text((tx - tw / 2, ty), text, fill=LINE_COLOR, font=font)
|
||||
|
||||
# 眼宽（左眼，标签在上）、两眼间距（标签在下）、脸宽（标签在上）—— 上下穿插
|
||||
hline(pts["left_outer"], pts["left_inner"],
|
||||
"眼宽", measure_result.eye_width_cm, above=True)
|
||||
hline(pts["left_inner"], pts["right_inner"],
|
||||
"间距", measure_result.inter_eye_cm, above=False)
|
||||
hline(pts["left_cheek"], pts["right_cheek"],
|
||||
"脸宽", measure_result.face_width_cm, above=True)
|
||||
|
||||
return canvas
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
import time
|
||||
import cv2
|
||||
from face_analysis.detector import detector
|
||||
from face_analysis.measure import measure_face
|
||||
|
||||
path = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg"
|
||||
out = sys.argv[2] if len(sys.argv) > 2 else "tests/output/annotated.png"
|
||||
img = cv2.imread(path)
|
||||
if img is None:
|
||||
print(f"无法读取图片: {path}")
|
||||
sys.exit(1)
|
||||
h, w = img.shape[:2]
|
||||
lms = detector.detect(img)
|
||||
if lms is None:
|
||||
print("未检出人脸")
|
||||
sys.exit(1)
|
||||
mask = None
|
||||
try:
|
||||
from face_analysis.hair_segmenter import get_segmenter
|
||||
mask = get_segmenter().segment_hair(img)
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[warn] 分割不可用，回退方案 A：{e}")
|
||||
|
||||
result = measure_face(lms, mask, w, h)
|
||||
t0 = time.time()
|
||||
canvas = create_annotated_image(img, result)
|
||||
dt = time.time() - t0
|
||||
os.makedirs(os.path.dirname(out), exist_ok=True)
|
||||
canvas.save(out)
|
||||
arr = np.asarray(canvas)
|
||||
print(f"saved {out} mode={canvas.mode} size={canvas.size} "
|
||||
f"transparent={bool((arr[:,:,3]==0).any())} opaque={bool((arr[:,:,3]>0).any())} "
|
||||
f"elapsed={dt*1000:.1f}ms")
|
||||
@@ -0,0 +1,215 @@
|
||||
"""BiSeNet (face-parsing.PyTorch) 网络结构，vendored。
|
||||
|
||||
源自 zllrunning/face-parsing.PyTorch，结构与权重 `79999_iter.pth`（CelebAMask-HQ
|
||||
19 类）严格对应，仅修改 resnet18 骨干加载为「优先本地权重」以适配内网离线。
|
||||
hair 类别索引 = 17。
|
||||
"""
|
||||
import os
|
||||
|
||||
import torch
|
||||
import torch.nn as nn
|
||||
import torch.nn.functional as F
|
||||
import torch.utils.model_zoo as modelzoo
|
||||
|
||||
resnet18_url = "https://download.pytorch.org/models/resnet18-5c106cde.pth"
|
||||
_LOCAL_RESNET18 = os.path.join(os.path.dirname(__file__), "weights", "resnet18-5c106cde.pth")
|
||||
|
||||
|
||||
def conv3x3(in_planes, out_planes, stride=1):
|
||||
return nn.Conv2d(in_planes, out_planes, kernel_size=3, stride=stride, padding=1, bias=False)
|
||||
|
||||
|
||||
class BasicBlock(nn.Module):
|
||||
def __init__(self, in_chan, out_chan, stride=1):
|
||||
super(BasicBlock, self).__init__()
|
||||
self.conv1 = conv3x3(in_chan, out_chan, stride)
|
||||
self.bn1 = nn.BatchNorm2d(out_chan)
|
||||
self.conv2 = conv3x3(out_chan, out_chan)
|
||||
self.bn2 = nn.BatchNorm2d(out_chan)
|
||||
self.relu = nn.ReLU(inplace=True)
|
||||
self.downsample = None
|
||||
if in_chan != out_chan or stride != 1:
|
||||
self.downsample = nn.Sequential(
|
||||
nn.Conv2d(in_chan, out_chan, kernel_size=1, stride=stride, bias=False),
|
||||
nn.BatchNorm2d(out_chan),
|
||||
)
|
||||
|
||||
def forward(self, x):
|
||||
residual = self.conv1(x)
|
||||
residual = F.relu(self.bn1(residual))
|
||||
residual = self.conv2(residual)
|
||||
residual = self.bn2(residual)
|
||||
shortcut = x
|
||||
if self.downsample is not None:
|
||||
shortcut = self.downsample(x)
|
||||
out = shortcut + residual
|
||||
out = self.relu(out)
|
||||
return out
|
||||
|
||||
|
||||
def create_layer_basic(in_chan, out_chan, bnum, stride=1):
|
||||
layers = [BasicBlock(in_chan, out_chan, stride=stride)]
|
||||
for _ in range(bnum - 1):
|
||||
layers.append(BasicBlock(out_chan, out_chan, stride=1))
|
||||
return nn.Sequential(*layers)
|
||||
|
||||
|
||||
class Resnet18(nn.Module):
|
||||
def __init__(self):
|
||||
super(Resnet18, self).__init__()
|
||||
self.conv1 = nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3, bias=False)
|
||||
self.bn1 = nn.BatchNorm2d(64)
|
||||
self.maxpool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
|
||||
self.layer1 = create_layer_basic(64, 64, bnum=2, stride=1)
|
||||
self.layer2 = create_layer_basic(64, 128, bnum=2, stride=2)
|
||||
self.layer3 = create_layer_basic(128, 256, bnum=2, stride=2)
|
||||
self.layer4 = create_layer_basic(256, 512, bnum=2, stride=2)
|
||||
self.init_weight()
|
||||
|
||||
def forward(self, x):
|
||||
x = self.conv1(x)
|
||||
x = F.relu(self.bn1(x))
|
||||
x = self.maxpool(x)
|
||||
x = self.layer1(x)
|
||||
feat8 = self.layer2(x) # 1/8
|
||||
feat16 = self.layer3(feat8) # 1/16
|
||||
feat32 = self.layer4(feat16) # 1/32
|
||||
return feat8, feat16, feat32
|
||||
|
||||
def init_weight(self):
|
||||
# 优先本地骨干权重（内网离线），缺失才回退 torch model_zoo（会查缓存）。
|
||||
if os.path.isfile(_LOCAL_RESNET18):
|
||||
state_dict = torch.load(_LOCAL_RESNET18, map_location="cpu")
|
||||
else:
|
||||
state_dict = modelzoo.load_url(resnet18_url)
|
||||
self_state_dict = self.state_dict()
|
||||
for k, v in state_dict.items():
|
||||
if "fc" in k:
|
||||
continue
|
||||
self_state_dict.update({k: v})
|
||||
self.load_state_dict(self_state_dict)
|
||||
|
||||
|
||||
class ConvBNReLU(nn.Module):
|
||||
def __init__(self, in_chan, out_chan, ks=3, stride=1, padding=1):
|
||||
super(ConvBNReLU, self).__init__()
|
||||
self.conv = nn.Conv2d(in_chan, out_chan, kernel_size=ks, stride=stride,
|
||||
padding=padding, bias=False)
|
||||
self.bn = nn.BatchNorm2d(out_chan)
|
||||
|
||||
def forward(self, x):
|
||||
x = self.conv(x)
|
||||
x = F.relu(self.bn(x))
|
||||
return x
|
||||
|
||||
|
||||
class BiSeNetOutput(nn.Module):
|
||||
def __init__(self, in_chan, mid_chan, n_classes):
|
||||
super(BiSeNetOutput, self).__init__()
|
||||
self.conv = ConvBNReLU(in_chan, mid_chan, ks=3, stride=1, padding=1)
|
||||
self.conv_out = nn.Conv2d(mid_chan, n_classes, kernel_size=1, bias=False)
|
||||
|
||||
def forward(self, x):
|
||||
x = self.conv(x)
|
||||
x = self.conv_out(x)
|
||||
return x
|
||||
|
||||
|
||||
class AttentionRefinementModule(nn.Module):
|
||||
def __init__(self, in_chan, out_chan):
|
||||
super(AttentionRefinementModule, self).__init__()
|
||||
self.conv = ConvBNReLU(in_chan, out_chan, ks=3, stride=1, padding=1)
|
||||
self.conv_atten = nn.Conv2d(out_chan, out_chan, kernel_size=1, bias=False)
|
||||
self.bn_atten = nn.BatchNorm2d(out_chan)
|
||||
self.sigmoid_atten = nn.Sigmoid()
|
||||
|
||||
def forward(self, x):
|
||||
feat = self.conv(x)
|
||||
atten = F.avg_pool2d(feat, feat.size()[2:])
|
||||
atten = self.conv_atten(atten)
|
||||
atten = self.bn_atten(atten)
|
||||
atten = self.sigmoid_atten(atten)
|
||||
out = torch.mul(feat, atten)
|
||||
return out
|
||||
|
||||
|
||||
class ContextPath(nn.Module):
|
||||
def __init__(self):
|
||||
super(ContextPath, self).__init__()
|
||||
self.resnet = Resnet18()
|
||||
self.arm16 = AttentionRefinementModule(256, 128)
|
||||
self.arm32 = AttentionRefinementModule(512, 128)
|
||||
self.conv_head32 = ConvBNReLU(128, 128, ks=3, stride=1, padding=1)
|
||||
self.conv_head16 = ConvBNReLU(128, 128, ks=3, stride=1, padding=1)
|
||||
self.conv_avg = ConvBNReLU(512, 128, ks=1, stride=1, padding=0)
|
||||
|
||||
def forward(self, x):
|
||||
feat8, feat16, feat32 = self.resnet(x)
|
||||
h8, w8 = feat8.size()[2:]
|
||||
h16, w16 = feat16.size()[2:]
|
||||
h32, w32 = feat32.size()[2:]
|
||||
|
||||
avg = F.avg_pool2d(feat32, feat32.size()[2:])
|
||||
avg = self.conv_avg(avg)
|
||||
avg_up = F.interpolate(avg, (h32, w32), mode="nearest")
|
||||
|
||||
feat32_arm = self.arm32(feat32)
|
||||
feat32_sum = feat32_arm + avg_up
|
||||
feat32_up = F.interpolate(feat32_sum, (h16, w16), mode="nearest")
|
||||
feat32_up = self.conv_head32(feat32_up)
|
||||
|
||||
feat16_arm = self.arm16(feat16)
|
||||
feat16_sum = feat16_arm + feat32_up
|
||||
feat16_up = F.interpolate(feat16_sum, (h8, w8), mode="nearest")
|
||||
feat16_up = self.conv_head16(feat16_up)
|
||||
|
||||
return feat8, feat16_up, feat32_up # feat8 未用，保持与权重结构一致
|
||||
|
||||
|
||||
class FeatureFusionModule(nn.Module):
|
||||
def __init__(self, in_chan, out_chan):
|
||||
super(FeatureFusionModule, self).__init__()
|
||||
self.convblk = ConvBNReLU(in_chan, out_chan, ks=1, stride=1, padding=0)
|
||||
self.conv1 = nn.Conv2d(out_chan, out_chan // 4, kernel_size=1, stride=1,
|
||||
padding=0, bias=False)
|
||||
self.conv2 = nn.Conv2d(out_chan // 4, out_chan, kernel_size=1, stride=1,
|
||||
padding=0, bias=False)
|
||||
self.relu = nn.ReLU(inplace=True)
|
||||
self.sigmoid = nn.Sigmoid()
|
||||
|
||||
def forward(self, fsp, fcp):
|
||||
fcat = torch.cat([fsp, fcp], dim=1)
|
||||
feat = self.convblk(fcat)
|
||||
atten = F.avg_pool2d(feat, feat.size()[2:])
|
||||
atten = self.conv1(atten)
|
||||
atten = self.relu(atten)
|
||||
atten = self.conv2(atten)
|
||||
atten = self.sigmoid(atten)
|
||||
feat_atten = torch.mul(feat, atten)
|
||||
feat_out = feat_atten + feat
|
||||
return feat_out
|
||||
|
||||
|
||||
class BiSeNet(nn.Module):
|
||||
def __init__(self, n_classes):
|
||||
super(BiSeNet, self).__init__()
|
||||
# 该权重(79999_iter.pth)的变体无独立 SpatialPath：
|
||||
# 直接用 ContextPath 的 resnet feat8（128ch）作为空间路径特征。
|
||||
self.cp = ContextPath()
|
||||
self.ffm = FeatureFusionModule(256, 256)
|
||||
self.conv_out = BiSeNetOutput(256, 256, n_classes)
|
||||
self.conv_out16 = BiSeNetOutput(128, 64, n_classes)
|
||||
self.conv_out32 = BiSeNetOutput(128, 64, n_classes)
|
||||
|
||||
def forward(self, x):
|
||||
h, w = x.size()[2:]
|
||||
feat_res8, feat_cp8, feat_cp16 = self.cp(x)
|
||||
feat_fuse = self.ffm(feat_res8, feat_cp8)
|
||||
|
||||
feat_out = self.conv_out(feat_fuse)
|
||||
feat_out16 = self.conv_out16(feat_cp8)
|
||||
feat_out32 = self.conv_out32(feat_cp16)
|
||||
feat_out = F.interpolate(feat_out, (h, w), mode="bilinear", align_corners=True)
|
||||
feat_out16 = F.interpolate(feat_out16, (h, w), mode="bilinear", align_corners=True)
|
||||
feat_out32 = F.interpolate(feat_out32, (h, w), mode="bilinear", align_corners=True)
|
||||
return feat_out, feat_out16, feat_out32
|
||||
@@ -0,0 +1,85 @@
|
||||
"""尺度校准：像素 → 厘米（虹膜直径法，眼宽降级）。
|
||||
|
||||
人类虹膜直径高度稳定（成人平均 11.7mm），作为天然标尺把像素距离换算成厘米。
|
||||
虹膜点（索引 469/471、474/476）需 refine_landmarks=True 才输出；缺失时降级用
|
||||
眼宽（外→内眼角，均值约 2.85cm）。详见技术方案 §3。
|
||||
"""
|
||||
from face_analysis.face_mesh_landmarks import (
|
||||
IRIS_LEFT_LEFT, IRIS_LEFT_RIGHT, IRIS_RIGHT_LEFT, IRIS_RIGHT_RIGHT,
|
||||
LEFT_EYE_OUTER, LEFT_EYE_INNER, RIGHT_EYE_INNER, RIGHT_EYE_OUTER,
|
||||
)
|
||||
|
||||
AVG_IRIS_DIAMETER_CM = 1.17 # 虹膜平均直径 11.7mm
|
||||
AVG_EYE_WIDTH_CM = 2.85 # 眼裂平均宽度约 28.5mm（降级标尺）
|
||||
|
||||
|
||||
def _lm_list(landmarks):
|
||||
"""兼容 NormalizedLandmarkList（有 .landmark）与裸 list 两种入参。"""
|
||||
return landmarks.landmark if hasattr(landmarks, "landmark") else landmarks
|
||||
|
||||
|
||||
def normalized_to_pixel(landmark, image_width, image_height):
|
||||
"""归一化坐标 → 像素坐标。"""
|
||||
return landmark.x * image_width, landmark.y * image_height
|
||||
|
||||
|
||||
def pixel_distance(p1, p2):
|
||||
"""两点像素欧氏距离。"""
|
||||
return ((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2) ** 0.5
|
||||
|
||||
|
||||
def _iris_diameter_px(lm, w, h):
|
||||
"""左右虹膜直径像素均值；任一边缘点缺失/为 0 返回 None。"""
|
||||
try:
|
||||
ll = normalized_to_pixel(lm[IRIS_LEFT_LEFT], w, h)
|
||||
lr = normalized_to_pixel(lm[IRIS_LEFT_RIGHT], w, h)
|
||||
rl = normalized_to_pixel(lm[IRIS_RIGHT_LEFT], w, h)
|
||||
rr = normalized_to_pixel(lm[IRIS_RIGHT_RIGHT], w, h)
|
||||
except (IndexError, KeyError):
|
||||
return None
|
||||
left_d = pixel_distance(ll, lr)
|
||||
right_d = pixel_distance(rl, rr)
|
||||
if left_d <= 0 or right_d <= 0:
|
||||
return None
|
||||
return (left_d + right_d) / 2
|
||||
|
||||
|
||||
def _eye_width_px(lm, w, h):
|
||||
"""左右眼宽（外→内眼角）像素均值，作为虹膜降级标尺。"""
|
||||
l = pixel_distance(normalized_to_pixel(lm[LEFT_EYE_OUTER], w, h),
|
||||
normalized_to_pixel(lm[LEFT_EYE_INNER], w, h))
|
||||
r = pixel_distance(normalized_to_pixel(lm[RIGHT_EYE_OUTER], w, h),
|
||||
normalized_to_pixel(lm[RIGHT_EYE_INNER], w, h))
|
||||
return (l + r) / 2
|
||||
|
||||
|
||||
def estimate_scale_factor(landmarks, image_width, image_height):
|
||||
"""估算 px_per_cm（每厘米对应像素数）。
|
||||
|
||||
优先用虹膜直径法；虹膜点不可用时降级用眼宽。返回正浮点数。
|
||||
"""
|
||||
lm = _lm_list(landmarks)
|
||||
iris_px = _iris_diameter_px(lm, image_width, image_height)
|
||||
if iris_px is not None:
|
||||
return iris_px / AVG_IRIS_DIAMETER_CM
|
||||
# 降级：眼宽法
|
||||
eye_px = _eye_width_px(lm, image_width, image_height)
|
||||
return eye_px / AVG_EYE_WIDTH_CM
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
import cv2
|
||||
from face_analysis.detector import detector
|
||||
|
||||
path = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg"
|
||||
img = cv2.imread(path)
|
||||
if img is None:
|
||||
print(f"无法读取图片: {path}")
|
||||
sys.exit(1)
|
||||
h, w = img.shape[:2]
|
||||
lms = detector.detect(img)
|
||||
if lms is None:
|
||||
print("未检出人脸")
|
||||
sys.exit(1)
|
||||
print(f"px_per_cm: {estimate_scale_factor(lms, w, h):.4f}")
|
||||
@@ -0,0 +1,61 @@
|
||||
"""MediaPipe Face Mesh 关键点检测封装（单例）。
|
||||
|
||||
封装经典 Solutions API（mp.solutions.face_mesh），模型权重内置于 pip 包，
|
||||
无需额外下载。开启 refine_landmarks=True 以获得虹膜点（尺度校准用），
|
||||
static_image_mode=True 适配单张图片推理，max_num_faces=1 只取最大/首个人脸。
|
||||
|
||||
详见技术方案 §8.2。
|
||||
"""
|
||||
import cv2
|
||||
import numpy as np
|
||||
import mediapipe as mp
|
||||
|
||||
mp_face_mesh = mp.solutions.face_mesh
|
||||
|
||||
|
||||
class FaceMeshDetector:
|
||||
"""MediaPipe Face Mesh 封装，单例模式（模块底部 detector）。"""
|
||||
|
||||
def __init__(self):
|
||||
self.face_mesh = mp_face_mesh.FaceMesh(
|
||||
static_image_mode=True,
|
||||
max_num_faces=1, # 仅检测单人（取最大脸）
|
||||
refine_landmarks=True, # 启用虹膜 + 唇部精细关键点
|
||||
min_detection_confidence=0.5,
|
||||
)
|
||||
|
||||
def detect(self, image: np.ndarray):
|
||||
"""检测人脸关键点。
|
||||
|
||||
Args:
|
||||
image: BGR numpy array（OpenCV 格式）。
|
||||
Returns:
|
||||
landmarks: NormalizedLandmarkList（.landmark 列表），或检测失败时 None。
|
||||
"""
|
||||
rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
|
||||
results = self.face_mesh.process(rgb)
|
||||
if results.multi_face_landmarks:
|
||||
return results.multi_face_landmarks[0]
|
||||
return None
|
||||
|
||||
def close(self):
|
||||
self.face_mesh.close()
|
||||
|
||||
|
||||
# 全局单例：模块加载时初始化一次，避免每请求重建（重建很慢）。
|
||||
detector = FaceMeshDetector()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
path = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg"
|
||||
img = cv2.imread(path)
|
||||
if img is None:
|
||||
print(f"无法读取图片: {path}")
|
||||
sys.exit(1)
|
||||
lms = detector.detect(img)
|
||||
if lms is None:
|
||||
print("detected landmarks: None（未检出人脸）")
|
||||
sys.exit(1)
|
||||
print(f"detected landmarks: {len(lms.landmark)}")
|
||||
@@ -0,0 +1,40 @@
|
||||
"""MediaPipe Face Mesh 关键点索引常量（四庭七眼测量用）。
|
||||
|
||||
MediaPipe Face Mesh 对 468 个点按固定拓扑编号；开启 refine_landmarks=True 后
|
||||
额外输出 10 个虹膜点（索引 468–477），总计 478 点。本模块集中定义本接口
|
||||
所需的全部索引，避免散落在各处的魔数。详见技术方案 §2。
|
||||
"""
|
||||
|
||||
# --- 四庭纵向中轴关键点 ---
|
||||
GLABELLA_9 = 9 # 眉间 / glabella（上点）
|
||||
GLABELLA_151 = 151 # 眉间 / glabella（下点），与 9 取中点作为眉心
|
||||
NOSE_BOTTOM = 94 # 鼻翼下缘 / subnasale（人中顶部）
|
||||
CHIN_TIP = 152 # 下巴尖 / menton（下颌最低点）
|
||||
|
||||
# --- 七眼横向关键点 ---
|
||||
LEFT_EYE_OUTER = 33 # 左眼外角
|
||||
LEFT_EYE_INNER = 133 # 左眼内角
|
||||
RIGHT_EYE_INNER = 362 # 右眼内角
|
||||
RIGHT_EYE_OUTER = 263 # 右眼外角
|
||||
LEFT_CHEEK = 234 # 左脸颧弓（脸宽左端）
|
||||
RIGHT_CHEEK = 454 # 右脸颧弓（脸宽右端）
|
||||
|
||||
# --- 鼻尖（solvePnP 用，可选） ---
|
||||
NOSE_TIP = 1 # 鼻尖（也有用 4 的版本）
|
||||
NOSE_TIP_ALT = 4
|
||||
|
||||
# --- 虹膜关键点（refine_landmarks=True 才输出，尺度校准用） ---
|
||||
IRIS_LEFT_CENTER = 468 # 左眼虹膜中心
|
||||
IRIS_LEFT_LEFT = 469 # 左虹膜左边缘
|
||||
IRIS_LEFT_RIGHT = 471 # 左虹膜右边缘
|
||||
IRIS_RIGHT_CENTER = 473 # 右眼虹膜中心
|
||||
IRIS_RIGHT_LEFT = 474 # 右虹膜左边缘
|
||||
IRIS_RIGHT_RIGHT = 476 # 右虹膜右边缘
|
||||
|
||||
# --- solvePnP 姿态估计用的 6 点（与通用 3D 头模一一对应，见 pose.py） ---
|
||||
MOUTH_LEFT = 61 # 左嘴角
|
||||
MOUTH_RIGHT = 291 # 右嘴角
|
||||
PNP_INDICES = [NOSE_TIP, CHIN_TIP, LEFT_EYE_OUTER, RIGHT_EYE_OUTER, MOUTH_LEFT, MOUTH_RIGHT]
|
||||
|
||||
# 含虹膜时的关键点总数
|
||||
NUM_LANDMARKS_WITH_IRIS = 478
|
||||
@@ -0,0 +1,166 @@
|
||||
"""方案 B：BiSeNet 头发分割 + 发际线/头顶定位。
|
||||
|
||||
加载 face-parsing BiSeNet（19 类，hair=17），对整图做像素级语义分割得到头发
|
||||
mask，再沿面部中轴线扫描得到真实发际线与头顶。GPU 可用时走 CUDA，否则 CPU。
|
||||
单例加载权重，避免每请求重载。详见技术方案 §1.4 / §4.0。
|
||||
"""
|
||||
import os
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
# ⚠️ torch / torchvision / BiSeNet 仅在 HairSegmenter.__init__ 内惰性导入，
|
||||
# 使本模块的纯 numpy 函数 locate_hairline_by_segmentation 可在无 torch/GPU
|
||||
# 的环境（如 Tier-1 合成几何测试、方案 A only 降级版）被安全导入。
|
||||
|
||||
_WEIGHTS = os.path.join(os.path.dirname(__file__), "weights", "79999_iter.pth")
|
||||
HAIR_CLASS = 17 # CelebAMask-HQ 19 类中 hair 的索引
|
||||
N_CLASSES = 19
|
||||
_INPUT_SIZE = 512 # BiSeNet 推理输入边长
|
||||
|
||||
|
||||
def _select_device(torch):
|
||||
"""选择推理设备：优先 CUDA，但实测一次小算子确认当前 GPU 架构被本 torch 支持。
|
||||
|
||||
场景：本机为 RTX 5090（sm_120/Blackwell），而 torch 2.2.2+cu121 仅编译到 sm_90，
|
||||
.cuda() 会在执行时抛 "no kernel image is available"。此处用一次小 matmul 探测，
|
||||
失败则回退 CPU（BiSeNet CPU 推理 ~0.3–1s/张，方案B 仍可用）。
|
||||
换装支持 sm_120 的 torch（cu128）后会自动改用 GPU，无需改代码。
|
||||
可用环境变量 FORCE_CPU=1 强制 CPU。
|
||||
"""
|
||||
import os as _os
|
||||
if _os.getenv("FORCE_CPU") == "1" or not torch.cuda.is_available():
|
||||
return torch.device("cpu")
|
||||
try:
|
||||
_ = (torch.zeros(8, 8, device="cuda") @ torch.zeros(8, 8, device="cuda")).cpu()
|
||||
return torch.device("cuda")
|
||||
except Exception: # noqa: BLE001 GPU 架构不被支持 → 回退 CPU
|
||||
return torch.device("cpu")
|
||||
|
||||
|
||||
class HairSegmenter:
|
||||
"""BiSeNet 头发分割封装。建议经 get_segmenter() 取单例。"""
|
||||
|
||||
def __init__(self, weights_path=_WEIGHTS):
|
||||
import torch
|
||||
import torchvision.transforms as transforms
|
||||
|
||||
from face_analysis.bisenet_model import BiSeNet
|
||||
|
||||
self._torch = torch
|
||||
self.device = _select_device(torch)
|
||||
self.net = BiSeNet(n_classes=N_CLASSES)
|
||||
state = torch.load(weights_path, map_location="cpu")
|
||||
self.net.load_state_dict(state)
|
||||
self.net.to(self.device)
|
||||
self.net.eval()
|
||||
self._to_tensor = transforms.Compose([
|
||||
transforms.ToTensor(),
|
||||
transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225)),
|
||||
])
|
||||
|
||||
def segment_hair(self, image_bgr):
|
||||
"""返回 hair_mask（H×W bool，True=头发），尺寸同输入原图。"""
|
||||
torch = self._torch
|
||||
h, w = image_bgr.shape[:2]
|
||||
rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB)
|
||||
resized = cv2.resize(rgb, (_INPUT_SIZE, _INPUT_SIZE),
|
||||
interpolation=cv2.INTER_LINEAR)
|
||||
inp = self._to_tensor(resized).unsqueeze(0).to(self.device)
|
||||
with torch.no_grad():
|
||||
out = self.net(inp)[0] # 主输出 (1, C, 512, 512)
|
||||
parsing = out.squeeze(0).argmax(0).cpu().numpy() # (512, 512) 类别图
|
||||
hair_small = (parsing == HAIR_CLASS).astype(np.uint8)
|
||||
# 还原到原图尺寸（最近邻保持类别边界）
|
||||
hair_mask = cv2.resize(hair_small, (w, h), interpolation=cv2.INTER_NEAREST)
|
||||
return hair_mask.astype(bool)
|
||||
|
||||
|
||||
_segmenter = None
|
||||
|
||||
|
||||
def get_segmenter():
|
||||
"""惰性单例：首次调用时加载权重（并占用显存），后续复用。"""
|
||||
global _segmenter
|
||||
if _segmenter is None:
|
||||
_segmenter = HairSegmenter()
|
||||
return _segmenter
|
||||
|
||||
|
||||
def locate_hairline_by_segmentation(hair_mask, brow_center_x, image_height):
|
||||
"""从头发 mask 定位发际线与头顶。
|
||||
|
||||
Args:
|
||||
hair_mask: H×W bool/uint8，True=头发。
|
||||
brow_center_x: 面部中轴线 x（像素）。
|
||||
image_height: 图高（保留参数，便于后续边界判断）。
|
||||
Returns:
|
||||
(hairline_y, hair_top_y) 像素坐标；失败返回 None（交给方案 A 兜底）。
|
||||
"""
|
||||
if hair_mask is None:
|
||||
return None
|
||||
mask = np.asarray(hair_mask).astype(bool)
|
||||
if mask.sum() == 0:
|
||||
return None
|
||||
|
||||
w = mask.shape[1]
|
||||
cx = int(round(brow_center_x))
|
||||
cx = max(0, min(cx, w - 1))
|
||||
# 中轴线附近窄列带（±3px）求稳，避免单列噪声
|
||||
band = mask[:, max(0, cx - 3): min(w, cx + 4)]
|
||||
col = band.any(axis=1)
|
||||
hair_rows = np.where(col)[0]
|
||||
if hair_rows.size == 0:
|
||||
return None
|
||||
|
||||
# 发际线：中轴线列带上头发区域最靠下的行（头发→皮肤交界，y 向下为正）
|
||||
hairline_y = int(hair_rows.max())
|
||||
# 头顶：整张头发 mask 的最高点（最小 y），用全图更鲁棒
|
||||
top_rows = np.where(mask.any(axis=1))[0]
|
||||
hair_top_y = int(top_rows.min())
|
||||
|
||||
# 合理性校验：头顶必须严格在发际线上方
|
||||
if hair_top_y >= hairline_y:
|
||||
return None
|
||||
return hairline_y, hair_top_y
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
from face_analysis.detector import detector
|
||||
from face_analysis.calibration import normalized_to_pixel
|
||||
from face_analysis.face_mesh_landmarks import GLABELLA_9, GLABELLA_151
|
||||
|
||||
path = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg"
|
||||
img = cv2.imread(path)
|
||||
if img is None:
|
||||
print(f"无法读取图片: {path}")
|
||||
sys.exit(1)
|
||||
h, w = img.shape[:2]
|
||||
segmenter = get_segmenter()
|
||||
print("device:", segmenter.device)
|
||||
mask = segmenter.segment_hair(img)
|
||||
print("hair pixels:", int(mask.sum()))
|
||||
|
||||
lms = detector.detect(img)
|
||||
if lms is None:
|
||||
print("未检出人脸，跳过定位")
|
||||
sys.exit(0)
|
||||
lm = lms.landmark
|
||||
bx = (normalized_to_pixel(lm[GLABELLA_9], w, h)[0]
|
||||
+ normalized_to_pixel(lm[GLABELLA_151], w, h)[0]) / 2
|
||||
by = (normalized_to_pixel(lm[GLABELLA_9], w, h)[1]
|
||||
+ normalized_to_pixel(lm[GLABELLA_151], w, h)[1]) / 2
|
||||
res = locate_hairline_by_segmentation(mask, bx, h)
|
||||
if res is None:
|
||||
print("定位失败（将回退方案 A）")
|
||||
else:
|
||||
hairline_y, hair_top_y = res
|
||||
print(f"brow_y={by:.1f} hairline_y={hairline_y} hair_top_y={hair_top_y}")
|
||||
print("自洽校验 hair_top_y < hairline_y < brow_y:",
|
||||
hair_top_y < hairline_y < by)
|
||||
|
||||
# dump mask 预览
|
||||
os.makedirs("tests/output", exist_ok=True)
|
||||
cv2.imwrite("tests/output/hair_mask.png", (mask.astype(np.uint8) * 255))
|
||||
print("mask 预览已存 tests/output/hair_mask.png")
|
||||
@@ -0,0 +1,254 @@
|
||||
"""四庭七眼测量核心：纵向定位（方案 B 主 / 方案 A 兜底）+ 七眼 + 厘米换算。
|
||||
|
||||
整合：
|
||||
- estimate_vertical_landmarks：方案 A，按三庭比例推算上/顶庭（兜底）。
|
||||
- 决策逻辑：优先方案 B（分割发际线/头顶），合理性校验不过则回退方案 A。
|
||||
- measure_seven_eyes：眼宽/脸宽/两眼间距实测。
|
||||
- measure_face：主入口，产出结构化结果 MeasureResult（含 to_response）。
|
||||
|
||||
详见技术方案 §4 / §5。本模块不依赖 torch，可在纯几何环境单独运行。
|
||||
"""
|
||||
from face_analysis.calibration import (
|
||||
estimate_scale_factor, normalized_to_pixel, pixel_distance, _lm_list,
|
||||
)
|
||||
from face_analysis.face_mesh_landmarks import (
|
||||
GLABELLA_9, GLABELLA_151, NOSE_BOTTOM, CHIN_TIP,
|
||||
LEFT_EYE_OUTER, LEFT_EYE_INNER, RIGHT_EYE_INNER, RIGHT_EYE_OUTER,
|
||||
LEFT_CHEEK, RIGHT_CHEEK,
|
||||
)
|
||||
from face_analysis.hair_segmenter import locate_hairline_by_segmentation
|
||||
|
||||
# 方案 A 推算比例常量（顶:上:中:下 = 0.22:0.25:0.28:0.25），见技术方案 §4.2
|
||||
_UPPER_RATIO = 0.25 / 0.265 # 上庭 ÷ 中下庭均值
|
||||
_TOP_RATIO = 0.22 / 0.28 # 顶庭 ÷ 中庭（≈ 0.786）
|
||||
|
||||
|
||||
def _brow_center(lm, w, h):
|
||||
"""眉心 = 索引 9 / 151 中点。"""
|
||||
g9 = normalized_to_pixel(lm[GLABELLA_9], w, h)
|
||||
g151 = normalized_to_pixel(lm[GLABELLA_151], w, h)
|
||||
return (g9[0] + g151[0]) / 2, (g9[1] + g151[1]) / 2
|
||||
|
||||
|
||||
def estimate_vertical_landmarks(landmarks, image_width, image_height):
|
||||
"""方案 A（兜底）：实测中/下庭，按比例推算上/顶庭。
|
||||
|
||||
返回 5 个纵向点像素坐标 + 各段像素高度。注意其循环论证局限：
|
||||
上/顶庭为估算值，不反映真实脸型（详见技术方案 §4.1）。
|
||||
"""
|
||||
lm = _lm_list(landmarks)
|
||||
w, h = image_width, image_height
|
||||
|
||||
brow_x, brow_y = _brow_center(lm, w, h)
|
||||
nose_bottom = normalized_to_pixel(lm[NOSE_BOTTOM], w, h)
|
||||
chin_tip = normalized_to_pixel(lm[CHIN_TIP], w, h)
|
||||
|
||||
middle_court_px = abs(brow_y - nose_bottom[1]) # 眉心 → 鼻翼下缘
|
||||
lower_court_px = abs(nose_bottom[1] - chin_tip[1]) # 鼻翼下缘 → 下巴尖
|
||||
|
||||
one_unit_px = (middle_court_px + lower_court_px) / 2 # 一等份 ≈ 中/下庭均值
|
||||
upper_court_px = one_unit_px * _UPPER_RATIO
|
||||
top_court_px = one_unit_px * _TOP_RATIO
|
||||
|
||||
hairline_y = brow_y - upper_court_px
|
||||
hair_top_y = hairline_y - top_court_px
|
||||
|
||||
return {
|
||||
"hair_top": (brow_x, hair_top_y),
|
||||
"hairline": (brow_x, hairline_y),
|
||||
"brow_center": (brow_x, brow_y),
|
||||
"nose_bottom": (nose_bottom[0], nose_bottom[1]),
|
||||
"chin_tip": (chin_tip[0], chin_tip[1]),
|
||||
"top_court_px": top_court_px,
|
||||
"upper_court_px": upper_court_px,
|
||||
"middle_court_px": middle_court_px,
|
||||
"lower_court_px": lower_court_px,
|
||||
}
|
||||
|
||||
|
||||
def _vertical_from_segmentation(lm, w, h, hair_mask):
|
||||
"""方案 B：用分割得到的发际线/头顶替换方案 A 的上/顶庭。
|
||||
|
||||
成功且通过合理性校验返回 vertical dict，否则返回 None。
|
||||
"""
|
||||
res = locate_hairline_by_segmentation(hair_mask, _brow_center(lm, w, h)[0], h)
|
||||
if res is None:
|
||||
return None
|
||||
hairline_y, hair_top_y = res
|
||||
|
||||
brow_x, brow_y = _brow_center(lm, w, h)
|
||||
nose_bottom = normalized_to_pixel(lm[NOSE_BOTTOM], w, h)
|
||||
chin_tip = normalized_to_pixel(lm[CHIN_TIP], w, h)
|
||||
|
||||
middle_court_px = abs(brow_y - nose_bottom[1])
|
||||
lower_court_px = abs(nose_bottom[1] - chin_tip[1])
|
||||
upper_court_px = brow_y - hairline_y # 发际线 → 眉心
|
||||
top_court_px = hairline_y - hair_top_y # 头顶 → 发际线
|
||||
|
||||
# 合理性校验：发际线在眉心上方、头顶在发际线上方、各庭为正
|
||||
if not (hair_top_y < hairline_y < brow_y):
|
||||
return None
|
||||
if upper_court_px <= 0 or top_court_px <= 0:
|
||||
return None
|
||||
if middle_court_px <= 0 or lower_court_px <= 0:
|
||||
return None
|
||||
|
||||
return {
|
||||
"hair_top": (brow_x, float(hair_top_y)),
|
||||
"hairline": (brow_x, float(hairline_y)),
|
||||
"brow_center": (brow_x, brow_y),
|
||||
"nose_bottom": (nose_bottom[0], nose_bottom[1]),
|
||||
"chin_tip": (chin_tip[0], chin_tip[1]),
|
||||
"top_court_px": top_court_px,
|
||||
"upper_court_px": upper_court_px,
|
||||
"middle_court_px": middle_court_px,
|
||||
"lower_court_px": lower_court_px,
|
||||
}
|
||||
|
||||
|
||||
def decide_vertical(landmarks, image_width, image_height, hair_mask):
|
||||
"""纵向定位决策：方案 B 优先，失败回退方案 A。
|
||||
|
||||
返回 (vertical_dict, hairline_source)，source ∈ {"segmentation","estimated"}。
|
||||
"""
|
||||
lm = _lm_list(landmarks)
|
||||
vb = _vertical_from_segmentation(lm, image_width, image_height, hair_mask)
|
||||
if vb is not None:
|
||||
return vb, "segmentation"
|
||||
return estimate_vertical_landmarks(landmarks, image_width, image_height), "estimated"
|
||||
|
||||
|
||||
def measure_seven_eyes(landmarks, image_width, image_height):
|
||||
"""七眼：眼宽（左右均值）、脸宽、两眼间距（像素）。"""
|
||||
lm = _lm_list(landmarks)
|
||||
w, h = image_width, image_height
|
||||
left_outer = normalized_to_pixel(lm[LEFT_EYE_OUTER], w, h)
|
||||
left_inner = normalized_to_pixel(lm[LEFT_EYE_INNER], w, h)
|
||||
right_inner = normalized_to_pixel(lm[RIGHT_EYE_INNER], w, h)
|
||||
right_outer = normalized_to_pixel(lm[RIGHT_EYE_OUTER], w, h)
|
||||
left_cheek = normalized_to_pixel(lm[LEFT_CHEEK], w, h)
|
||||
right_cheek = normalized_to_pixel(lm[RIGHT_CHEEK], w, h)
|
||||
|
||||
left_eye = pixel_distance(left_outer, left_inner)
|
||||
right_eye = pixel_distance(right_inner, right_outer)
|
||||
return {
|
||||
"eye_width_px": (left_eye + right_eye) / 2,
|
||||
"face_width_px": pixel_distance(left_cheek, right_cheek),
|
||||
"inter_eye_distance_px": pixel_distance(left_inner, right_inner),
|
||||
# 标注图用的横向点像素坐标（不进 to_response）
|
||||
"points": {
|
||||
"left_outer": left_outer, "left_inner": left_inner,
|
||||
"right_inner": right_inner, "right_outer": right_outer,
|
||||
"left_cheek": left_cheek, "right_cheek": right_cheek,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class MeasureResult:
|
||||
"""测量结果，提供 to_response() 输出与接口文档同构的 data 字段。"""
|
||||
|
||||
def __init__(self, vertical, eyes, px_per_cm, hairline_source, head_pose):
|
||||
self.vertical = vertical
|
||||
self.eyes = eyes
|
||||
self.px_per_cm = px_per_cm
|
||||
self.hairline_source = hairline_source
|
||||
self.head_pose = head_pose # (yaw, pitch, roll) 或 None
|
||||
|
||||
# 各庭厘米
|
||||
self.top_cm = vertical["top_court_px"] / px_per_cm
|
||||
self.upper_cm = vertical["upper_court_px"] / px_per_cm
|
||||
self.middle_cm = vertical["middle_court_px"] / px_per_cm
|
||||
self.lower_cm = vertical["lower_court_px"] / px_per_cm
|
||||
self.face_total_cm = self.top_cm + self.upper_cm + self.middle_cm + self.lower_cm
|
||||
|
||||
# 七眼厘米
|
||||
self.eye_width_cm = eyes["eye_width_px"] / px_per_cm
|
||||
self.face_width_cm = eyes["face_width_px"] / px_per_cm
|
||||
self.inter_eye_cm = eyes["inter_eye_distance_px"] / px_per_cm
|
||||
|
||||
def to_response(self):
|
||||
total_px = (self.vertical["top_court_px"] + self.vertical["upper_court_px"]
|
||||
+ self.vertical["middle_court_px"] + self.vertical["lower_court_px"])
|
||||
fw_px = self.eyes["face_width_px"]
|
||||
|
||||
def pt(name):
|
||||
x, y = self.vertical[name]
|
||||
return {"x": int(round(x)), "y": int(round(y))}
|
||||
|
||||
data = {
|
||||
"face_total_height_cm": round(self.face_total_cm, 2),
|
||||
"four_courts": {
|
||||
"top_court_cm": round(self.top_cm, 2),
|
||||
"upper_court_cm": round(self.upper_cm, 2),
|
||||
"middle_court_cm": round(self.middle_cm, 2),
|
||||
"lower_court_cm": round(self.lower_cm, 2),
|
||||
"ratios": {
|
||||
"top_court": round(self.vertical["top_court_px"] / total_px, 3),
|
||||
"upper_court": round(self.vertical["upper_court_px"] / total_px, 3),
|
||||
"middle_court": round(self.vertical["middle_court_px"] / total_px, 3),
|
||||
"lower_court": round(self.vertical["lower_court_px"] / total_px, 3),
|
||||
},
|
||||
},
|
||||
"seven_eyes": {
|
||||
"eye_width_cm": round(self.eye_width_cm, 2),
|
||||
"face_width_cm": round(self.face_width_cm, 2),
|
||||
"inter_eye_distance_cm": round(self.inter_eye_cm, 2),
|
||||
"ratios": {
|
||||
"eye_width": round(self.eyes["eye_width_px"] / fw_px, 3),
|
||||
"inter_eye_distance": round(self.eyes["inter_eye_distance_px"] / fw_px, 3),
|
||||
},
|
||||
},
|
||||
"landmarks": {
|
||||
"hair_top": pt("hair_top"),
|
||||
"hairline": pt("hairline"),
|
||||
"brow_center": pt("brow_center"),
|
||||
"nose_bottom": pt("nose_bottom"),
|
||||
"chin_tip": pt("chin_tip"),
|
||||
},
|
||||
"hairline_source": self.hairline_source,
|
||||
}
|
||||
if self.head_pose is not None:
|
||||
yaw, pitch, roll = self.head_pose
|
||||
data["head_pose"] = {
|
||||
"yaw": round(yaw, 2), "pitch": round(pitch, 2), "roll": round(roll, 2),
|
||||
}
|
||||
return data
|
||||
|
||||
|
||||
def measure_face(landmarks, hair_mask, image_width, image_height, head_pose=None):
|
||||
"""主入口：纵向决策 + 七眼 + 尺度换算 → MeasureResult。"""
|
||||
vertical, source = decide_vertical(landmarks, image_width, image_height, hair_mask)
|
||||
eyes = measure_seven_eyes(landmarks, image_width, image_height)
|
||||
px_per_cm = estimate_scale_factor(landmarks, image_width, image_height)
|
||||
return MeasureResult(vertical, eyes, px_per_cm, source, head_pose)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
import json
|
||||
import cv2
|
||||
from face_analysis.detector import detector
|
||||
from face_analysis.pose import estimate_head_pose
|
||||
|
||||
path = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg"
|
||||
img = cv2.imread(path)
|
||||
if img is None:
|
||||
print(f"无法读取图片: {path}")
|
||||
sys.exit(1)
|
||||
h, w = img.shape[:2]
|
||||
lms = detector.detect(img)
|
||||
if lms is None:
|
||||
print("未检出人脸")
|
||||
sys.exit(1)
|
||||
|
||||
# 尝试分割（若 torch 不可用则走方案 A）
|
||||
mask = None
|
||||
try:
|
||||
from face_analysis.hair_segmenter import get_segmenter
|
||||
mask = get_segmenter().segment_hair(img)
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f"[warn] 分割不可用，回退方案 A：{e}")
|
||||
|
||||
pose = estimate_head_pose(lms, w, h)
|
||||
result = measure_face(lms, mask, w, h, head_pose=pose)
|
||||
print(json.dumps(result.to_response(), ensure_ascii=False, indent=2))
|
||||
@@ -0,0 +1,101 @@
|
||||
"""头部姿态估计（cv2.solvePnP）+ 正面照校验。
|
||||
|
||||
用通用 3D 头模与 6 个 MediaPipe 关键点求解欧拉角（yaw/pitch/roll，单位：度），
|
||||
阈值即可写成业务可读的「yaw>15° 拒绝」，并把角度返回前端做拍照引导。
|
||||
详见技术方案 §9。
|
||||
"""
|
||||
import os
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
|
||||
from face_analysis.face_mesh_landmarks import PNP_INDICES
|
||||
|
||||
# 正面照判定阈值（度），可由环境变量覆盖，便于上线后按真实数据标定（见技术方案 §11）。
|
||||
# ⚠️ 标定说明：基于通用 6 点 3D 头模 + solvePnP，对明显正面但相机略带俯仰/个体
|
||||
# 脸型差异的真实照片，解出的 yaw/pitch 常落在 15~25°（roll 较稳定，多在 5° 内）。
|
||||
# 因此默认阈值放宽到 30°，只拦截明显侧脸（真实侧脸 yaw 通常 40°+），
|
||||
# 避免误杀正常上传图。生产可通过环境变量随时收紧/放宽，无需改代码。
|
||||
YAW_THRESHOLD = float(os.getenv("FRONTAL_YAW_THR", "30"))
|
||||
PITCH_THRESHOLD = float(os.getenv("FRONTAL_PITCH_THR", "30"))
|
||||
ROLL_THRESHOLD = float(os.getenv("FRONTAL_ROLL_THR", "30"))
|
||||
|
||||
# 通用 3D 头部模型（单位 mm，近似），与 PNP_INDICES 一一对应：
|
||||
# 鼻尖(1) / 下巴(152) / 左眼外角(33) / 右眼外角(263) / 左嘴角(61) / 右嘴角(291)
|
||||
# ⚠️ 采用「相机坐标系」约定：x 向右、y 向下、z 向场景内（远离观察者）。
|
||||
# 与 MediaPipe 像素坐标（y 下）一致，且 +z 指向人脸背面，
|
||||
# 这样正面照解出的旋转矩阵≈单位阵，欧拉角≈0。
|
||||
# 若只翻 y 不翻 z（或都不翻），会残留 ~180° 翻转使正面图被误判。
|
||||
_MODEL_POINTS = np.array([
|
||||
(0.0, 0.0, 0.0), # 鼻尖
|
||||
(0.0, 63.6, 12.5), # 下巴（在鼻尖下方 → y 正）
|
||||
(-43.3, -32.7, 26.0), # 左眼外角（在鼻尖上方 → y 负，且凹于鼻尖 → z 正）
|
||||
(43.3, -32.7, 26.0), # 右眼外角
|
||||
(-28.9, 28.9, 24.1), # 左嘴角
|
||||
(28.9, 28.9, 24.1), # 右嘴角
|
||||
], dtype=np.float64)
|
||||
|
||||
|
||||
def estimate_head_pose(landmarks, image_width, image_height):
|
||||
"""求解头部欧拉角，返回 (yaw, pitch, roll)（度）。solvePnP 失败返回 None。"""
|
||||
lm = landmarks.landmark if hasattr(landmarks, "landmark") else landmarks
|
||||
image_points = np.array([
|
||||
(lm[i].x * image_width, lm[i].y * image_height)
|
||||
for i in PNP_INDICES
|
||||
], dtype=np.float64)
|
||||
|
||||
focal = float(image_width) # 近似焦距
|
||||
cam_matrix = np.array([[focal, 0, image_width / 2],
|
||||
[0, focal, image_height / 2],
|
||||
[0, 0, 1]], dtype=np.float64)
|
||||
dist = np.zeros((4, 1)) # 假设无畸变
|
||||
|
||||
success, rvec, _tvec = cv2.solvePnP(
|
||||
_MODEL_POINTS, image_points, cam_matrix, dist,
|
||||
flags=cv2.SOLVEPNP_ITERATIVE,
|
||||
)
|
||||
if not success:
|
||||
return None
|
||||
rot, _ = cv2.Rodrigues(rvec)
|
||||
# 在「相机坐标系」(x右 y下 z内) 下抽取 Tait-Bryan 欧拉角，物理含义对齐：
|
||||
# yaw = 绕 Y（竖轴）转 → 左右扭头
|
||||
# pitch = 绕 X（横轴）转 → 上下点头
|
||||
# roll = 绕 Z（光轴）转 → 面内倾斜
|
||||
sy = (rot[0, 0] ** 2 + rot[1, 0] ** 2) ** 0.5
|
||||
yaw = float(np.degrees(np.arctan2(-rot[2, 0], sy)))
|
||||
pitch = float(np.degrees(np.arctan2(rot[2, 1], rot[2, 2])))
|
||||
roll = float(np.degrees(np.arctan2(rot[1, 0], rot[0, 0])))
|
||||
return yaw, pitch, roll
|
||||
|
||||
|
||||
def check_frontal_face(landmarks, image_width, image_height,
|
||||
yaw_thr=YAW_THRESHOLD, pitch_thr=PITCH_THRESHOLD,
|
||||
roll_thr=ROLL_THRESHOLD):
|
||||
"""正面照判定：yaw/pitch/roll 绝对值均在阈值内才算正面。
|
||||
|
||||
solvePnP 解算失败时返回 True（不拦截，交由后续逻辑），避免误杀。
|
||||
"""
|
||||
pose = estimate_head_pose(landmarks, image_width, image_height)
|
||||
if pose is None:
|
||||
return True
|
||||
yaw, pitch, roll = pose
|
||||
return abs(yaw) <= yaw_thr and abs(pitch) <= pitch_thr and abs(roll) <= roll_thr
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
from face_analysis.detector import detector
|
||||
|
||||
path = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg"
|
||||
img = cv2.imread(path)
|
||||
if img is None:
|
||||
print(f"无法读取图片: {path}")
|
||||
sys.exit(1)
|
||||
h, w = img.shape[:2]
|
||||
lms = detector.detect(img)
|
||||
if lms is None:
|
||||
print("未检出人脸")
|
||||
sys.exit(1)
|
||||
yaw, pitch, roll = estimate_head_pose(lms, w, h)
|
||||
frontal = check_frontal_face(lms, w, h)
|
||||
print(f"yaw={yaw:.2f} pitch={pitch:.2f} roll={roll:.2f} frontal={frontal}")
|
||||
Reference in New Issue
Block a user