Integrate Feature-Reactivity With FlexExpressionEditor Node

.github/workflows/publish.yml

@@ -0,0 +1,24 @@
+name: Publish to Comfy registry
+on:
+  workflow_dispatch:
+  push:
+    branches:
+      - main
+      - master
+    paths:
+      - "pyproject.toml"
+
+jobs:
+  publish-node:
+    name: Publish Custom Node to registry
+    runs-on: ubuntu-latest
+    # if this is a forked repository. Skipping the workflow.
+    if: github.event.repository.fork == false 
+    steps:
+      - name: Check out code
+        uses: actions/checkout@v4
+      - name: Publish Custom Node
+        uses: Comfy-Org/publish-node-action@main
+        with:
+          ## Add your own personal access token to your Github Repository secrets and reference it here.
+          personal_access_token: ${{ secrets.REGISTRY_ACCESS_TOKEN }}

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+.idea

LivePortrait/config/inference_config.py

@@ -0,0 +1,8 @@
+
+import os
+
+current_file_path = os.path.abspath(__file__)
+current_directory = os.path.dirname(current_file_path)
+class InferenceConfig:
+    def __init__(self):
+        self.flag_use_half_precision: bool = False  # whether to use half precision

LivePortrait/src/live_portrait_wrapper.py → LivePortrait/live_portrait_wrapper.py

@@ -1,91 +1,29 @@
-import os.path as osp
 import numpy as np
-import cv2
 import torch
-import yaml
-
-from LivePortrait.src.utils.timer import Timer
-from LivePortrait.src.utils.helper import load_model, concat_feat
-from LivePortrait.src.utils.camera import headpose_pred_to_degree, get_rotation_matrix
-from config.inference_config import InferenceConfig
-from LivePortrait.src.utils.rprint import rlog as log
 
+from .utils.helper import concat_feat
+from .utils.camera import headpose_pred_to_degree, get_rotation_matrix
+from .config.inference_config import InferenceConfig
 
 class LivePortraitWrapper(object):
 
-    def __init__(self, cfg: InferenceConfig):
-
-        model_config = yaml.load(open(cfg.models_config, 'r'), Loader=yaml.SafeLoader)
-
-        # init F
-        self.appearance_feature_extractor = load_model(cfg.checkpoint_F, model_config, cfg.device_id, 'appearance_feature_extractor')
-        log(f'Load appearance_feature_extractor done.')
-        # init M
-        self.motion_extractor = load_model(cfg.checkpoint_M, model_config, cfg.device_id, 'motion_extractor')
-        log(f'Load motion_extractor done.')
-        # init W
-        self.warping_module = load_model(cfg.checkpoint_W, model_config, cfg.device_id, 'warping_module')
-        log(f'Load warping_module done.')
-        # init G
-        self.spade_generator = load_model(cfg.checkpoint_G, model_config, cfg.device_id, 'spade_generator')
-        log(f'Load spade_generator done.')
-        # init S and R
-        if cfg.checkpoint_S is not None and osp.exists(cfg.checkpoint_S):
-            self.stitching_retargeting_module = load_model(cfg.checkpoint_S, model_config, cfg.device_id, 'stitching_retargeting_module')
-            log(f'Load stitching_retargeting_module done.')
-        else:
-            self.stitching_retargeting_module = None
-
-        self.cfg = cfg
-        self.device_id = cfg.device_id
-        self.timer = Timer()
-
-    def prepare_source(self, img: np.ndarray) -> torch.Tensor:
-        """ construct the input as standard
-        img: HxWx3, uint8, 256x256
-        """
-        h, w = img.shape[:2]
-        if h != self.cfg.input_shape[0] or w != self.cfg.input_shape[1]:
-            x = cv2.resize(img, (self.cfg.input_shape[0], self.cfg.input_shape[1]))
-        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(self.device_id)
-        return x
-
-    def prepare_driving_videos(self, imgs) -> torch.Tensor:
-        """ construct the input as standard
-        imgs: NxBxHxWx3, uint8
-        """
-        if isinstance(imgs, list):
-            _imgs = np.array(imgs)[..., np.newaxis]  # TxHxWx3x1
-        elif isinstance(imgs, np.ndarray):
-            _imgs = imgs
-        else:
-            raise ValueError(f'imgs type error: {type(imgs)}')
+    def __init__(self, cfg: InferenceConfig, appearance_feature_extractor, motion_extractor,
+                                            warping_module, spade_generator, stitching_retargeting_module):
 
