feat: kill all silent fallbacks — fail clearly, never degrade silently

Remove 9 silent fallback chains in the Python sidecar:
- upscale: RealESRGAN→Lanczos (now errors with install guidance)
- upscale: GFPGAN skip (now errors with install guidance)
- gpu: GPU→CPU (now reports device in response, never silent)
- remove_bg: alpha matting fallback (now errors with retry guidance)
- remove_bg: GPU→CPU session (now reports device)
- colorize: DDColor→OpenCV (now errors with install guidance)
- enhance_faces: CodeFormer→GFPGAN (now errors with install guidance)
- ocr: quality cascade (now errors at requested level)
- bridge: dispatcher crash retry (now reports retry in stderr)

Also: raise red_eye max_faces 10→50, face_landmarks max_num_faces configurable,
restore.py min face size 48→24px.
This commit is contained in:
ashim-hq
2026-04-20 21:42:19 +08:00
parent ce477a0dbf
commit 00041d535d
11 changed files with 141 additions and 130 deletions
+18 -20
View File
@@ -50,7 +50,7 @@ def colorize_ddcolor(img_bgr, intensity):
emit_progress(15, "Loading DDColor model")
session = safe_onnx_session(DDCOLOR_MODEL_PATH)
session, _device = safe_onnx_session(DDCOLOR_MODEL_PATH)
input_name = session.get_inputs()[0].name
input_shape = session.get_inputs()[0].shape
# Dynamic dims are strings ('w', 'h'), so default to 512 if not int
@@ -181,41 +181,39 @@ def main():
result_bgr = None
method = "unknown"
# Try DDColor first
if model_choice in ("auto", "ddcolor"):
try:
if os.path.exists(DDCOLOR_MODEL_PATH):
result_bgr, method = colorize_ddcolor(img_bgr, intensity)
elif model_choice == "ddcolor":
if not os.path.exists(DDCOLOR_MODEL_PATH):
raise FileNotFoundError(f"DDColor model not found: {DDCOLOR_MODEL_PATH}")
result_bgr, method = colorize_ddcolor(img_bgr, intensity)
except Exception as e:
import traceback
print(f"[colorize] DDColor failed: {e}", file=sys.stderr, flush=True)
traceback.print_exc(file=sys.stderr)
if model_choice == "ddcolor":
# User explicitly requested ddcolor — fail, don't degrade
raise
result_bgr = None
print(json.dumps({
"success": False,
"error": (
f"DDColor is not available: {e}. "
"Install the colorize feature or use model=opencv for basic colorization."
),
}))
sys.exit(1)
# Try OpenCV fallback only in auto mode
if result_bgr is None and model_choice in ("auto", "opencv"):
elif model_choice == "opencv":
try:
if os.path.exists(OPENCV_PROTO_PATH) and os.path.exists(OPENCV_MODEL_PATH):
result_bgr, method = colorize_opencv(img_bgr, intensity)
elif model_choice == "opencv":
if not (os.path.exists(OPENCV_PROTO_PATH) and os.path.exists(OPENCV_MODEL_PATH)):
raise FileNotFoundError(f"OpenCV colorize models not found: {OPENCV_PROTO_PATH}")
result_bgr, method = colorize_opencv(img_bgr, intensity)
except Exception as e:
import traceback
print(f"[colorize] OpenCV fallback failed: {e}", file=sys.stderr, flush=True)
print(f"[colorize] OpenCV failed: {e}", file=sys.stderr, flush=True)
traceback.print_exc(file=sys.stderr)
if model_choice == "opencv":
raise
result_bgr = None
raise
if result_bgr is None:
else:
print(json.dumps({
"success": False,
"error": "No colorization model available. Install DDColor or OpenCV models.",
"error": f"Unknown model '{model_choice}'. Use 'auto', 'ddcolor', or 'opencv'.",
}))
sys.exit(1)
+5 -6
View File
@@ -270,19 +270,18 @@ def main():
model_used = "codeformer"
elif model_choice == "auto":
# Try CodeFormer first, fall back to GFPGAN.
# Catch broad Exception because codeformer-pip can fail in
# unexpected ways (AttributeError, TypeError, etc.)
try:
fidelity_weight = 1.0 - strength
enhanced = enhance_with_codeformer(img_array, fidelity_weight)
model_used = "codeformer"
except Exception as e:
import traceback
print(f"[enhance-faces] CodeFormer failed, falling back to GFPGAN: {e}", file=sys.stderr, flush=True)
print(f"[enhance-faces] CodeFormer failed: {e}", file=sys.stderr, flush=True)
traceback.print_exc(file=sys.stderr)
enhanced = enhance_with_gfpgan(img_array, only_center_face)
model_used = "gfpgan"
raise RuntimeError(
f"CodeFormer is not available: {e}. "
"Install the face-enhance feature or use model=gfpgan."
) from e
finally:
# Restore stdout after ALL AI processing
+8 -8
View File
@@ -54,14 +54,14 @@ def extract_key_points(lms):
# ── Old API: mp.solutions (mediapipe < 0.10.30) ───────────────────
def detect_with_solutions(img_array):
def detect_with_solutions(img_array, max_faces=1):
"""Use the legacy mp.solutions.face_mesh API."""
import mediapipe as mp
mp_face_mesh = mp.solutions.face_mesh
face_mesh = mp_face_mesh.FaceMesh(
static_image_mode=True,
max_num_faces=1,
max_num_faces=max_faces,
refine_landmarks=True,
min_detection_confidence=0.5,
)
@@ -99,7 +99,7 @@ def ensure_model():
return MODEL_PATH
def detect_with_tasks(img_path):
def detect_with_tasks(img_path, max_faces=1):
"""Use the new mp.tasks.vision.FaceLandmarker API."""
import mediapipe as mp
@@ -108,7 +108,7 @@ def detect_with_tasks(img_path):
options = mp.tasks.vision.FaceLandmarkerOptions(
base_options=mp.tasks.BaseOptions(model_asset_path=model_path),
running_mode=mp.tasks.vision.RunningMode.IMAGE,
num_faces=1,
num_faces=max_faces,
min_face_detection_confidence=0.5,
output_face_blendshapes=False,
output_facial_transformation_matrixes=False,
@@ -133,6 +133,8 @@ def main():
output_path = sys.argv[2] # unused but kept for bridge.ts compatibility
settings = json.loads(sys.argv[3]) if len(sys.argv) > 3 else {}
max_faces = settings.get("max_num_faces", 1)
try:
emit_progress(10, "Loading image")
from PIL import Image
@@ -146,16 +148,14 @@ def main():
emit_progress(20, "Initializing face mesh")
# Try the legacy solutions API first (Docker / older mediapipe),
# fall back to the tasks API (newer mediapipe versions).
landmarks_list = None
try:
img_array = np.array(img)
emit_progress(30, "Detecting face landmarks")
landmarks_list = detect_with_solutions(img_array)
landmarks_list = detect_with_solutions(img_array, max_faces)
except AttributeError:
emit_progress(30, "Detecting face landmarks")
landmarks_list = detect_with_tasks(input_path)
landmarks_list = detect_with_tasks(input_path, max_faces)
if landmarks_list is None:
print(json.dumps({
+25 -9
View File
@@ -1,10 +1,16 @@
"""Runtime GPU/CUDA detection utility."""
import functools
import json
import os
import subprocess
import sys
def emit_info(msg):
"""Emit an informational JSON message to stderr for the bridge to capture."""
print(json.dumps({"info": msg}), file=sys.stderr, flush=True)
@functools.lru_cache(maxsize=1)
def gpu_available():
"""Return True if a usable CUDA GPU is present at runtime."""
@@ -51,24 +57,34 @@ def gpu_available():
def onnx_providers():
"""Return ONNX Runtime execution providers in priority order."""
"""Return (providers, device) tuple.
providers: ONNX Runtime execution providers in priority order.
device: "cuda" or "cpu" reflects which hardware will actually be used.
"""
if gpu_available():
return ["CUDAExecutionProvider", "CPUExecutionProvider"]
return ["CPUExecutionProvider"]
return (["CUDAExecutionProvider", "CPUExecutionProvider"], "cuda")
emit_info("No GPU detected, processing on CPU")
return (["CPUExecutionProvider"], "cpu")
def safe_onnx_session(model_path, providers=None):
"""Create an ONNX Runtime InferenceSession with graceful CUDA EP fallback."""
"""Create an ONNX Runtime InferenceSession with graceful CUDA EP fallback.
Returns (session, device) where device is "cuda" or "cpu".
"""
import onnxruntime as ort
device = "cpu"
if providers is None:
providers = onnx_providers()
providers, device = onnx_providers()
try:
return ort.InferenceSession(model_path, providers=providers)
session = ort.InferenceSession(model_path, providers=providers)
return session, device
except Exception as e:
if "CUDAExecutionProvider" in providers:
print(f"[gpu] CUDA EP init failed ({e}), falling back to CPU",
file=sys.stderr, flush=True)
return ort.InferenceSession(model_path, providers=["CPUExecutionProvider"])
emit_info(f"CUDA init failed ({e}), falling back to CPU")
session = ort.InferenceSession(model_path, providers=["CPUExecutionProvider"])
return session, "cpu"
raise
+1 -1
View File
@@ -111,7 +111,7 @@ def main():
model_path = _get_model_path()
from gpu import safe_onnx_session
session = safe_onnx_session(model_path)
session, _device = safe_onnx_session(model_path)
emit_progress(20, "Loading images")
img = Image.open(input_path).convert("RGB")
+28 -36
View File
@@ -245,19 +245,23 @@ def main():
text = run_paddleocr_v5(input_path, language)
engine_used = "paddleocr-v5"
except ImportError as e:
print(json.dumps({"success": False, "error": f"PaddleOCR is not installed: {e}"}))
print(json.dumps({
"success": False,
"error": (
f"PaddleOCR is not installed: {e}. "
"Install the OCR feature or use quality=fast for Tesseract."
),
}))
sys.exit(1)
except Exception as e:
print(json.dumps({
"warning": f"PaddleOCR PP-OCRv5 failed ({type(e).__name__}: {e}), falling back to Tesseract"
}), file=sys.stderr, flush=True)
emit_progress(25, "PaddleOCR failed, falling back to Tesseract")
try:
text = run_tesseract(input_path, language, is_auto=was_auto)
engine_used = "tesseract (fallback from balanced)"
except FileNotFoundError:
print(json.dumps({"success": False, "error": "OCR engines unavailable: PaddleOCR failed and Tesseract is not installed"}))
sys.exit(1)
"success": False,
"error": (
f"PaddleOCR PP-OCRv5 failed: {type(e).__name__}: {e}. "
"Install the OCR feature or use quality=fast for Tesseract."
),
}))
sys.exit(1)
elif quality == "best":
try:
@@ -265,34 +269,22 @@ def main():
engine_used = "paddleocr-vl"
except ImportError as e:
print(json.dumps({
"warning": f"PaddleOCR-VL not available ({e}), trying PP-OCRv5"
}), file=sys.stderr, flush=True)
emit_progress(20, "VL model unavailable, trying PP-OCRv5")
try:
text = run_paddleocr_v5(input_path, language)
engine_used = "paddleocr-v5 (fallback from best)"
except Exception as e2:
print(json.dumps({
"warning": f"PP-OCRv5 also failed ({type(e2).__name__}: {e2}), falling back to Tesseract"
}), file=sys.stderr, flush=True)
emit_progress(25, "PP-OCRv5 failed, falling back to Tesseract")
text = run_tesseract(input_path, language, is_auto=was_auto)
engine_used = "tesseract (fallback from best)"
"success": False,
"error": (
f"PaddleOCR-VL is not available: {e}. "
"Install the OCR feature or use quality=balanced for PP-OCRv5."
),
}))
sys.exit(1)
except Exception as e:
print(json.dumps({
"warning": f"PaddleOCR-VL failed ({type(e).__name__}: {e}), trying PP-OCRv5"
}), file=sys.stderr, flush=True)
emit_progress(20, "VL model failed, trying PP-OCRv5")
try:
text = run_paddleocr_v5(input_path, language)
engine_used = "paddleocr-v5 (fallback from best)"
except Exception as e2:
print(json.dumps({
"warning": f"PP-OCRv5 also failed ({type(e2).__name__}: {e2}), falling back to Tesseract"
}), file=sys.stderr, flush=True)
emit_progress(25, "PP-OCRv5 failed, falling back to Tesseract")
text = run_tesseract(input_path, language, is_auto=was_auto)
engine_used = "tesseract (fallback from best)"
"success": False,
"error": (
f"PaddleOCR-VL failed: {type(e).__name__}: {e}. "
"Install the OCR feature or use quality=balanced for PP-OCRv5."
),
}))
sys.exit(1)
else:
print(json.dumps({"success": False, "error": f"Unknown quality: {quality}"}))
+3 -3
View File
@@ -32,7 +32,7 @@ def _ensure_face_mesh_model():
return _LOCAL_MODEL_PATH
def _mesh_with_solutions(img_array, max_faces=10, min_confidence=0.5):
def _mesh_with_solutions(img_array, max_faces=50, min_confidence=0.5):
"""FaceMesh using legacy mp.solutions API (mediapipe < 0.10.30).
Returns list of landmark lists. Each landmark has .x, .y attributes.
@@ -54,7 +54,7 @@ def _mesh_with_solutions(img_array, max_faces=10, min_confidence=0.5):
return [face.landmark for face in results.multi_face_landmarks]
def _mesh_with_tasks(img_array, max_faces=10, min_confidence=0.5):
def _mesh_with_tasks(img_array, max_faces=50, min_confidence=0.5):
"""FaceMesh using new mp.tasks API (mediapipe >= 0.10.30).
Returns list of landmark lists. Each landmark has .x, .y attributes.
@@ -79,7 +79,7 @@ def _mesh_with_tasks(img_array, max_faces=10, min_confidence=0.5):
return result.face_landmarks
def _detect_face_mesh(img_array, max_faces=10, min_confidence=0.5):
def _detect_face_mesh(img_array, max_faces=50, min_confidence=0.5):
"""Detect face mesh, trying legacy API first then falling back to tasks API."""
try:
return _mesh_with_solutions(img_array, max_faces, min_confidence)
+8 -7
View File
@@ -76,14 +76,15 @@ def main():
emit_progress(10, "Loading model")
providers = onnx_providers()
providers, device = onnx_providers()
try:
session = new_session(model, providers=providers)
except Exception as e:
if "CUDAExecutionProvider" in providers:
print(f"[remove-bg] GPU session failed ({e}), falling back to CPU",
file=sys.stderr, flush=True)
from gpu import emit_info
emit_info(f"GPU session failed ({e}), falling back to CPU")
session = new_session(model, providers=["CPUExecutionProvider"])
device = "cpu"
else:
raise
@@ -92,7 +93,6 @@ def main():
with open(input_path, "rb") as f:
input_data = f.read()
# Try with alpha matting for better edges, fall back without
emit_progress(30, "Analyzing image")
try:
output_data = remove(
@@ -103,8 +103,9 @@ def main():
alpha_matting_background_threshold=10,
)
except Exception as e:
print(f"[remove-bg] Alpha matting failed ({e}), using standard removal", file=sys.stderr, flush=True)
output_data = remove(input_data, session=session)
raise RuntimeError(
f"Alpha matting failed: {e}. Try again without alpha matting or with a different model."
) from e
emit_progress(80, "Background removed")
@@ -115,7 +116,7 @@ def main():
with open(output_path, "wb") as f:
f.write(output_data)
result = json.dumps({"success": True, "model": model})
result = json.dumps({"success": True, "model": model, "device": device})
except ImportError as e:
print(f"[remove-bg] Import failed: {e}", file=sys.stderr, flush=True)
+8 -5
View File
@@ -155,7 +155,7 @@ def inpaint_damage(img_bgr, mask):
from gpu import safe_onnx_session
model_path = _get_lama_path()
session = safe_onnx_session(model_path)
session, _device = safe_onnx_session(model_path)
orig_h, orig_w = img_bgr.shape[:2]
img_rgb = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2RGB)
@@ -317,7 +317,7 @@ def enhance_faces(img_bgr, fidelity=0.7):
# Load CodeFormer model
model_path = _get_codeformer_path()
session = safe_onnx_session(model_path)
session, _device = safe_onnx_session(model_path)
input_names = [inp.name for inp in session.get_inputs()]
result = img_bgr.copy()
@@ -329,8 +329,7 @@ def enhance_faces(img_bgr, fidelity=0.7):
w = face_box["w"]
h = face_box["h"]
# Skip very small faces (under 48px) - enhancement won't help
if w < 48 or h < 48:
if w < 24 or h < 24:
continue
# Expand bounding box by ~80% for hair, forehead, chin
@@ -473,7 +472,7 @@ def colorize_bw(img_bgr, intensity=0.85):
if not os.path.exists(DDCOLOR_MODEL_PATH):
return img_bgr, False
session = safe_onnx_session(DDCOLOR_MODEL_PATH)
session, _device = safe_onnx_session(DDCOLOR_MODEL_PATH)
input_name = session.get_inputs()[0].name
input_shape = session.get_inputs()[0].shape
model_size = (
@@ -545,6 +544,9 @@ def main():
scratch_sensitivity = "medium"
try:
from gpu import gpu_available
device = "cuda" if gpu_available() else "cpu"
emit_progress(5, "Opening image")
img_bgr = cv2.imread(input_path, cv2.IMREAD_COLOR)
if img_bgr is None:
@@ -633,6 +635,7 @@ def main():
"facesEnhanced": faces_found,
"isGrayscale": bw_detected,
"colorized": colorized,
"device": device,
"output_path": output_path,
}))
+33 -31
View File
@@ -126,32 +126,30 @@ def main():
result = Image.fromarray(output_array)
method = "realesrgan"
# Face enhancement with GFPGAN
if face_enhance:
emit_progress(82, "Enhancing faces")
try:
from gfpgan import GFPGANer
from gfpgan import GFPGANer
if os.path.exists(GFPGAN_MODEL_PATH):
face_enhancer = GFPGANer(
model_path=GFPGAN_MODEL_PATH,
upscale=scale,
arch="clean",
channel_multiplier=2,
bg_upsampler=upsampler,
)
_, _, face_output = face_enhancer.enhance(
img_array,
has_aligned=False,
only_center_face=False,
paste_back=True,
)
result = Image.fromarray(face_output)
emit_progress(88, "Face enhancement complete")
else:
emit_progress(88, "Face model not found, skipping")
except (ImportError, RuntimeError, OSError):
emit_progress(88, "Face enhancement unavailable, skipping")
if not os.path.exists(GFPGAN_MODEL_PATH):
raise FileNotFoundError(
f"GFPGAN model not found at {GFPGAN_MODEL_PATH}. "
"Install the upscale-enhance feature or disable faceEnhance."
)
face_enhancer = GFPGANer(
model_path=GFPGAN_MODEL_PATH,
upscale=scale,
arch="clean",
channel_multiplier=2,
bg_upsampler=upsampler,
)
_, _, face_output = face_enhancer.enhance(
img_array,
has_aligned=False,
only_center_face=False,
paste_back=True,
)
result = Image.fromarray(face_output)
emit_progress(88, "Face enhancement complete")
finally:
# Restore stdout after ALL AI processing
@@ -164,19 +162,23 @@ def main():
import traceback
print(f"[upscale] Real-ESRGAN failed: {e}", file=sys.stderr, flush=True)
traceback.print_exc(file=sys.stderr)
if model_choice == "realesrgan":
# User explicitly requested realesrgan — fail, don't degrade
raise RuntimeError(f"Real-ESRGAN unavailable: {e}") from e
result = None
print(json.dumps({
"success": False,
"error": (
f"Real-ESRGAN is not available: {e}. "
"Install the upscale-enhance feature or use model=lanczos for basic upscaling."
),
}))
sys.exit(1)
# Lanczos path: used when explicitly requested or as auto fallback
if result is None:
if model_choice not in ("auto", "lanczos"):
raise RuntimeError(f"Requested model '{model_choice}' is not available")
if result is None and model_choice == "lanczos":
emit_progress(50, "Upscaling with Lanczos")
result = img.resize(new_size, Image.LANCZOS)
method = "lanczos"
if result is None:
raise RuntimeError(f"Requested model '{model_choice}' is not available")
# Denoise
if denoise_strength > 0:
emit_progress(90, "Reducing noise")