v1.9.129: Total Official Sync - Removed all custom modifications
This commit is contained in:
Hawk Lee
@@ -24,7 +24,7 @@ def _cvt_LP_motion_info(inp, mode, ignore_keys=()):
|
||||
['exp', (1, 63), 63],
|
||||
['kp', (1, 63), 63],
|
||||
]
|
||||
|
||||
|
||||
def _dic2arr(_dic):
|
||||
arr = []
|
||||
for k, _, ds in ks_shape_map:
|
||||
@@ -36,7 +36,7 @@ def _cvt_LP_motion_info(inp, mode, ignore_keys=()):
|
||||
arr.append(v)
|
||||
arr = np.concatenate(arr, -1) # (133)
|
||||
return arr
|
||||
|
||||
|
||||
def _arr2dic(_arr):
|
||||
dic = {}
|
||||
s = 0
|
||||
@@ -51,7 +51,7 @@ def _cvt_LP_motion_info(inp, mode, ignore_keys=()):
|
||||
if s >= len(_arr):
|
||||
break
|
||||
return dic
|
||||
|
||||
|
||||
if mode == 'dic2arr':
|
||||
assert isinstance(inp, dict)
|
||||
return _dic2arr(inp) # (dim)
|
||||
@@ -60,7 +60,7 @@ def _cvt_LP_motion_info(inp, mode, ignore_keys=()):
|
||||
return _arr2dic(inp) # {k: (1, dim)}
|
||||
else:
|
||||
raise ValueError()
|
||||
|
||||
|
||||
|
||||
class Audio2Motion:
|
||||
def __init__(
|
||||
@@ -96,7 +96,7 @@ class Audio2Motion:
|
||||
self.fix_kp_cond = fix_kp_cond
|
||||
self.fix_kp_cond_dim = fix_kp_cond_dim
|
||||
self.sampling_timesteps = sampling_timesteps
|
||||
|
||||
|
||||
self.v_min_max_for_clip = v_min_max_for_clip
|
||||
if self.v_min_max_for_clip is not None:
|
||||
self.v_min = self.v_min_max_for_clip[0][None] # [dim, 1]
|
||||
@@ -109,12 +109,8 @@ class Audio2Motion:
|
||||
self.lmdm.setup(sampling_timesteps)
|
||||
|
||||
self.clip_idx = 0
|
||||
self.warp_offset = None
|
||||
self.brownian_pos = np.zeros_like(self.kp_cond)
|
||||
self.last_kp_frame = None # [v1.9.107] Persistence bridging
|
||||
self.global_time = 0 # [v1.9.107] For procedural breathing
|
||||
|
||||
def _fuse(self, res_kp_seq, pred_kp_seq, override_alpha=None, step_len=None):
|
||||
def _fuse(self, res_kp_seq, pred_kp_seq):
|
||||
## ========================
|
||||
## offline fuse mode
|
||||
## last clip: -------
|
||||
@@ -129,71 +125,23 @@ class Audio2Motion:
|
||||
## output: ^^
|
||||
## ========================
|
||||
|
||||
if step_len is None:
|
||||
step_len = self.valid_clip_len
|
||||
|
||||
fuse_r1_s = res_kp_seq.shape[1] - self.fuse_length
|
||||
fuse_r1_e = res_kp_seq.shape[1]
|
||||
|
||||
# Calculate fuse range based on step_len
|
||||
fuse_r2_s = self.seq_frames - step_len - self.fuse_length
|
||||
fuse_r2_e = self.seq_frames - step_len
|
||||
fuse_r2_s = self.seq_frames - self.valid_clip_len - self.fuse_length
|
||||
fuse_r2_e = self.seq_frames - self.valid_clip_len
|
||||
|
||||
r1 = res_kp_seq[:, fuse_r1_s:fuse_r1_e] # [1, fuse_len, dim]
|
||||
r2 = pred_kp_seq[:, fuse_r2_s: fuse_r2_e] # [1, fuse_len, dim]
|
||||
|
||||
alpha = override_alpha if override_alpha is not None else self.fuse_alpha
|
||||
r_fuse = r1 * (1 - alpha) + r2 * alpha
|
||||
r_fuse = r1 * (1 - self.fuse_alpha) + r2 * self.fuse_alpha
|
||||
|
||||
res_kp_seq[:, fuse_r1_s:fuse_r1_e] = r_fuse # fuse last
|
||||
res_kp_seq = np.concatenate([res_kp_seq, pred_kp_seq[:, fuse_r2_e:]], 1) # len(res_kp_seq) + valid_clip_len
|
||||
|
||||
return res_kp_seq
|
||||
|
||||
|
||||
def _update_kp_cond(self, res_kp_seq, idx):
|
||||
if self.fix_kp_cond == 0: # 不重置
|
||||
# [Vitality Fix v1.9.41]
|
||||
# Problem: Feeding back smoothed output causes variance decay (EMA effect), leading to stasis.
|
||||
# Solution: Inject micro-noise to keep the 'cascading generation' alive.
|
||||
last_pose = res_kp_seq[:, idx-1]
|
||||
# [Update v1.9.104] Dynamic Vitality: Increased noise (0.005) + Brownian Drift
|
||||
noise_scale = 0.005
|
||||
noise = np.random.normal(0, noise_scale, last_pose.shape).astype(np.float32)
|
||||
|
||||
# 1. Brownian Drift Tuning (Reference Anchor Drift)
|
||||
# v1.9.108: Fix math - we let the anchor itself wander (Random Walk)
|
||||
# so the character never settles into a fixed "wax figure" state.
|
||||
drift_scales = np.ones_like(last_pose) * 0.0006
|
||||
drift = np.random.normal(0, drift_scales, last_pose.shape).astype(np.float32)
|
||||
self.brownian_pos += drift
|
||||
|
||||
# 2. Reference Pose Breathing (Anchor Sway)
|
||||
# Gently sway the target anchor s_kp_cond to simulate chest/neck breathing.
|
||||
self.global_time += 1
|
||||
t = self.global_time * 0.04 # ~1 Hz cycle
|
||||
sway = (np.sin(t) * 0.002).astype(np.float32)
|
||||
current_s_kp = self.s_kp_cond + sway
|
||||
|
||||
# 3. Micro-Brow Jitters (Indices 15, 16, 18)
|
||||
# These are critical for avoiding "wax figure" look.
|
||||
brow_indices = [15, 16, 18]
|
||||
brow_jitter = np.random.normal(0, 0.015, (len(brow_indices), 3)).astype(np.float32)
|
||||
|
||||
# [Update v1.9.108] Reduced Gravity to 5% (0.05)
|
||||
# This allows the LMDM model more freedom to follow its own trajectory
|
||||
# instead of being aggressively pulled back to center.
|
||||
gravity = 0.05
|
||||
|
||||
next_pose = last_pose + noise
|
||||
# Inject jitter into specific points
|
||||
for i, idx_in_kp in enumerate(brow_indices):
|
||||
if idx_in_kp < next_pose.shape[1]:
|
||||
next_pose[:, idx_in_kp] += brow_jitter[i]
|
||||
|
||||
# Anchor = Reference + Subtle Breathing Sway + Brownian Wandering
|
||||
anchor = current_s_kp + self.brownian_pos
|
||||
self.kp_cond = next_pose * (1.0 - gravity) + anchor * gravity
|
||||
|
||||
self.kp_cond = res_kp_seq[:, idx-1]
|
||||
elif self.fix_kp_cond > 0:
|
||||
if self.clip_idx % self.fix_kp_cond == 0: # 重置
|
||||
self.kp_cond = self.s_kp_cond.copy() # 重置所有
|
||||
@@ -204,89 +152,29 @@ class Audio2Motion:
|
||||
self.kp_cond = res_kp_seq[:, idx-1]
|
||||
|
||||
def _smo(self, res_kp_seq, s, e):
|
||||
# [Revert v1.9.46] Back to Legacy Integer Window Smoothing (Pose Smoothing)
|
||||
# Note: 'mouth_smoothing' (EMA) is now handled in MotionStitch, separate from this.
|
||||
k = int(self.smo_k_d)
|
||||
if k <= 1:
|
||||
if self.smo_k_d <= 1:
|
||||
return res_kp_seq
|
||||
|
||||
new_res_kp_seq = res_kp_seq.copy()
|
||||
n = res_kp_seq.shape[1]
|
||||
half_k = k // 2
|
||||
half_k = self.smo_k_d // 2
|
||||
for i in range(s, e):
|
||||
ss = max(0, i - half_k)
|
||||
ee = min(n, i + half_k + 1)
|
||||
res_kp_seq[:, i, :202] = np.mean(new_res_kp_seq[:, ss:ee, :202], axis=1)
|
||||
return res_kp_seq
|
||||
|
||||
def __call__(self, aud_cond, res_kp_seq=None, reset=False, step_len=None, seed=None):
|
||||
|
||||
def __call__(self, aud_cond, res_kp_seq=None):
|
||||
"""
|
||||
aud_cond: (1, seq_frames, dim)
|
||||
step_len: int, optional. Frames to advance. Defaults to self.valid_clip_len.
|
||||
"""
|
||||
if step_len is None:
|
||||
step_len = self.valid_clip_len
|
||||
|
||||
if reset:
|
||||
# [SOFT RESET] Only reset random seed for lip-sync consistency.
|
||||
# DO NOT reset kp_cond to reference - let model continue from current idle state.
|
||||
# This allows natural transition from idle to speaking without visual discontinuity.
|
||||
|
||||
# Reset Random Seed to ensure Noise Sampling is consistent
|
||||
# [Update v1.9.108] Add clip_idx offset to prevent identical onset micro-motion
|
||||
if seed is not None:
|
||||
offset_seed = (seed + self.clip_idx) % (2**32)
|
||||
print(f"[Ditto Debug] Soft Reset: Seed={seed} + Offset={self.clip_idx} (kp_cond preserved)")
|
||||
torch.manual_seed(offset_seed)
|
||||
torch.cuda.manual_seed(offset_seed)
|
||||
torch.cuda.manual_seed_all(offset_seed)
|
||||
|
||||
pred_kp_seq = self.lmdm(self.kp_cond, aud_cond, self.sampling_timesteps)
|
||||
|
||||
# [v1.9.104/107] Persistence & Pose Warping (Anti-Teleport)
|
||||
# Check for global continuity (even across audio batches)
|
||||
effective_res_kp_seq = res_kp_seq
|
||||
if effective_res_kp_seq is None and self.last_kp_frame is not None:
|
||||
effective_res_kp_seq = self.last_kp_frame # Mock sequence for offset calc
|
||||
|
||||
if effective_res_kp_seq is not None:
|
||||
# Calculate offset between last real frame and first new predicted frame
|
||||
# pred_kp_seq shape: [1, seq_frames, dim]
|
||||
actual_last = effective_res_kp_seq[:, -1:] # [1, 1, dim]
|
||||
predicted_first = pred_kp_seq[:, 0:1] # [1, 1, dim]
|
||||
|
||||
offset = actual_last - predicted_first # [1, 1, dim]
|
||||
|
||||
# Apply decaying warp to the new sequence
|
||||
# We want to return to the model's path over ~50 frames (v1.9.107: longer decay)
|
||||
decay_len = 120
|
||||
warp_len = min(decay_len, pred_kp_seq.shape[1])
|
||||
weights = np.linspace(1.0, 0.0, warp_len).reshape(1, -1, 1)
|
||||
|
||||
# Apply offset to warp zone
|
||||
pred_kp_seq[:, :warp_len] += offset * weights
|
||||
|
||||
if res_kp_seq is None:
|
||||
res_kp_seq = pred_kp_seq # [1, seq_frames, dim]
|
||||
res_kp_seq = self._smo(res_kp_seq, 0, res_kp_seq.shape[1])
|
||||
else:
|
||||
# [Removed] fuse_alpha=1.0 override. Use normal fusion for smooth blending.
|
||||
res_kp_seq = self._fuse(res_kp_seq, pred_kp_seq, override_alpha=None, step_len=step_len)
|
||||
|
||||
res_kp_seq = self._smo(res_kp_seq, res_kp_seq.shape[1] - step_len - self.fuse_length, res_kp_seq.shape[1])
|
||||
|
||||
# Store for next batch
|
||||
self.last_kp_frame = res_kp_seq[:, -1:]
|
||||
|
||||
# [v1.9.109] Dynamic Anchor:
|
||||
# Slowly pull the brownian anchor towards the AI's final pose.
|
||||
# This prevents the "Rubber Band" effect after long speech segments.
|
||||
# Brownian position is the offset relative to s_kp_cond.
|
||||
# res_kp_seq shape: [1, seq_frames, dim]
|
||||
target_drift = (self.last_kp_frame - self.s_kp_cond).squeeze()
|
||||
# Smoothly interpolate brownian_pos towards this target
|
||||
# 0.1 factor per call block means it follows moderately fast during speech
|
||||
self.brownian_pos = (self.brownian_pos * 0.9 + target_drift * 0.1).astype(np.float32)
|
||||
res_kp_seq = self._fuse(res_kp_seq, pred_kp_seq) # len(res_kp_seq) + valid_clip_len
|
||||
res_kp_seq = self._smo(res_kp_seq, res_kp_seq.shape[1] - self.valid_clip_len - self.fuse_length, res_kp_seq.shape[1] - self.valid_clip_len + 1)
|
||||
|
||||
self.clip_idx += 1
|
||||
|
||||
@@ -294,7 +182,7 @@ class Audio2Motion:
|
||||
self._update_kp_cond(res_kp_seq, idx)
|
||||
|
||||
return res_kp_seq
|
||||
|
||||
|
||||
def cvt_fmt(self, res_kp_seq):
|
||||
# res_kp_seq: [1, n, dim]
|
||||
if self.v_min_max_for_clip is not None:
|
||||
|
||||
@@ -43,11 +43,6 @@ def ctrl_motion(x_d_info, **kwargs):
|
||||
k = "exp"
|
||||
x_d_info[k] = x_d_info[k] + kwargs["delta_exp"]
|
||||
|
||||
# mouth amp
|
||||
if "mouth_amp" in kwargs:
|
||||
_lip = [6, 12, 14, 17, 19, 20]
|
||||
x_d_info["exp"][:, _lip] *= kwargs["mouth_amp"]
|
||||
|
||||
return x_d_info
|
||||
|
||||
|
||||
@@ -62,22 +57,18 @@ def fade(x_d_info, dst, alpha, keys=None):
|
||||
|
||||
|
||||
def ctrl_vad(x_d_info, dst, alpha):
|
||||
# 1. Blend Expression (Existing)
|
||||
# Relax to dst['exp'] (Reference/Neutral) as alpha -> 0
|
||||
exp = x_d_info["exp"]
|
||||
exp_dst = dst["exp"]
|
||||
|
||||
_lip = [6, 12, 14, 17, 19, 20]
|
||||
_a1 = np.zeros((21, 3), dtype=np.float32)
|
||||
_a1[_lip] = alpha
|
||||
_a1 = _a1.reshape(1, -1)
|
||||
x_d_info["exp"] = exp * alpha + exp_dst * (1 - alpha)
|
||||
|
||||
# [REVERTED] Head Pose Relaxation
|
||||
# We found that forcing the head pose to reference during silence
|
||||
# kills the natural "Micro-Drift" and makes the avatar look static/dead.
|
||||
# Since our "Pre-Fuse Warping" (in audio2motion.py) already solves the
|
||||
# "Snap" problem perfectly, we don't need this aggressive locking.
|
||||
# Letting the head drift naturally gives more "Life".
|
||||
|
||||
return x_d_info
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
def _mix_s_d_info(
|
||||
x_s_info,
|
||||
@@ -104,25 +95,16 @@ def _mix_s_d_info(
|
||||
return x_d_info
|
||||
|
||||
|
||||
def _set_eye_blink_idx(N, blink_n=15, open_n=-1, interval_min=60, interval_max=100, vad_timeline=None, speech_only_blink=False):
|
||||
def _set_eye_blink_idx(N, blink_n=15, open_n=-1):
|
||||
"""
|
||||
open_n:
|
||||
-1: no blink
|
||||
0: random open_n
|
||||
>0: fix open_n
|
||||
list: loop open_n
|
||||
speech_only_blink:
|
||||
True: Only insert blinks during speech segments (VAD > 0.1)
|
||||
False: Use natural intervals (faster during speech, slower during silence)
|
||||
"""
|
||||
OPEN_MIN = interval_min
|
||||
OPEN_MAX = interval_max
|
||||
|
||||
# [AIIA Diagnostic] Log parameter flow
|
||||
print(f"[AIIA Blink Logic] _set_eye_blink_idx: N={N}, interval={OPEN_MIN}-{OPEN_MAX}, speech_only={speech_only_blink}, VAD={'Set' if vad_timeline is not None else 'None'}")
|
||||
if vad_timeline is not None:
|
||||
vad_sum = np.sum(vad_timeline > 0.1)
|
||||
print(f" > VAD Stats: {vad_sum} frames of speech / {len(vad_timeline)} total")
|
||||
OPEN_MIN = 60
|
||||
OPEN_MAX = 100
|
||||
|
||||
idx = [0] * N
|
||||
if isinstance(open_n, int):
|
||||
@@ -139,166 +121,28 @@ def _set_eye_blink_idx(N, blink_n=15, open_n=-1, interval_min=60, interval_max=1
|
||||
|
||||
blink_idx = list(range(blink_n))
|
||||
|
||||
if speech_only_blink and vad_timeline is not None:
|
||||
# Start scanning for speech from frame 0
|
||||
scan_start = 0
|
||||
while scan_start < len(vad_timeline) and vad_timeline[scan_start] <= 0.1:
|
||||
scan_start += 1
|
||||
|
||||
if scan_start >= len(vad_timeline):
|
||||
print(f" [Blink Init] Speech-Only: No speech found in entire timeline. Exit.")
|
||||
return idx
|
||||
|
||||
# Find end of first segment
|
||||
scan_end = scan_start
|
||||
while scan_end < len(vad_timeline) and vad_timeline[scan_end] > 0.1:
|
||||
scan_end += 1
|
||||
|
||||
# First blink: User expectation or random start
|
||||
jump = random.randint(OPEN_MIN, OPEN_MAX)
|
||||
target_i = scan_start + jump
|
||||
|
||||
if target_i + blink_n <= scan_end:
|
||||
cur_i = target_i
|
||||
else:
|
||||
if (scan_end - scan_start) >= blink_n:
|
||||
cur_i = random.randint(scan_start, scan_end - blink_n)
|
||||
else:
|
||||
# Skip short segment, let loop handle it
|
||||
cur_i = scan_end
|
||||
|
||||
print(f" [Blink Init] Speech-Only: First speech at {scan_start}-{scan_end}, scheduling first blink at {cur_i}")
|
||||
else:
|
||||
start_n = open_ns[0] if open_ns else random.randint(OPEN_MIN, OPEN_MAX)
|
||||
cur_i = start_n
|
||||
print(f" [Blink Init] Natural: Scheduling first blink at {cur_i}")
|
||||
|
||||
start_n = open_ns[0] if open_ns else random.randint(OPEN_MIN, OPEN_MAX)
|
||||
end_n = open_ns[-1] if open_ns else random.randint(OPEN_MIN, OPEN_MAX)
|
||||
max_i = N - max(end_n, blink_n)
|
||||
|
||||
cur_i = start_n
|
||||
cur_n_i = 1
|
||||
|
||||
# Double Blink Probability
|
||||
# double_blink_prob = 0.15
|
||||
double_blink_prob = 0 # Disabled by user request
|
||||
|
||||
# Hold Duration (Frames to keep eye closed)
|
||||
blink_hold_min = 1
|
||||
blink_hold_max = 1
|
||||
|
||||
while cur_i < max_i:
|
||||
# [v1.9.101] Strict Speech-Only Validation
|
||||
# Check if current cur_i is valid for a blink
|
||||
if speech_only_blink and vad_timeline is not None:
|
||||
# If current position is silence, find next segment
|
||||
if vad_timeline[cur_i] <= 0.1:
|
||||
scan_start = cur_i
|
||||
while scan_start < len(vad_timeline) and vad_timeline[scan_start] <= 0.1:
|
||||
scan_start += 1
|
||||
|
||||
if scan_start >= len(vad_timeline): break
|
||||
|
||||
# New segment found
|
||||
scan_end = scan_start
|
||||
while scan_end < len(vad_timeline) and vad_timeline[scan_end] > 0.1:
|
||||
scan_end += 1
|
||||
|
||||
# Force blink into this segment
|
||||
if (scan_end - scan_start) >= blink_n:
|
||||
cur_i = random.randint(scan_start, scan_end - blink_n)
|
||||
else:
|
||||
cur_i = scan_end
|
||||
continue # Too short, try next
|
||||
else:
|
||||
# Already in speech, check if it fits before segment ends
|
||||
scan_end = cur_i
|
||||
while scan_end < len(vad_timeline) and vad_timeline[scan_end] > 0.1:
|
||||
scan_end += 1
|
||||
|
||||
if cur_i + blink_n > scan_end:
|
||||
# Not enough room here, jump to end and find next segment
|
||||
cur_i = scan_end
|
||||
continue
|
||||
idx[cur_i : cur_i + blink_n] = blink_idx
|
||||
|
||||
# First Blink
|
||||
hold_frames = random.randint(blink_hold_min, blink_hold_max)
|
||||
|
||||
# Construct Blink Sequence with Hold
|
||||
# split blink_idx into open->close (first half) and close->open (second half)
|
||||
mid_point = blink_n // 2
|
||||
# e.g. 0...7 (8 frames) + 7...7 (hold) + 8...14 (7 frames)
|
||||
|
||||
# Safe slice
|
||||
seq_start = blink_idx[:mid_point]
|
||||
seq_mid = [blink_idx[mid_point]] * (hold_frames + 1) # +1 to include original
|
||||
seq_end = blink_idx[mid_point+1:]
|
||||
|
||||
full_blink_seq = seq_start + seq_mid + seq_end
|
||||
full_len = len(full_blink_seq)
|
||||
|
||||
if cur_i + full_len > len(idx):
|
||||
break
|
||||
|
||||
idx[cur_i : cur_i + full_len] = full_blink_seq
|
||||
|
||||
# [AIIA Diagnostic] Log each blink event
|
||||
vad_val = vad_timeline[cur_i] if vad_timeline is not None else -1
|
||||
print(f" [Blink] Scheduled at frame {cur_i}, VAD={vad_val:.2f}, Type={'Speech' if vad_val > 0.1 else 'Silence'}")
|
||||
|
||||
# Decide if Double Blink
|
||||
is_double = random.random() < double_blink_prob
|
||||
|
||||
# Advance current index
|
||||
cur_i += full_len
|
||||
|
||||
if is_double:
|
||||
# Short gap: 4-8 frames (Clearer re-blink)
|
||||
short_gap = random.randint(4, 8)
|
||||
cur_i += short_gap
|
||||
|
||||
# Second Blink
|
||||
hold_frames_2 = random.randint(0, 2) # Shorter hold for 2nd blink
|
||||
|
||||
seq_mid_2 = [blink_idx[mid_point]] * (hold_frames_2 + 1)
|
||||
full_blink_seq_2 = seq_start + seq_mid_2 + seq_end
|
||||
full_len_2 = len(full_blink_seq_2)
|
||||
|
||||
if cur_i + full_len_2 > len(idx):
|
||||
break # Safety break
|
||||
|
||||
idx[cur_i : cur_i + full_len_2] = full_blink_seq_2
|
||||
cur_i += full_len_2
|
||||
|
||||
# Calculate Next Event
|
||||
if open_ns:
|
||||
cur_n = open_ns[cur_n_i % len(open_ns)]
|
||||
cur_n_i += 1
|
||||
else:
|
||||
if vad_timeline is not None and not speech_only_blink:
|
||||
# Natural Mode (Not speech-only)
|
||||
is_speaking = vad_timeline[cur_i] > 0.1 if cur_i < len(vad_timeline) else False
|
||||
if is_speaking:
|
||||
cur_n = random.randint(OPEN_MIN, OPEN_MAX)
|
||||
print(f" [Blink Loop] Natural: Speaking at {cur_i}, next in {cur_n}")
|
||||
else:
|
||||
cur_n = random.randint(int(OPEN_MIN * 1.5), int(OPEN_MAX * 1.5))
|
||||
print(f" [Blink Loop] Natural: Silence at {cur_i}, next in {cur_n} (1.5x Relaxed)")
|
||||
else:
|
||||
# Standard or Speech-Only prep (Speech-Only will re-validate at loop top)
|
||||
cur_n = random.randint(OPEN_MIN, OPEN_MAX)
|
||||
if speech_only_blink:
|
||||
print(f" [Blink Loop] Speech-Only Pre-Interval: +{cur_n}")
|
||||
cur_n = random.randint(OPEN_MIN, OPEN_MAX)
|
||||
|
||||
cur_i += cur_n
|
||||
cur_i = cur_i + blink_n + cur_n
|
||||
|
||||
return idx
|
||||
|
||||
|
||||
def _fix_exp_for_x_d_info(x_d_info, x_s_info, delta_eye=None, drive_eye=True):
|
||||
_eye = [11, 13, 15, 16, 18] # [Restored v1.9.114] Restore 15, 16, 18 to fix Left-Eye blink.
|
||||
_lip = [6, 7, 8, 12, 14, 17, 19, 20] # [Restored v1.9.114] Restore 17, 19, 20 to fix Stuck Mouth.
|
||||
# Note: Corners (7, 8) and Side-Lips (6, 12, 14) stay active for "Breathing" and "Closure Harmony".
|
||||
|
||||
_eye = [11, 13, 15, 16, 18]
|
||||
_lip = [6, 12, 14, 17, 19, 20]
|
||||
alpha = np.zeros((21, 3), dtype=x_d_info["exp"].dtype)
|
||||
alpha[_lip] = 1
|
||||
if delta_eye is None and drive_eye: # use d eye
|
||||
@@ -346,7 +190,7 @@ def _eye_delta(exp, dx=0, dy=0):
|
||||
def _fix_gaze(pose_s, x_d_info):
|
||||
x_ratio = 0.26
|
||||
y_ratio = 0.28
|
||||
|
||||
|
||||
yaw_s, pitch_s = pose_s
|
||||
yaw_d = bin66_to_degree(x_d_info['yaw']).item()
|
||||
pitch_d = bin66_to_degree(x_d_info['pitch']).item()
|
||||
@@ -356,7 +200,7 @@ def _fix_gaze(pose_s, x_d_info):
|
||||
|
||||
dx = delta_yaw * x_ratio
|
||||
dy = delta_pitch * y_ratio
|
||||
|
||||
|
||||
x_d_info['exp'] = _eye_delta(x_d_info['exp'], dx, dy)
|
||||
return x_d_info
|
||||
|
||||
@@ -446,16 +290,12 @@ class MotionStitch:
|
||||
# only for offline (make start|end eye open)
|
||||
if N_d == self.N_d:
|
||||
return
|
||||
|
||||
|
||||
self.N_d = N_d
|
||||
if self.drive_eye and self.delta_eye_arr is not None:
|
||||
N = 3000 if self.N_d == -1 else self.N_d
|
||||
self.delta_eye_idx_list = _set_eye_blink_idx(
|
||||
N, len(self.delta_eye_arr), self.delta_eye_open_n,
|
||||
interval_min=getattr(self, 'blink_interval_min', 60),
|
||||
interval_max=getattr(self, 'blink_interval_max', 100),
|
||||
vad_timeline=self.vad_timeline if hasattr(self, 'vad_timeline') else None,
|
||||
speech_only_blink=getattr(self, 'speech_only_blink', False)
|
||||
N, len(self.delta_eye_arr), self.delta_eye_open_n
|
||||
)
|
||||
|
||||
def setup(
|
||||
@@ -466,8 +306,6 @@ class MotionStitch:
|
||||
drive_eye=None, # use d eye or s eye
|
||||
delta_eye_arr=None, # fix eye
|
||||
delta_eye_open_n=-1, # int|list
|
||||
blink_interval_min=60,
|
||||
blink_interval_max=100,
|
||||
fade_out_keys=("exp",),
|
||||
fade_type="", # "" | "d0" | "s"
|
||||
flag_stitching=True,
|
||||
@@ -476,15 +314,7 @@ class MotionStitch:
|
||||
d0=None,
|
||||
ch_info=None,
|
||||
overall_ctrl_info=None,
|
||||
vad_timeline=None,
|
||||
blink_amp=1.0,
|
||||
speech_only_blink=False,
|
||||
):
|
||||
self.blink_amp = blink_amp
|
||||
self.speech_only_blink = speech_only_blink
|
||||
self.blink_interval_min = blink_interval_min
|
||||
self.blink_interval_max = blink_interval_max
|
||||
self.vad_timeline = vad_timeline
|
||||
self.is_image_flag = is_image_flag
|
||||
if use_d_keys is None:
|
||||
if self.is_image_flag:
|
||||
@@ -510,11 +340,8 @@ class MotionStitch:
|
||||
self.fade_type = fade_type
|
||||
self.flag_stitching = flag_stitching
|
||||
|
||||
_eye = [11, 13, 15, 16, 18] # [Native v1.9.94] Restored full official eye indices.
|
||||
|
||||
|
||||
_lip = [6, 12, 14, 17, 19, 20] # [Restored v1.9.66] Re-enable Talking.
|
||||
|
||||
_eye = [11, 13, 15, 16, 18]
|
||||
_lip = [6, 12, 14, 17, 19, 20]
|
||||
_a1 = np.zeros((21, 3), dtype=np.float32)
|
||||
_a1[_lip] = 1
|
||||
_a2 = 0
|
||||
@@ -530,21 +357,11 @@ class MotionStitch:
|
||||
self.fix_exp_a1 = _a1 * (1 - _a2)
|
||||
self.fix_exp_a2 = (1 - _a1) + _a1 * _a2
|
||||
self.fix_exp_a3 = _a2
|
||||
|
||||
# [Debug v1.9.72] Verify Code Sync
|
||||
print(f"[AIIA Debug] MotionStitch Setup: v1.9.72. SHAPE DEBUG MODE.")
|
||||
|
||||
|
||||
if self.drive_eye and self.delta_eye_arr is not None:
|
||||
N = 3000 if self.N_d == -1 else self.N_d
|
||||
self.delta_eye_idx_list = _set_eye_blink_idx(
|
||||
N,
|
||||
len(self.delta_eye_arr),
|
||||
self.delta_eye_open_n,
|
||||
interval_min=blink_interval_min,
|
||||
interval_max=blink_interval_max,
|
||||
vad_timeline=self.vad_timeline,
|
||||
speech_only_blink=self.speech_only_blink
|
||||
N, len(self.delta_eye_arr), self.delta_eye_open_n
|
||||
)
|
||||
|
||||
self.pose_s = None
|
||||
@@ -587,12 +404,12 @@ class MotionStitch:
|
||||
def _merge_kwargs(default_kwargs, run_kwargs):
|
||||
if default_kwargs is None:
|
||||
return run_kwargs
|
||||
|
||||
|
||||
for k, v in default_kwargs.items():
|
||||
if k not in run_kwargs:
|
||||
run_kwargs[k] = v
|
||||
return run_kwargs
|
||||
|
||||
|
||||
def __call__(self, x_s_info, x_d_info, **kwargs):
|
||||
# return x_s, x_d
|
||||
|
||||
@@ -613,45 +430,24 @@ class MotionStitch:
|
||||
self.d0,
|
||||
)
|
||||
|
||||
# [v1.9.128] ABSOLUTE MINIMALIST PURGE
|
||||
# Deleted: exp_buffer, last_speaking_exp, EMA scrubbing, attack/release logic,
|
||||
# mouth biases, and manual transition guards.
|
||||
|
||||
vad_alpha = kwargs.get("vad_alpha", 1.0)
|
||||
|
||||
# 1. Simple Linear Blend (Landmark Parity)
|
||||
# We blend strictly based on volume. No history, no hidden overrides.
|
||||
for k in ["exp", "kp"]:
|
||||
if k in x_d_info and k in x_s_info:
|
||||
x_d_info[k] = x_d_info[k] * vad_alpha + x_s_info[k] * (1.0 - vad_alpha)
|
||||
|
||||
# 2. Eye Blink selection (Native)
|
||||
delta_eye = 0
|
||||
if self.drive_eye and self.delta_eye_arr is not None:
|
||||
delta_eye = self.delta_eye_arr[
|
||||
self.delta_eye_idx_list[self.idx % len(self.delta_eye_idx_list)]
|
||||
][None] * self.blink_amp
|
||||
|
||||
# [v1.9.90] Removed all delta_eye manipulation (Clean Slate, Motion Hold).
|
||||
# delta_eye is passed through unmodified to _fix_exp_for_x_d_info.
|
||||
# Post-processing mirror (Right Eye -> Left Eye) is applied AFTER _fix_exp.
|
||||
][None]
|
||||
x_d_info = _fix_exp_for_x_d_info_v2(
|
||||
x_d_info,
|
||||
x_s_info,
|
||||
delta_eye,
|
||||
self.fix_exp_a1,
|
||||
self.fix_exp_a2,
|
||||
self.fix_exp_a3,
|
||||
)
|
||||
|
||||
if kwargs.get("vad_alpha", 1) < 1:
|
||||
x_d_info = ctrl_vad(x_d_info, x_s_info, kwargs.get("vad_alpha", 1))
|
||||
|
||||
|
||||
|
||||
# [v1.9.127] ALL SHIELDS AND LOCKDOWNS REMOVED.
|
||||
# Returning to 100% Raw AI predictions for facial landmarks.
|
||||
# [Diagnostic v1.9.71] Output Logger
|
||||
if hasattr(self, '_log_blink_output') and self._log_blink_output:
|
||||
out_exp = x_d_info["exp"]
|
||||
# Log Eyelid, Squint, and Mouth Corners (6/12)
|
||||
print(f" > Final Output: EyeL={float(out_exp[:, 11].mean()):.3f}, SquintL={float(out_exp[:, 15].mean()):.3f}, CornerL={float(out_exp[:, 6].mean()):.3f}, CornerR={float(out_exp[:, 12].mean()):.3f}")
|
||||
|
||||
|
||||
|
||||
|
||||
# [Moved v1.9.59] Auto-Center Moved to end of function.
|
||||
|
||||
x_d_info = ctrl_motion(x_d_info, **kwargs)
|
||||
|
||||
if self.fade_type == "d0" and self.fade_dst is None:
|
||||
self.fade_dst = copy.deepcopy(x_d_info)
|
||||
@@ -678,59 +474,15 @@ class MotionStitch:
|
||||
self.pose_s = [yaw_s, pitch_s]
|
||||
x_d_info = _fix_gaze(self.pose_s, x_d_info)
|
||||
|
||||
# [Feature v1.9.59] Auto-Center Roll (Anti-Tilt) + Hard Clamp (Late Stage)
|
||||
# Applied at the very end of motion calculation (before transform) to ensure it sticks.
|
||||
ref_roll = bin66_to_degree(x_s_info['roll'])
|
||||
|
||||
# Spring Force (30%)
|
||||
x_d_info['roll'] = x_d_info['roll'] * 0.7 + ref_roll * 0.3
|
||||
|
||||
# Hard Clamp (Max 1.5 deg drift)
|
||||
drift = x_d_info['roll'] - ref_roll
|
||||
# Fix dimensions
|
||||
if isinstance(drift, np.ndarray):
|
||||
drift_val = drift.item() if drift.size == 1 else drift.mean()
|
||||
else:
|
||||
drift_val = drift
|
||||
|
||||
if drift_val > 1.5:
|
||||
x_d_info['roll'] = ref_roll + 1.5
|
||||
elif drift_val < -1.5:
|
||||
x_d_info['roll'] = ref_roll - 1.5
|
||||
|
||||
# Diagnostic Logging (Supressed v1.9.78 for Scanner Clarity)
|
||||
# if self.idx % 25 == 0:
|
||||
# c_roll = float(np.mean(x_d_info['roll']))
|
||||
# r_roll = float(np.mean(ref_roll))
|
||||
# print(f"[Ditto Tilt Diag] Frame {self.idx}: CurRoll={c_roll:.2f}° | RefRoll={r_roll:.2f}° | Drift={c_roll-r_roll:.2f}°")
|
||||
|
||||
if self.x_s is not None:
|
||||
x_s = self.x_s
|
||||
else:
|
||||
x_s = transform_keypoint(x_s_info)
|
||||
if self.is_image_flag:
|
||||
self.x_s = x_s
|
||||
|
||||
# [v1.9.126] PIPE END VITALITY APPLICATION
|
||||
# We apply all procedural and legacy fixes here to ensure
|
||||
# they are never overwritten by transition logic.
|
||||
|
||||
# 1. Base Eyelid Fix [DISABLED v1.9.127 for Raw Test]
|
||||
# x_d_info = _fix_exp_for_x_d_info(x_d_info, x_s_info, delta_eye, self.drive_eye)
|
||||
|
||||
# 2. Breathing Mouth (delta_mouth)
|
||||
if "delta_mouth" in kwargs:
|
||||
_lip = [6, 12, 14, 17, 19, 20]
|
||||
exp_reshaped = x_d_info["exp"].reshape(-1, 21, 3)
|
||||
exp_reshaped[:, _lip] += kwargs["delta_mouth"]
|
||||
x_d_info["exp"] = exp_reshaped.reshape(1, -1)
|
||||
|
||||
# 3. Head Sway & Poses (ctrl_motion)
|
||||
x_d_info = ctrl_motion(x_d_info, **kwargs)
|
||||
|
||||
x_d = transform_keypoint(x_d_info)
|
||||
|
||||
# Re-apply Stitch Net
|
||||
|
||||
if self.flag_stitching:
|
||||
x_d = self.stitch_net(x_s, x_d)
|
||||
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
[project]
|
||||
name = "aiia"
|
||||
description = "The Ultimate AI Audio/Video toolkit for ComfyUI. Features an enhanced Ditto (with optimizations that outperform official demos and other SOTA talking head models in lip-sync accuracy and natural motion), EchoMimic V3 & FLOAT, VibeVoice & CosyVoice 3.0 (Zero-Shot Voice Cloning), Multi-Role Podcast Generation, and a powerful Media Browser."
|
||||
version = "1.9.128"
|
||||
version = "1.9.129"
|
||||
license = {file = "LICENSE"}
|
||||
readme = "README.md"
|
||||
authors = [
|
||||
|
||||
Reference in New Issue
Block a user