Update expression_editor.py

兼容cuda\mps\cpu平台

README.md

@@ -7,6 +7,31 @@
 > [推荐：mixlab-nodes](https://github.com/shadowcz007/comfyui-mixlab-nodes)
 
 
+### 更新
+
+expression_editor：
+
+![alt text](./example/1722665711975.png)
+
+[示例工作流](./example/expression_workflow.json)
+
+表情代码：修改自[ComfyUI-AdvancedLivePortrait](https://github.com/PowerHouseMan/ComfyUI-AdvancedLivePortrait)
+
+face crop 模型参考[comfyui-ultralytics-yolo](https://github.com/shadowcz007/comfyui-ultralytics-yolo)
+
+下载  [face_yolov8m.pt 或者 face_yolov8n.pt](https://github.com/ultralytics/assets/releases/) 到 ```models/ultralytics/bbox/```
+
+ExpressionVideoNode:
+
+可以创建2个表情之间的插值动画了！
+![alt text](./example/8dec8f644e71dc376a320e92c6563dc.png)
+
+
+### 教程
+[![image](https://github.com/user-attachments/assets/ab5d45c3-aeae-458f-bc8b-b017e7d1138c)](https://www.bilibili.com/video/BV1JW421R7sP/?buvid=ZE4865E83C2A9F8547C08310ED8406E72D1B&is_story_h5=false&mid=hSf%2B8X%2BJL2Hq%2F3zyc4No3A%3D%3D&p=1&plat_id=116&share_from=ugc&share_medium=iphone&share_plat=ios&share_session_id=B4772702-1A00-4E2D-8993-4725A2F52BB1&share_source=WEIXIN&share_tag=s_i&spmid=united.player-video-detail.0.0&timestamp=1720927658&unique_k=LAUWKu1&up_id=43149384&vd_source=6b8c7c3af882b1b8460fa6fa0ce1c69d)
+
+
+
 
 ### workflow
 

__init__.py

@@ -1,11 +1,14 @@
 from .nodes.live_portrait import LivePortraitNode,FaceCropInfo,Retargeting,LivePortraitVideoNode
-
+from .nodes.expression_editor import ExpressionEditor,ExpressionVideoNode,ExpressionVideo2VideoNode
 
 NODE_CLASS_MAPPINGS = {
     "LivePortraitNode": LivePortraitNode, 
     "LivePortraitVideoNode":LivePortraitVideoNode,
     "FaceCropInfo":FaceCropInfo,
-    "Retargeting":Retargeting
+    "Retargeting":Retargeting,
+    "ExpressionEditor_":ExpressionEditor,
+    "ExpressionVideoNode":ExpressionVideoNode,
+    "ExpressionVideo2VideoNode":ExpressionVideo2VideoNode
 }
 
 # dict = { "key":value }
@@ -14,7 +17,10 @@ NODE_DISPLAY_NAME_MAPPINGS = {
    "LivePortraitNode":"Live Portrait",
    "LivePortraitVideoNode":"Live Portrait for Video",
    "FaceCropInfo":"Face Crop Info",
-   "Retargeting":"Retargeting"
+   "Retargeting":"Retargeting",
+   "ExpressionEditor_":"Expression Editor",
+   "ExpressionVideoNode":"Expression Video",
+   "ExpressionVideo2VideoNode":"Expression Video 2 Video"
 }
 
 # web ui的节点功能

nodes/LivePortrait/deviceutils.py

@@ -0,0 +1,18 @@
+import torch
+
+device_name = (
+    "cuda"
+    if torch.cuda.is_available()
+    else "mps"
+    if torch.backends.mps.is_available()
+    else "cpu"
+)
+print(f'@@device:{device_name}')
+
+device_providers = (
+    ["CUDAExecutionProvider", "CPUExecutionProvider"]
+    if torch.cuda.is_available()
+    else ["MPSExecutionProvider", "CPUExecutionProvider"]
+    if torch.backends.mps.is_available()
+    else ["CPUExecutionProvider"]
+)

nodes/LivePortrait/speed.py

@@ -12,19 +12,45 @@ import time
 import numpy as np
 from LivePortrait.src.utils.helper import load_model, concat_feat
 from LivePortrait.src.config.inference_config import InferenceConfig
+import deviceutils
 
 
 def initialize_inputs(batch_size=1):
     """
     Generate random input tensors and move them to GPU
     """
-    feature_3d = torch.randn(batch_size, 32, 16, 64, 64).cuda().half()
-    kp_source = torch.randn(batch_size, 21, 3).cuda().half()
-    kp_driving = torch.randn(batch_size, 21, 3).cuda().half()
-    source_image = torch.randn(batch_size, 3, 256, 256).cuda().half()
-    generator_input = torch.randn(batch_size, 256, 64, 64).cuda().half()
-    eye_close_ratio = torch.randn(batch_size, 3).cuda().half()
-    lip_close_ratio = torch.randn(batch_size, 2).cuda().half()
+
+    if deviceutils.device_name != "cuda":
+        feature_3d = torch.randn(batch_size, 32, 16, 64, 64).cuda().half()
+        kp_source = torch.randn(batch_size, 21, 3).cuda().half()
+        kp_driving = torch.randn(batch_size, 21, 3).cuda().half()
+        source_image = torch.randn(batch_size, 3, 256, 256).cuda().half()
+        generator_input = torch.randn(batch_size, 256, 64, 64).cuda().half()
+        eye_close_ratio = torch.randn(batch_size, 3).cuda().half()
+        lip_close_ratio = torch.randn(batch_size, 2).cuda().half()
+    elif deviceutils.device_name == "mps":
+        feature_3d = torch.randn(batch_size, 32, 16, 64, 64).half()
+        feature_3d = feature_3d.to(deviceutils.device_name)
+        kp_source = torch.randn(batch_size, 21, 3).half()
+        kp_source = kp_source.to(deviceutils.device_name)
+        kp_driving = torch.randn(batch_size, 21, 3).half()
+        kp_driving = kp_driving.to(deviceutils.device_name)
+        source_image = torch.randn(batch_size, 3, 256, 256).half()
+        source_image = source_image.to(deviceutils.device_name)
+        generator_input = torch.randn(batch_size, 256, 64, 64).half()
+        generator_input = generator_input.to(deviceutils.device_name)
+        eye_close_ratio = torch.randn(batch_size, 3).half()
+        eye_close_ratio = eye_close_ratio.to(deviceutils.device_name)
+        lip_close_ratio = torch.randn(batch_size, 2).half()
+        lip_close_ratio = lip_close_ratio.to(deviceutils.device_name)
+    else:
+        feature_3d = torch.randn(batch_size, 32, 16, 64, 64).half()
+        kp_source = torch.randn(batch_size, 21, 3).half()
+        kp_driving = torch.randn(batch_size, 21, 3).half()
+        source_image = torch.randn(batch_size, 3, 256, 256).half()
+        generator_input = torch.randn(batch_size, 256, 64, 64).half()
+        eye_close_ratio = torch.randn(batch_size, 3).half()
+        lip_close_ratio = torch.randn(batch_size, 2).half()
     feat_stitching = concat_feat(kp_source, kp_driving).half()
     feat_eye = concat_feat(kp_source, eye_close_ratio).half()
     feat_lip = concat_feat(kp_source, lip_close_ratio).half()
@@ -105,34 +131,65 @@ def measure_inference_times(compiled_models, stitching_retargeting_module, input
 
     with torch.no_grad():
         for _ in range(100):
-            torch.cuda.synchronize()
+            if deviceutils.device_name == "cuda":
+                torch.cuda.synchronize()
+            elif deviceutils.device_name == "mps":
+                torch.mps.synchronize()
+            else:
+                torch.cpu.synchronize()
+
             overall_start = time.time()
 
             start = time.time()
             compiled_models['Appearance Feature Extractor'](inputs['source_image'])
-            torch.cuda.synchronize()
+            if deviceutils.device_name == "cuda":
+                torch.cuda.synchronize()
+            elif deviceutils.device_name == "mps":
+                torch.mps.synchronize()
+            else:
+                torch.cpu.synchronize()
             times['Appearance Feature Extractor'].append(time.time() - start)
 
             start = time.time()
             compiled_models['Motion Extractor'](inputs['source_image'])
-            torch.cuda.synchronize()
+            if deviceutils.device_name == "cuda":
+                torch.cuda.synchronize()
+            elif deviceutils.device_name == "mps":
+                torch.mps.synchronize()
+            else:
+                torch.cpu.synchronize()
             times['Motion Extractor'].append(time.time() - start)
 
             start = time.time()
             compiled_models['Warping Network'](inputs['feature_3d'], inputs['kp_driving'], inputs['kp_source'])
-            torch.cuda.synchronize()
+            if deviceutils.device_name == "cuda":
+                torch.cuda.synchronize()
+            elif deviceutils.device_name == "mps":
+                torch.mps.synchronize()
+            else:
+                torch.cpu.synchronize()
             times['Warping Network'].append(time.time() - start)
 
             start = time.time()
             compiled_models['SPADE Decoder'](inputs['generator_input'])  # Adjust input as required
-            torch.cuda.synchronize()
+            if deviceutils.device_name == "cuda":
+                torch.cuda.synchronize()
+            elif deviceutils.device_name == "mps":
+                torch.mps.synchronize()
+            else:
+                torch.cpu.synchronize()
             times['SPADE Decoder'].append(time.time() - start)
 
             start = time.time()
             stitching_retargeting_module['stitching'](inputs['feat_stitching'])
             stitching_retargeting_module['eye'](inputs['feat_eye'])
             stitching_retargeting_module['lip'](inputs['feat_lip'])
-            torch.cuda.synchronize()
+            if deviceutils.device_name == "cuda":
+                torch.cuda.synchronize()
+            elif deviceutils.device_name == "mps":
+                torch.mps.synchronize()
+            else:
+                torch.cpu.synchronize()
             times['Retargeting Models'].append(time.time() - start)
 
             overall_times.append(time.time() - overall_start)

nodes/LivePortrait/src/live_portrait_wrapper.py

@@ -18,6 +18,8 @@ from .utils.retargeting_utils import calc_eye_close_ratio, calc_lip_close_ratio
 from LivePortrait.src.config.inference_config import InferenceConfig
 from LivePortrait.src.utils.rprint import rlog as log
 
+from .. import deviceutils
+
 
 class LivePortraitWrapper(object):
 
@@ -71,7 +73,12 @@ class LivePortraitWrapper(object):
             raise ValueError(f'img ndim should be 3 or 4: {x.ndim}')
         x = np.clip(x, 0, 1)  # clip to 0~1
         x = torch.from_numpy(x).permute(0, 3, 1, 2)  # 1xHxWx3 -> 1x3xHxW
-        x = x.cuda(self.device_id)
+        if deviceutils.device_name == "cuda":
+            x = x.cuda(self.device_id)
+        elif deviceutils.device_name == "mps":
+            x = x.to('mps')
+        else:
+            x = x.to('cpu')
         return x
 
     def prepare_driving_videos(self, imgs) -> torch.Tensor:
@@ -88,7 +95,14 @@ class LivePortraitWrapper(object):
         y = _imgs.astype(np.float32) / 255.
         y = np.clip(y, 0, 1)  # clip to 0~1
         y = torch.from_numpy(y).permute(0, 4, 3, 1, 2)  # TxHxWx3x1 -> Tx1x3xHxW
-        y = y.cuda(self.device_id)
+
+        if deviceutils.device_name == "cuda":
+            y = y.cuda(self.device_id)
+        elif deviceutils.device_name == "mps":
+            y = y.to('mps')
+        else:
+            y = y.to('cpu')
+
 
         return y
 
@@ -97,7 +111,7 @@ class LivePortraitWrapper(object):
         x: Bx3xHxW, normalized to 0~1
         """
         with torch.no_grad():
-            with torch.autocast(device_type='cuda', dtype=torch.float16, enabled=self.cfg.flag_use_half_precision):
+            with torch.autocast(device_type="cuda" if deviceutils.device_name == "cuda" else "cpu", dtype=torch.float16, enabled=self.cfg.flag_use_half_precision):
                 feature_3d = self.appearance_feature_extractor(x)
 
         return feature_3d.float()
@@ -108,8 +122,8 @@ class LivePortraitWrapper(object):
         flag_refine_info: whether to trandform the pose to degrees and the dimention of the reshape
         return: A dict contains keys: 'pitch', 'yaw', 'roll', 't', 'exp', 'scale', 'kp'
         """
-        with torch.no_grad():
-            with torch.autocast(device_type='cuda', dtype=torch.float16, enabled=self.cfg.flag_use_half_precision):
+        with torch.no_grad():#mps不支持autocast
+            with torch.autocast(device_type="cuda" if deviceutils.device_name == "cuda" else "cpu", dtype=torch.float16, enabled=self.cfg.flag_use_half_precision):
                 kp_info = self.motion_extractor(x)
 
             if self.cfg.flag_use_half_precision:
@@ -282,7 +296,7 @@ class LivePortraitWrapper(object):
         """
         # The line 18 in Algorithm 1: D(W(f_s; x_s, x′_d,i)）
         with torch.no_grad():
-            with torch.autocast(device_type='cuda', dtype=torch.float16, enabled=self.cfg.flag_use_half_precision):
+            with torch.autocast(device_type="cuda" if deviceutils.device_name == "cuda" else "cpu", dtype=torch.float16, enabled=self.cfg.flag_use_half_precision):
                 # get decoder input
                 ret_dct = self.warping_module(feature_3d, kp_source=kp_source, kp_driving=kp_driving)
                 # decode
@@ -318,17 +332,31 @@ class LivePortraitWrapper(object):
 
     def calc_combined_eye_ratio(self, input_eye_ratio, source_lmk):
         eye_close_ratio = calc_eye_close_ratio(source_lmk[None])
-        eye_close_ratio_tensor = torch.from_numpy(eye_close_ratio).float().cuda(self.device_id)
-        input_eye_ratio_tensor = torch.Tensor([input_eye_ratio[0][0]]).reshape(1, 1).cuda(self.device_id)
+        eye_close_ratio_tensor = torch.from_numpy(eye_close_ratio).float()
+        input_eye_ratio_tensor = torch.Tensor([input_eye_ratio[0][0]]).reshape(1, 1)
+        if deviceutils.device_name == "cuda":
+            eye_close_ratio_tensor = eye_close_ratio_tensor.to(f'cuda:{self.device_id}')
+            input_eye_ratio_tensor = input_eye_ratio_tensor.to(f'cuda:{self.device_id}')
+        else:
+            eye_close_ratio_tensor = eye_close_ratio_tensor.to(deviceutils.device_name)
+            input_eye_ratio_tensor = input_eye_ratio_tensor.to(deviceutils.device_name)
+
+
         # [c_s,eyes, c_d,eyes,i]
         combined_eye_ratio_tensor = torch.cat([eye_close_ratio_tensor, input_eye_ratio_tensor], dim=1)
         return combined_eye_ratio_tensor
 
     def calc_combined_lip_ratio(self, input_lip_ratio, source_lmk):
         lip_close_ratio = calc_lip_close_ratio(source_lmk[None])
-        lip_close_ratio_tensor = torch.from_numpy(lip_close_ratio).float().cuda(self.device_id)
+        lip_close_ratio_tensor = torch.from_numpy(lip_close_ratio).float()
         # [c_s,lip, c_d,lip,i]
-        input_lip_ratio_tensor = torch.Tensor([input_lip_ratio[0]]).cuda(self.device_id)
+        input_lip_ratio_tensor = torch.Tensor([input_lip_ratio[0]])
+        if deviceutils.device_name == "cuda":
+            lip_close_ratio_tensor = lip_close_ratio_tensor.to(f'cuda:{self.device_id}')
+            input_lip_ratio_tensor = input_lip_ratio_tensor.to(f'cuda:{self.device_id}')
+        else:
+            lip_close_ratio_tensor = lip_close_ratio_tensor.to(deviceutils.device_name)
+            input_lip_ratio_tensor = input_lip_ratio_tensor.to(deviceutils.device_name)
         if input_lip_ratio_tensor.shape != [1, 1]:
             input_lip_ratio_tensor = input_lip_ratio_tensor.reshape(1, 1)
         combined_lip_ratio_tensor = torch.cat([lip_close_ratio_tensor, input_lip_ratio_tensor], dim=1)

nodes/LivePortrait/src/modules/dense_motion.py

@@ -8,6 +8,7 @@ from torch import nn
 import torch.nn.functional as F
 import torch
 from .util import Hourglass, make_coordinate_grid, kp2gaussian
+from ... import deviceutils
 
 
 class DenseMotionNetwork(nn.Module):
@@ -59,7 +60,9 @@ class DenseMotionNetwork(nn.Module):
         heatmap = gaussian_driving - gaussian_source  # (bs, num_kp, d, h, w)
 
         # adding background feature
-        zeros = torch.zeros(heatmap.shape[0], 1, spatial_size[0], spatial_size[1], spatial_size[2]).type(heatmap.type()).to(heatmap.device)
+        print(f'heatmap.type():{heatmap.type()},heatmap.device:{heatmap.device}')
+        ty = heatmap.type() if deviceutils.device_name == 'cuda' else 'torch.FloatTensor'
+        zeros = torch.zeros(heatmap.shape[0], 1, spatial_size[0], spatial_size[1], spatial_size[2]).type(ty).to(heatmap.device)
         heatmap = torch.cat([zeros, heatmap], dim=1)
         heatmap = heatmap.unsqueeze(2)         # (bs, 1+num_kp, 1, d, h, w)
         return heatmap

nodes/LivePortrait/src/utils/cropper.py

@@ -5,7 +5,11 @@ from PIL import Image
 import os.path as osp
 from typing import List, Union, Tuple
 from dataclasses import dataclass, field
-import cv2; cv2.setNumThreads(0); cv2.ocl.setUseOpenCL(False)
+import cv2;
+
+from ... import deviceutils
+
+cv2.setNumThreads(0); cv2.ocl.setUseOpenCL(False)
 
 from .landmark_runner import LandmarkRunner
 from .face_analysis_diy import FaceAnalysisDIY
@@ -30,7 +34,6 @@ class Trajectory:
     frame_rgb_lst: Union[Tuple, List, np.ndarray] = field(default_factory=list)  # frame list
     frame_rgb_crop_lst: Union[Tuple, List, np.ndarray] = field(default_factory=list)  # frame crop list
 
-
 class Cropper(object):
     def __init__(self, **kwargs) -> None:
         device_id = kwargs.get('device_id', 0)
@@ -41,7 +44,7 @@ class Cropper(object):
         self.landmark_runner = LandmarkRunner(
             # ckpt_path=make_abs_path('../../pretrained_weights/liveportrait/landmark.onnx'),
             ckpt_path=landmark_runner_ckpt,
-            onnx_provider='cuda',
+            onnx_provider=deviceutils.device_name,
             device_id=device_id
         )
         self.landmark_runner.warmup()
@@ -50,7 +53,7 @@ class Cropper(object):
             name='buffalo_l',
             # root=make_abs_path('../../pretrained_weights/insightface'),
             root=insightface_pretrained_weights,
-            providers=["CUDAExecutionProvider"]
+            providers=deviceutils.device_providers
         )
         self.face_analysis_wrapper.prepare(ctx_id=device_id, det_size=(512, 512))
         self.face_analysis_wrapper.warmup()

nodes/LivePortrait/src/utils/helper.py

@@ -8,7 +8,7 @@ import os
 import os.path as osp
 import cv2
 import torch
-from rich.console import Console
+# from rich.console import Console
 from collections import OrderedDict
 
 from LivePortrait.src.modules.spade_generator import SPADEDecoder
@@ -17,6 +17,7 @@ from LivePortrait.src.modules.motion_extractor import MotionExtractor
 from LivePortrait.src.modules.appearance_feature_extractor import AppearanceFeatureExtractor
 from LivePortrait.src.modules.stitching_retargeting_network import StitchingRetargetingNetwork
 from .rprint import rlog as log
+from ... import deviceutils
 
 
 def suffix(filename):
@@ -67,7 +68,15 @@ def squeeze_tensor_to_numpy(tensor):
 
 def dct2cuda(dct: dict, device_id: int):
     for key in dct:
-        dct[key] = torch.tensor(dct[key]).cuda(device_id)
+        tensor = torch.tensor(dct[key])
+        if deviceutils.device_name == "cuda":
+            tensor = tensor.to(f'cuda:{device_id}')
+        elif deviceutils.device_name == "mps":
+            tensor = tensor.to('mps')
+        else:
+            tensor = tensor.to('cpu')
+        dct[key] = tensor
+        # dct[key] = torch.tensor(dct[key]).cuda(device_id)
     return dct
 
 
@@ -96,13 +105,13 @@ def load_model(ckpt_path, model_config, device, model_type):
     model_params = model_config['model_params'][f'{model_type}_params']
 
     if model_type == 'appearance_feature_extractor':
-        model = AppearanceFeatureExtractor(**model_params).cuda(device)
+        model = AppearanceFeatureExtractor(**model_params)
     elif model_type == 'motion_extractor':
-        model = MotionExtractor(**model_params).cuda(device)
+        model = MotionExtractor(**model_params)
     elif model_type == 'warping_module':
-        model = WarpingNetwork(**model_params).cuda(device)
+        model = WarpingNetwork(**model_params)
     elif model_type == 'spade_generator':
-        model = SPADEDecoder(**model_params).cuda(device)
+        model = SPADEDecoder(**model_params)
     elif model_type == 'stitching_retargeting_module':
         # Special handling for stitching and retargeting module
         config = model_config['model_params']['stitching_retargeting_module_params']
@@ -110,17 +119,26 @@ def load_model(ckpt_path, model_config, device, model_type):
 
         stitcher = StitchingRetargetingNetwork(**config.get('stitching'))
         stitcher.load_state_dict(remove_ddp_dumplicate_key(checkpoint['retarget_shoulder']))
-        stitcher = stitcher.cuda(device)
+        if deviceutils.device_name == "cuda":
+            stitcher = stitcher.cuda(device)
+        else:
+            stitcher = stitcher.to(deviceutils.device_name)
         stitcher.eval()
 
         retargetor_lip = StitchingRetargetingNetwork(**config.get('lip'))
         retargetor_lip.load_state_dict(remove_ddp_dumplicate_key(checkpoint['retarget_mouth']))
-        retargetor_lip = retargetor_lip.cuda(device)
+        if deviceutils.device_name == "cuda":
+            retargetor_lip = retargetor_lip.cuda(device)
+        else:
+            retargetor_lip = retargetor_lip.to(deviceutils.device_name)
         retargetor_lip.eval()
 
         retargetor_eye = StitchingRetargetingNetwork(**config.get('eye'))
         retargetor_eye.load_state_dict(remove_ddp_dumplicate_key(checkpoint['retarget_eye']))
-        retargetor_eye = retargetor_eye.cuda(device)
+        if deviceutils.device_name == "cuda":
+            retargetor_eye = retargetor_eye.cuda(device)
+        else:
+            retargetor_eye = retargetor_eye.to(deviceutils.device_name)
         retargetor_eye.eval()
 
         return {
@@ -131,6 +149,11 @@ def load_model(ckpt_path, model_config, device, model_type):
     else:
         raise ValueError(f"Unknown model type: {model_type}")
 
+    if deviceutils.device_name == "cuda":
+        model = model.to(f'cuda:{device}')
+    else:
+        model = model.to(deviceutils.device_name)
+
     model.load_state_dict(torch.load(ckpt_path, map_location=lambda storage, loc: storage))
     model.eval()
     return model

nodes/LivePortrait/src/utils/landmark_runner.py

@@ -1,7 +1,11 @@
 # coding: utf-8
 
 import os.path as osp
-import cv2; cv2.setNumThreads(0); cv2.ocl.setUseOpenCL(False)
+import cv2;
+
+from ... import deviceutils
+
+cv2.setNumThreads(0); cv2.ocl.setUseOpenCL(False)
 import torch
 import numpy as np
 import onnxruntime
@@ -32,7 +36,7 @@ class LandmarkRunner(object):
         self.dsize = kwargs.get('dsize', 224)
         self.timer = Timer()
         # print('---------------------------------#onnx_provider',ckpt_path)
-        if onnx_provider.lower() == 'cuda':
+        if deviceutils.device_name == 'cuda':
             self.session = onnxruntime.InferenceSession(
                 ckpt_path, providers=[
                     ('CUDAExecutionProvider', {'device_id': device_id})

nodes/LivePortrait/src/utils/retargeting_utils.py

@@ -6,6 +6,8 @@ Functions to compute distance ratios between specific pairs of facial landmarks
 import numpy as np
 import torch
 
+from LivePortrait import deviceutils
+
 
 def calculate_distance_ratio(lmk: np.ndarray, idx1: int, idx2: int, idx3: int, idx4: int, eps: float = 1e-6) -> np.ndarray:
     """
@@ -58,8 +60,15 @@ def calc_lip_close_ratio(lmk: np.ndarray) -> np.ndarray:
 def compute_eye_delta(frame_idx, input_eye_ratios, source_landmarks, portrait_wrapper, kp_source):
     input_eye_ratio = input_eye_ratios[frame_idx][0][0]
     eye_close_ratio = calc_eye_close_ratio(source_landmarks[None])
-    eye_close_ratio_tensor = torch.from_numpy(eye_close_ratio).float().cuda(portrait_wrapper.device_id)
-    input_eye_ratio_tensor = torch.Tensor([input_eye_ratio]).reshape(1, 1).cuda(portrait_wrapper.device_id)
+    eye_close_ratio_tensor = torch.from_numpy(eye_close_ratio).float()
+    input_eye_ratio_tensor = torch.Tensor([input_eye_ratio]).reshape(1, 1)
+    if deviceutils.device_name == "cuda":
+        eye_close_ratio_tensor = eye_close_ratio_tensor.to(f'cuda:{portrait_wrapper.device_id}')
+        input_eye_ratio_tensor = input_eye_ratio_tensor.to(f'cuda:{portrait_wrapper.device_id}')
+    else:
+        eye_close_ratio_tensor = eye_close_ratio_tensor.to(deviceutils.device_name)
+        input_eye_ratio_tensor = input_eye_ratio_tensor.to(deviceutils.device_name)
+
     combined_eye_ratio_tensor = torch.cat([eye_close_ratio_tensor, input_eye_ratio_tensor], dim=1)
     # print(combined_eye_ratio_tensor.mean())
     eye_delta = portrait_wrapper.retarget_eye(kp_source, combined_eye_ratio_tensor)
@@ -69,8 +78,14 @@ def compute_eye_delta(frame_idx, input_eye_ratios, source_landmarks, portrait_wr
 def compute_lip_delta(frame_idx, input_lip_ratios, source_landmarks, portrait_wrapper, kp_source):
     input_lip_ratio = input_lip_ratios[frame_idx][0]
     lip_close_ratio = calc_lip_close_ratio(source_landmarks[None])
-    lip_close_ratio_tensor = torch.from_numpy(lip_close_ratio).float().cuda(portrait_wrapper.device_id)
-    input_lip_ratio_tensor = torch.Tensor([input_lip_ratio]).cuda(portrait_wrapper.device_id)
+    lip_close_ratio_tensor = torch.from_numpy(lip_close_ratio).float()
+    input_lip_ratio_tensor = torch.Tensor([input_lip_ratio])
+    if deviceutils.device_name == "cuda":
+        lip_close_ratio_tensor = lip_close_ratio_tensor.to(f'cuda:{portrait_wrapper.device_id}')
+        input_lip_ratio_tensor = input_lip_ratio_tensor.to(f'cuda:{portrait_wrapper.device_id}')
+    else:
+        lip_close_ratio_tensor = lip_close_ratio_tensor.to(deviceutils.device_name)
+        input_lip_ratio_tensor = input_lip_ratio_tensor.to(deviceutils.device_name)
     combined_lip_ratio_tensor = torch.cat([lip_close_ratio_tensor, input_lip_ratio_tensor], dim=1)
     lip_delta = portrait_wrapper.retarget_lip(kp_source, combined_lip_ratio_tensor)
     return lip_delta

nodes/expression_editor.py

@@ -0,0 +1,719 @@
+import os
+import sys
+import numpy as np
+import torch
+import cv2
+from PIL import Image
+import folder_paths
+import copy,json
+from ultralytics import YOLO
+from scipy.interpolate import CubicSpline
+
+
+current_file_path = os.path.abspath(__file__)
+current_directory = os.path.dirname(current_file_path)
+sys.path.append(current_directory)
+from LivePortrait.src.live_portrait_wrapper import LivePortraitWrapper
+from LivePortrait.src.utils.camera import get_rotation_matrix
+
+def tensor2pil(image):
+    return Image.fromarray(np.clip(255. * image.cpu().numpy().squeeze(), 0, 255).astype(np.uint8))
+def pil2tensor(image):
+    return torch.from_numpy(np.array(image).astype(np.float32) / 255.0).unsqueeze(0)
+def rgb_crop(rgb, region):
+    return rgb[region[1]:region[3], region[0]:region[2]]
+
+def get_rgb_size(rgb):
+    return rgb.shape[1], rgb.shape[0]
+def create_transform_matrix(x, y, scale=1):
+    return np.float32([[scale, 0, x], [0, scale, y]])
+
+def get_model_dir(m):
+    try:
+        return folder_paths.get_folder_paths(m)[0]
+    except:
+        return os.path.join(folder_paths.models_dir, m)
+
+def calc_crop_limit(center, img_size, crop_size):
+    pos = center - crop_size / 2
+    if pos < 0:
+        crop_size += pos * 2
+        pos = 0
+
+    pos2 = pos + crop_size
+
+    if img_size < pos2:
+        crop_size -= (pos2 - img_size) * 2
+        pos2 = img_size
+        pos = pos2 - crop_size
+
+    return pos, pos2, crop_size
+
+ 
+# 2个字典之间的插值
+def interpolate_dicts(from_dict, to_dict, interpolations_num, interpolation_type='linear'):
+    def linear_interpolate(val1, val2, alpha):
+        return val1 + (val2 - val1) * alpha
+    
+    def nearest_neighbor_interpolate(val1, val2, alpha):
+        return val1 if alpha < 0.5 else val2
+    
+    keys = list(from_dict.keys())  # Convert dict_keys to a list
+    from_values = np.array([from_dict[key] for key in keys])
+    to_values = np.array([to_dict[key] for key in keys])
+    interpolated_dicts = []
+    
+    if interpolation_type == 'cubic':
+        cs = CubicSpline([0, 1], np.vstack([from_values, to_values]), axis=0)
+    
+    for i in range(interpolations_num):
+        alpha = i / (interpolations_num - 1)
+        interpolated_dict = {}
+        
+        for key in keys:
+            if interpolation_type == 'linear':
+                interpolated_dict[key] = linear_interpolate(from_dict[key], to_dict[key], alpha)
+            elif interpolation_type == 'nearest':
+                interpolated_dict[key] = nearest_neighbor_interpolate(from_dict[key], to_dict[key], alpha)
+            elif interpolation_type == 'cubic':
+                interpolated_dict[key] = cs(alpha)[keys.index(key)]
+            # 添加更多插值方法的条件
+        
+        interpolated_dicts.append(interpolated_dict)
+    
+    return interpolated_dicts
+
+def update_expression_json(new_dict):
+    default_keys = {
+        "rotate_pitch": 0,
+        "rotate_yaw": 0,
+        "rotate_roll": 0,
+        "blink": 0,
+        "eyebrow": 0,
+        "wink": 0,
+        "pupil_x": 0,
+        "pupil_y": 0,
+        "aaa": 0,
+        "eee": 0,
+        "woo": 0,
+        "smile": 0,
+        "src_weight": 0
+    }
+
+    for key in new_dict:
+        if key in default_keys:
+            default_keys[key] = new_dict[key]
+
+    return default_keys
+
+
+# 修改模型路径 ， 沿用 comfyui-liveportrait
+class InferenceConfig:
+    def __init__(self,
+                    models_config,
+                    checkpoint_F,
+                    checkpoint_M,
+                    checkpoint_G,
+                    checkpoint_W,
+                    checkpoint_S,
+                    mask_crop = None,
+                    flag_use_half_precision=True,
+                    flag_lip_zero=True,
+                    lip_zero_threshold=0.03,
+                    flag_eye_retargeting=False,
+                    flag_lip_retargeting=False,
+                    flag_stitching=True,
+                    flag_relative=True,
+                    anchor_frame=0,
+                    input_shape=(256, 256),
+                    output_format='mp4',
+                    output_fps=30,
+                    crf=15,
+                    flag_write_result=True,
+                    flag_pasteback=True,
+                    flag_write_gif=False,
+                    size_gif=256,
+                    ref_max_shape=1280,
+                    ref_shape_n=2,
+                    device_id=0,
+                    flag_do_crop=True,
+                    flag_do_rot=True):
+        self.models_config = models_config
+        self.checkpoint_F = checkpoint_F
+        self.checkpoint_M = checkpoint_M
+        self.checkpoint_G = checkpoint_G
+        self.checkpoint_W = checkpoint_W
+        self.checkpoint_S = checkpoint_S
+        self.flag_use_half_precision = flag_use_half_precision
+        self.flag_lip_zero = flag_lip_zero
+        self.lip_zero_threshold = lip_zero_threshold
+        self.flag_eye_retargeting = flag_eye_retargeting
+        self.flag_lip_retargeting = flag_lip_retargeting
+        self.flag_stitching = flag_stitching
+        self.flag_relative = flag_relative
+        self.anchor_frame = anchor_frame
+        self.input_shape = input_shape
+        self.output_format = output_format
+        self.output_fps = output_fps
+        self.crf = crf
+        self.flag_write_result = flag_write_result
+        self.flag_pasteback = flag_pasteback
+        self.flag_write_gif = flag_write_gif
+        self.size_gif = size_gif
+        self.ref_max_shape = ref_max_shape
+        self.ref_shape_n = ref_shape_n
+        self.device_id = device_id
+        self.flag_do_crop = flag_do_crop
+        self.flag_do_rot = flag_do_rot
+        self.mask_crop=mask_crop
+
+liveportrait_model=get_model_dir('liveportrait')
+
+inference_cfg = InferenceConfig(
+        models_config=os.path.join(current_directory,'LivePortrait','src','config','models.yaml'),
+        checkpoint_F=os.path.join(liveportrait_model,'base_models','appearance_feature_extractor.pth'),
+        checkpoint_M=os.path.join(liveportrait_model,'base_models','motion_extractor.pth') ,
+        checkpoint_G=os.path.join(liveportrait_model,'base_models','spade_generator.pth') ,
+        checkpoint_W=os.path.join(liveportrait_model,'base_models','warping_module.pth'),
+        checkpoint_S=os.path.join(liveportrait_model,'retargeting_models','stitching_retargeting_module.pth')
+    )
+
+class PreparedSrcImg:
+    def __init__(self, src_rgb, crop_trans_m, x_s_info, f_s_user, x_s_user, mask_ori):
+        self.src_rgb = src_rgb
+        self.crop_trans_m = crop_trans_m
+        self.x_s_info = x_s_info
+        self.f_s_user = f_s_user
+        self.x_s_user = x_s_user
+        self.mask_ori = mask_ori
+
+class LP_Engine:
+    pipeline = None
+    bbox_model = None
+    mask_img = None
+
+    def detect_face(self, image_rgb):
+
+        crop_factor = 1.7
+        bbox_drop_size = 10
+
+        if self.bbox_model == None:
+            bbox_model_path = os.path.join(get_model_dir("ultralytics"), "face_yolov8n.pt")
+            
+            # 沿用 comfyui-ultralytics-yolo
+            for fp in [os.path.join("bbox","face_yolov8m.pt"),os.path.join("bbox","face_yolov8n.pt"),"face_yolov8n.pt","face_yolov8m.pt",]:
+                np=os.path.join(get_model_dir("ultralytics"), fp)
+                if os.path.isfile(np):
+                    bbox_model_path=np
+
+            self.bbox_model = YOLO(bbox_model_path)
+
+        pred = self.bbox_model(image_rgb, conf=0.7, device="")
+        bboxes = pred[0].boxes.xyxy.cpu().numpy()
+
+        w, h = get_rgb_size(image_rgb)
+
+        # for x, label in zip(segmasks, detected_results[0]):
+        for x1, y1, x2, y2 in bboxes:
+            bbox_w = x2 - x1
+            bbox_h = y2 - y1
+
+            crop_w = bbox_w * crop_factor
+            crop_h = bbox_h * crop_factor
+
+            crop_w = max(crop_h, crop_w)
+            crop_h = crop_w
+
+            kernel_x = x1 + bbox_w / 2
+            kernel_y = y1 + bbox_h / 2
+
+            new_x1, new_x2, crop_w = calc_crop_limit(kernel_x, w, crop_w)
+
+            if crop_w < crop_h:
+                crop_h = crop_w
+
+            new_y1, new_y2, crop_h = calc_crop_limit(kernel_y, h, crop_h)
+
+            if crop_h < crop_w:
+                crop_w = crop_h
+                new_x1, new_x2, crop_w = calc_crop_limit(kernel_x, w, crop_w)
+
+            return [int(new_x1), int(new_y1), int(new_x2), int(new_y2)]
+
+        print("Failed to detect face!!")
+        return [0, 0, w, h]
+
+    def crop_face(self, rgb_img):
+        region = self.detect_face(rgb_img)
+        face_image = rgb_crop(rgb_img, region)
+        return face_image, region
+
+    def get_pipeline(self):
+        if self.pipeline == None:
+            print("Load pipeline...")
+            self.pipeline = LivePortraitWrapper(cfg=inference_cfg)
+
+        return self.pipeline
+
+    def prepare_src_image(self, img):
+        h, w = img.shape[:2]
+        input_shape = [256,256]
+        if h != input_shape[0] or w != input_shape[1]:
+            x = cv2.resize(img, (input_shape[0], input_shape[1]), interpolation = cv2.INTER_LINEAR)
+        else:
+            x = img.copy()
+
+        if x.ndim == 3:
+            x = x[np.newaxis].astype(np.float32) / 255.  # HxWx3 -> 1xHxWx3, normalized to 0~1
+        elif x.ndim == 4:
+            x = x.astype(np.float32) / 255.  # BxHxWx3, normalized to 0~1
+        else:
+            raise ValueError(f'img ndim should be 3 or 4: {x.ndim}')
+        x = np.clip(x, 0, 1)  # clip to 0~1
+        x = torch.from_numpy(x).permute(0, 3, 1, 2)  # 1xHxWx3 -> 1x3xHxW
+        x = x.cuda()
+        return x
+
+    def GetMask(self):
+        if self.mask_img is None:
+            path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "./LivePortrait/src/utils/resources/mask_template.png")
+            self.mask_img = cv2.imread(path, cv2.IMREAD_COLOR)
+        return self.mask_img
+
+    def prepare_source(self, source_image, is_video = False):
+        print("Prepare source...")
+        engine = self.get_pipeline()
+        source_image_np = (source_image * 255).byte().numpy()
+        img_rgb = source_image_np[0]
+        face_img, crop_region = self.crop_face(img_rgb)
+
+        scale = face_img.shape[0] / 512.
+        crop_trans_m = create_transform_matrix(crop_region[0], crop_region[1], scale)
+        mask_ori = cv2.warpAffine(self.GetMask(), crop_trans_m, get_rgb_size(img_rgb), cv2.INTER_LINEAR)
+        mask_ori = mask_ori.astype(np.float32) / 255.
+
+        psi_list = []
+        for img_rgb in source_image_np:
+            face_img = rgb_crop(img_rgb, crop_region)
+            i_s = self.prepare_src_image(face_img)
+            x_s_info = engine.get_kp_info(i_s)
+            f_s_user = engine.extract_feature_3d(i_s)
+            x_s_user = engine.transform_keypoint(x_s_info)
+            psi = PreparedSrcImg(img_rgb, crop_trans_m, x_s_info, f_s_user, x_s_user, mask_ori)
+            if is_video == False:
+                return psi
+            psi_list.append(psi)
+
+        return psi_list
+
+    def prepare_driving_video(self, face_images):
+        print("Prepare driving video...")
+        pipeline = self.get_pipeline()
+        f_img_np = (face_images * 255).byte().numpy()
+
+        out_list = []
+        for f_img in f_img_np:
+            i_d = pipeline.prepare_source(f_img)
+            d_info = pipeline.get_kp_info(i_d)
+            #out_list.append((d_info, get_rotation_matrix(d_info['pitch'], d_info['yaw'], d_info['roll'])))
+            out_list.append(d_info)
+
+        return out_list
+
+    def calc_fe(_, x_d_new, eyes, eyebrow, wink, pupil_x, pupil_y, mouth, eee, woo, smile,
+                rotate_pitch, rotate_yaw, rotate_roll):
+
+        x_d_new[0, 20, 1] += smile * -0.01
+        x_d_new[0, 14, 1] += smile * -0.02
+        x_d_new[0, 17, 1] += smile * 0.0065
+        x_d_new[0, 17, 2] += smile * 0.003
+        x_d_new[0, 13, 1] += smile * -0.00275
+        x_d_new[0, 16, 1] += smile * -0.00275
+        x_d_new[0, 3, 1] += smile * -0.0035
+        x_d_new[0, 7, 1] += smile * -0.0035
+
+        x_d_new[0, 19, 1] += mouth * 0.001
+        x_d_new[0, 19, 2] += mouth * 0.0001
+        x_d_new[0, 17, 1] += mouth * -0.0001
+        rotate_pitch -= mouth * 0.05
+
+        x_d_new[0, 20, 2] += eee * -0.001
+        x_d_new[0, 20, 1] += eee * -0.001
+        #x_d_new[0, 19, 1] += eee * 0.0006
+        x_d_new[0, 14, 1] += eee * -0.001
+
+        x_d_new[0, 14, 1] += woo * 0.001
+        x_d_new[0, 3, 1] += woo * -0.0005
+        x_d_new[0, 7, 1] += woo * -0.0005
+        x_d_new[0, 17, 2] += woo * -0.0005
+
+        x_d_new[0, 11, 1] += wink * 0.001
+        x_d_new[0, 13, 1] += wink * -0.0003
+        x_d_new[0, 17, 0] += wink * 0.0003
+        x_d_new[0, 17, 1] += wink * 0.0003
+        x_d_new[0, 3, 1] += wink * -0.0003
+        rotate_roll -= wink * 0.1
+        rotate_yaw -= wink * 0.1
+
+        if 0 < pupil_x:
+            x_d_new[0, 11, 0] += pupil_x * 0.0007
+            x_d_new[0, 15, 0] += pupil_x * 0.001
+        else:
+            x_d_new[0, 11, 0] += pupil_x * 0.001
+            x_d_new[0, 15, 0] += pupil_x * 0.0007
+
+        x_d_new[0, 11, 1] += pupil_y * -0.001
+        x_d_new[0, 15, 1] += pupil_y * -0.001
+        eyes -= pupil_y / 2.
+
+        x_d_new[0, 11, 1] += eyes * -0.001
+        x_d_new[0, 13, 1] += eyes * 0.0003
+        x_d_new[0, 15, 1] += eyes * -0.001
+        x_d_new[0, 16, 1] += eyes * 0.0003
+
+
+        if 0 < eyebrow:
+            x_d_new[0, 1, 1] += eyebrow * 0.001
+            x_d_new[0, 2, 1] += eyebrow * -0.001
+        else:
+            x_d_new[0, 1, 0] += eyebrow * -0.001
+            x_d_new[0, 2, 0] += eyebrow * 0.001
+            x_d_new[0, 1, 1] += eyebrow * 0.0003
+            x_d_new[0, 2, 1] += eyebrow * -0.0003
+
+
+        return torch.Tensor([rotate_pitch, rotate_yaw, rotate_roll])
+g_engine = LP_Engine()
+
+class ExpressionSet:
+    def __init__(self, erst = None, es = None):
+        if es != None:
+            self.e = copy.deepcopy(es.e)  # [:, :, :]
+            self.r = copy.deepcopy(es.r)  # [:]
+            self.s = copy.deepcopy(es.s)
+            self.t = copy.deepcopy(es.t)
+        elif erst != None:
+            self.e = erst[0]
+            self.r = erst[1]
+            self.s = erst[2]
+            self.t = erst[3]
+        else:
+            self.e = torch.from_numpy(np.zeros((1, 21, 3))).float().to(device='cuda')
+            self.r = torch.Tensor([0, 0, 0])
+            self.s = 0
+            self.t = 0
+    def div(self, value):
+        self.e /= value
+        self.r /= value
+        self.s /= value
+        self.t /= value
+    def add(self, other):
+        self.e += other.e
+        self.r += other.r
+        self.s += other.s
+        self.t += other.t
+    def sub(self, other):
+        self.e -= other.e
+        self.r -= other.r
+        self.s -= other.s
+        self.t -= other.t
+    def mul(self, value):
+        self.e *= value
+        self.r *= value
+        self.s *= value
+        self.t *= value
+
+    #def apply_ratio(self, ratio):        self.exp *= ratio
+
+def expression_run(psi,expression_json):
+    rotate_yaw = -expression_json["rotate_yaw"]
+
+    pipeline = g_engine.get_pipeline()
+
+    s_info = psi.x_s_info
+    #delta_new = copy.deepcopy()
+    s_exp = s_info['exp'] * expression_json["src_weight"]
+    s_exp[0, 5] = s_info['exp'][0, 5]
+    s_exp += s_info['kp']
+
+    es = ExpressionSet()
+
+    # if sample_image != None:
+    #     if id(self.sample_image) != id(sample_image):
+    #         self.sample_image = sample_image
+    #         d_image_np = (sample_image * 255).byte().numpy()
+    #         d_face, _ = g_engine.crop_face(d_image_np[0])
+    #         i_d = pipeline.prepare_source(d_face)
+    #         self.d_info = pipeline.get_kp_info(i_d)
+    #         self.d_info['exp'][0, 5, 0] = 0
+    #         self.d_info['exp'][0, 5, 1] = 0
+
+    #     # delta_new += s_exp * (1 - sample_ratio) + self.d_info['exp'] * sample_ratio
+    #     es.e += self.d_info['exp'] * sample_ratio
+
+    es.r = g_engine.calc_fe(es.e, 
+                            expression_json["blink"], 
+                            expression_json["eyebrow"], 
+                            expression_json["wink"],
+                            expression_json["pupil_x"], 
+                            expression_json["pupil_y"], 
+                            expression_json["aaa"], 
+                            expression_json["eee"], 
+                            expression_json["woo"], 
+                            expression_json['smile'],
+                            expression_json['rotate_pitch'], 
+                            rotate_yaw, 
+                            expression_json['rotate_roll']
+                            )
+
+    new_rotate = get_rotation_matrix(s_info['pitch'] + es.r[0], s_info['yaw'] + es.r[1],
+                                         s_info['roll'] + es.r[2])
+    x_d_new = (s_info['scale'] * (1 + es.s)) * ((s_exp + es.e) @ new_rotate) + s_info['t']
+
+    x_d_new = pipeline.stitching(psi.x_s_user, x_d_new)
+
+    crop_out = pipeline.warp_decode(psi.f_s_user, psi.x_s_user, x_d_new)
+    crop_out = pipeline.parse_output(crop_out['out'])[0]
+
+    crop_with_fullsize = cv2.warpAffine(crop_out, psi.crop_trans_m, get_rgb_size(psi.src_rgb), cv2.INTER_LINEAR)
+    out = np.clip(psi.mask_ori * crop_with_fullsize + (1 - psi.mask_ori) * psi.src_rgb, 0, 255).astype(np.uint8)
+
+    return pil2tensor(out)
+
+
+#  
+class ExpressionVideoNode:
+    def __init__(self):
+        self.src_image = None
+    @classmethod
+    def INPUT_TYPES(s):
+        
+        return {"required": {
+                        "src_image": ("IMAGE",), 
+                        "from_expression":("STRING", {"forceInput": True,"dynamicPrompts": False}),
+                        "to_expression":("STRING", {"forceInput": True,"dynamicPrompts": False}),
+                        "interpolation_type":( ['linear', 'nearest', 'cubic'], 
+                                    {"default": "cubic"}),
+                        "interpolations_num":("INT", {
+                                "default": 1, 
+                                "min": 1, #Minimum value
+                                "max": 204800000000, #Maximum value
+                                "step": 1, #Slider's step
+                                "display": "number" # Cosmetic only: display as "number" or "slider"
+                            })
+                        },
+                }
+    
+    RETURN_TYPES = ("IMAGE",)
+    RETURN_NAMES = ("frames",)
+
+    FUNCTION = "run"
+
+    # OUTPUT_NODE = True
+
+    CATEGORY = "♾️Mixlab/Video/LivePortrait"
+
+    INPUT_IS_LIST = False
+    OUTPUT_IS_LIST = (False,) #x list
+  
+    def run(self,src_image,from_expression, to_expression,interpolation_type,interpolations_num ):
+        
+        if id(src_image) != id(self.src_image):
+            self.psi = g_engine.prepare_source(src_image)
+            self.src_image = src_image
+
+        from_expression = json.loads(from_expression)
+        to_expression = json.loads(to_expression)
+
+        from_expression=update_expression_json(from_expression)
+        to_expression=update_expression_json(to_expression)
+        # print('from_expression',from_expression)
+        # print('to_expression',to_expression)
+        exps=interpolate_dicts(from_expression,to_expression,interpolations_num,interpolation_type)
+        
+        result=[]
+
+        for exp in exps:
+            out_img = expression_run(self.psi,exp)
+            result.append(out_img)
+
+        result=torch.cat(result, dim=0)
+
+        return (result,)
+
+
+#  
+class ExpressionVideo2VideoNode:
+    def __init__(self):
+        self.src_image = None
+    @classmethod
+    def INPUT_TYPES(s):
+        
+        return {"required": {
+                        "src_frames": ("IMAGE",), #batch
+                        "from_expression":("STRING", {"forceInput": True,"dynamicPrompts": False}),
+                        "to_expression":("STRING", {"forceInput": True,"dynamicPrompts": False}),
+                        "interpolation_type":( ['linear', 'nearest', 'cubic'], 
+                                    {"default": "cubic"}), 
+                        },
+                }
+    
+    RETURN_TYPES = ("IMAGE",)
+    RETURN_NAMES = ("frames",)
+
+    FUNCTION = "run"
+
+    # OUTPUT_NODE = True
+
+    CATEGORY = "♾️Mixlab/Video/LivePortrait"
+
+    INPUT_IS_LIST = False
+    OUTPUT_IS_LIST = (False,) #x list
+  
+    def run(self,src_frames,from_expression, to_expression,interpolation_type ):
+        
+        images = [src_frames[i:i + 1, ...] for i in range(src_frames.shape[0])]
+        interpolations_num=len(images)
+        from_expression = json.loads(from_expression)
+        to_expression = json.loads(to_expression)
+
+        from_expression=update_expression_json(from_expression)
+        to_expression=update_expression_json(to_expression)
+      
+        exps=interpolate_dicts(from_expression,to_expression,interpolations_num,interpolation_type)
+        
+        result=[]
+
+        for i in range(interpolations_num):
+
+            src_image = images[i]
+            exp=exps[i]
+
+            self.psi = g_engine.prepare_source(src_image)
+            self.src_image = src_image
+
+            out_img = expression_run(self.psi,exp)
+            result.append(out_img)
+
+        result=torch.cat(result, dim=0)
+
+        return (result,)
+
+class ExpressionEditor:
+    def __init__(self):
+        self.sample_image = None
+        self.src_image = None
+
+    @classmethod
+    def INPUT_TYPES(s):
+        display = "number"
+        #display = "slider"
+        return {
+            "required": {
+                "src_image": ("IMAGE",), 
+
+                "rotate_pitch": ("FLOAT", {"default": 0, "min": -50, "max": 50, "step": 0.2, "display": display}),
+                "rotate_yaw": ("FLOAT", {"default": 0, "min": -50, "max": 50, "step": 0.2, "display": display}),
+                "rotate_roll": ("FLOAT", {"default": 0, "min": -50, "max": 50, "step": 0.2, "display": display}),
+
+                "blink": ("FLOAT", {"default": 0, "min": -30, "max": 15, "step": 0.2, "display": display}),
+                "eyebrow": ("FLOAT", {"default": 0, "min": -30, "max": 25, "step": 0.2, "display": display}),
+                "wink": ("FLOAT", {"default": 0, "min": -10, "max": 25, "step": 0.2, "display": display}),
+                
+                "pupil_x": ("FLOAT", {"default": 0, "min": -15, "max": 15, "step": 0.2, "display": display}),
+                "pupil_y": ("FLOAT", {"default": 0, "min": -15, "max": 15, "step": 0.2, "display": display}),
+                
+                "aaa": ("FLOAT", {"default": 0, "min": -30, "max": 120, "step": 1, "display": display}),
+                "eee": ("FLOAT", {"default": 0, "min": -20, "max": 15, "step": 0.2, "display": display}),
+                "woo": ("FLOAT", {"default": 0, "min": -20, "max": 15, "step": 0.2, "display": display}),
+                
+                "smile": ("FLOAT", {"default": 0, "min": -0.3, "max": 1.3, "step": 0.01, "display": display}),
+
+                "src_weight": ("FLOAT", {"default": 1, "min": 0, "max": 1, "step": 0.01, "display": display}),
+                # "sample_ratio": ("FLOAT", {"default": 1, "min": 0, "max": 1, "step": 0.01, "display": display}),
+            },
+
+            "optional": {
+                
+                "expression_json":("STRING", {"forceInput": True,"dynamicPrompts": False}),
+                # "sample_image": ("IMAGE",), 
+            },
+        }
+
+    RETURN_TYPES = ("IMAGE","STRING",)
+    RETURN_NAMES = ("image","expression_json",)
+
+    FUNCTION = "run"
+
+    # OUTPUT_NODE = True
+
+    CATEGORY = "♾️Mixlab/Face"
+
+    # INPUT_IS_LIST = False
+    # OUTPUT_IS_LIST = (False,)
+
+    def run(self,src_image, rotate_pitch, rotate_yaw, rotate_roll, blink, eyebrow, wink, pupil_x, pupil_y, aaa,
+             eee, woo, smile,
+            src_weight, expression_json=None):
+        
+        
+        if expression_json!=None:
+            try:
+                dict_obj = json.loads(expression_json)
+                if "rotate_pitch" in dict_obj:
+                    rotate_pitch=dict_obj["rotate_pitch"]
+                if "rotate_yaw" in dict_obj:
+                    rotate_yaw=dict_obj["rotate_yaw"]
+                if "rotate_roll" in dict_obj:
+                    rotate_roll=dict_obj["rotate_roll"]
+                if "blink" in dict_obj:
+                    blink=dict_obj["blink"]
+                if "eyebrow" in dict_obj:
+                    eyebrow=dict_obj["eyebrow"]
+                if "wink" in dict_obj:
+                    wink=dict_obj["wink"]
+                if "pupil_x" in dict_obj:
+                    pupil_x=dict_obj["pupil_x"]
+                if "pupil_y" in dict_obj:
+                    pupil_y=dict_obj["pupil_y"]
+                if "aaa" in dict_obj:
+                    aaa=dict_obj["aaa"]
+                if "eee" in dict_obj:
+                    eee=dict_obj["eee"]
+                if "woo" in dict_obj:
+                    woo=dict_obj["woo"]
+                if "smile" in dict_obj:
+                    smile=dict_obj["smile"]
+                if "src_weight" in dict_obj:
+                    src_weight=dict_obj["src_weight"]
+            except:
+                print("#expression_json",expression_json)
+
+        # print('#expression_json',expression_json)
+
+        expression_json={
+            "rotate_pitch":rotate_pitch,
+            "rotate_yaw":rotate_yaw,
+            "rotate_roll":rotate_roll, 
+            "blink":blink, 
+            "eyebrow":eyebrow, 
+            "wink":wink, 
+            "pupil_x":pupil_x, 
+            "pupil_y":pupil_y, 
+            "aaa":aaa, 
+            "eee":eee, 
+            "woo":woo, 
+            "smile":smile,
+            "src_weight":src_weight
+        }
+
+        rotate_yaw = -rotate_yaw
+
+        if id(src_image) != id(self.src_image):
+            self.psi = g_engine.prepare_source(src_image)
+            self.src_image = src_image
+       
+        out_img = expression_run(self.psi,expression_json)
+
+        return (out_img,json.dumps(expression_json) ,)
+

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "comfyui-liveportrait"
 description = "The ComfyUI version of [a/LivePortrait](https://github.com/KwaiVGI/LivePortrait)."
-version = "1.3.1"
+version = "1.4.0"
 license = "LICENSE"
 dependencies = ["numpy>=1.26.4", "opencv-python-headless", "imageio>=2.34.2", "lmdb>=1.4.1", "timm>=1.0.7", "rich>=13.7.1", "ffmpeg>=1.4", "onnxruntime-gpu>=1.18.0", "onnx>=1.16.1", "scikit-image>=0.24.0", "albumentations>=1.4.10", "matplotlib>=3.9.0", "imageio-ffmpeg>=0.5.1"]
 

requirements.txt

@@ -3,6 +3,7 @@ opencv-python-headless
 imageio-ffmpeg>=0.5.1
 lmdb>=1.4.1
 timm>=1.0.7
-rich>=13.7.1
+
 albumentations>=1.4.10
-insightface
+insightface
+ultralytics