xslandClaude Opus 4.8
0ddfa83743
- 端点 POST /api/v1/hairline/grow（app.py，纯新增，不影响接口1-5） - 编排模块 face_analysis/hairline_grow.py：复用接口9 遮罩 → HTTP 调 change_hair(8801) → 按遮罩贴回 - 双生成后端 gen_backend：swaphair(换发型LoRA) / hairgrow(区域生发inpaint) - swap_mode：ext_mask(接口9遮罩作换发型遮罩) / as_is；融合 feather/alpha_gradient/seamless - 参数全在测试页可调；可视化按算法文档4步：最终遮罩→生成全帧→严格贴回→接缝融合 - 附启停脚本 scripts/restart_if11_backends.sh、算法文档、测试图 注：change_hair 侧 swapHair 的 ext_mask/denoising_strength 改造在 change_hair 仓库，向后兼容。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
338 lines
14 KiB
Python
338 lines
14 KiB
Python
"""接口11：发际线生发。
|
||
|
||
输入一张发际线较高 / 头发稀少的正脸图 + 发际线类型 ID（= change_hair 的 hair_id，
|
||
如 chang_tuoyuan/chang_bolang/...），输出同一个人、同一发型、按该发际线类型压低发际线
|
||
后的图片。管线（见 docs/发际线增强算法.md）：
|
||
|
||
1. 用接口9 的算法算出头发遮罩（含额头闭合区域，外缘内缩 erode_cm）。
|
||
—— seg_model 选 bisenet/segformer，mask_type 选 eroded(内缩)/closed(未内缩闭合区域)。
|
||
2. 调 change_hair 换发型服务（/api/swapHair/v1）生成该发际线类型的图。返回的结果图已被
|
||
change_hair 用 M_inv 贴回、与输入原图**同分辨率同对齐**，可直接按遮罩合成。
|
||
两种取图模式（swap_mode）：
|
||
- ext_mask：把步骤1 的遮罩作为 ext_mask 传给 swapHair，让 webui 精确重绘该区域
|
||
（忠于算法文档「换发型的遮罩用接口9 遮罩」）。
|
||
- as_is：不改 change_hair，swapHair 用它自己的内部遮罩，贴回时再裁到接口9 遮罩。
|
||
3. 严格按接口9 遮罩把生成图贴回原图（遮罩外=原图，纹丝不动）。
|
||
4. 融合接缝：blend_method 选 feather(高斯羽化) / alpha_gradient(距离变换内渐变) /
|
||
seamless(泊松无缝克隆)；feather_px、edge_erode_px 控制过渡细节。
|
||
|
||
对外返回每一步可视化（base64，data URI），供测试页逐步展示。经网关时 *_base64 字段会被
|
||
落盘改写为 *_url。
|
||
"""
|
||
import base64
|
||
import os
|
||
import time
|
||
from uuid import uuid4
|
||
|
||
import cv2
|
||
import numpy as np
|
||
|
||
from face_analysis.detector import detector
|
||
from face_analysis.calibration import estimate_scale_factor
|
||
from face_analysis.head_mask import (
|
||
NoFaceError,
|
||
_baseline_points,
|
||
_upper_region_mask,
|
||
_bisenet_hair_mask,
|
||
_segformer_hair_mask,
|
||
_fill_to_baseline,
|
||
_erode,
|
||
_largest_cc,
|
||
_overlay,
|
||
)
|
||
|
||
# change_hair 服务地址（可用环境变量覆盖）
|
||
SWAP_URL = os.getenv("SWAP_HAIR_URL", "http://127.0.0.1:8801/api/swapHair/v1")
|
||
HAIRGROW_URL = os.getenv("HAIR_GROW_URL", "http://127.0.0.1:8801/api/hairGrow/v1")
|
||
SWAP_TIMEOUT = float(os.getenv("SWAP_HAIR_TIMEOUT", "300"))
|
||
|
||
DEFAULTS = {
|
||
"gen_backend": "swaphair", # swaphair(换发型LoRA) | hairgrow(区域生发inpaint)
|
||
"is_hr": False,
|
||
"seg_model": "segformer", # bisenet | segformer
|
||
"mask_type": "eroded", # eroded | closed
|
||
"erode_cm": 1.2,
|
||
"swap_mode": "ext_mask", # ext_mask | as_is（仅 swaphair）
|
||
"denoising_strength": 0.6, # 仅 swaphair
|
||
"hairgrow_strength": 0.75, # 仅 hairgrow
|
||
"blend_method": "feather", # feather | alpha_gradient | seamless
|
||
"feather_px": 15,
|
||
"edge_erode_px": 3,
|
||
}
|
||
|
||
|
||
class SwapError(Exception):
|
||
"""调用 change_hair 换发型服务失败。"""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 编码
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _jpg_b64(bgr):
|
||
ok, buf = cv2.imencode(".jpg", bgr, [cv2.IMWRITE_JPEG_QUALITY, 92])
|
||
return "data:image/jpeg;base64," + base64.b64encode(buf.tobytes()).decode()
|
||
|
||
|
||
def _png_b64(bgr_or_gray):
|
||
ok, buf = cv2.imencode(".png", bgr_or_gray)
|
||
return "data:image/png;base64," + base64.b64encode(buf.tobytes()).decode()
|
||
|
||
|
||
def _gray_b64(gray_float):
|
||
"""0~1 的浮点图 → 灰度 PNG data URI。"""
|
||
g = np.clip(gray_float * 255.0, 0, 255).astype(np.uint8)
|
||
return _png_b64(g)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 步骤1：接口9 头发遮罩（复用 head_mask 构件）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def compute_mask(image_bgr, landmarks, seg_model, mask_type, erode_cm, px_per_cm):
|
||
"""算出布尔遮罩 + 可视化。
|
||
|
||
seg_model: bisenet | segformer；mask_type: eroded(内缩) | closed(闭合区域未内缩)。
|
||
返回 (mask_bool, viz_dict)。
|
||
"""
|
||
h, w = image_bgr.shape[:2]
|
||
r = int(round(max(0.0, erode_cm) * px_per_cm))
|
||
baseline_pts = _baseline_points(landmarks, w, h)
|
||
upper = _upper_region_mask(baseline_pts, w, h)
|
||
|
||
if seg_model == "bisenet":
|
||
hair_mask = _bisenet_hair_mask(image_bgr, landmarks, w, h)
|
||
elif seg_model == "segformer":
|
||
hair_mask = _segformer_hair_mask(image_bgr)
|
||
else:
|
||
raise ValueError(f"未知 seg_model: {seg_model}")
|
||
|
||
top_fill = _fill_to_baseline(hair_mask, upper) # 含额头，延伸到图底
|
||
closed = _largest_cc(top_fill & upper) # 闭合区域：头发+额头，底=基线
|
||
eroded = _largest_cc(_erode(top_fill, r) & upper) # 外缘内缩 r、底线不动
|
||
mask_bool = eroded if mask_type == "eroded" else closed
|
||
|
||
# 只输出最终遮罩（叠加图 + 纯遮罩），不展开接口9 内部子步骤
|
||
viz = {
|
||
"erode_px": r,
|
||
"hair_pixels": int(hair_mask.sum()),
|
||
"mask_pixels": int(mask_bool.sum()),
|
||
"mask_overlay_base64": _jpg_b64(_overlay(image_bgr, mask_bool, (0, 0, 255))),
|
||
"mask_base64": _png_b64((mask_bool.astype(np.uint8)) * 255),
|
||
}
|
||
return mask_bool, viz
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 步骤2：调 change_hair 换发型
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _call_swap(image_bgr, hairline_id, is_hr, ext_mask_bool, denoising_strength):
|
||
"""调 change_hair /api/swapHair/v1，返回与输入同分辨率同对齐的换发型结果（BGR）。
|
||
|
||
ext_mask_bool 非 None 时作为 ext_mask 传入（swap_mode=ext_mask）。
|
||
denoising_strength：webui img2img 重绘强度（越大生发越激进），透传给换发型。
|
||
"""
|
||
import requests
|
||
|
||
ok, ibuf = cv2.imencode(".jpg", image_bgr, [cv2.IMWRITE_JPEG_QUALITY, 95])
|
||
payload = {
|
||
"hair_id": hairline_id,
|
||
"task_id": "if11-" + uuid4().hex[:12],
|
||
"is_hr": "true" if is_hr else "false",
|
||
"user_img_path": "data:image/jpeg;base64," + base64.b64encode(ibuf.tobytes()).decode(),
|
||
"output_format": "base64",
|
||
"denoising_strength": float(denoising_strength),
|
||
}
|
||
if ext_mask_bool is not None:
|
||
mbuf = cv2.imencode(".png", (ext_mask_bool.astype(np.uint8)) * 255)[1]
|
||
payload["ext_mask"] = "data:image/png;base64," + base64.b64encode(mbuf.tobytes()).decode()
|
||
|
||
try:
|
||
resp = requests.post(SWAP_URL, json=payload, timeout=SWAP_TIMEOUT)
|
||
except Exception as ex: # noqa: BLE001
|
||
raise SwapError(f"换发型服务不可达（{SWAP_URL}）：{ex}")
|
||
|
||
try:
|
||
j = resp.json()
|
||
except Exception: # noqa: BLE001
|
||
raise SwapError(f"换发型服务返回非 JSON（HTTP {resp.status_code}）：{resp.text[:200]}")
|
||
|
||
if j.get("state") != 0 or not j.get("data"):
|
||
raise SwapError(f"换发型失败：{j.get('msg', j)}")
|
||
|
||
b64 = j["data"]
|
||
if "," in b64 and b64.startswith("data:"):
|
||
b64 = b64.split(",", 1)[1]
|
||
result = cv2.imdecode(np.frombuffer(base64.b64decode(b64), np.uint8), cv2.IMREAD_COLOR)
|
||
if result is None:
|
||
raise SwapError("换发型结果解码失败")
|
||
|
||
# 保险：与原图对齐（change_hair 已贴回原尺寸，若极端情况尺寸不符则拉回）
|
||
if result.shape[:2] != image_bgr.shape[:2]:
|
||
result = cv2.resize(result, (image_bgr.shape[1], image_bgr.shape[0]),
|
||
interpolation=cv2.INTER_LANCZOS4)
|
||
return result
|
||
|
||
|
||
def _call_hairgrow(image_bgr, mask_bool, strength):
|
||
"""调 change_hair /api/hairGrow/v1（区域生发 inpaint），在遮罩区域长出头发。
|
||
|
||
返回与输入同分辨率的结果（BGR）。hairGrow 内部已做贴回与颜色迁移，
|
||
这里再套接口11 的遮罩羽化贴回以保证遮罩外严格不动。
|
||
"""
|
||
import requests
|
||
|
||
ok, ibuf = cv2.imencode(".jpg", image_bgr, [cv2.IMWRITE_JPEG_QUALITY, 95])
|
||
mbuf = cv2.imencode(".png", (mask_bool.astype(np.uint8)) * 255)[1]
|
||
payload = {
|
||
"img": "data:image/jpeg;base64," + base64.b64encode(ibuf.tobytes()).decode(),
|
||
"mask": "data:image/png;base64," + base64.b64encode(mbuf.tobytes()).decode(),
|
||
"strength": float(strength),
|
||
"output_format": "base64",
|
||
}
|
||
try:
|
||
resp = requests.post(HAIRGROW_URL, json=payload, timeout=SWAP_TIMEOUT)
|
||
except Exception as ex: # noqa: BLE001
|
||
raise SwapError(f"区域生发服务不可达（{HAIRGROW_URL}）：{ex}")
|
||
try:
|
||
j = resp.json()
|
||
except Exception: # noqa: BLE001
|
||
raise SwapError(f"区域生发返回非 JSON（HTTP {resp.status_code}）：{resp.text[:200]}")
|
||
if j.get("state") != 0 or not j.get("result"):
|
||
raise SwapError(f"区域生发失败：{j.get('msg', j)}")
|
||
b64 = j["result"]
|
||
if "," in b64 and b64.startswith("data:"):
|
||
b64 = b64.split(",", 1)[1]
|
||
result = cv2.imdecode(np.frombuffer(base64.b64decode(b64), np.uint8), cv2.IMREAD_COLOR)
|
||
if result is None:
|
||
raise SwapError("区域生发结果解码失败")
|
||
if result.shape[:2] != image_bgr.shape[:2]:
|
||
result = cv2.resize(result, (image_bgr.shape[1], image_bgr.shape[0]),
|
||
interpolation=cv2.INTER_LANCZOS4)
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 步骤3+4：按遮罩贴回 + 接缝融合
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _feather_alpha(mask_bool, blend_method, feather_px, edge_erode_px):
|
||
"""由布尔遮罩生成 0~1 的 alpha（贴图权重）。遮罩外恒为 0（原图纹丝不动）。"""
|
||
m = mask_bool.astype(np.uint8)
|
||
if edge_erode_px > 0:
|
||
k = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2 * edge_erode_px + 1,) * 2)
|
||
m = cv2.erode(m, k)
|
||
|
||
fp = max(1, int(feather_px))
|
||
if blend_method == "alpha_gradient":
|
||
# 距离变换：过渡只发生在遮罩内侧（边界 0 → 内部 feather_px 处 1），遮罩外严格为 0
|
||
dist = cv2.distanceTransform(m, cv2.DIST_L2, 3)
|
||
alpha = np.clip(dist / fp, 0.0, 1.0)
|
||
else: # feather（高斯羽化，默认）
|
||
ksz = fp * 2 + 1
|
||
alpha = cv2.GaussianBlur(m.astype(np.float32), (ksz, ksz), sigmaX=fp / 2.0)
|
||
alpha = np.clip(alpha, 0.0, 1.0)
|
||
return alpha
|
||
|
||
|
||
def _composite(orig, swap_result, mask_bool, blend_method, feather_px, edge_erode_px):
|
||
"""把 swap_result 按遮罩贴回 orig，返回 (final_bgr, alpha_float or None)。"""
|
||
if blend_method == "seamless":
|
||
m = mask_bool.astype(np.uint8)
|
||
if edge_erode_px > 0:
|
||
k = cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (2 * edge_erode_px + 1,) * 2)
|
||
m = cv2.erode(m, k)
|
||
if m.sum() < 10:
|
||
return orig.copy(), None
|
||
ys, xs = np.where(m > 0)
|
||
center = (int((xs.min() + xs.max()) / 2), int((ys.min() + ys.max()) / 2))
|
||
final = cv2.seamlessClone(swap_result, orig, m * 255, center, cv2.NORMAL_CLONE)
|
||
return final, None
|
||
|
||
alpha = _feather_alpha(mask_bool, blend_method, feather_px, edge_erode_px)
|
||
a3 = alpha[:, :, None]
|
||
final = (orig.astype(np.float32) * (1 - a3) + swap_result.astype(np.float32) * a3)
|
||
return np.clip(final, 0, 255).astype(np.uint8), alpha
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主入口
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def generate_hairline_grow(image_bgr, hairline_id, is_hr=False, seg_model="segformer",
|
||
mask_type="eroded", erode_cm=1.2, swap_mode="ext_mask",
|
||
blend_method="feather", feather_px=15, edge_erode_px=3,
|
||
denoising_strength=0.6, gen_backend="swaphair",
|
||
hairgrow_strength=0.75):
|
||
"""接口11 完整管线。返回可直接进 ok() 的 data dict。未检出人脸抛 NoFaceError。
|
||
|
||
gen_backend：生成后端。swaphair=换发型LoRA（应用发际线类型）；
|
||
hairgrow=区域生发inpaint（在遮罩内长出头发、压低发际线）。
|
||
"""
|
||
h, w = image_bgr.shape[:2]
|
||
landmarks = detector.detect(image_bgr)
|
||
if landmarks is None:
|
||
raise NoFaceError()
|
||
px_per_cm = estimate_scale_factor(landmarks, w, h)
|
||
|
||
# 步骤1：接口9 遮罩
|
||
t0 = time.time()
|
||
mask_bool, mask_viz = compute_mask(
|
||
image_bgr, landmarks, seg_model, mask_type, erode_cm, px_per_cm)
|
||
t_mask = time.time() - t0
|
||
|
||
# 步骤2：生成（按后端）
|
||
t0 = time.time()
|
||
if gen_backend == "hairgrow":
|
||
swap_result = _call_hairgrow(image_bgr, mask_bool, hairgrow_strength)
|
||
else:
|
||
ext_mask = mask_bool if swap_mode == "ext_mask" else None
|
||
swap_result = _call_swap(image_bgr, hairline_id, is_hr, ext_mask, denoising_strength)
|
||
t_swap = time.time() - t0
|
||
|
||
# 步骤3：严格按遮罩硬贴回（无融合，用于对比）
|
||
hard_paste = image_bgr.copy()
|
||
hard_paste[mask_bool] = swap_result[mask_bool]
|
||
|
||
# 步骤4：接缝融合
|
||
t0 = time.time()
|
||
final, alpha = _composite(
|
||
image_bgr, swap_result, mask_bool, blend_method, feather_px, edge_erode_px)
|
||
t_blend = time.time() - t0
|
||
|
||
data = {
|
||
"hairline_id": hairline_id,
|
||
"gen_backend": gen_backend,
|
||
"hairgrow_strength": round(float(hairgrow_strength), 3),
|
||
"is_hr": is_hr,
|
||
"seg_model": seg_model,
|
||
"mask_type": mask_type,
|
||
"erode_cm": round(float(erode_cm), 2),
|
||
"swap_mode": swap_mode,
|
||
"blend_method": blend_method,
|
||
"feather_px": int(feather_px),
|
||
"edge_erode_px": int(edge_erode_px),
|
||
"denoising_strength": round(float(denoising_strength), 3),
|
||
"px_per_cm": round(float(px_per_cm), 4),
|
||
"erode_px": mask_viz["erode_px"],
|
||
"hair_pixels": mask_viz["hair_pixels"],
|
||
"mask_pixels": mask_viz["mask_pixels"],
|
||
"image_size": {"width": w, "height": h},
|
||
"timings_ms": {
|
||
"mask": int(t_mask * 1000),
|
||
"swap": int(t_swap * 1000),
|
||
"blend": int(t_blend * 1000),
|
||
},
|
||
"steps": {
|
||
"input_base64": _jpg_b64(image_bgr),
|
||
"mask_overlay_base64": mask_viz["mask_overlay_base64"],
|
||
"mask_base64": mask_viz["mask_base64"],
|
||
"swap_raw_base64": _jpg_b64(swap_result),
|
||
"hard_paste_base64": _jpg_b64(hard_paste),
|
||
"alpha_base64": _gray_b64(alpha) if alpha is not None else mask_viz["mask_base64"],
|
||
"final_base64": _jpg_b64(final),
|
||
},
|
||
}
|
||
return data
|