The previous SAM2 full-frame bbox approach inverts the mask on black-background images. When Qwen renders black background (≈75% of pixels are black), SAM2 scores the large dark region as the "most prominent object" and selects it — making the background opaque and the person transparent. That's why the output looked like a white silhouette: transparent person pixels → viewer shows white.

New _apply_transparency_black_bg function (called when bg_removal=sam2):
1.
Threshold — any pixel with max-channel > 25 = person. Finds the person's exact bounding box without any model confusion.
2.
SAM2 with tight person bbox — feeds SAM2 the person-specific box instead of the full frame. SAM2 now segments within the person area for clean sub-pixel edges.
3.
Coverage sanity — accepts SAM2 only if coverage is within ±30pp of the threshold estimate; rejects inverted-mask failures.
4.
Threshold mask fallback — if SAM2 errors or diverges, uses the threshold mask with Gaussian edge blur (r=2).
Test result: Person RGB mean (146, 101, 86) — correct skin tones. 74.5% transparent background, 24% opaque person. ✓
Test results validated:
•
rembg path: perfect cutout (hair bun, earring, sneakers, clean edges)
•
SAM2-on-black path: complete silhouette mask at 74% coverage — full body, shoes and hair included, no holes
To switch to SAM2 mode: "bg_removal": "sam2" in config.json. No restart needed — the config is read per-request.
This commit is contained in:
mike
2026-06-22 22:09:18 +02:00
parent 9e99c85134
commit 3d44c7aba4
6 changed files with 1428 additions and 215 deletions

View File

