xslandClaude
9678b54267
- 参数类型从 int 改为 string（逗号分隔），自动解析去重排序 - 越界/非法值返回 1007 - _parse_hair_styles() 辅助函数：解析 + 去重 + 范围校验 - service 层 hair_style 参数改为 hair_styles: list[int] - 接口2 和 接口7 同步更新 - 服务已重启，smoke test 通过 Co-Authored-By: Claude <noreply@anthropic.com>
273 lines
12 KiB
Python
273 lines
12 KiB
Python
"""接口2 服务层：模型单例 + 性别贴图映射 + 「照片→N 张发际线预览图」管线。
|
||
|
||
把 head3d 的 extract_hairline 步骤包成单例复用（避免每请求重建模型），再按性别
|
||
对每张贴图调 render.render_hairline_overlay 生成预览图。
|
||
"""
|
||
from __future__ import annotations
|
||
import glob
|
||
import os
|
||
|
||
import cv2
|
||
import numpy as np
|
||
|
||
from . import constants as C
|
||
from . import comfyui
|
||
from .face_landmarks import FaceLandmarker
|
||
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, build_overlay_layer
|
||
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
|
||
|
||
logger = logging.getLogger("hair.worker")
|
||
|
||
_REPO = os.path.dirname(os.path.dirname(__file__))
|
||
_TEXTURE_DIR = os.path.join(_REPO, "hairline_texture")
|
||
_BLACK_TEXTURE_DIR = os.path.join(_REPO, "hairline_texture_black")
|
||
|
||
# ⚠️ 本 worker 是 RTX 5090(sm_120)，torch 2.2.2(cu121) 只编到 sm_90，CUDA 跑算子会报
|
||
# "no kernel image"。SegFormer 默认走 CPU（~2.5s/张）。换 torch cu128 后可设 SEG_DEVICE=cuda。
|
||
_SEG_DEVICE = os.getenv("SEG_DEVICE", "cpu")
|
||
|
||
_landmarker = None
|
||
_parser = None
|
||
_texture_map = None
|
||
|
||
|
||
def get_landmarker() -> FaceLandmarker:
|
||
global _landmarker
|
||
if _landmarker is None:
|
||
_landmarker = FaceLandmarker(static_image_mode=True)
|
||
return _landmarker
|
||
|
||
|
||
def get_parser() -> FaceParser:
|
||
global _parser
|
||
if _parser is None:
|
||
_parser = FaceParser(device=_SEG_DEVICE)
|
||
return _parser
|
||
|
||
|
||
def _gender_key(stem: str):
|
||
"""文件名 stem → (gender, key)；非 girl_/man_ 前缀返回 (None, None)。"""
|
||
if stem.startswith("girl_"):
|
||
return "female", stem[5:].replace(" ", "").strip()
|
||
if stem.startswith("man_"):
|
||
return "male", stem[4:].replace(" ", "").strip()
|
||
return None, None
|
||
|
||
|
||
def get_texture_map() -> dict:
|
||
"""扫描 hairline_texture/ 建 {gender: [(key, path)]}，按 key 排序、缓存。
|
||
|
||
文件名规范化去空格（如 `man_ inverse_arc.png` → key `inverse_arc`）。
|
||
"""
|
||
global _texture_map
|
||
if _texture_map is not None:
|
||
return _texture_map
|
||
mapping: dict[str, list] = {"female": [], "male": []}
|
||
for path in sorted(glob.glob(os.path.join(_TEXTURE_DIR, "*.png"))):
|
||
stem = os.path.splitext(os.path.basename(path))[0]
|
||
gender, key = _gender_key(stem)
|
||
if gender:
|
||
mapping[gender].append((key, path))
|
||
for g in mapping:
|
||
mapping[g].sort(key=lambda kp: kp[0])
|
||
_texture_map = mapping
|
||
return _texture_map
|
||
|
||
|
||
def extract_502(image_bgr: np.ndarray):
|
||
"""照片(BGR) → (points502 MP序, valid17)。无人脸返回 (None, None)。"""
|
||
ctx = extract_context(image_bgr)
|
||
if ctx is None:
|
||
return None, None
|
||
return ctx["points"], ctx["valid"]
|
||
|
||
|
||
def extract_context(image_bgr: np.ndarray):
|
||
"""照片(BGR) → {landmarks, parse_map, points, valid}。无人脸返回 None。"""
|
||
rgb = cv2.cvtColor(image_bgr, cv2.COLOR_BGR2RGB)
|
||
landmarks = get_landmarker().detect(rgb)
|
||
if landmarks is None:
|
||
return None
|
||
parse_map = get_parser().parse(rgb)
|
||
hairline_2d, valid = sample_hairline(landmarks, parse_map)
|
||
hairline_2d = smooth_hairline(hairline_2d, valid)
|
||
hairline_3d = lift_hairline_to_3d(landmarks, hairline_2d)
|
||
middle_3d = build_middle_row(landmarks, hairline_3d)
|
||
points = assemble_full(landmarks, middle_3d, hairline_3d)
|
||
return {"landmarks": landmarks, "parse_map": parse_map, "points": points, "valid": valid}
|
||
|
||
|
||
def _black_texture_path(white_path: str) -> str:
|
||
"""白贴图路径 → 同名黑贴图路径（hairline_texture_black/）。"""
|
||
return os.path.join(_BLACK_TEXTURE_DIR, os.path.basename(white_path))
|
||
|
||
|
||
def generate_previews(image_bgr: np.ndarray, gender: str):
|
||
"""生成该性别全部发际线预览图（仅预览，不生发）。
|
||
|
||
Returns: list[dict] {"hairline_type", "image_bgr", "order"}；无人脸返回 None。
|
||
"""
|
||
if gender not in ("male", "female"):
|
||
raise ValueError(f"gender 必须是 male/female，收到 {gender!r}")
|
||
ctx = extract_context(image_bgr)
|
||
if ctx is None:
|
||
return None
|
||
uv, ext_faces = load_ext_mesh()
|
||
results = []
|
||
for order, (key, path) in enumerate(get_texture_map()[gender], start=1):
|
||
preview = render_hairline_overlay(image_bgr, ctx["points"], ext_faces, uv,
|
||
load_texture_rgba(path))
|
||
results.append({"hairline_type": key, "image_bgr": preview, "order": order})
|
||
return results
|
||
|
||
|
||
def generate_grow_results(image_bgr: np.ndarray, gender: str, use_mask: bool = True,
|
||
prompt: str = None, hair_styles: list[int] | None = None,
|
||
workflow_path: str | None = None):
|
||
"""指定发际线类型：预览图(白线) + 生发图(ComfyUI)。
|
||
|
||
hair_styles（1-indexed 列表）：指定生成哪几张发际线（按贴图排序）。female: 1..5，male: 1..4。
|
||
为 None 时生成全部（兼容旧调用）。
|
||
use_mask（默认 True）：是否启用 inpaint 遮罩，用于测试对比（同接口3）。
|
||
False 时用**干净原图 + 空遮罩**送 ComfyUI（不烧黑色模板线）。
|
||
prompt（默认 None）：ComfyUI 提示词，非 None 时替换工作流节点60文本。
|
||
workflow_path（默认 None）：ComfyUI 工作流 JSON 路径，None 用默认 add_hair.json。
|
||
Returns: list[dict] {"hairline_type","order","image_bgr"(预览), "grown_png"(bytes 或 None)}。
|
||
无人脸返回 None。某张 ComfyUI 失败时该项 grown_png=None，不抛异常。
|
||
"""
|
||
if gender not in ("male", "female"):
|
||
raise ValueError(f"gender 必须是 male/female，收到 {gender!r}")
|
||
ctx = extract_context(image_bgr)
|
||
if ctx is None:
|
||
return None
|
||
uv, ext_faces = load_ext_mesh()
|
||
|
||
textures = get_texture_map()[gender] # [(key, path), ...] 已排序
|
||
if hair_styles is not None:
|
||
items = [(s, textures[s - 1]) for s in hair_styles]
|
||
else:
|
||
items = list(enumerate(textures, start=1))
|
||
|
||
# 禁用遮罩：干净原图 + 空遮罩，与模板无关 → 只跑一次 ComfyUI，下面 N 项复用
|
||
shared_grown = None
|
||
if not use_mask:
|
||
try:
|
||
h, w = image_bgr.shape[:2]
|
||
buf = io.BytesIO()
|
||
compose_comfy_rgba(image_bgr, np.zeros((h, w), np.uint8)).save(buf, format="PNG")
|
||
shared_grown = comfyui.run(buf.getvalue(), prompt=prompt, workflow_path=workflow_path)
|
||
except Exception as e: # noqa: BLE001
|
||
logger.warning("接口2 生发图失败(无遮罩)：%s", e)
|
||
|
||
results = []
|
||
for order, (key, white_path) in items:
|
||
white = load_texture_rgba(white_path)
|
||
preview = render_hairline_overlay(image_bgr, ctx["points"], ext_faces, uv, white)
|
||
|
||
if not use_mask:
|
||
grown_png = shared_grown
|
||
else:
|
||
grown_png = None
|
||
try:
|
||
black = load_texture_rgba(_black_texture_path(white_path))
|
||
marked, mask = build_inpaint_mask(
|
||
image_bgr, ctx["landmarks"], ctx["parse_map"], ctx["points"], black)
|
||
buf = io.BytesIO()
|
||
compose_comfy_rgba(marked, mask).save(buf, format="PNG")
|
||
grown_png = comfyui.run(buf.getvalue(), prompt=prompt, workflow_path=workflow_path)
|
||
except Exception as e: # noqa: BLE001 单张失败不拖垮整请求
|
||
logger.warning("接口2 生发图失败 type=%s：%s", key, e)
|
||
|
||
results.append({"hairline_type": key, "order": order,
|
||
"image_bgr": preview, "grown_png": grown_png})
|
||
return results
|
||
|
||
|
||
def generate_hairline_pngs(image_bgr: np.ndarray, gender: str):
|
||
"""接口5：该性别全部发际线叠图(同接口2预览) + 最佳(order1)发际线曲线的面部中间点。
|
||
|
||
Returns: {"images":[{hairline_type,order,image_bgr}], "best_center":(x,y)}；无人脸 None。
|
||
"""
|
||
if gender not in ("male", "female"):
|
||
raise ValueError(f"gender 必须是 male/female，收到 {gender!r}")
|
||
ctx = extract_context(image_bgr)
|
||
if ctx is None:
|
||
return None
|
||
h, w = image_bgr.shape[:2]
|
||
uv, ext_faces = load_ext_mesh()
|
||
lm = ctx["landmarks"]
|
||
# 面部中轴 x = 眉心(9/151 中点)
|
||
face_cx = float((lm[9, 0] + lm[151, 0]) / 2 * w)
|
||
|
||
textures = get_texture_map()[gender]
|
||
images, best_center = [], None
|
||
for order, (key, path) in enumerate(textures, start=1):
|
||
white = load_texture_rgba(path)
|
||
preview = render_hairline_overlay(image_bgr, ctx["points"], ext_faces, uv, white)
|
||
images.append({"hairline_type": key, "order": order, "image_bgr": preview})
|
||
if order == 1: # 最佳发际线曲线的中点（面部中轴处的发际线 y）
|
||
overlay = build_overlay_layer(h, w, ctx["points"], ext_faces, uv, white)
|
||
ys, xs = np.where(overlay[:, :, 3] > 40)
|
||
if xs.size:
|
||
near = np.abs(xs - face_cx) <= max(2, int(w * 0.02))
|
||
col_ys = ys[near] if near.any() else ys[np.argsort(np.abs(xs - face_cx))[:20]]
|
||
best_center = (int(round(face_cx)), int(round(float(col_ys.mean()))))
|
||
return {"images": images, "best_center": best_center}
|
||
|
||
|
||
def generate_grow_b(marked_bgr: np.ndarray, use_mask: bool = True, prompt: str = None):
|
||
"""接口3：检测医生手绘发际线 → 遮罩 → 送 ComfyUI 生发（仅需划线图一张）。
|
||
|
||
检测路径只用来**建遮罩**；ComfyUI 输入图用 **marked 原图**（含医生手绘线，
|
||
工作流提示词会清除黑线再生发）。
|
||
|
||
use_mask（默认 True）：是否启用自动检测的遮罩，用于测试对比。
|
||
- True：检测手绘线 → 建遮罩 → alpha=255−mask（透明区=重绘区，节点44 画黄色参考区）。
|
||
- False：跳过检测，直接送划线图，alpha 全 255（空遮罩，节点26 mask 为空），
|
||
模型仅凭医生黑线参考生发。无需改工作流，唯一变量是遮罩。
|
||
Returns: {"grown_png": bytes 或 None, "status": "ok"|"no_face"|"no_line"}。
|
||
"""
|
||
h, w = marked_bgr.shape[:2]
|
||
if use_mask:
|
||
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"}
|
||
line_w = max(2, int(w * 0.006))
|
||
curve_mask = path_to_curve_mask(path, h, w, thickness=max(3, line_w))
|
||
mask = mask_from_curve(curve_mask, landmarks, parse_map)
|
||
else:
|
||
mask = np.zeros((h, w), np.uint8) # 空遮罩：alpha 全 255，跳过检测
|
||
|
||
buf = io.BytesIO()
|
||
compose_comfy_rgba(marked_bgr, mask).save(buf, format="PNG") # marked 原图 + 遮罩
|
||
grown_png = comfyui.run(buf.getvalue(), prompt=prompt)
|
||
return {"grown_png": grown_png, "status": "ok"}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
g = sys.argv[2] if len(sys.argv) > 2 else "female"
|
||
img = cv2.imread(sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/frontal.jpg")
|
||
os.makedirs("tests/output", exist_ok=True)
|
||
print("texture map:", {k: [kp[0] for kp in v] for k, v in get_texture_map().items()})
|
||
res = generate_previews(img, g)
|
||
if res is None:
|
||
print("无人脸")
|
||
sys.exit(1)
|
||
for r in res:
|
||
out = f"tests/output/preview_{g}_{r['hairline_type']}.png"
|
||
cv2.imwrite(out, r["image_bgr"])
|
||
print(f" order={r['order']} type={r['hairline_type']} -> {out}")
|