| |
| """InsightFace `inswapper_128` face swap with OpenVINO - single file example. |
| |
| Everything runs on OpenVINO (no PyTorch / onnxruntime needed at inference time): |
| |
| 1. retinaface_10g -> face detection + 5 point landmarks |
| 2. arcface_w600k_r50-> 512-d face embedding of the *source* face |
| 3. inswapper_128 -> the swap itself (128x128 target crop + source embedding) |
| |
| The OpenVINO IRs are published in the Hugging Face repo `HelloSun/inswapper` |
| in 16-bit (FP16), 8-bit (W8A8) and 4-bit (u4 weight-only) precisions. They are |
| downloaded automatically on first use and cached in `~/.cache/inswapper_openvino`. |
| |
| IMPORTANT: the InsightFace models are published for **non-commercial research |
| use only** (see the README). Do not use them to impersonate people. |
| |
| Command line |
| ------------ |
| python inswapper_ov.py -s source.jpg -t target.jpg -o out.jpg |
| python inswapper_ov.py -s source.jpg -t target.jpg -o out.jpg -m int8 |
| python inswapper_ov.py -s source.jpg -t target.jpg -o out.jpg --all-faces |
| python inswapper_ov.py -s source.jpg -t target.jpg -o out.jpg -d GPU |
| |
| Python API |
| ---------- |
| from inswapper_ov import FaceSwapperOpenVINO |
| swapper = FaceSwapperOpenVINO(precision='int8') |
| out = swapper.swap('source.jpg', 'target.jpg') |
| """ |
|
|
| import argparse |
| import os |
| import pathlib |
| import shutil |
| import sys |
| import time |
|
|
| import cv2 |
| import numpy as np |
| import openvino as ov |
|
|
| HF_REPO = 'HelloSun/inswapper' |
| CACHE_DIR = pathlib.Path(os.environ.get('INSWAPPER_OV_CACHE', pathlib.Path.home() / '.cache/inswapper_openvino')) |
|
|
| |
| PRECISIONS = { |
| 'fp16': ('fp16', '16-bit FP16 weights/activations'), |
| 'int8': ('int8', '8-bit W8A8 (NNCF PTQ + SmoothQuant)'), |
| 'int4': ('int4', '4-bit u4 weight-only (NNCF LWC), FP16 activations'), |
| 'int4-mixed': ('int4_mixed', '4-bit/8-bit mixed precision (75% of the weights INT4)'), |
| } |
| AUX_FILES = {'det': 'retinaface', 'rec': 'arcface'} |
|
|
| |
| |
| |
| ARCFACE_DST = np.array([[38.2946, 51.6963], [73.5318, 51.5014], [56.0252, 71.7366], |
| [41.5493, 92.3655], [70.7299, 92.2041]], dtype=np.float32) |
|
|
|
|
| def estimate_norm(lmk, image_size=112): |
| """Similarity transform that maps the 5 landmarks onto the arcface template.""" |
| assert lmk.shape == (5, 2) |
| assert image_size % 112 == 0 or image_size % 128 == 0 |
| if image_size % 112 == 0: |
| ratio, diff_x = float(image_size) / 112.0, 0.0 |
| else: |
| ratio, diff_x = float(image_size) / 128.0, 8.0 * (float(image_size) / 128.0) |
| dst = ARCFACE_DST * ratio |
| dst[:, 0] += diff_x |
| src = np.asarray(lmk, dtype=np.float32) |
| |
| src_mean, dst_mean = src.mean(axis=0), dst.mean(axis=0) |
| src_c, dst_c = src - src_mean, dst - dst_mean |
| A = dst_c.T @ src_c |
| U, S, Vt = np.linalg.svd(A) |
| R = U @ Vt |
| if np.linalg.det(R) < 0: |
| U[:, -1] *= -1 |
| R = U @ Vt |
| scale = S.sum() / (src_c ** 2).sum() |
| t = dst_mean - scale * R @ src_mean |
| return np.concatenate([scale * R, t[:, None]], axis=1).astype(np.float32) |
|
|
|
|
| def norm_crop(img, landmark, image_size=128): |
| """Crop + warp the face described by `landmark` to `image_size` x `image_size` (BGR).""" |
| M = estimate_norm(landmark, image_size) |
| return cv2.warpAffine(img, M, (image_size, image_size), borderValue=0.0), M |
|
|
|
|
| |
| |
| |
| def _blob(img, mean, std, size, swap_rb=True): |
| return cv2.dnn.blobFromImage(img, 1.0 / std, size, (mean, mean, mean), swapRB=swap_rb) |
|
|
|
|
| def distance2bbox(points, distance): |
| x1 = points[:, 0] - distance[:, 0] |
| y1 = points[:, 1] - distance[:, 1] |
| x2 = points[:, 0] + distance[:, 2] |
| y2 = points[:, 1] + distance[:, 3] |
| return np.stack([x1, y1, x2, y2], axis=-1) |
|
|
|
|
| def distance2kps(points, distance): |
| preds = [] |
| for i in range(0, distance.shape[1], 2): |
| preds.append(points[:, i % 2] + distance[:, i]) |
| preds.append(points[:, i % 2 + 1] + distance[:, i + 1]) |
| return np.stack(preds, axis=-1) |
|
|
|
|
| class RetinaFace: |
| """retinaface_10g detection (single input scale, NMS, 5 landmarks).""" |
|
|
| FEAT_STRIDES = [8, 16, 32] |
| FMC = 3 |
| NUM_ANCHORS = 2 |
|
|
| def __init__(self, compiled, det_size=(640, 640), det_thresh=0.5, nms_thresh=0.4): |
| self.compiled = compiled |
| self.det_size = tuple(det_size) |
| self.det_thresh = det_thresh |
| self.nms_thresh = nms_thresh |
| self.center_cache = {} |
|
|
| def _anchor_centers(self, height, width, stride): |
| key = (height, width, stride) |
| if key not in self.center_cache: |
| centers = np.stack(np.mgrid[:height, :width][::-1], axis=-1).astype(np.float32) |
| centers = (centers * stride).reshape(-1, 2) |
| if self.NUM_ANCHORS > 1: |
| centers = np.stack([centers] * self.NUM_ANCHORS, axis=1).reshape(-1, 2) |
| self.center_cache[key] = centers |
| return self.center_cache[key] |
|
|
| def detect(self, img): |
| """Return (bboxes[n,5] = x1,y1,x2,y2,score, kpss[n,5,2]).""" |
| in_w, in_h = self.det_size |
| im_ratio = img.shape[0] / img.shape[1] |
| model_ratio = in_h / in_w |
| if im_ratio > model_ratio: |
| new_h, new_w = in_h, int(in_h / im_ratio) |
| else: |
| new_w, new_h = in_w, int(in_w * im_ratio) |
| det_scale = new_h / img.shape[0] |
| resized = cv2.resize(img, (new_w, new_h)) |
| det_img = np.zeros((in_h, in_w, 3), dtype=np.uint8) |
| det_img[:new_h, :new_w, :] = resized |
| blob = _blob(det_img, 127.5, 128.0, (in_w, in_h)) |
| outs = self.compiled([blob]) |
|
|
| scores_all, bboxes_all, kpss_all = [], [], [] |
| for idx, stride in enumerate(self.FEAT_STRIDES): |
| scores = outs[idx] |
| bbox_preds = outs[idx + self.FMC] * stride |
| kps_preds = outs[idx + self.FMC * 2] * stride |
| height, width = blob.shape[2] // stride, blob.shape[3] // stride |
| centers = self._anchor_centers(height, width, stride) |
| pos = np.where(scores >= self.det_thresh)[0] |
| if pos.size == 0: |
| continue |
| scores_all.append(scores[pos]) |
| bboxes_all.append(distance2bbox(centers, bbox_preds)[pos]) |
| kps = distance2kps(centers, kps_preds).reshape(-1, 5, 2) |
| kpss_all.append(kps[pos]) |
| if not scores_all: |
| return np.empty((0, 5), np.float32), np.empty((0, 5, 2), np.float32) |
| scores = np.vstack(scores_all) |
| order = scores.ravel().argsort()[::-1] |
| bboxes = np.vstack(bboxes_all) / det_scale |
| kpss = np.vstack(kpss_all) / det_scale |
| pre_det = np.hstack((bboxes, scores)).astype(np.float32)[order, :] |
| kpss = kpss[order, :, :] |
| keep = self._nms(pre_det) |
| return pre_det[keep, :], kpss[keep, :, :] |
|
|
| def _nms(self, dets): |
| x1, y1, x2, y2, scores = dets[:, 0], dets[:, 1], dets[:, 2], dets[:, 3], dets[:, 4] |
| areas = (x2 - x1 + 1) * (y2 - y1 + 1) |
| order = scores.argsort()[::-1] |
| keep = [] |
| while order.size > 0: |
| i = order[0] |
| keep.append(i) |
| xx1 = np.maximum(x1[i], x1[order[1:]]) |
| yy1 = np.maximum(y1[i], y1[order[1:]]) |
| xx2 = np.minimum(x2[i], x2[order[1:]]) |
| yy2 = np.minimum(y2[i], y2[order[1:]]) |
| w = np.maximum(0.0, xx2 - xx1 + 1) |
| h = np.maximum(0.0, yy2 - yy1 + 1) |
| inter = w * h |
| ovr = inter / (areas[i] + areas[order[1:]] - inter) |
| order = order[np.where(ovr <= self.nms_thresh)[0] + 1] |
| return keep |
|
|
|
|
| class ArcFace: |
| """arcface_w600k_r50 embedding (512-d, L2-normalised).""" |
|
|
| def __init__(self, compiled): |
| self.compiled = compiled |
|
|
| def embedding(self, img_bgr, landmark): |
| aimg, _ = norm_crop(img_bgr, landmark, 112) |
| blob = _blob(aimg, 127.5, 127.5, (112, 112)) |
| feat = self.compiled([blob])[0][0] |
| return feat / np.linalg.norm(feat) |
|
|
|
|
| class INSwapper: |
| """inswapper_128: target crop (128x128 RGB) + source embedding -> swapped face.""" |
|
|
| def __init__(self, compiled, emap): |
| self.compiled = compiled |
| self.emap = emap |
|
|
| def project(self, embedding): |
| latent = np.dot(embedding.reshape(1, -1), self.emap) |
| return (latent / np.linalg.norm(latent)).astype(np.float32) |
|
|
| def swap_face(self, img_bgr, target_landmark, latent): |
| """Swap one face and paste it back; returns a new BGR image.""" |
| aimg, M = norm_crop(img_bgr, target_landmark, 128) |
| blob = _blob(aimg, 0.0, 255.0, (128, 128)) |
| pred = self.compiled([blob, latent])[0][0] |
| img_fake = pred.transpose(1, 2, 0) |
| bgr_fake = np.clip(255 * img_fake, 0, 255).astype(np.uint8)[:, :, ::-1] |
|
|
| fake_diff = bgr_fake.astype(np.float32) - aimg.astype(np.float32) |
| fake_diff = np.abs(fake_diff).mean(axis=2) |
| fake_diff[:2, :] = 0 |
| fake_diff[-2:, :] = 0 |
| fake_diff[:, :2] = 0 |
| fake_diff[:, -2:] = 0 |
| IM = cv2.invertAffineTransform(M) |
| size = (img_bgr.shape[1], img_bgr.shape[0]) |
| img_white = np.full(aimg.shape[:2], 255, dtype=np.float32) |
| bgr_fake = cv2.warpAffine(bgr_fake, IM, size, borderValue=0.0) |
| img_white = cv2.warpAffine(img_white, IM, size, borderValue=0.0) |
| fake_diff = cv2.warpAffine(fake_diff, IM, size, borderValue=0.0) |
| img_white[img_white > 20] = 255 |
| fake_diff[fake_diff < 10] = 0 |
| fake_diff[fake_diff >= 10] = 255 |
| mask_h_inds, mask_w_inds = np.where(img_white == 255) |
| mask_h = np.max(mask_h_inds) - np.min(mask_h_inds) |
| mask_w = np.max(mask_w_inds) - np.min(mask_w_inds) |
| mask_size = int(np.sqrt(mask_h * mask_w)) |
| k = max(mask_size // 10, 10) |
| img_mask = cv2.erode(img_white, np.ones((k, k), np.uint8), iterations=1) |
| fake_diff = cv2.dilate(fake_diff, np.ones((2, 2), np.uint8), iterations=1) |
| k = max(mask_size // 20, 5) |
| img_mask = cv2.GaussianBlur(img_mask, (2 * k + 1, 2 * k + 1), 0) |
| fake_diff = cv2.GaussianBlur(fake_diff, (11, 11), 0) |
| img_mask = np.reshape(img_mask / 255., [img_mask.shape[0], img_mask.shape[1], 1]) |
| merged = img_mask * bgr_fake + (1 - img_mask) * img_bgr.astype(np.float32) |
| return merged.astype(np.uint8) |
|
|
|
|
| |
| def _download(rel_path): |
| """Fetch a model file from the Hub into the local cache (xml + bin together).""" |
| from huggingface_hub import hf_hub_download |
| paths = [rel_path] |
| if rel_path.endswith('.xml'): |
| paths.append(rel_path[:-4] + '.bin') |
| CACHE_DIR.mkdir(parents=True, exist_ok=True) |
| for rel in paths: |
| dst = CACHE_DIR / pathlib.Path(rel).name |
| if dst.exists() and dst.stat().st_size > 0: |
| continue |
| print(f'[inswapper_ov] downloading {rel} from {HF_REPO} ...') |
| cached = hf_hub_download(repo_id=HF_REPO, filename=rel) |
| shutil.copy(cached, dst) |
| return str(CACHE_DIR / pathlib.Path(rel_path).name) |
|
|
|
|
| def model_name(kind, precision): |
| """kind in {'swap', 'det', 'rec'} -> file stem, e.g. 'inswapper_int4'. |
| |
| 'int4-mixed' only exists for the swapper; the detector / recogniser fall back to |
| their pure INT4 builds. |
| """ |
| if precision not in PRECISIONS: |
| raise ValueError(f'unknown precision "{precision}", choose from {list(PRECISIONS)}') |
| suffix = PRECISIONS[precision][0] |
| if kind == 'swap': |
| return f'inswapper_{suffix}' |
| return f'{AUX_FILES[kind]}_{"int4" if suffix == "int4_mixed" else suffix}' |
|
|
|
|
| def resolve_model(kind, precision, model_dir=None): |
| """-> local .xml path (from model_dir, the cache, or the Hugging Face Hub).""" |
| name = model_name(kind, precision) |
| if model_dir: |
| local = pathlib.Path(model_dir) / f'{name}.xml' |
| if local.exists() and local.with_suffix('.bin').exists(): |
| return str(local) |
| xml = CACHE_DIR / f'{name}.xml' |
| if not xml.exists() or not xml.with_suffix('.bin').exists(): |
| xml = _download(f'models/{name}.xml') |
| return str(xml) |
|
|
|
|
| def resolve_emap(model_dir=None): |
| if model_dir: |
| local = pathlib.Path(model_dir) / 'inswapper_emap.npy' |
| if local.exists(): |
| return np.load(str(local)) |
| npy = CACHE_DIR / 'inswapper_emap.npy' |
| if not npy.exists(): |
| npy = _download('models/inswapper_emap.npy') |
| return np.load(str(npy)) |
|
|
|
|
| class FaceSwapperOpenVINO: |
| """High level API: `swap(source_img, target_img, face_index=0, all_faces=False)`.""" |
|
|
| def __init__(self, precision='fp16', device='CPU', det_size=(640, 640), model_dir=None, |
| num_threads=None, precision_hint='f32'): |
| core = ov.Core() |
| props = {} |
| if num_threads: |
| props['INFERENCE_NUM_THREADS'] = num_threads |
| if precision_hint: |
| props['INFERENCE_PRECISION_HINT'] = precision_hint |
|
|
| def load(kind): |
| path = resolve_model(kind, precision, model_dir) |
| return core.compile_model(core.read_model(path), device, props) |
|
|
| self.detector = RetinaFace(load('det'), det_size=det_size) |
| self.recognizer = ArcFace(load('rec')) |
| self.swapper = INSwapper(load('swap'), resolve_emap(model_dir)) |
| self.compiled_swap = self.swapper.compiled |
| self.precision = precision |
| print(f'[inswapper_ov] precision={PRECISIONS[precision][0]} ({PRECISIONS[precision][1]}) device={device}') |
|
|
| |
| @staticmethod |
| def _read(path): |
| img = cv2.imread(str(path), cv2.IMREAD_COLOR) |
| if img is None: |
| raise FileNotFoundError(f'cannot read image: {path}') |
| return img |
|
|
| def detect(self, img): |
| return self.detector.detect(img) |
|
|
| def embedding(self, img, face_index=0): |
| _, kpss = self.detector.detect(img) |
| if len(kpss) == 0: |
| raise RuntimeError('no face detected') |
| kps = kpss[min(face_index, len(kpss) - 1)] |
| return self.recognizer.embedding(img, kps), kps |
|
|
| |
| def swap(self, source, target, out=None, face_index=0, all_faces=False, verbose=True): |
| """Swap the face of `source` into `target`. |
| |
| source/target can be a path or a BGR ndarray. Returns the BGR result. |
| """ |
| src = self._read(source) if isinstance(source, (str, pathlib.Path)) else source |
| tgt = self._read(target) if isinstance(target, (str, pathlib.Path)) else target |
|
|
| _, src_kpss = self.detector.detect(src) |
| if len(src_kpss) == 0: |
| raise RuntimeError('no face detected in the source image') |
| src_kps = src_kpss[0] |
| embedding = self.recognizer.embedding(src, src_kps) |
| latent = self.swapper.project(embedding) |
|
|
| bboxes, kpss = self.detector.detect(tgt) |
| if len(kpss) == 0: |
| raise RuntimeError('no face detected in the target image') |
| idx = range(len(kpss)) if all_faces else [min(face_index, len(kpss) - 1)] |
| result = tgt |
| for i in idx: |
| result = self.swapper.swap_face(result, kpss[i], latent) |
| if verbose: |
| print(f'[inswapper_ov] detected {len(kpss)} face(s), swapped {len(list(idx))}') |
| if out is not None: |
| cv2.imwrite(str(out), result) |
| return result |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description='InsightFace inswapper_128 face swap with OpenVINO', |
| formatter_class=argparse.RawDescriptionHelpFormatter) |
| parser.add_argument('-s', '--source', required=True, help='image providing the face') |
| parser.add_argument('-t', '--target', required=True, help='image to swap into') |
| parser.add_argument('-o', '--output', default='result.jpg') |
| parser.add_argument('-m', '--model', default='fp16', choices=list(PRECISIONS), |
| help='16-bit (fp16) / 8-bit (int8) / 4-bit (int4)') |
| parser.add_argument('-d', '--device', default='CPU', help='CPU / GPU / NPU / AUTO') |
| parser.add_argument('--all-faces', action='store_true', help='swap every detected face') |
| parser.add_argument('--face-index', type=int, default=0) |
| parser.add_argument('--det-size', type=int, nargs=2, default=[640, 640]) |
| parser.add_argument('--model-dir', default=None, help='folder with the local IR files') |
| parser.add_argument('--num-threads', type=int, default=0) |
| args = parser.parse_args() |
|
|
| t0 = time.time() |
| swapper = FaceSwapperOpenVINO(precision=args.model, device=args.device, det_size=tuple(args.det_size), |
| model_dir=args.model_dir, num_threads=args.num_threads or None) |
| swapper.swap(args.source, args.target, out=args.output, face_index=args.face_index, |
| all_faces=args.all_faces) |
| print(f'[inswapper_ov] wrote {args.output} in {time.time() - t0:.2f}s') |
| return 0 |
|
|
|
|
| if __name__ == '__main__': |
| sys.exit(main()) |