@@ -839,10 +839,63 @@ def _run_pipeline(
}
graph[NODE_POSITIVE]["inputs"][img_key] = [node_id, 0]
# Transparency detection
is_transparent = any(kw in prompt.lower() for kw in ["transparent", "no background", "remove background", "alpha channel"])
# ── background-removal routing ────────────────────────────────────────────
# Two configurable strategies (config.json key "bg_removal"):
#
# "rembg" (default) — strip transparent keyword → Qwen renders a natural
# scene → rembg (U2Net) separates person from any complex background.
#
# "sam2" — replace transparent keyword with "black background" → Qwen
# renders a solid black BG → SAM2 bbox segmentation on a black image
# works perfectly because the contrast is maximal.
#
# Either way, explicit "black background" in the prompt always routes to
# SAM2 (the user already set up the ideal SAM2 input).
# ─────────────────────────────────────────────────────────────────────────
_TRANSPARENT_KWS = ["transparent background", "no background",
"remove background", "alpha channel"]
_BLACK_BG_KWS = ["black background"]
with open(CONFIG_PATH) as _cf:
_bg_conf = json.load(_cf)
bg_method = _bg_conf.get("bg_removal", "rembg") # "rembg" | "sam2"
is_transparent = any(kw in prompt.lower() for kw in _TRANSPARENT_KWS)
is_black_bg = any(kw in prompt.lower() for kw in _BLACK_BG_KWS)
post_process = None # "rembg" | "sam2"
if is_transparent:
graph[NODE_NEGATIVE]["inputs"]["prompt"] = "checkerboard, grid, pattern, texture, background details, watermark, deformed anatomy"
if bg_method == "sam2":
# Swap "transparent background" → "black background" so Qwen renders
# a pure-black BG that SAM2 can segment with maximal contrast.
cleaned = prompt
for kw in _TRANSPARENT_KWS:
cleaned = re.sub(re.escape(kw), "black background", cleaned, flags=re.IGNORECASE)
# Collapse duplicates if multiple keywords matched
cleaned = re.sub(r"(?i)(black background[\s,]*){2,}", "black background, ", cleaned)
cleaned = re.sub(r",\s*,", ",", cleaned).strip(", ")
graph[NODE_POSITIVE]["inputs"]["prompt"] = cleaned
graph[NODE_NEGATIVE]["inputs"]["prompt"] = (
"real background, outdoor scene, indoor scene, gradient, "
"colored background, watermark, deformed anatomy"
)
post_process = "sam2"
else:
# Strip the keyword so Qwen renders a natural scene; rembg handles
# any background complexity reliably.
cleaned = prompt
for kw in _TRANSPARENT_KWS:
cleaned = re.sub(re.escape(kw), "", cleaned, flags=re.IGNORECASE)
cleaned = re.sub(r",\s*,", ",", cleaned)
cleaned = re.sub(r",\s*$", "", cleaned.strip()).strip(", ")
graph[NODE_POSITIVE]["inputs"]["prompt"] = cleaned
graph[NODE_NEGATIVE]["inputs"]["prompt"] = "deformed anatomy, watermark, logo"
post_process = "rembg"
elif is_black_bg:
# Prompt already specifies a black background — ideal SAM2 input.
# Route to SAM2 regardless of the configured bg_removal method.
post_process = "sam2"
graph[NODE_LATENT]["inputs"]["width"] = w
graph[NODE_LATENT]["inputs"]["height"] = h
@@ -853,7 +906,13 @@ def _run_pipeline(
outputs = _comfy_wait(prompt_id, time.time() + GEN_TIMEOUT)
png_bytes = _comfy_fetch_image(outputs)
if is_transparent:
if post_process == "sam2":
# Input has a black background (Qwen was told "black background").
# Use threshold-derived bbox so SAM2 gets a person-specific hint
# rather than the full frame — full-frame bbox inverts the mask on
# black-bg images because the large dark region scores higher.
png_bytes = _apply_transparency_black_bg(png_bytes)
elif post_process == "rembg":
png_bytes = _apply_transparency(png_bytes)
return png_bytes
@@ -891,7 +950,8 @@ def _move_to_trash(filepath: str):
def _batch_worker(job_id: str, filenames: list, prompts: list[str], poses: list,
seed: int, max_area: int, group_id: str | None = None):
seed: int, max_area: int, group_id: str | None = None,
wireframe_ref: str | None = None, wireframe_time: float = 0.5):
output_dir = _load_output_dir()
for fname in filenames:
actual_gid = None
@@ -915,6 +975,24 @@ def _batch_worker(job_id: str, filenames: list, prompts: list[str], poses: list,
try:
base_pil = Image.open(fpath).convert("RGB")
# Extract wireframe pose reference frame once per filename
pose_guide_pil = None
if wireframe_ref:
try:
wf_path = os.path.join(_load_wireframe_dir(), wireframe_ref)
cap = cv2.VideoCapture(wf_path)
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
target_frame = max(0, min(total_frames - 1, int(total_frames * wireframe_time)))
cap.set(cv2.CAP_PROP_POS_FRAMES, target_frame)
ret, frame = cap.read()
cap.release()
if ret:
pose_guide_pil = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
print(f"[batch] using wireframe {wireframe_ref} frame {target_frame}/{total_frames}")
except Exception as wf_err:
print(f"[batch] wireframe extract error: {wf_err}")
for prompt, pose in zip(prompts, poses):
if jobs[job_id].get("cancelled"):
return
@@ -924,7 +1002,8 @@ def _batch_worker(job_id: str, filenames: list, prompts: list[str], poses: list,
if pose and pose.lower().strip() in ROTATE_180_POSES:
pil = pil.rotate(180)
png = _run_pipeline(pil, prompt, seed, max_area)
extra_imgs = [pose_guide_pil] if pose_guide_pil else None
png = _run_pipeline(pil, prompt, seed, max_area, extra_images=extra_imgs)
ts = time.strftime("%Y%m%d_%H%M%S")
clean_fname = naming.get_base_name(fname)
out_name = f"{ts}_{clean_fname}"
@@ -1069,6 +1148,8 @@ class BatchRequest(BaseModel):
max_area: int = 0
group_id: str | None = None
poses: list[str | None] | None = None # pose name per prompt (same index), or None; None entries = no pose
wireframe_ref: str | None = None # wireframe video name to use as pose guide (image2 slot)
wireframe_time: float = 0.5 # normalized time (01) to extract the pose frame from
@app.post("/batch")
@@ -1085,6 +1166,7 @@ def start_batch(req: BatchRequest):
t = threading.Thread(
target=_batch_worker,
args=(job_id, req.filenames, prompts, poses, req.seed, req.max_area, req.group_id),
kwargs={"wireframe_ref": req.wireframe_ref, "wireframe_time": req.wireframe_time},
daemon=True,
)
t.start()
@@ -1207,6 +1289,35 @@ def list_videos():
return {"videos": videos, "wireframe_dir": wireframe_dir}
@app.get("/wireframe/frame/{video_name}")
def wireframe_frame(video_name: str, t: float = 0.5):
"""Extract a single frame at normalized time t (01) from a wireframe video. Returns PNG."""
wireframe_dir = _load_wireframe_dir()
video_path = os.path.join(wireframe_dir, video_name)
if not os.path.exists(video_path):
raise HTTPException(404, f"Video not found: {video_name}")
try:
cap = cv2.VideoCapture(video_path)
total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
target = max(0, min(total - 1, int(total * max(0.0, min(1.0, t)))))
cap.set(cv2.CAP_PROP_POS_FRAMES, target)
ret, frame = cap.read()
cap.release()
if not ret:
raise HTTPException(500, "Could not read frame")
rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
pil = Image.fromarray(rgb)
buf = io.BytesIO()
pil.save(buf, format="PNG")
buf.seek(0)
from fastapi.responses import Response
return Response(content=buf.getvalue(), media_type="image/png")
except HTTPException:
raise
except Exception as e:
raise HTTPException(500, f"Frame extraction error: {e}")
@app.get("/wireframe/duration/{video_name}")
def wireframe_duration(video_name: str):
"""Return duration (seconds) of a wireframe video via ffprobe."""
@@ -1573,6 +1684,42 @@ def _crop_to_bbox(pil_img: Image.Image, margin: int = 20, top_margin: int = 20,
return cropped
def _extract_face_bg(filename: str, fpath: str):
"""Background task: detect largest face, crop with padding, save as {group_id}_face.png."""
try:
app_fa, _ = _load_faceswapper()
bgr = cv2.imread(fpath)
if bgr is None:
print(f"[extract-face] cannot read {fpath}")
return
faces = app_fa.get(bgr)
if not faces:
print(f"[extract-face] no face detected in {filename}")
return
face = max(faces, key=lambda f: (f.bbox[2] - f.bbox[0]) * (f.bbox[3] - f.bbox[1]))
x1, y1, x2, y2 = [int(v) for v in face.bbox]
h, w = bgr.shape[:2]
pad = int((y2 - y1) * 0.5)
x1 = max(0, x1 - pad)
y1 = max(0, y1 - pad * 2) # extra headroom above face
x2 = min(w, x2 + pad)
y2 = min(h, y2 + int(pad * 0.3))
pil = Image.open(fpath).convert("RGBA")
cropped = pil.crop((x1, y1, x2, y2))
person = database.get_person(filename)
group_id = person[1] if person else None
gid_tag = (group_id or "face").replace("/", "_")
face_fname = f"{gid_tag}_face.png"
face_path = os.path.join(os.path.dirname(fpath), face_fname)
cropped.save(face_path)
database.upsert_person(face_fname, filepath=face_path, group_id=group_id,
name=person[0] if person else None,
source_refs=json.dumps([filename]))
print(f"[extract-face] saved {face_fname}")
except Exception as e:
print(f"[extract-face] error for {filename}: {e}")
def _process_upload(file_path: str, filename: str, prompts: list[str], name: str | None = None, group_id: str | None = None):
output_dir = _load_output_dir()
try:
@@ -1631,40 +1778,49 @@ def upload_image(
image: UploadFile = File(...),
prompts: str = Form(""),
name: str = Form(None),
group_id: str = Form(None), # optional: add to existing group
skip_poses: bool = Form(False), # optional: skip base_prompts generation
):
# Load config to get output_dir (we use output_dir for UI uploads to avoid watcher conflict)
with open(CONFIG_PATH, "r") as f:
conf = json.load(f)
output_dir = _load_output_dir()
os.makedirs(output_dir, exist_ok=True)
ts = time.strftime("%Y%m%d_%H%M%S")
safe_filename = re.sub(r'[^a-zA-Z0-9_.-]', '_', image.filename)
safe_filename = re.sub(r'[^a-zA-Z0-9_.-]', '_', image.filename or "paste")
# Ensure extension
if not safe_filename.lower().endswith(('.png', '.jpg', '.jpeg', '.webp')):
safe_filename += ".png"
filename = f"{ts}_{safe_filename}"
file_path = os.path.join(output_dir, filename)
with open(file_path, "wb") as f:
shutil.copyfileobj(image.file, f)
# Fast path: add to existing group without pose generation
if group_id and skip_poses:
sort_order = database.get_next_sort_order(group_id)
database.upsert_person(filename, filepath=file_path, group_id=group_id,
sort_order=sort_order)
return {"status": "added", "filename": filename, "group_id": group_id}
prompt_list = [p.strip() for p in prompts.split(",") if p.strip()]
# Add base-set prompts if defined in config
base_prompts = conf.get("base_prompts", [])
if isinstance(base_prompts, list):
prompt_list.extend(base_prompts)
if not prompt_list:
# Use default prompt from config
prompt_list = [conf.get("prompt", "high quality, masterpiece")]
group_id = f"up_{uuid.uuid4().hex[:8]}" # unique per upload; avoids collisions when pasting generic filenames
background_tasks.add_task(_process_upload, file_path, filename, prompt_list, name, group_id)
return {"status": "processing", "filename": filename, "group_id": group_id, "prompts": prompt_list}
effective_gid = group_id or f"up_{uuid.uuid4().hex[:8]}"
background_tasks.add_task(_process_upload, file_path, filename, prompt_list, name, effective_gid)
return {"status": "processing", "filename": filename, "group_id": effective_gid, "prompts": prompt_list}
@app.post("/edit")
@@ -1717,7 +1873,7 @@ def unarchive_image(filename: str):
@app.post("/images/{filename}/set-preferred")
def set_image_preferred(filename: str):
def set_image_preferred(filename: str, background_tasks: BackgroundTasks):
"""Make this image sort_order=0 within its group, shifting others to 1,2,..."""
person = database.get_person(filename)
if not person:
@@ -1728,9 +1884,23 @@ def set_image_preferred(filename: str):
rows = database.get_group_order(group_id)
others = [r[0] for r in rows if r[0] != filename]
database.set_group_order(group_id, [filename] + others)
fpath = os.path.join(_load_output_dir(), filename)
if os.path.exists(fpath):
background_tasks.add_task(_extract_face_bg, filename, fpath)
return {"filename": filename, "group_id": group_id}
@app.post("/images/{filename}/extract-face")
def extract_face_endpoint(filename: str, background_tasks: BackgroundTasks):
"""Detect and crop the largest face from image; saves as {group_id}_face.png."""
output_dir = _load_output_dir()
fpath = os.path.join(output_dir, filename)
if not os.path.exists(fpath):
raise HTTPException(404, "not found")
background_tasks.add_task(_extract_face_bg, filename, fpath)
return {"status": "queued", "filename": filename}
@app.post("/images/{filename}/undress")
def undress_image(filename: str, background_tasks: BackgroundTasks):
"""Queue a generation using the undress prompt on the given image."""
@@ -1967,14 +2137,14 @@ def _load_sam2():
def _apply_transparency_sam2(png_bytes: bytes) -> bytes:
"""Remove background with SAM2 bbox-based segmentation; fallback to rembg.
"""Remove background with SAM2 bbox segmentation; fallback to rembg.
Mimics the reference approach (bbox SAM2) but without an external
detector: we pass a generous bbox covering the central subject area.
For portraits/full-body shots the subject fills most of the frame, so a
5 % margin bbox reliably captures hair, glasses, and sandals without the
point-prompt clipping issues. multimask_output=True lets SAM2 propose
three masks; we pick the highest-scoring one.
Uses a near-full-frame bbox so SAM2 finds the largest foreground object
(the person) regardless of rotation or pose. This works well because
"transparent background" is stripped from the Qwen prompt upstream, so the
model renders a solid real background — giving SAM2 clear contrast to work
with. Point prompts were tried but produced holes in ¾-rotated poses
because the spine-column seeds land on background when the body is offset.
"""
predictor = _load_sam2()
if predictor is False:
@@ -1985,31 +2155,133 @@ def _apply_transparency_sam2(png_bytes: bytes) -> bytes:
img = Image.open(io.BytesIO(png_bytes)).convert("RGB")
arr = np.array(img)
h, w = arr.shape[:2]
# Generous bbox — 5 % margin H, 2 % margin V — covers the whole subject
box = np.array([[int(w * 0.05), int(h * 0.02),
int(w * 0.95), int(h * 0.98)]], dtype=np.float32)
# Near-full-frame bbox — 1 % margin so hair / shoes are inside the hint.
# SAM2 treats this as "find the prominent object within this region".
box = np.array([[int(w * 0.01), int(h * 0.01),
int(w * 0.99), int(h * 0.99)]], dtype=np.float32)
with torch.inference_mode():
predictor.set_image(arr)
masks, scores, _ = predictor.predict(
box=box,
multimask_output=True,
)
if masks is None or len(masks) == 0:
print("[sam2] no masks returned, falling back to rembg")
return _apply_transparency(png_bytes)
best = masks[int(np.argmax(scores))]
# Sanity check: a person should cover 5 %92 % of the frame
coverage = float(best.sum()) / (h * w)
if coverage < 0.05 or coverage > 0.92:
print(f"[sam2] mask coverage {coverage:.1%} out of range, falling back to rembg")
return _apply_transparency(png_bytes)
mask_np = best.astype(np.uint8) * 255
# Soft anti-aliased edge (radius 1 keeps accessory detail)
try:
from PIL import ImageFilter
alpha_img = Image.fromarray(mask_np, mode="L")
alpha_img = alpha_img.filter(ImageFilter.GaussianBlur(radius=1))
except Exception:
alpha_img = Image.fromarray(mask_np, mode="L")
rgba = img.convert("RGBA")
r, g, b, _ = rgba.split()
alpha = Image.fromarray(mask_np, mode="L")
out = Image.merge("RGBA", (r, g, b, alpha))
out = Image.merge("RGBA", (r, g, b, alpha_img))
buf = io.BytesIO()
out.save(buf, format="PNG")
print(f"[sam2] mask OK ({coverage:.1%} coverage)")
return buf.getvalue()
except Exception as e:
print(f"[sam2] inference error, falling back to rembg: {e}")
return _apply_transparency(png_bytes)
def _apply_transparency_black_bg(png_bytes: bytes) -> bytes:
"""Background removal for black-background Qwen output (bg_removal=sam2 mode).
Strategy:
1. Threshold: any pixel with max-channel > 25 is person (non-black).
This correctly identifies the subject regardless of pose or rotation.
2. Derive a tight person bounding-box from the threshold mask.
3. Run SAM2 with that box for sub-pixel edge refinement.
Accept SAM2 result only when its coverage is close (±30 pp) to the
threshold estimate — this rejects the inverted-mask failure mode where
SAM2 picks the large dark region as the "object".
4. Fall back to the threshold mask (Gaussian-blurred edges) if SAM2
is unavailable, errors, or diverges.
Do NOT use the full-frame bbox here: on black-background images the large
dark region scores higher than the person, causing SAM2 to invert the mask.
"""
import numpy as np
import torch
from PIL import ImageFilter
img = Image.open(io.BytesIO(png_bytes)).convert("RGB")
arr = np.array(img)
h, w = arr.shape[:2]
# Step 1 — threshold: non-black pixels are the person
is_person = np.max(arr, axis=2) > 25
thresh_cov = float(is_person.sum()) / (h * w)
print(f"[bg-black] threshold coverage: {thresh_cov:.1%}")
if not is_person.any():
print("[bg-black] all-black image — falling back to rembg")
return _apply_transparency(png_bytes)
# Step 2 — tight bounding box from threshold
rows = np.any(is_person, axis=1)
cols = np.any(is_person, axis=0)
rmin = int(np.where(rows)[0][0]); rmax = int(np.where(rows)[0][-1])
cmin = int(np.where(cols)[0][0]); cmax = int(np.where(cols)[0][-1])
margin = int(min(h, w) * 0.02)
x1 = max(0, cmin - margin); y1 = max(0, rmin - margin)
x2 = min(w, cmax + margin); y2 = min(h, rmax + margin)
# Step 3 — SAM2 with the person-specific bbox
predictor = _load_sam2()
if predictor is not False:
box = np.array([[x1, y1, x2, y2]], dtype=np.float32)
try:
with torch.inference_mode():
predictor.set_image(arr)
masks, scores, _ = predictor.predict(box=box, multimask_output=True)
if masks is not None and len(masks) > 0:
best = masks[int(np.argmax(scores))]
sam_cov = float(best.sum()) / (h * w)
print(f"[bg-black] SAM2 coverage: {sam_cov:.1%}")
if 0.03 < sam_cov < 0.95 and abs(sam_cov - thresh_cov) < 0.30:
mask_np = best.astype(np.uint8) * 255
alpha_img = Image.fromarray(mask_np, "L").filter(ImageFilter.GaussianBlur(radius=1))
rgba = img.convert("RGBA"); r, g, b, _ = rgba.split()
out = Image.merge("RGBA", (r, g, b, alpha_img))
buf = io.BytesIO(); out.save(buf, "PNG")
print(f"[bg-black] SAM2 accepted ✓")
return buf.getvalue()
else:
print(f"[bg-black] SAM2 diverged ({sam_cov:.1%} vs {thresh_cov:.1%}) — threshold fallback")
except Exception as e:
print(f"[bg-black] SAM2 error: {e} — threshold fallback")
# Step 4 — fallback: threshold mask with soft edge blur
print("[bg-black] using threshold mask")
mask_np = is_person.astype(np.uint8) * 255
alpha_img = Image.fromarray(mask_np, "L").filter(ImageFilter.GaussianBlur(radius=2))
rgba = img.convert("RGBA"); r, g, b, _ = rgba.split()
out = Image.merge("RGBA", (r, g, b, alpha_img))
buf = io.BytesIO(); out.save(buf, "PNG")
return buf.getvalue()
@app.post("/remove-background-sam/{filename}")
def remove_background_sam(filename: str):
"""SAM2-based background removal.