diff --git a/app.py b/app.py index 6539a78..1ddf691 100644 --- a/app.py +++ b/app.py @@ -559,9 +559,9 @@ async def hair_grow( "message": "success", "request_id": "mock-request-id", "data": { - "best_hairline_image_url": SAMPLE_IMAGE_URL, - "hair_growth_image_url": SAMPLE_IMAGE_URL, - "hairline_type": "花瓣形", + "best_hairline_image_base64": "iVBORw0KGgo...（原图）", + "hair_growth_image_base64": "iVBORw0KGgo...（生发图）", + "hairline_type": "custom", }, } } @@ -585,12 +585,46 @@ async def hair_grow_b( original_image_url: Optional[str] = Form(default=None, description="原始用户照片 URL"), original_image_base64: Optional[str] = Form(default=None, description="原始用户照片 base64"), ): - data = { - "best_hairline_image_url": SAMPLE_IMAGE_URL, - "hair_growth_image_url": SAMPLE_IMAGE_URL, - "hairline_type": "花瓣形", - } - return ok(data) + # 1. 两组图各三选一取图 + marked_raw, e = await resolve_image_bytes(marked_image_file, marked_image_url, marked_image_base64) + if e is not None: + return e + orig_raw, e = await resolve_image_bytes(original_image_file, original_image_url, original_image_base64) + if e is not None: + return e + if len(marked_raw) > MAX_FILE_BYTES or len(orig_raw) > MAX_FILE_BYTES: + return err(1006, "文件超出 1 MB 限制") + + marked = cv2.imdecode(np.frombuffer(marked_raw, np.uint8), cv2.IMREAD_COLOR) + original = cv2.imdecode(np.frombuffer(orig_raw, np.uint8), cv2.IMREAD_COLOR) + if marked is None or original is None: + return err(1008, "图片格式不支持（仅 JPG / PNG）") + + h, w = marked.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 fastapi.concurrency import run_in_threadpool + from hairline.service import generate_grow_b + + res = await run_in_threadpool(generate_grow_b, marked, original) + if res["status"] == "no_face": + return err(1001, "无法识别人像") + if res["status"] == "no_line": + return err(1001, "未检测到发际线划线，请确认划线图额头有清晰的手绘发际线") + + grown_b64 = base64.b64encode(res["grown_png"]).decode() if res["grown_png"] else None + data = { + "best_hairline_image_base64": base64.b64encode(orig_raw).decode(), # 原图原样 + "hair_growth_image_base64": grown_b64, + "hairline_type": "custom", + } + return ok(data) + except Exception as ex: # noqa: BLE001 + logger.exception("接口3 处理异常") + return err(1007, f"处理失败：{ex}") # --------------------------------------------------------------------------- diff --git a/docs/接口3-B端生发-技术实现方案.md b/docs/接口3-B端生发-技术实现方案.md new file mode 100644 index 0000000..08f55fc --- /dev/null +++ b/docs/接口3-B端生发-技术实现方案.md @@ -0,0 +1,93 @@ +# 接口 3：B 端生发 — 技术实现方案（马克笔发际线检测 + 生发） + +> 在 **高性能 worker（GPU 机）** 实现，与接口 1/2 同机。对外经网关代理。 +> B 端：医生在患者额头**用马克笔画出规划的发际线**，拍照上传。系统**检测这条手绘线**， +> 据此生成生发图。检测算法移植自 `/home/xsl/headmark` 的调研结论（黑帽 + Dijkstra）。 + +--- + +## 0. 契约（对齐 `接口文档.md` 接口3，不变） + +`POST /api/v1/hair/grow-b` + +| 输入 | 说明 | +|------|------| +| `marked_image_*` | 已划线（医生标注发际线）的图，三选一，必填 | +| `original_image_*` | 原始用户照片，三选一，必填 | + +| 输出 data | 决策 | +|-----------|------| +| `best_hairline_image_url` | **= 原图 original** 原样返回（worker 返回 `best_hairline_image_base64`） | +| `hair_growth_image_url` | **生发后图片**（ComfyUI，worker 返回 `hair_growth_image_base64`） | +| `hairline_type` | 固定 **`"custom"`**（手绘定制） | + +> 落盘改 URL 由网关做（架构同接口1/2）。 + +--- + +## 1. 马克笔发际线检测（核心，源自 headmark 调研） + +headmark `docs/detection_research.md` 结论：全局灰度阈值不可用（笔迹平均灰度反而高于阈值、 +与皮肤阴影分布重叠）；推荐 **黑帽响应图 + 端点锚定 Dijkstra 最小路径**，实测误差 ≤0.5px(GT锚点)。 +本项目用 **MediaPipe 锚点**（非 GT）实测平均 3.2px、中位 0px —— 对生成遮罩足够（线会膨胀成带）。 + +``` +detect_marker_hairline(marked_bgr, landmarks, parse_map): + [1] ROI = forehead_upper_region(landmarks) ∩ head_silhouette(parse_map) # 复用接口2 mask.py + [2] 黑帽响应 bh = MORPH_BLACKHAT(gray, ksize=max(15,int(w*0.025)|1))；ROI 外置 0 + [3] 锚点 = MediaPipe 21(左鬓角)/251(右鬓角)，各自小窗口(≈w*3%)内吸附到 bh 最大处 + [4] 代价 cost = bh.max()-bh+1；ROI 外设 1e6； + path = skimage.graph.route_through_array(cost, 左锚, 右锚, fully_connected, geometric) + [5] 拒识：path 平均 bh 响应 < 阈值(可调) → None（上层返回 1001 "未检测到发际线划线"） + return path # (N,2) row,col +``` + +- 依赖：**`scikit-image==0.24.0`**。⚠️ 0.25+ 强依赖 numpy≥2，会顶掉 mediapipe 的 numpy<2 → + mediapipe/SegFormer 全崩。**必须锁 0.24.x**。 +- 复用接口2：`forehead_upper_region` / `head_silhouette`（`hairline/mask.py`）、SegFormer / MediaPipe 单例。 + +## 2. 遮罩 + 原图重画干净线 + +- **遮罩**：path → 画成 curve_mask → 复用接口2 `_above_curve_region` + `head` + `_clean_mask` + 得到"发际线以上闭合区域"。 +- **ComfyUI 输入图**：用 **原图 original**（按需缩放到 marked 尺寸对齐坐标），**重画一条干净黑线** + （检测 path 膨胀成线宽），避免医生手绘的毛刺/杂线干扰生成。 +- 合成 RGBA：RGB=重画线的原图，alpha=255−mask（透明=重绘区）。复用 `compose_comfy_rgba`。 + +## 3. 生发（复用接口2 ComfyUI 客户端） + +`hairline/comfyui.run(rgba_png)` → 跑 `add_hair.json`（Flux-2）→ 生发图 PNG。同步。 + +## 4. worker handler（`/api/v1/hair/grow-b`） + +``` +1. marked + original 各三选一取图（复用 resolve_image_bytes）+ 校验(大小/解码/分辨率) +2. 在 marked 上：landmarks(MediaPipe)+parse(SegFormer) → detect_marker_hairline + - 无人脸 → 1001；未检测到画线 → 1001 "未检测到发际线划线" +3. 遮罩 + 原图重画线 → RGBA → comfyui.run → 生发图 +4. return ok({ best_hairline_image_base64: 原图, hair_growth_image_base64: 生发图, + hairline_type: "custom" }) +异常 → 1007；重活 run_in_threadpool。 +``` + +## 5. 开发步骤 + +| 阶段 | 内容 | 验证 | +|------|------|------| +| **M1 检测** | `hairline/marker_detect.py`（黑帽+锚点+Dijkstra+拒识） | headmark test_image：检测线贴合真值；无线图被拒识 | +| **M2 遮罩+重画** | path→遮罩(复用) + 原图重画干净线 + RGBA 合成 | 目视：干净线在原图、遮罩贴合 | +| **M3 接 app** | grow-b 真实实现 + 输出字段 + 1001 | curl：best=原图/grown 合法PNG/type=custom；无线→1001 | +| **M4 测试** | 检测/mask 单测 + mock-ComfyUI 集成 + 真机冒烟 | pytest 绿；真机出生发图 | + +## 6. 风险 + +1. **锚点偏差/路径端点偏移**：MediaPipe 21/251 吸附后仍可能在鬓角端有偏移（实测 max~42px，少数点）。 + 膨胀成带 + 遮罩闭合可吸收；必要时改进吸附窗口或端点截断。 +2. **没画线/画线极浅**：靠拒识阈值（路径平均黑帽响应）兜底，阈值需在更多真实图上标定。 +3. **marked 与 original 尺寸/对齐不一致**：按 marked 坐标系处理，original 缩放对齐；若两图非同源（不同姿态）会错位——约定二者为"同一张照片的划线版/原始版"。 +4. **抬头纹/眉毛/发丝干扰**：黑帽 + ROI + Dijkstra 平滑已大幅抑制（调研验证抬头纹零干扰），极端情况可在代价图抑制头发区域。 + +--- + +> **文档版本**: v1.0 ｜ **创建日期**: 2026-06-15 ｜ 检测来源: headmark（黑帽+Dijkstra）｜ +> 生发: 复用接口2 ComfyUI(add_hair.json) ｜ 运行位置: worker(GPU) + 本机 ComfyUI(8182) diff --git a/hairline/marker_detect.py b/hairline/marker_detect.py new file mode 100644 index 0000000..739475b --- /dev/null +++ b/hairline/marker_detect.py @@ -0,0 +1,102 @@ +"""接口3：马克笔手绘发际线检测（黑帽响应图 + 端点锚定 Dijkstra 最小路径）。 + +源自 /home/xsl/headmark 调研结论：全局灰度阈值不可用（笔迹平均灰度反高于阈值、 +与皮肤阴影分布重叠）；黑帽变换响应"比局部邻域暗的细结构"，叠加 ROI + 两鬓角锚点间 +最小代价路径，对抬头纹/眉毛/发丝鲁棒。复用接口2 的 ROI（额头上部 ∩ 头部分割）。 +""" +from __future__ import annotations + +import cv2 +import numpy as np +from skimage.graph import route_through_array + +from .mask import forehead_upper_region, head_silhouette + +# 鬓角锚点（MediaPipe canonical 索引）：21 左、251 右 +ANCHOR_LEFT = 21 +ANCHOR_RIGHT = 251 +# 拒识阈值：路径平均黑帽响应低于此值 → 判"未检测到画线"（待真实图标定） +MIN_MEAN_RESPONSE = 8.0 + + +def _blackhat(gray: np.ndarray, w: int) -> np.ndarray: + k = max(15, int(w * 0.025) | 1) + kernel = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (k, k)) + return cv2.morphologyEx(gray, cv2.MORPH_BLACKHAT, kernel).astype(np.float32) + + +def _snap_anchor(bh_roi: np.ndarray, x: int, y: int, w: int): + """在 (x,y) 周围窗口内吸附到黑帽响应最大处，返回 (row, col)。""" + win = max(8, int(w * 0.03)) + h, ww = bh_roi.shape + x0, x1 = max(0, x - win), min(ww, x + win) + y0, y1 = max(0, y - win), min(h, y + win) + sub = bh_roi[y0:y1, x0:x1] + if sub.size == 0 or sub.max() <= 0: + return (int(np.clip(y, 0, h - 1)), int(np.clip(x, 0, ww - 1))) + dy, dx = np.unravel_index(int(np.argmax(sub)), sub.shape) + return (y0 + dy, x0 + dx) + + +def detect_marker_hairline(marked_bgr: np.ndarray, landmarks_mp: np.ndarray, + parse_map: np.ndarray, min_mean_response: float = MIN_MEAN_RESPONSE): + """检测手绘发际线，返回路径 (N,2) row,col；未检出/被拒识返回 None。""" + h, w = marked_bgr.shape[:2] + roi = cv2.bitwise_and(forehead_upper_region(landmarks_mp, w, h), + head_silhouette(parse_map)) > 0 + if roi.sum() == 0: + return None + + gray = cv2.cvtColor(marked_bgr, cv2.COLOR_BGR2GRAY) + bh = _blackhat(gray, w) + bh_roi = bh * roi + + al = _snap_anchor(bh_roi, int(landmarks_mp[ANCHOR_LEFT, 0] * w), + int(landmarks_mp[ANCHOR_LEFT, 1] * h), w) + ar = _snap_anchor(bh_roi, int(landmarks_mp[ANCHOR_RIGHT, 0] * w), + int(landmarks_mp[ANCHOR_RIGHT, 1] * h), w) + + cost = (bh.max() - bh) + 1.0 + cost[~roi] = 1e6 # 禁止路径走出 ROI + path, _ = route_through_array(cost, al, ar, fully_connected=True, geometric=True) + path = np.asarray(path) + + # 拒识：路径平均黑帽响应过低 → 没画线（强行找出的伪路径） + if float(bh[path[:, 0], path[:, 1]].mean()) < min_mean_response: + return None + return path + + +def path_to_curve_mask(path: np.ndarray, h: int, w: int, thickness: int = 3) -> np.ndarray: + """把路径画成曲线 mask（uint8 0/255），用作遮罩下边界 / 重画干净线。""" + m = np.zeros((h, w), np.uint8) + pts = path[:, ::-1].reshape(-1, 1, 2) # (row,col)→(x,y) + cv2.polylines(m, [pts], False, 255, thickness, lineType=cv2.LINE_AA) + return m + + +if __name__ == "__main__": + import sys + from .service import get_landmarker, get_parser + + path_img = sys.argv[1] if len(sys.argv) > 1 else "/home/xsl/headmark/test_image/input1.png" + img = cv2.imread(path_img) + if img is None: + print(f"无法读取 {path_img}"); sys.exit(1) + h, w = img.shape[:2] + rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + lm = get_landmarker().detect(rgb) + if lm is None: + print("未检出人脸"); sys.exit(1) + pm = get_parser().parse(rgb) + p = detect_marker_hairline(img, lm, pm) + if p is None: + print("未检测到发际线划线（拒识）"); sys.exit(0) + print(f"检测到画线：{len(p)} 点") + vis = img.copy() + cv2.polylines(vis, [p[:, ::-1].reshape(-1, 1, 2)], False, (0, 0, 255), 2) + import os + os.makedirs("tests/output", exist_ok=True) + name = os.path.splitext(os.path.basename(path_img))[0] + cv2.imwrite(f"tests/output/marker_{name}.png", vis) + print(f"saved tests/output/marker_{name}.png") diff --git a/hairline/mask.py b/hairline/mask.py index f726437..4a76550 100644 --- a/hairline/mask.py +++ b/hairline/mask.py @@ -83,27 +83,29 @@ def _clean_mask(mask: np.ndarray, w: int) -> np.ndarray: return out +def mask_from_curve(curve_mask: np.ndarray, landmarks_mp: np.ndarray, + parse_map: np.ndarray) -> np.ndarray: + """由发际线曲线 + ROI(额头上部 ∩ 头部) 围成"曲线以上"闭合遮罩（uint8 0..255）。 + + 接口2(模板渲染曲线) 与 接口3(检测路径曲线) 共用。 + """ + h, w = curve_mask.shape[:2] + roi = cv2.bitwise_and(forehead_upper_region(landmarks_mp, w, h), + head_silhouette(parse_map)) + above = _above_curve_region(curve_mask, h, w) + return _clean_mask(cv2.bitwise_and(roi, above), w) + + def build_inpaint_mask(photo_bgr: np.ndarray, landmarks_mp: np.ndarray, parse_map: np.ndarray, points502: np.ndarray, black_texture_rgba: np.ndarray): - """返回 (marked_bgr 划线图, mask uint8 0..255 重绘区)。""" + """接口2：返回 (marked_bgr 划线图, mask uint8 0..255 重绘区)。""" h, w = photo_bgr.shape[:2] uv, ext_faces = load_ext_mesh() - - # ④ 渲染黑线：marked = 烧进照片；curve_mask = 曲线像素 marked = render_hairline_overlay(photo_bgr, points502, ext_faces, uv, black_texture_rgba) overlay = build_overlay_layer(h, w, points502, ext_faces, uv, black_texture_rgba) curve_mask = (overlay[:, :, 3] > 40).astype(np.uint8) * 255 - - # ①②③ ROI - upper = forehead_upper_region(landmarks_mp, w, h) - head = head_silhouette(parse_map) - roi = cv2.bitwise_and(upper, head) - - # ⑤ ROI ∩ 曲线以上 → 清理 - above = _above_curve_region(curve_mask, h, w) - mask = cv2.bitwise_and(roi, above) - mask = _clean_mask(mask, w) + mask = mask_from_curve(curve_mask, landmarks_mp, parse_map) return marked, mask diff --git a/hairline/service.py b/hairline/service.py index 1130749..aeea964 100644 --- a/hairline/service.py +++ b/hairline/service.py @@ -17,7 +17,8 @@ from .face_parsing import FaceParser from .hairline_2d import sample_hairline, smooth_hairline from .lift_3d import lift_hairline_to_3d, build_middle_row, assemble_full from .render import load_ext_mesh, load_texture_rgba, render_hairline_overlay -from .mask import build_inpaint_mask, compose_comfy_rgba +from .mask import build_inpaint_mask, compose_comfy_rgba, mask_from_curve +from .marker_detect import detect_marker_hairline, path_to_curve_mask import io import logging @@ -160,6 +161,42 @@ def generate_grow_results(image_bgr: np.ndarray, gender: str): return results +def generate_grow_b(marked_bgr: np.ndarray, original_bgr: np.ndarray): + """接口3：检测医生手绘发际线 → 遮罩 → 原图重画干净线 → ComfyUI 生发。 + + Returns: {"grown_png": bytes 或 None, "status": "ok"|"no_face"|"no_line"}。 + """ + rgb = cv2.cvtColor(marked_bgr, cv2.COLOR_BGR2RGB) + landmarks = get_landmarker().detect(rgb) + if landmarks is None: + return {"grown_png": None, "status": "no_face"} + parse_map = get_parser().parse(rgb) + path = detect_marker_hairline(marked_bgr, landmarks, parse_map) + if path is None: + return {"grown_png": None, "status": "no_line"} + + h, w = marked_bgr.shape[:2] + # 原图对齐到 marked 坐标系（同一张照片的原始版/划线版，尺寸应一致） + orig = original_bgr + if orig.shape[:2] != (h, w): + orig = cv2.resize(orig, (w, h), interpolation=cv2.INTER_AREA) + + # 在原图上重画干净黑线（膨胀成笔迹宽度），替代医生手绘的毛刺 + line_w = max(2, int(w * 0.006)) + marked_clean = orig.copy() + cv2.polylines(marked_clean, [path[:, ::-1].reshape(-1, 1, 2)], False, + (0, 0, 0), line_w, lineType=cv2.LINE_AA) + + # 遮罩：检测路径曲线 + ROI 闭合 + curve_mask = path_to_curve_mask(path, h, w, thickness=max(3, line_w)) + mask = mask_from_curve(curve_mask, landmarks, parse_map) + + buf = io.BytesIO() + compose_comfy_rgba(marked_clean, mask).save(buf, format="PNG") + grown_png = comfyui.run(buf.getvalue()) + return {"grown_png": grown_png, "status": "ok"} + + if __name__ == "__main__": import sys g = sys.argv[2] if len(sys.argv) > 2 else "female" diff --git a/requirements.txt b/requirements.txt index 0e9ce3c..4e818cc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -22,5 +22,9 @@ torchvision==0.17.2 # MediaPipe Tasks(FaceLandmarker) 用已装的 mediapipe；新增 SegFormer 人脸分割： transformers==4.45.2 # SegFormer 人脸分割（jonathandinu/face-parsing，本地权重） +# 接口3：B端生发（马克笔发际线检测） +# ⚠️ 必须 0.24.x —— 0.25+ 强依赖 numpy>=2，会顶掉 mediapipe 需要的 numpy<2 +scikit-image==0.24.0 # route_through_array（黑帽响应图上的 Dijkstra 最小路径） + # 测试 pytest==8.3.3 diff --git a/tests/fixtures/marked_hairline.jpg b/tests/fixtures/marked_hairline.jpg new file mode 100644 index 0000000..2f5a5cd Binary files /dev/null and b/tests/fixtures/marked_hairline.jpg differ diff --git a/tests/test_api.py b/tests/test_api.py index 75b340e..5ef7366 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -103,6 +103,39 @@ def test_grow_female_returns_5(client, monkeypatch): assert "image_url" not in results[0] +GROWB = "/api/v1/hair/grow-b" + + +def test_growb_missing_original_1007(client): + files = {"marked_image_file": ("m.jpg", open(fixture("marked_hairline.jpg"), "rb"), "application/octet-stream")} + r = client.post(GROWB, headers=H, files=files) + assert r.json()["code"] == 1007 + + +def test_growb_no_line_1001(client): + f = lambda: open(fixture("frontal.jpg"), "rb") + files = {"marked_image_file": ("m.jpg", f(), "application/octet-stream"), + "original_image_file": ("o.jpg", f(), "application/octet-stream")} + r = client.post(GROWB, headers=H, files=files) + assert r.json()["code"] == 1001 + + +def test_growb_success(client, monkeypatch): + import hairline.comfyui as comfy + monkeypatch.setattr(comfy, "run", lambda *a, **k: _PNG_1x1) + fm = open(fixture("marked_hairline.jpg"), "rb") + fo = open(fixture("marked_hairline.jpg"), "rb") + files = {"marked_image_file": ("m.jpg", fm, "application/octet-stream"), + "original_image_file": ("o.jpg", fo, "application/octet-stream")} + body = client.post(GROWB, headers=H, files=files).json() + assert body["code"] == 0, body + d = body["data"] + assert d["hairline_type"] == "custom" + assert base64.b64decode(d["hair_growth_image_base64"])[:8] == b"\x89PNG\r\n\x1a\n" + assert d["best_hairline_image_base64"] # 原图原样(非空) + assert "best_hairline_image_url" not in d + + def test_success_structure(client): r = _post(client, "frontal.jpg") body = r.json() diff --git a/tests/test_marker.py b/tests/test_marker.py new file mode 100644 index 0000000..99ad945 --- /dev/null +++ b/tests/test_marker.py @@ -0,0 +1,42 @@ +"""接口3 马克笔检测测试：辅助函数(纯numpy) + 真实样本检测/拒识。""" +import cv2 +import numpy as np +from conftest import fixture + +from hairline.marker_detect import ( + detect_marker_hairline, path_to_curve_mask, _snap_anchor, +) +from hairline.service import get_landmarker, get_parser + + +def test_path_to_curve_mask(): + path = np.array([[10, 5], [10, 50], [10, 90]]) # 水平线 row=10 + m = path_to_curve_mask(path, 100, 100, thickness=3) + assert m[10, 50] == 255 + assert m[90, 50] == 0 + + +def test_snap_anchor_moves_to_peak(): + bh = np.zeros((100, 100), np.float32) + bh[40, 30] = 99.0 # 峰值在 (40,30) + r, c = _snap_anchor(bh, x=33, y=42, w=1000) # 起点附近 → 应吸到峰值 + assert (r, c) == (40, 30) + + +def _ctx(path_img): + img = cv2.imread(path_img) + rgb = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + lm = get_landmarker().detect(rgb) + pm = get_parser().parse(rgb) + return img, lm, pm + + +def test_detect_real_marked(): + img, lm, pm = _ctx(fixture("marked_hairline.jpg")) + path = detect_marker_hairline(img, lm, pm) + assert path is not None and len(path) > 50 # 检测到画线 + + +def test_reject_clean_photo(): + img, lm, pm = _ctx(fixture("frontal.jpg")) + assert detect_marker_hairline(img, lm, pm) is None # 无画线 → 拒识