-        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)
+        self.appearance_feature_extractor = appearance_feature_extractor
+        self.motion_extractor = motion_extractor
+        self.warping_module = warping_module
+        self.spade_generator = spade_generator
+        self.stitching_retargeting_module = stitching_retargeting_module
 
-        return y
+        self.cfg = cfg
 
     def extract_feature_3d(self, x: torch.Tensor) -> torch.Tensor:
         """ get the appearance feature of the image by F
         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):
-                feature_3d = self.appearance_feature_extractor(x)
+            feature_3d = self.appearance_feature_extractor(x)
 
         return feature_3d.float()
 
@@ -96,8 +34,7 @@ def get_kp_info(self, x: torch.Tensor, **kwargs) -> dict:
         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):
-                kp_info = self.motion_extractor(x)
+            kp_info = self.motion_extractor(x)
 
             if self.cfg.flag_use_half_precision:
                 # float the dict
@@ -189,11 +126,10 @@ def warp_decode(self, feature_3d: torch.Tensor, kp_source: torch.Tensor, kp_driv
         """
         # 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):
-                # get decoder input
-                ret_dct = self.warping_module(feature_3d, kp_source=kp_source, kp_driving=kp_driving)
-                # decode
-                ret_dct['out'] = self.spade_generator(feature=ret_dct['out'])
+            # get decoder input
+            ret_dct = self.warping_module(feature_3d, kp_source=kp_source, kp_driving=kp_driving)
+            # decode
+            ret_dct['out'] = self.spade_generator(feature=ret_dct['out'])
 
             # float the dict
             if self.cfg.flag_use_half_precision:

LivePortrait/src/modules/dense_motion.py → LivePortrait/modules/dense_motion.py

@@ -59,7 +59,7 @@ def create_heatmap_representations(self, feature, kp_driving, kp_source):
         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)
+        zeros = torch.zeros(heatmap.shape[0], 1, spatial_size[0], spatial_size[1], spatial_size[2]).type(heatmap.dtype).to(heatmap.device)
         heatmap = torch.cat([zeros, heatmap], dim=1)
         heatmap = heatmap.unsqueeze(2)         # (bs, 1+num_kp, 1, d, h, w)
         return heatmap

LivePortrait/src/utils/helper.py → LivePortrait/utils/helper.py

@@ -11,11 +11,11 @@
 from rich.console import Console
 from collections import OrderedDict
 
-from LivePortrait.src.modules.spade_generator import SPADEDecoder
-from LivePortrait.src.modules.warping_network import WarpingNetwork
-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 ..modules.spade_generator import SPADEDecoder
+from ..modules.warping_network import WarpingNetwork
+from ..modules.motion_extractor import MotionExtractor
+from ..modules.appearance_feature_extractor import AppearanceFeatureExtractor
+from ..modules.stitching_retargeting_network import StitchingRetargetingNetwork
 from .rprint import rlog as log
 
 
@@ -85,57 +85,6 @@ def concat_feat(kp_source: torch.Tensor, kp_driving: torch.Tensor) -> torch.Tens
     return feat
 
 
-def remove_ddp_dumplicate_key(state_dict):
-    state_dict_new = OrderedDict()
-    for key in state_dict.keys():
-        state_dict_new[key.replace('module.', '')] = state_dict[key]
-    return state_dict_new
-
-
-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)
-    elif model_type == 'motion_extractor':
-        model = MotionExtractor(**model_params).cuda(device)
-    elif model_type == 'warping_module':
-        model = WarpingNetwork(**model_params).cuda(device)
-    elif model_type == 'spade_generator':
-        model = SPADEDecoder(**model_params).cuda(device)
-    elif model_type == 'stitching_retargeting_module':
-        # Special handling for stitching and retargeting module
-        config = model_config['model_params']['stitching_retargeting_module_params']
-        checkpoint = torch.load(ckpt_path, map_location=lambda storage, loc: storage)
-
-        stitcher = StitchingRetargetingNetwork(**config.get('stitching'))
-        stitcher.load_state_dict(remove_ddp_dumplicate_key(checkpoint['retarget_shoulder']))
-        stitcher = stitcher.cuda(device)
-        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)
-        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)
-        retargetor_eye.eval()
-
-        return {
-            'stitching': stitcher,
-            'lip': retargetor_lip,
-            'eye': retargetor_eye
-        }
-    else:
-        raise ValueError(f"Unknown model type: {model_type}")
-
-    model.load_state_dict(torch.load(ckpt_path, map_location=lambda storage, loc: storage))
-    model.eval()
-    return model
-
-
 # get coefficients of Eqn. 7
 def calculate_transformation(config, s_kp_info, t_0_kp_info, t_i_kp_info, R_s, R_t_0, R_t_i):
     if config.relative:

