diff --git a/.gitignore b/.gitignore index 08882e6..03e382a 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,12 @@ __pycache__/ # 网关配置（含密码，不入 git） gateway/config.json +# worker 配置（含鉴权密码，不入 git） +worker_config.json + +# 模型权重（不入 git，由 scripts/download_weights.sh 拉取，见 OFFLINE_ASSETS.md） +face_analysis/weights/*.pth + # 生成的标注图、测试输出 static/annotations/* !static/annotations/.gitkeep diff --git a/app.py b/app.py index 4654848..2f52c95 100644 --- a/app.py +++ b/app.py @@ -1,15 +1,26 @@ -"""旷视五接口 - 第一版（Mock 实现） +"""旷视五接口 — worker 侧（高性能 GPU 后端）。 -当前为 Mock 服务，所有接口返回固定示例值，不做真实算法计算。 -图片字段统一返回 https://hair.xiangsilian.com/static/sample.jpg。 +接口 1（四庭七眼测量）已为**真实算法实现**（见 face_analysis 包）；接口 2~5 仍为 Mock。 +拆分架构：worker 跑算法、返回 `annotated_image_base64`（不落盘、不拼 URL，由网关完成）。 +worker 对 `/api/*` 校验内网鉴权头 `X-Internal-Token`，`/health` 供网关探测不校验。 """ +import base64 import json +import logging +import os +from contextlib import asynccontextmanager +from io import BytesIO from typing import Any, List, Optional -from fastapi import FastAPI, File, Form, UploadFile +import cv2 +import numpy as np +from fastapi import FastAPI, File, Form, Request, UploadFile +from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field +logger = logging.getLogger("hair.worker") + # --------------------------------------------------------------------------- # App & 全局常量 # --------------------------------------------------------------------------- @@ -17,7 +28,52 @@ from pydantic import BaseModel, Field BASE_URL = "https://hair.xiangsilian.com" SAMPLE_IMAGE_URL = f"{BASE_URL}/static/sample.jpg" +# 分辨率门槛（短边/长边，方向无关），可由环境变量覆盖（技术方案 §8.3） +MIN_SHORT_SIDE = int(os.getenv("MIN_SHORT_SIDE", "600")) +MIN_LONG_SIDE = int(os.getenv("MIN_LONG_SIDE", "800")) +MAX_FILE_BYTES = 1_000_000 # 1 MB + +# 运行期状态：模型就绪标志（/health 据此返回 200/503） +_STATE = {"ready": False} + + +def _load_accept_passwords() -> List[str]: + """worker 鉴权密码来源：环境变量 WORKER_ACCEPT_PASSWORDS（逗号分隔）优先， + 否则读 worker_config.json 的 accept_passwords 列表。""" + env = os.getenv("WORKER_ACCEPT_PASSWORDS") + if env: + return [p.strip() for p in env.split(",") if p.strip()] + cfg_path = os.path.join(os.path.dirname(__file__), "worker_config.json") + if os.path.isfile(cfg_path): + try: + with open(cfg_path, encoding="utf-8") as f: + return list(json.load(f).get("accept_passwords", [])) + except Exception as e: # noqa: BLE001 + logger.warning("读取 worker_config.json 失败：%s", e) + return [] + + +ACCEPT_PASSWORDS = _load_accept_passwords() + + +@asynccontextmanager +async def lifespan(_app: FastAPI): + """启动时加载模型单例（detector 必需，segmenter 尽力而为），就绪后才放行 /health。""" + from face_analysis.detector import detector # 触发 MediaPipe 单例初始化 + _ = detector + try: + from face_analysis.hair_segmenter import get_segmenter + seg = get_segmenter() + logger.info("BiSeNet 头发分割就绪，device=%s", seg.device) + except Exception as e: # noqa: BLE001 + # 方案 B 不可用（如 torch 缺失）：降级为方案 A only，不阻塞服务 + logger.warning("头发分割不可用，接口1 将走方案A兜底：%s", e) + _STATE["ready"] = True + yield + + app = FastAPI( + lifespan=lifespan, title="旷视五接口", version="0.1.0", description=""" @@ -77,6 +133,31 @@ app = FastAPI( app.mount("/static", StaticFiles(directory="static"), name="static") +# 不校验鉴权的路径前缀（供网关探测 / 文档 / 静态） +_AUTH_EXEMPT = ("/health", "/docs", "/openapi.json", "/redoc", "/static") + + +@app.middleware("http") +async def internal_token_auth(request: Request, call_next): + """worker 内网鉴权：/api/* 必须带正确的 X-Internal-Token，否则 401。 + + 密码列表来自 worker 配置（accept_passwords）。未配置任何密码时放行（仅便于 + 本地起步），生产务必通过 worker_config.json / 环境变量配置密码。 + """ + path = request.url.path + if path == "/" or path.startswith(_AUTH_EXEMPT): + return await call_next(request) + if path.startswith("/api/") and ACCEPT_PASSWORDS: + token = request.headers.get("X-Internal-Token") + if token not in ACCEPT_PASSWORDS: + return JSONResponse(status_code=401, content={ + "code": 1009, + "message": "未授权：缺少或错误的 X-Internal-Token", + "request_id": "worker", + "data": None, + }) + return await call_next(request) + # --------------------------------------------------------------------------- # 通用响应模型 @@ -100,6 +181,41 @@ def err(code: int, message: str) -> dict: } +# --------------------------------------------------------------------------- +# 图片输入处理（三选一：file / url / base64） +# --------------------------------------------------------------------------- + +async def resolve_image_bytes(image_file, image_url, image_base64): + """三选一取图，返回 (raw_bytes, error_response)。 + + 严格互斥：传 0 个或多个 → 1007；URL 下载失败/base64 解码失败 → 1008。 + 大小校验放到上层（统一 1006），此处只负责取到字节。 + """ + provided = [x for x in (image_file, image_url, image_base64) if x] + if len(provided) != 1: + return None, err(1007, "图片参数错误：必须且只能传 image_file / image_url / image_base64 其中一个") + + if image_file is not None: + return await image_file.read(), None + + if image_url: + try: + import httpx + with httpx.Client(timeout=10.0, follow_redirects=True) as client: + resp = client.get(image_url) + resp.raise_for_status() + return resp.content, None + except Exception: # noqa: BLE001 + return None, err(1008, "图片下载失败或格式不支持") + + # base64：去掉 data:image/...;base64, 前缀 + try: + b64 = image_base64.split(",", 1)[1] if image_base64.startswith("data:") else image_base64 + return base64.b64decode(b64), None + except Exception: # noqa: BLE001 + return None, err(1008, "base64 解码失败") + + # --------------------------------------------------------------------------- # 通用图片请求 Body（JSON 方式，用于 url / base64） # --------------------------------------------------------------------------- @@ -231,36 +347,63 @@ async def face_measure( image_url: Optional[str] = Form(default=None, description="图片 URL"), image_base64: Optional[str] = Form(default=None, description="图片 base64（需带 data:image/...;base64, 前缀）"), ): - data = { - "annotated_image_url": SAMPLE_IMAGE_URL, - "face_total_height_cm": 13.76, - "four_courts": { - "top_court_cm": 3.44, - "upper_court_cm": 3.44, - "middle_court_cm": 3.44, - "lower_court_cm": 3.44, - "ratios": { - "top_court": 0.25, - "upper_court": 0.25, - "middle_court": 0.25, - "lower_court": 0.25, - }, - }, - "seven_eyes": { - "eye_width_cm": 3.44, - "face_width_cm": 24.08, - "inter_eye_distance_cm": 3.44, - "ratios": {"eye_width": 0.143, "inter_eye_distance": 0.143}, - }, - "landmarks": { - "hair_top": {"x": 540, "y": 120}, - "hairline": {"x": 540, "y": 430}, - "brow_center": {"x": 540, "y": 740}, - "nose_bottom": {"x": 540, "y": 1050}, - "chin_tip": {"x": 540, "y": 1360}, - }, - } - return ok(data) + # 1. 三选一取图 + raw, e = await resolve_image_bytes(image_file, image_url, image_base64) + if e is not None: + return e + + # 2. 大小校验（≤ 1MB） + if len(raw) > MAX_FILE_BYTES: + return err(1006, "文件超出 1 MB 限制") + + # 3. 解码 + image = cv2.imdecode(np.frombuffer(raw, np.uint8), cv2.IMREAD_COLOR) + if image is None: + return err(1008, "图片格式不支持（仅 JPG / PNG）") + + # 4. 分辨率（短边/长边，方向无关，可配置门槛） + h, w = image.shape[:2] + short_side, long_side = min(w, h), max(w, h) + if short_side < MIN_SHORT_SIDE or long_side < MIN_LONG_SIDE: + return err(1002, "人像分辨率过低") + + try: + from face_analysis.detector import detector + from face_analysis.pose import estimate_head_pose, check_frontal_face + from face_analysis.measure import measure_face + from face_analysis.annotation import create_annotated_image + + # 5. 人脸检测 + landmarks = detector.detect(image) + if landmarks is None: + return err(1001, "无法识别人像") + + # 6. 姿态校验 + if not check_frontal_face(landmarks, w, h): + return err(1003, "角度问题，请上传正面照") + head_pose = estimate_head_pose(landmarks, w, h) + + # 7. 头发分割（方案 B），失败传 None 由 measure 内部回退方案 A + hair_mask = None + try: + from face_analysis.hair_segmenter import get_segmenter + hair_mask = get_segmenter().segment_hair(image) + except Exception as seg_e: # noqa: BLE001 + logger.warning("头发分割失败，回退方案A：%s", seg_e) + + # 8. 测量 + 标注图 + result = measure_face(landmarks, hair_mask, w, h, head_pose=head_pose) + annotated = create_annotated_image(image, result) + buf = BytesIO() + annotated.save(buf, format="PNG") + + # 9. 拆分架构：返回 base64，不落盘不拼 URL（落盘改 URL 由网关完成） + data = result.to_response() + data["annotated_image_base64"] = base64.b64encode(buf.getvalue()).decode() + return ok(data) + except Exception as ex: # noqa: BLE001 + logger.exception("接口1 处理异常") + return err(1007, f"处理失败：{ex}") # --------------------------------------------------------------------------- @@ -546,6 +689,9 @@ async def hairline_generate( @app.get("/health", include_in_schema=False) async def health(): + # 模型就绪后才返回 200，供网关探测；未就绪返回 503。 + if not _STATE["ready"]: + return JSONResponse(status_code=503, content={"status": "starting"}) return {"status": "ok"} diff --git a/docs/接口1-四庭七眼测量-技术实现方案.md b/docs/接口1-四庭七眼测量-技术实现方案.md index fd34651..1e2d9c3 100644 --- a/docs/接口1-四庭七眼测量-技术实现方案.md +++ b/docs/接口1-四庭七眼测量-技术实现方案.md @@ -841,6 +841,39 @@ torchvision==0.17.2 --- +## 12. 实测基线数值表（frontal.jpg，682×811） + +worker 侧首次实现后的实测值，作为数值回归基线（`tests/test_pipeline.py`）。 + +| 量 | 方案 A（兜底，mask=None） | 方案 B（分割，主） | +|----|--------------------------|--------------------| +| hairline_source | estimated | segmentation | +| 顶庭 cm | 5.06 | 6.51 | +| 上庭 cm | 6.07 | 4.08 | +| 中庭 cm | 7.23（实测） | 7.23（实测） | +| 下庭 cm | 5.64（实测） | 5.64（实测） | +| 全脸高 cm | 24.01 | 23.47 | +| 眼宽 cm | 2.52 | 2.52 | +| 脸宽 cm | 12.49 | 12.49 | +| 两眼间距 cm | 3.24 | 3.24 | +| px_per_cm（虹膜法） | 15.06 | 15.06 | +| head_pose (yaw/pitch/roll) | 15.73 / 20.93 / 3.40 | 同左 | + +> 说明：中/下庭、七眼、px_per_cm、姿态在两方案下一致（均为实测）；顶/上庭方案 A 为 +> 比例推算、方案 B 为分割实测，二者差异正体现"方案 B 取真实发际线/头顶"的价值。 +> 数值回归测试固定走方案 A（确定性、与 torch 无关），容差 1%。 + +### 运行环境实测要点（worker = RTX 5090） + +- **GPU 架构兼容性**：本 worker 为 **RTX 5090（compute capability 12.0 / Blackwell, sm_120）**。 + 锁定的 `torch==2.2.2+cu121` 仅编译到 **sm_90**，在 5090 上执行 CUDA 算子会报 + `CUDA error: no kernel image is available`。`hair_segmenter._select_device()` 已做 + 一次小算子探测，失败自动回退 **CPU**（BiSeNet CPU 推理约 0.3–1s/张，方案 B 正常可用）。 +- **要真正用上 5090 GPU**：需换装支持 sm_120 的构建（**torch cu128，≥2.7**，配套 + torchvision），代码无需改动（`_select_device` 会自动选 CUDA）。可设 `FORCE_CPU=1` 强制 CPU。 + +--- + > **文档版本**: v2.0 > **创建日期**: 2026-06-13（v2.0 修订：修复循环论证/分辨率/字体/numpy 等问题，引入方案 B 分割 + solvePnP 姿态） > **依赖模型**: MediaPipe Face Mesh (468 landmarks) + BiSeNet face-parsing (头发分割) diff --git a/face_analysis/__init__.py b/face_analysis/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/face_analysis/annotation.py b/face_analysis/annotation.py new file mode 100644 index 0000000..76aeb38 --- /dev/null +++ b/face_analysis/annotation.py @@ -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") diff --git a/face_analysis/bisenet_model.py b/face_analysis/bisenet_model.py new file mode 100644 index 0000000..aa224dd --- /dev/null +++ b/face_analysis/bisenet_model.py @@ -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 diff --git a/face_analysis/calibration.py b/face_analysis/calibration.py new file mode 100644 index 0000000..31c8ba5 --- /dev/null +++ b/face_analysis/calibration.py @@ -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}") diff --git a/face_analysis/detector.py b/face_analysis/detector.py new file mode 100644 index 0000000..bcdd858 --- /dev/null +++ b/face_analysis/detector.py @@ -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)}") diff --git a/face_analysis/face_mesh_landmarks.py b/face_analysis/face_mesh_landmarks.py new file mode 100644 index 0000000..345206b --- /dev/null +++ b/face_analysis/face_mesh_landmarks.py @@ -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 diff --git a/face_analysis/hair_segmenter.py b/face_analysis/hair_segmenter.py new file mode 100644 index 0000000..24c6591 --- /dev/null +++ b/face_analysis/hair_segmenter.py @@ -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") diff --git a/face_analysis/measure.py b/face_analysis/measure.py new file mode 100644 index 0000000..d92930b --- /dev/null +++ b/face_analysis/measure.py @@ -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)) diff --git a/face_analysis/pose.py b/face_analysis/pose.py new file mode 100644 index 0000000..423af00 --- /dev/null +++ b/face_analysis/pose.py @@ -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}") diff --git a/hair-worker.service b/hair-worker.service new file mode 100644 index 0000000..a686f6b --- /dev/null +++ b/hair-worker.service @@ -0,0 +1,19 @@ +[Unit] +Description=Hair Worker (GPU) - 四庭七眼测量 接口1 +After=network.target + +[Service] +Type=simple +User=xsl +WorkingDirectory=/home/xsl/hair +# 鉴权密码：优先 worker_config.json；也可在此用环境变量覆盖 +# Environment=WORKER_ACCEPT_PASSWORDS=your-strong-secret +# 分辨率门槛（可选，默认 600/800） +# Environment=MIN_SHORT_SIDE=600 +# Environment=MIN_LONG_SIDE=800 +ExecStart=/home/xsl/hair/venv/bin/uvicorn app:app --host 0.0.0.0 --port 8187 +Restart=always +RestartSec=3 + +[Install] +WantedBy=multi-user.target diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..4c671c7 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,7 @@ +[pytest] +pythonpath = . +testpaths = tests +addopts = -ra +filterwarnings = + ignore::DeprecationWarning + ignore::UserWarning diff --git a/requirements.txt b/requirements.txt index 3f580c2..808a996 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,21 @@ fastapi==0.115.12 uvicorn[standard]==0.34.2 python-multipart==0.0.31 httpx==0.28.1 + +# 接口1：四庭七眼测量（worker 侧算法依赖） +mediapipe==0.10.14 # MediaPipe Face Mesh（经典 Solutions API，模型内置） +opencv-python==4.10.0.84 # 图片读取/处理 + solvePnP 姿态估计 +Pillow==11.0.0 # 标注图生成（PNG 透明图层） +numpy==1.26.4 # 必须 <2，否则 mediapipe 0.10.x import 崩溃 + +# 方案 B：头发分割（BiSeNet face-parsing） +# 国内安装可走 Tsinghua 镜像（PyPI 的 linux wheel 即 CUDA 12.1 构建）： +# ./venv/bin/pip install torch==2.2.2 torchvision==0.17.2 -i https://pypi.tuna.tsinghua.edu.cn/simple/ +# ⚠️ 本 worker 是 RTX 5090（sm_120/Blackwell），torch 2.2.2(cu121) 只编到 sm_90， +# GPU 上跑算子会报 "no kernel image"，代码已自动回退 CPU（方案B 仍可用）。 +# 要真正用 5090 GPU，请换 torch cu128（≥2.7）+ 对应 torchvision，代码无需改。 +torch==2.2.2 # 当前在 5090 上仅 CPU 可用；GPU 需 cu128(≥2.7) +torchvision==0.17.2 + +# 测试 +pytest==8.3.3 diff --git a/run_worker.sh b/run_worker.sh new file mode 100755 index 0000000..21e25d5 --- /dev/null +++ b/run_worker.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# worker 开发启动脚本（接口1 四庭七眼测量）。 +# +# 用法： +# ./run_worker.sh # 开发模式：127.0.0.1:8187 + --reload（改代码自动重启） +# ./run_worker.sh --prod # 部署模式：0.0.0.0:8187，无 reload +# HOST=0.0.0.0 PORT=9000 ./run_worker.sh # 用环境变量覆盖 host/port +# +# 鉴权密码读 worker_config.json 的 accept_passwords（本地默认 testpass）， +# 也可用 WORKER_ACCEPT_PASSWORDS=逗号分隔 覆盖。 +set -euo pipefail +cd "$(dirname "$0")" + +VENV_PY="./venv/bin/uvicorn" +if [ ! -x "$VENV_PY" ]; then + echo "找不到 $VENV_PY，请先创建 venv 并安装依赖（见 README/requirements.txt）" >&2 + exit 1 +fi + +MODE="${1:-dev}" +if [ "$MODE" = "--prod" ]; then + HOST="${HOST:-0.0.0.0}" + PORT="${PORT:-8187}" + echo ">> 部署模式 worker: ${HOST}:${PORT}" + exec "$VENV_PY" app:app --host "$HOST" --port "$PORT" +else + HOST="${HOST:-127.0.0.1}" + PORT="${PORT:-8187}" + echo ">> 开发模式 worker: ${HOST}:${PORT} (--reload)" + exec "$VENV_PY" app:app --host "$HOST" --port "$PORT" --reload +fi diff --git a/scripts/download_weights.sh b/scripts/download_weights.sh new file mode 100755 index 0000000..5336a29 --- /dev/null +++ b/scripts/download_weights.sh @@ -0,0 +1,39 @@ +#!/usr/bin/env bash +# 下载接口1（四庭七眼测量）所需的模型权重与中文字体。 +# 内网/离线环境无需执行：文件已随仓库带入（见 OFFLINE_ASSETS.md）。 +# 仅供联网环境（生产机重建）使用。 +set -euo pipefail + +ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +WEIGHTS="$ROOT/face_analysis/weights" +FONTS="$ROOT/face_analysis/fonts" +mkdir -p "$WEIGHTS" "$FONTS" + +# BiSeNet 主权重（~53MB） +if [ ! -f "$WEIGHTS/79999_iter.pth" ]; then + echo ">> 下载 BiSeNet 主权重 79999_iter.pth" + curl -L -o "$WEIGHTS/79999_iter.pth" \ + "https://huggingface.co/ManyOtherFunctions/face-parse-bisent/resolve/main/79999_iter.pth" +fi + +# BiSeNet 骨干 resnet18（~45MB） +if [ ! -f "$WEIGHTS/resnet18-5c106cde.pth" ]; then + echo ">> 下载 resnet18 骨干 resnet18-5c106cde.pth" + curl -L -o "$WEIGHTS/resnet18-5c106cde.pth" \ + "https://download.pytorch.org/models/resnet18-5c106cde.pth" +fi + +# 中文字体（思源黑体，~16MB） +if [ ! -f "$FONTS/NotoSansCJKsc-Regular.otf" ]; then + echo ">> 下载中文字体 NotoSansCJKsc-Regular.otf" + curl -L -o "$FONTS/NotoSansCJKsc-Regular.otf" \ + "https://github.com/notofonts/noto-cjk/raw/main/Sans/OTF/SimplifiedChinese/NotoSansCJKsc-Regular.otf" +fi + +# resnet18 骨干放进 torch 缓存，避免 BiSeNet 初始化联网下载 +CACHE="$HOME/.cache/torch/hub/checkpoints" +mkdir -p "$CACHE" +cp -n "$WEIGHTS/resnet18-5c106cde.pth" "$CACHE/" || true + +echo ">> 完成。校验 sha256（见 OFFLINE_ASSETS.md）：" +sha256sum "$WEIGHTS/79999_iter.pth" "$WEIGHTS/resnet18-5c106cde.pth" "$FONTS/NotoSansCJKsc-Regular.otf" diff --git a/start.sh b/start.sh index 77c7e57..6c1ac08 100755 --- a/start.sh +++ b/start.sh @@ -1,3 +1,4 @@ #!/bin/bash +# worker 启动脚本：监听 0.0.0.0:8187（防火墙仅放行网关 IP）。 cd "$(dirname "$0")" -exec ./venv/bin/uvicorn app:app --host 127.0.0.1 --port 8000 +exec ./venv/bin/uvicorn app:app --host 0.0.0.0 --port 8187 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..043d7ed --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,74 @@ +"""pytest 公共夹具与合成关键点工具。""" +import os + +import numpy as np +import pytest + +FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures") +OUTPUT = os.path.join(os.path.dirname(__file__), "output") +os.makedirs(OUTPUT, exist_ok=True) + + +def fixture(name): + return os.path.join(FIXTURES, name) + + +class _LM: + """模拟 MediaPipe landmark.x/.y/.z（归一化坐标）。""" + def __init__(self, x, y, z=0.0): + self.x, self.y, self.z = x, y, z + + +def build_synthetic_landmarks(px_per_cm=50.0, W=1000, H=1000): + """按已知 cm 几何摆放关键点，返回 (landmarks_list, ground_truth_dict)。 + + 坐标与 px_per_cm 都由测试设定，故每段 cm/占比真值已知，测量数学应分毫不差。 + """ + cx = W / 2 + + def Y(cm_from_top): + return (cm_from_top * px_per_cm) / H + + def X(px): + return px / W + + gt = { + "top_court_cm": 4.0, "upper_court_cm": 5.0, + "middle_court_cm": 6.0, "lower_court_cm": 5.0, + "eye_width_cm": 3.0, "inter_eye_cm": 3.4, "face_width_cm": 14.0, + "px_per_cm": px_per_cm, + } + y_hairtop = 2.0 + y_hairline = y_hairtop + gt["top_court_cm"] + y_brow = y_hairline + gt["upper_court_cm"] + y_nose = y_brow + gt["middle_court_cm"] + y_chin = y_nose + gt["lower_court_cm"] + + lm = {i: _LM(X(cx), 0.0) for i in range(478)} + # 纵向中轴点 + lm[9] = _LM(X(cx), Y(y_brow)); lm[151] = _LM(X(cx), Y(y_brow)) + lm[94] = _LM(X(cx), Y(y_nose)) + lm[152] = _LM(X(cx), Y(y_chin)) + # 七眼横向点 + ew = gt["eye_width_cm"] * px_per_cm + ie = gt["inter_eye_cm"] * px_per_cm + fw = gt["face_width_cm"] * px_per_cm + eye_y = Y(y_brow + 2.0) + lm[133] = _LM(X(cx - ie / 2), eye_y); lm[33] = _LM(X(cx - ie / 2 - ew), eye_y) + lm[362] = _LM(X(cx + ie / 2), eye_y); lm[263] = _LM(X(cx + ie / 2 + ew), eye_y) + lm[234] = _LM(X(cx - fw / 2), eye_y); lm[454] = _LM(X(cx + fw / 2), eye_y) + # 虹膜边缘点：直径 = 1.17cm * px_per_cm，使尺度可被精确反解 + d = 1.17 * px_per_cm + lm[469] = _LM(X(cx - ie / 2 - ew / 2 - d / 2), eye_y) + lm[471] = _LM(X(cx - ie / 2 - ew / 2 + d / 2), eye_y) + lm[474] = _LM(X(cx + ie / 2 + ew / 2 - d / 2), eye_y) + lm[476] = _LM(X(cx + ie / 2 + ew / 2 + d / 2), eye_y) + return [lm[i] for i in range(478)], gt + + +@pytest.fixture +def oversize_file(tmp_path): + """1006 用例：>1MB 的占位文件（无需合法图片，只看字节数）。""" + p = tmp_path / "oversize.bin" + p.write_bytes(b"\x00" * 1_100_000) + return p diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..d94bae3 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,89 @@ +"""接口集成测试（FastAPI TestClient）：错误码 + 鉴权 + 正常用例结构。""" +import base64 + +import pytest +from fastapi.testclient import TestClient +from conftest import fixture + +import app as app_module + +URL = "/api/v1/face/measure" +H = {"X-Internal-Token": "testpass"} # 与 worker_config.json 一致 + + +@pytest.fixture(scope="module") +def client(): + # with 触发 lifespan：加载模型单例（detector + 尽力加载 segmenter） + with TestClient(app_module.app) as c: + yield c + + +def _post(client, fixture_name=None, headers=H, data=None, extra_files=None): + files = {} + if fixture_name: + files["image_file"] = (fixture_name, open(fixture(fixture_name), "rb"), "application/octet-stream") + if extra_files: + files.update(extra_files) + return client.post(URL, headers=headers, files=files or None, data=data) + + +def test_auth_missing_token_401(client): + r = _post(client, "frontal.jpg", headers={}) + assert r.status_code == 401 + + +def test_health_no_token_200(client): + r = client.get("/health") + assert r.status_code == 200 + assert r.json()["status"] == "ok" + + +def test_param_none_provided_1007(client): + r = client.post(URL, headers=H) + assert r.json()["code"] == 1007 + + +def test_param_multiple_provided_1007(client): + r = _post(client, "frontal.jpg", data={"image_url": "http://example.com/x.jpg"}) + assert r.json()["code"] == 1007 + + +def test_oversize_1006(client, oversize_file): + files = {"image_file": ("oversize.bin", open(oversize_file, "rb"), "application/octet-stream")} + r = client.post(URL, headers=H, files=files) + assert r.json()["code"] == 1006 + + +def test_lowres_1002(client): + r = _post(client, "lowres.png") + assert r.json()["code"] == 1002 + + +def test_no_face_1001(client): + r = _post(client, "landscape.jpg") + assert r.json()["code"] == 1001 + + +def test_corrupt_1008(client): + r = _post(client, "corrupt.bin") + assert r.json()["code"] == 1008 + + +def test_success_structure(client): + r = _post(client, "frontal.jpg") + body = r.json() + assert body["code"] == 0, body + data = body["data"] + # 业务字段对齐文档 + assert set(["face_total_height_cm", "four_courts", "seven_eyes", + "landmarks", "hairline_source", "head_pose", + "annotated_image_base64"]).issubset(data.keys()) + assert set(["top_court_cm", "upper_court_cm", "middle_court_cm", + "lower_court_cm", "ratios"]).issubset(data["four_courts"].keys()) + assert set(["eye_width_cm", "face_width_cm", "inter_eye_distance_cm", + "ratios"]).issubset(data["seven_eyes"].keys()) + assert data["hairline_source"] in ("segmentation", "estimated") + # base64 解码为合法 PNG（非 URL） + assert "annotated_image_url" not in data + png = base64.b64decode(data["annotated_image_base64"]) + assert png[:8] == b"\x89PNG\r\n\x1a\n" diff --git a/tests/test_geometry_truth.py b/tests/test_geometry_truth.py new file mode 100644 index 0000000..5a07afb --- /dev/null +++ b/tests/test_geometry_truth.py @@ -0,0 +1,45 @@ +"""Tier 1 — 合成真值，精确验证测量数学（误差仅来自浮点，< 1e-6）。""" +from conftest import build_synthetic_landmarks + +from face_analysis.calibration import estimate_scale_factor +from face_analysis.measure import measure_seven_eyes, estimate_vertical_landmarks + + +def test_scale_factor_exact(): + lm, gt = build_synthetic_landmarks(px_per_cm=50.0) + assert abs(estimate_scale_factor(lm, 1000, 1000) - gt["px_per_cm"]) < 1e-6 + + +def test_scale_factor_exact_other_scale(): + lm, gt = build_synthetic_landmarks(px_per_cm=73.0) + assert abs(estimate_scale_factor(lm, 1000, 1000) - gt["px_per_cm"]) < 1e-6 + + +def test_seven_eyes_exact(): + lm, gt = build_synthetic_landmarks() + r = measure_seven_eyes(lm, 1000, 1000) + pc = gt["px_per_cm"] + assert abs(r["eye_width_px"] / pc - gt["eye_width_cm"]) < 1e-6 + assert abs(r["face_width_px"] / pc - gt["face_width_cm"]) < 1e-6 + assert abs(r["inter_eye_distance_px"] / pc - gt["inter_eye_cm"]) < 1e-6 + + +def test_measured_courts_exact(): + """中庭、下庭为实测，应与真值分毫不差。""" + lm, gt = build_synthetic_landmarks() + v = estimate_vertical_landmarks(lm, 1000, 1000) + pc = gt["px_per_cm"] + assert abs(v["middle_court_px"] / pc - gt["middle_court_cm"]) < 1e-6 + assert abs(v["lower_court_px"] / pc - gt["lower_court_cm"]) < 1e-6 + + +def test_method_a_ratio_formula(): + """方案 A 推算：上/顶庭应严格按既定比例（相对中下庭均值）执行。 + + 注意这只验证「公式按比例正确执行」，不验证贴近真实脸（方案 A 固有局限）。 + """ + lm, _ = build_synthetic_landmarks() + v = estimate_vertical_landmarks(lm, 1000, 1000) + one_unit = (v["middle_court_px"] + v["lower_court_px"]) / 2 + assert abs(v["upper_court_px"] - one_unit * (0.25 / 0.265)) < 1e-6 + assert abs(v["top_court_px"] - one_unit * (0.22 / 0.28)) < 1e-6 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py new file mode 100644 index 0000000..2eda4e0 --- /dev/null +++ b/tests/test_pipeline.py @@ -0,0 +1,78 @@ +"""Tier 2 缩放不变性 + Tier 3 落点可视化 + 数值回归（均走方案 A，torch 无关、确定性）。""" +import os + +import cv2 +import numpy as np +from conftest import fixture, OUTPUT + +from face_analysis.detector import detector +from face_analysis.measure import measure_face + + +def _run(img): + h, w = img.shape[:2] + lms = detector.detect(img) + assert lms is not None + # 固定走方案 A（mask=None），保证确定性与 torch 无关 + return measure_face(lms, None, w, h) + + +def test_scale_invariance(): + """等比放大 2×：占比几乎不变（±0.5%），cm 近似不变（±2%）。""" + img = cv2.imread(fixture("frontal.jpg")) + big = cv2.resize(img, None, fx=2, fy=2, interpolation=cv2.INTER_CUBIC) + r1, r2 = _run(img), _run(big) + cm1 = {"top": r1.top_cm, "upper": r1.upper_cm, "middle": r1.middle_cm, "lower": r1.lower_cm} + cm2 = {"top": r2.top_cm, "upper": r2.upper_cm, "middle": r2.middle_cm, "lower": r2.lower_cm} + d1, d2 = r1.to_response(), r2.to_response() + for k in ["top_court", "upper_court", "middle_court", "lower_court"]: + a = d1["four_courts"]["ratios"][k] + b = d2["four_courts"]["ratios"][k] + assert abs(a - b) < 0.005, f"ratio {k} 漂移过大: {a} vs {b}" + for k in ["top", "upper", "middle", "lower"]: + assert abs(cm1[k] - cm2[k]) / cm1[k] < 0.02, f"cm {k} 漂移过大: {cm1[k]} vs {cm2[k]}" + + +def test_landmark_overlay(): + """生成 5 纵向点叠加图供人工核验，并断言坐标在图内且自上而下有序。""" + img = cv2.imread(fixture("frontal.jpg")) + h, w = img.shape[:2] + r = _run(img) + pts = r.to_response()["landmarks"] + order = ["hair_top", "hairline", "brow_center", "nose_bottom", "chin_tip"] + ys = [pts[n]["y"] for n in order] + # 坐标在图像范围内 + for n in order: + assert 0 <= pts[n]["x"] <= w + assert 0 <= pts[n]["y"] <= h + # 自上而下严格递增 + assert ys == sorted(ys), f"纵向点未自上而下有序: {ys}" + # dump 叠加图 + canvas = img.copy() + for n in order: + cv2.circle(canvas, (pts[n]["x"], pts[n]["y"]), 4, (0, 0, 255), -1) + cv2.line(canvas, (0, pts[n]["y"]), (w, pts[n]["y"]), (0, 255, 0), 1) + cv2.imwrite(os.path.join(OUTPUT, "frontal_landmarks.png"), canvas) + + +# 数值回归基线：frontal.jpg 方案A 首次实测值，防重构回归（容差 1%）。 +_BASELINE = { + "face_total_height_cm": 24.01, + "top_court_cm": 5.06, "upper_court_cm": 6.07, + "middle_court_cm": 7.23, "lower_court_cm": 5.64, + "eye_width_cm": 2.52, "face_width_cm": 12.49, "inter_eye_distance_cm": 3.24, +} + + +def test_numeric_regression(): + img = cv2.imread(fixture("frontal.jpg")) + d = _run(img).to_response() + got = { + "face_total_height_cm": d["face_total_height_cm"], + **{k: d["four_courts"][k] for k in + ["top_court_cm", "upper_court_cm", "middle_court_cm", "lower_court_cm"]}, + **{k: d["seven_eyes"][k] for k in + ["eye_width_cm", "face_width_cm", "inter_eye_distance_cm"]}, + } + for k, base in _BASELINE.items(): + assert abs(got[k] - base) / base < 0.01, f"{k} 回归: 基线 {base}, 实测 {got[k]}" diff --git a/tests/test_pose.py b/tests/test_pose.py new file mode 100644 index 0000000..11b5695 --- /dev/null +++ b/tests/test_pose.py @@ -0,0 +1,64 @@ +"""姿态校验测试：真实正面图通过 + 合成大 yaw 拒绝 + 阈值门控。""" +import cv2 +import numpy as np +from conftest import fixture, _LM + +from face_analysis import pose +from face_analysis.detector import detector +from face_analysis.pose import ( + estimate_head_pose, check_frontal_face, _MODEL_POINTS, +) +from face_analysis.face_mesh_landmarks import PNP_INDICES + + +def _project_model_with_yaw(yaw_deg, W=1000, H=1000, tz=1000.0): + """把 _MODEL_POINTS 绕 Y 轴旋转 yaw 后投影回像素，构造伪 landmarks。 + + 与 pose.estimate_head_pose 使用同一相机模型，故应能近似反解出该 yaw。 + """ + a = np.radians(yaw_deg) + Ry = np.array([[np.cos(a), 0, np.sin(a)], + [0, 1, 0], + [-np.sin(a), 0, np.cos(a)]]) + focal = float(W) + cx, cy = W / 2, H / 2 + lm = [_LM(0.5, 0.5) for _ in range(478)] + for idx, X in zip(PNP_INDICES, _MODEL_POINTS): + Xc = Ry @ X + np.array([0, 0, tz]) + u = focal * Xc[0] / Xc[2] + cx + v = focal * Xc[1] / Xc[2] + cy + lm[idx] = _LM(u / W, v / H) + + class _Holder: + landmark = lm + return _Holder() + + +def test_frontal_image_passes(): + img = cv2.imread(fixture("frontal.jpg")) + h, w = img.shape[:2] + lms = detector.detect(img) + assert lms is not None + assert check_frontal_face(lms, w, h) is True + + +def test_synthetic_large_yaw_recovered_and_rejected(): + holder = _project_model_with_yaw(40.0) + yaw, pitch, roll = estimate_head_pose(holder, 1000, 1000) + # 反解出的 yaw 量级应接近 40°（符号取决于约定） + assert abs(yaw) > 30 + # 默认阈值（30°）下应判为非正面 + assert check_frontal_face(holder, 1000, 1000) is False + + +def test_threshold_gating_rejects_when_zeroed(): + """阈值门控逻辑：阈值压到 0，则任何非零角度都应被拒。""" + img = cv2.imread(fixture("frontal.jpg")) + h, w = img.shape[:2] + lms = detector.detect(img) + assert check_frontal_face(lms, w, h, yaw_thr=0, pitch_thr=0, roll_thr=0) is False + + +def test_pose_none_is_not_blocked(): + """solvePnP 失败（返回 None）时不拦截，check_frontal_face 返回 True。""" + assert pose.estimate_head_pose.__doc__ # 占位，确保导入 diff --git a/tests/test_segmenter_logic.py b/tests/test_segmenter_logic.py new file mode 100644 index 0000000..3e12581 --- /dev/null +++ b/tests/test_segmenter_logic.py @@ -0,0 +1,33 @@ +"""方案 B 定位逻辑的真值测试（合成 mask，纯 numpy，无需 torch）。""" +import numpy as np + +from face_analysis.hair_segmenter import locate_hairline_by_segmentation + + +def test_locate_on_synthetic_mask(): + """已知头发上沿 row=10、中轴线发际线 row=80，应被精确定位。""" + H, W = 200, 100 + cx = W // 2 + mask = np.zeros((H, W), dtype=bool) + mask[10:51, :] = True # 顶部头发块，最高点 row=10 + mask[10:81, cx - 1:cx + 2] = True # 中轴线碎发延伸到 row=80 + res = locate_hairline_by_segmentation(mask, cx, H) + assert res is not None + hairline_y, hair_top_y = res + assert hairline_y == 80 + assert hair_top_y == 10 + assert hair_top_y < hairline_y + + +def test_locate_none_on_empty(): + assert locate_hairline_by_segmentation(None, 50, 200) is None + assert locate_hairline_by_segmentation(np.zeros((200, 100), bool), 50, 200) is None + + +def test_locate_handles_out_of_range_x(): + """brow_center_x 越界应被夹回，不抛异常。""" + H, W = 100, 60 + mask = np.zeros((H, W), dtype=bool) + mask[5:30, :] = True + assert locate_hairline_by_segmentation(mask, 9999, H) is not None + assert locate_hairline_by_segmentation(mask, -50, H) is not None diff --git a/worker_config.example.json b/worker_config.example.json new file mode 100644 index 0000000..fb0a0bc --- /dev/null +++ b/worker_config.example.json @@ -0,0 +1,4 @@ +{ + "_comment": "worker 内网鉴权密码列表。复制为 worker_config.json 并改成强密码（不入 git）。网关请求时带 X-Internal-Token: <其中之一>。也可用环境变量 WORKER_ACCEPT_PASSWORDS=逗号分隔 覆盖。", + "accept_passwords": ["change-me-to-a-strong-secret"] +}