fix(ai): support both old and new mediapipe APIs for airgapped Docker (#69)

MediaPipe >= 0.10.30 removed the mp.solutions namespace. This broke
face blur, face enhance, red-eye removal, and photo restoration for
users running newer mediapipe versions (closes #43).

All 5 Python scripts that use mediapipe now try the legacy mp.solutions
API first and fall back to the new mp.tasks API on AttributeError.
Model files (blaze_face_short_range.task, face_landmarker.task) are
pre-downloaded during Docker build into /opt/models/mediapipe/ so the
image works fully airgapped. Local dev auto-downloads to .models/.

Co-authored-by: stirling-image <stirling-image@users.noreply.github.com>
This commit is contained in:
stirling-image
2026-04-14 16:18:17 +08:00
committed by GitHub
co-authored by stirling-image
parent 5be8be3dc3
commit 519541867e
6 changed files with 117 additions and 39 deletions
+11 -8
View File
@@ -12,19 +12,22 @@ def emit_progress(percent, stage):
# ── Model path for new mp.tasks API ─────────────────────────────────
_FACE_DETECT_MODEL_URL = "https://storage.googleapis.com/mediapipe-models/face_detector/blaze_face_short_range/float16/latest/blaze_face_short_range.task"
_MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_FACE_DETECT_MODEL_PATH = os.path.join(_MODEL_DIR, "blaze_face_short_range.task")
_DOCKER_MODEL_PATH = "/opt/models/mediapipe/blaze_face_short_range.task"
_LOCAL_MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_LOCAL_MODEL_PATH = os.path.join(_LOCAL_MODEL_DIR, "blaze_face_short_range.task")
def _ensure_face_detect_model():
"""Download the face detector model if not present."""
if os.path.exists(_FACE_DETECT_MODEL_PATH):
return _FACE_DETECT_MODEL_PATH
os.makedirs(_MODEL_DIR, exist_ok=True)
"""Resolve face detector model. Docker path first, then local dev."""
if os.path.exists(_DOCKER_MODEL_PATH):
return _DOCKER_MODEL_PATH
if os.path.exists(_LOCAL_MODEL_PATH):
return _LOCAL_MODEL_PATH
os.makedirs(_LOCAL_MODEL_DIR, exist_ok=True)
import urllib.request
emit_progress(15, "Downloading face detection model")
urllib.request.urlretrieve(_FACE_DETECT_MODEL_URL, _FACE_DETECT_MODEL_PATH)
return _FACE_DETECT_MODEL_PATH
urllib.request.urlretrieve(_FACE_DETECT_MODEL_URL, _LOCAL_MODEL_PATH)
return _LOCAL_MODEL_PATH
def _detect_with_solutions(img_array, min_confidence):
+11 -8
View File
@@ -40,19 +40,22 @@ CODEFORMER_MODEL_PATH = os.environ.get(
# ── Model path for new mp.tasks API ─────────────────────────────────
_FACE_DETECT_MODEL_URL = "https://storage.googleapis.com/mediapipe-models/face_detector/blaze_face_short_range/float16/latest/blaze_face_short_range.task"
_MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_FACE_DETECT_MODEL_PATH = os.path.join(_MODEL_DIR, "blaze_face_short_range.task")
_DOCKER_MODEL_PATH = "/opt/models/mediapipe/blaze_face_short_range.task"
_LOCAL_MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_LOCAL_MODEL_PATH = os.path.join(_LOCAL_MODEL_DIR, "blaze_face_short_range.task")
def _ensure_face_detect_model():
"""Download the face detector model if not present."""
if os.path.exists(_FACE_DETECT_MODEL_PATH):
return _FACE_DETECT_MODEL_PATH
os.makedirs(_MODEL_DIR, exist_ok=True)
"""Resolve face detector model. Docker path first, then local dev."""
if os.path.exists(_DOCKER_MODEL_PATH):
return _DOCKER_MODEL_PATH
if os.path.exists(_LOCAL_MODEL_PATH):
return _LOCAL_MODEL_PATH
os.makedirs(_LOCAL_MODEL_DIR, exist_ok=True)
import urllib.request
emit_progress(15, "Downloading face detection model")
urllib.request.urlretrieve(_FACE_DETECT_MODEL_URL, _FACE_DETECT_MODEL_PATH)
return _FACE_DETECT_MODEL_PATH
urllib.request.urlretrieve(_FACE_DETECT_MODEL_URL, _LOCAL_MODEL_PATH)
return _LOCAL_MODEL_PATH
def detect_faces_mediapipe(img_array, sensitivity):
+4 -1
View File
@@ -79,12 +79,15 @@ def detect_with_solutions(img_array):
# ── New API: mp.tasks (mediapipe >= 0.10.30) ───────────────────────
MODEL_URL = "https://storage.googleapis.com/mediapipe-models/face_landmarker/face_landmarker/float16/latest/face_landmarker.task"
_DOCKER_MODEL_PATH = "/opt/models/mediapipe/face_landmarker.task"
MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
MODEL_PATH = os.path.join(MODEL_DIR, "face_landmarker.task")
def ensure_model():
"""Download the face landmarker model if not present."""
"""Resolve face landmarker model. Docker path first, then local dev."""
if os.path.exists(_DOCKER_MODEL_PATH):
return _DOCKER_MODEL_PATH
if os.path.exists(MODEL_PATH):
return MODEL_PATH
os.makedirs(MODEL_DIR, exist_ok=True)
+11 -8
View File
@@ -12,19 +12,22 @@ def emit_progress(percent, stage):
# ── Model path for new mp.tasks API ─────────────────────────────────
_FACE_MESH_MODEL_URL = "https://storage.googleapis.com/mediapipe-models/face_landmarker/face_landmarker/float16/latest/face_landmarker.task"
_MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_FACE_MESH_MODEL_PATH = os.path.join(_MODEL_DIR, "face_landmarker.task")
_DOCKER_MODEL_PATH = "/opt/models/mediapipe/face_landmarker.task"
_LOCAL_MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_LOCAL_MODEL_PATH = os.path.join(_LOCAL_MODEL_DIR, "face_landmarker.task")
def _ensure_face_mesh_model():
"""Download the face landmarker model if not present."""
if os.path.exists(_FACE_MESH_MODEL_PATH):
return _FACE_MESH_MODEL_PATH
os.makedirs(_MODEL_DIR, exist_ok=True)
"""Resolve face landmarker model. Docker path first, then local dev."""
if os.path.exists(_DOCKER_MODEL_PATH):
return _DOCKER_MODEL_PATH
if os.path.exists(_LOCAL_MODEL_PATH):
return _LOCAL_MODEL_PATH
os.makedirs(_LOCAL_MODEL_DIR, exist_ok=True)
import urllib.request
emit_progress(15, "Downloading face mesh model")
urllib.request.urlretrieve(_FACE_MESH_MODEL_URL, _FACE_MESH_MODEL_PATH)
return _FACE_MESH_MODEL_PATH
urllib.request.urlretrieve(_FACE_MESH_MODEL_URL, _LOCAL_MODEL_PATH)
return _LOCAL_MODEL_PATH
def _mesh_with_solutions(img_array, max_faces=10, min_confidence=0.5):
+11 -8
View File
@@ -220,19 +220,22 @@ def _get_codeformer_path():
# ── Model path for new mp.tasks API ─────────────────────────────────
_FACE_DETECT_MODEL_URL = "https://storage.googleapis.com/mediapipe-models/face_detector/blaze_face_short_range/float16/latest/blaze_face_short_range.task"
_FACE_DETECT_MODEL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_FACE_DETECT_MODEL_PATH = os.path.join(_FACE_DETECT_MODEL_DIR, "blaze_face_short_range.task")
_FACE_DETECT_DOCKER_PATH = "/opt/models/mediapipe/blaze_face_short_range.task"
_FACE_DETECT_LOCAL_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "..", ".models")
_FACE_DETECT_LOCAL_PATH = os.path.join(_FACE_DETECT_LOCAL_DIR, "blaze_face_short_range.task")
def _ensure_face_detect_model():
"""Download the face detector model if not present."""
if os.path.exists(_FACE_DETECT_MODEL_PATH):
return _FACE_DETECT_MODEL_PATH
os.makedirs(_FACE_DETECT_MODEL_DIR, exist_ok=True)
"""Resolve face detector model. Docker path first, then local dev."""
if os.path.exists(_FACE_DETECT_DOCKER_PATH):
return _FACE_DETECT_DOCKER_PATH
if os.path.exists(_FACE_DETECT_LOCAL_PATH):
return _FACE_DETECT_LOCAL_PATH
os.makedirs(_FACE_DETECT_LOCAL_DIR, exist_ok=True)
import urllib.request
emit_progress(15, "Downloading face detection model")
urllib.request.urlretrieve(_FACE_DETECT_MODEL_URL, _FACE_DETECT_MODEL_PATH)
return _FACE_DETECT_MODEL_PATH
urllib.request.urlretrieve(_FACE_DETECT_MODEL_URL, _FACE_DETECT_LOCAL_PATH)
return _FACE_DETECT_LOCAL_PATH
def enhance_faces(img_bgr, fidelity=0.7):