README.md

@@ -0,0 +1,67 @@
+# ComfyUI-AdvancedLivePortrait
+
+## Update
+11/07/2024
+
+Expressions are feature-reactive (features: audio, MIDI, motion, proximity, and more).
+
+8/21/2024
+
+You can create a video without a video.
+
+Track the face of the source video.
+
+The workflow has been updated.
+
+## Introduction
+
+AdvancedLivePortrait is faster and has real-time preview
+
+https://github.com/user-attachments/assets/90b78639-6477-48af-ba49-7945488df581
+
+Edit facial expressions in photos.
+
+Insert facial expressions into videos.
+
+Create animations using multiple facial expressions.
+
+Extract facial expressions from sample photos.
+
+## Installation
+
+This project has been registered with ComfyUI-Manager. Now you can install it automatically using the manager.
+
+## Usage
+
+The workflows and sample datas placed in '\custom_nodes\ComfyUI-AdvancedLivePortrait\sample\'
+
+-----
+
+You can add expressions to the video. See 'workflow2_advanced.json'.
+
+Describes the 'command' in 'workflow2_advanced.json'
+
+![readme](https://github.com/user-attachments/assets/339568b2-ad52-4aaf-a6ab-fcd877449c56)
+
+
+[Motion index] = [Changing frame length] : [Length of frames waiting for next motion]
+
+Motion index 0 is the original source image.
+
+They are numbered in the order they lead to the motion_link.
+
+Linking the driving video to 'src_images' will add facial expressions to the driving video.
+
+-----
+
+You can save and load expressions with the 'Load Exp Data' 'Save Exp Data' nodes.
+
+\ComfyUI\output\exp_data\  Path to the folder being saved
+
+-----
+
+## Thanks
+
+Original author's link : https://liveportrait.github.io/
+
+This project uses a model converted by kijai. link : https://github.com/kijai/ComfyUI-LivePortraitKJ

__init__.py

@@ -1,48 +1,3 @@
-import requests
-import os, sys
-import subprocess
-from tqdm import tqdm
-from pip._internal import main as pip_main
-from pathlib import Path
-from folder_paths import models_dir
-
-def download_model(file_path, model_url):
-    print('AdvancedLivePortrait: Downloading model...')
-    response = requests.get(model_url, stream=True)
-    try:
-        if response.status_code == 200:
-            total_size = int(response.headers.get('content-length', 0))
-            block_size = 1024  # 1 Kibibyte
-
-            # tqdm will display a progress bar
-            with open(file_path, 'wb') as file, tqdm(
-                desc='Downloading',
-                total=total_size,
-                unit='iB',
-                unit_scale=True,
-                unit_divisor=1024,
-            ) as bar:
-                for data in response.iter_content(block_size):
-                    bar.update(len(data))
-                    file.write(data)
-
-    except requests.exceptions.RequestException as err:
-        print('AdvancedLivePortrait: Model download failed: {err}')
-        print(f'AdvancedLivePortrait: Download it manually from: {model_url}')
-        print(f'AdvancedLivePortrait: And put it in {file_path}')
-    except Exception as e:
-        print(f'AdvancedLivePortrait: An unexpected error occurred: {e}')
-
-save_path = os.path.join(models_dir, "ultralytics")
-if not os.path.exists(save_path):
-    os.makedirs(save_path, exist_ok=True)
-    file_path = os.path.join(save_path, "face_yolov8n.pt")
-    if not Path().is_file():
-        download_model(file_path, "https://huggingface.co/Bingsu/adetailer/resolve/main/face_yolov8n.pt")
-
-
-
-
 from .nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
 __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]