nodes/havvk-ComfyUI_AIIA
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: Introduce Audio Smart Chunker for optimized long audio splitting and CosyVoice integration

Browse Source
This commit is contained in:
Hawk Lee
2025-12-26 22:59:05 +08:00
parent 5f5063d57b
commit 633aaf44db
3 changed files with 136 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
__init__.py
+3
View File
@@ -64,6 +64,9 @@ _load_nodes_from_module(".aiia_audio_merger", "aiia_audio_merger")
# 10. 处理 aiia_cosyvoice_nodes.py
_load_nodes_from_module(".aiia_cosyvoice_nodes", "aiia_cosyvoice_nodes")
# 11. 处理 aiia_audio_processor.py
_load_nodes_from_module(".aiia_audio_processor", "aiia_audio_processor")
# 告诉 ComfyUI 这个节点包有一个包含网页资源的 'js' 目录
WEB_DIRECTORY = "js"
aiia_audio_processor.py
+93
View File
@@ -0,0 +1,93 @@
import torch
import numpy as np
class AIIA_Audio_Silence_Splitter:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio": ("AUDIO",),
"max_duration": ("FLOAT", {"default": 27.0, "min": 5.0, "max": 29.0, "step": 0.1, "tooltip": "每个片段的最大长度，建议设置为27秒以留出余量"}),
"silence_threshold": ("FLOAT", {"default": 0.001, "min": 0.0001, "max": 0.1, "step": 0.0001, "tooltip": "判定为静音的能量阈值"}),
"min_silence_duration": ("FLOAT", {"default": 0.3, "min": 0.05, "max": 2.0, "step": 0.05, "tooltip": "判定为有效停顿的最短静音时长"}),
}
}
RETURN_TYPES = ("WHISPER_CHUNKS", "INT")
RETURN_NAMES = ("whisper_chunks", "chunk_count")
FUNCTION = "split_audio"
CATEGORY = "AIIA/audio"
def split_audio(self, audio, max_duration, silence_threshold, min_silence_duration):
waveform = audio["waveform"]
sample_rate = audio["sample_rate"]
# 计算单通道能量
energy = torch.abs(waveform).mean(dim=1).squeeze()
if energy.ndim > 1: energy = energy.mean(dim=0) # 处理 batch
# 1. 识别静音区间
is_silence = energy < silence_threshold
# 转换为 numpy 处理更方便
sil_np = is_silence.cpu().numpy()
# 寻找静音区间的起始和结束索引
diff = np.diff(sil_np.astype(int))
starts = np.where(diff == 1)[0]
ends = np.where(diff == -1)[0]
if sil_np[0]: starts = np.insert(starts, 0, 0)
if sil_np[-1]: ends = np.append(ends, len(sil_np) - 1)
# 过滤过短的静音
min_sil_samples = int(min_sil_duration * sample_rate)
valid_silence_gaps = []
for s, e in zip(starts, ends):
if e - s >= min_sil_samples:
valid_silence_gaps.append((s / sample_rate, e / sample_rate))
# 2. 贪心算法构建片段
total_duration = waveform.shape[-1] / sample_rate
chunks = []
current_start = 0.0
while current_start < total_duration:
target_end = current_start + max_duration
if target_end >= total_duration:
chunks.append([current_start, total_duration])
break
# 在 [current_start, target_end] 范围内寻找最后一个有效的静音点
best_gap_mid = -1
for g_start, g_end in valid_silence_gaps:
g_mid = (g_start + g_end) / 2
if current_start < g_mid <= target_end:
best_gap_mid = g_mid
elif g_mid > target_end:
break
if best_gap_mid != -1:
chunks.append([current_start, best_gap_mid])
current_start = best_gap_mid
else:
# 如果没找到静音点，强行截断并打印警告
print(f"警告: 在 {current_start:.1f}s 后的 {max_duration}s 内未找到静音点，强行截断。")
chunks.append([current_start, target_end])
current_start = target_end
# 3. 构造输出
whisper_chunks_data = {
"text": "",
"chunks": [
{"timestamp": [round(c[0], 3), round(c[1], 3)], "text": f"Chunk {i}", "speaker": "AIIA_SMART_CHUNK"}
for i, c in enumerate(chunks)
],
"language": ""
}
print(f"--- [AIIA Smart Splitter] 已将 {total_duration:.1f}s 音频划分为 {len(chunks)} 个片段 ---")
return (whisper_chunks_data, len(chunks))
NODE_CLASS_MAPPINGS = {"AIIA_Audio_Silence_Splitter": AIIA_Audio_Silence_Splitter}
NODE_DISPLAY_NAME_MAPPINGS = {"AIIA_Audio_Silence_Splitter": "Audio Smart Chunker (Silence-based)"}
aiia_cosyvoice_nodes.py
+40 -63
View File
@@ -48,12 +48,10 @@ class AIIA_CosyVoice_VoiceConversion:
cosyvoice_model = model["model"]
sample_rate = cosyvoice_model.sample_rate
# 1. 准备参考音频
target_waveform = target_audio["waveform"]
if target_audio["sample_rate"] != sample_rate:
import torchaudio
resampler = torchaudio.transforms.Resample(target_audio["sample_rate"], sample_rate)
target_waveform = resampler(target_waveform)
target_waveform = torchaudio.transforms.Resample(target_audio["sample_rate"], sample_rate)(target_waveform)
max_target_samples = 30 * sample_rate
if target_waveform.shape[-1] > max_target_samples:
@@ -65,20 +63,16 @@ class AIIA_CosyVoice_VoiceConversion:
if target_np.ndim == 2: target_np = target_np.T
sf.write(target_path, target_np, sample_rate)
# 2. 准备源音频
source_waveform = source_audio["waveform"]
if source_audio["sample_rate"] != sample_rate:
import torchaudio
resampler = torchaudio.transforms.Resample(source_audio["sample_rate"], sample_rate)
source_waveform = resampler(source_waveform)
source_waveform = torchaudio.transforms.Resample(source_audio["sample_rate"], sample_rate)(source_waveform)
source_waveform = source_waveform.squeeze()
if source_waveform.ndim == 1: source_waveform = source_waveform.unsqueeze(0)
total_samples = source_waveform.shape[-1]
# 硬限制计算：核心片段 + 重叠部分 < 30秒
MAX_TOTAL_SEC = 29.8 # 留一点点 Buffer
MAX_TOTAL_SEC = 29.8
chunk_samples = chunk_size * sample_rate
overlap_samples = overlap_size * sample_rate
max_total_samples = int(MAX_TOTAL_SEC * sample_rate)
@@ -88,79 +82,62 @@ class AIIA_CosyVoice_VoiceConversion:
if os.path.exists(target_path): os.unlink(target_path)
return ({"waveform": result_waveform.unsqueeze(0), "sample_rate": sample_rate},)
print(f"[AIIA CosyVoice] Long audio detected. Max allowed total chunk len: {MAX_TOTAL_SEC}s")
# 3. 智能分块逻辑
chunks_to_process = []
current_start = 0
search_range = 2 * sample_rate
is_pre_chunked = whisper_chunks and any(c.get("speaker") == "AIIA_SMART_CHUNK" for c in whisper_chunks.get("chunks", []))
while current_start < total_samples:
# 核心目标终点
target_end_time = (current_start + chunk_samples) / sample_rate
best_split_time = target_end_time
# 语义优先：寻找最接近的缝隙
if whisper_chunks and "chunks" in whisper_chunks:
closest_gap_dist = float('inf')
for i in range(len(whisper_chunks["chunks"]) - 1):
gap_start = whisper_chunks["chunks"][i]["timestamp"][1]
gap_end = whisper_chunks["chunks"][i+1]["timestamp"][0]
gap_mid = (gap_start + gap_end) / 2
# 关键修复：确保 (split_point + overlap) 不超过 30s 限制
if (gap_mid * sample_rate + overlap_samples) - current_start > max_total_samples:
continue
dist = abs(gap_mid - target_end_time)
if dist < 5: # 5秒内
if dist < closest_gap_dist:
if is_pre_chunked:
print(f"[AIIA CosyVoice] Detected Smart Chunker input. Following pre-defined segments strictly.")
for c in whisper_chunks["chunks"]:
s_time, e_time = c["timestamp"]
s_idx = int(s_time * sample_rate)
e_idx = int(e_time * sample_rate)
e_idx_with_overlap = min(e_idx + overlap_samples, total_samples)
chunks_to_process.append(source_waveform[:, s_idx:e_idx_with_overlap])
else:
current_start = 0
search_range = 2 * sample_rate
while current_start < total_samples:
target_end_time = (current_start + chunk_samples) / sample_rate
best_split_time = target_end_time
if whisper_chunks and "chunks" in whisper_chunks:
closest_gap_dist = float('inf')
for i in range(len(whisper_chunks["chunks"]) - 1):
gap_mid = (whisper_chunks["chunks"][i]["timestamp"][1] + whisper_chunks["chunks"][i+1]["timestamp"][0]) / 2
if (gap_mid * sample_rate + overlap_samples) - current_start > max_total_samples: continue
dist = abs(gap_mid - target_end_time)
if dist < 5 and dist < closest_gap_dist:
closest_gap_dist = dist
best_split_time = gap_mid
split_point = int(best_split_time * sample_rate)
# 物理微调：静音探测
split_point = self._find_best_split_point(source_waveform, split_point, search_range)
# 严格限制：确保不越界且总长度（含重叠）不超过 30s
split_point = min(split_point, total_samples)
if (split_point + overlap_samples) - current_start > max_total_samples:
split_point = current_start + max_total_samples - overlap_samples - 100
actual_end = min(split_point + overlap_samples, total_samples)
# 检查是否是最后一块
if actual_end >= total_samples:
chunks_to_process.append(source_waveform[:, current_start:])
break
else:
chunks_to_process.append(source_waveform[:, current_start:actual_end])
current_start = split_point
# 如果剩余长度太短（小于1秒），合并到最后一块
if total_samples - current_start < sample_rate:
chunks_to_process[-1] = source_waveform[:, total_samples - chunks_to_process[-1].shape[-1] - (total_samples - current_start): ]
break
split_point = int(best_split_time * sample_rate)
split_point = self._find_best_split_point(source_waveform, split_point, search_range)
split_point = min(split_point, total_samples)
if (split_point + overlap_samples) - current_start > max_total_samples:
split_point = current_start + max_total_samples - overlap_samples - 100
actual_end = min(split_point + overlap_samples, total_samples)
if actual_end >= total_samples:
chunks_to_process.append(source_waveform[:, current_start:])
break
else:
chunks_to_process.append(source_waveform[:, current_start:actual_end])
current_start = split_point
if total_samples - current_start < sample_rate:
break
final_segments = []
for i, chunk in enumerate(chunks_to_process):
print(f"[AIIA CosyVoice] Processing chunk {i+1}/{len(chunks_to_process)}, actual input len: {chunk.shape[-1]/sample_rate:.2f}s")
converted_chunk = self._inference_single_chunk(cosyvoice_model, chunk, target_path, speed, sample_rate, seed)
if not final_segments:
final_segments.append(converted_chunk)
else:
prev_chunk = final_segments[-1]
chunk_overlap_samples = int(overlap_samples / speed)
if chunk_overlap_samples > 0 and prev_chunk.shape[-1] > chunk_overlap_samples:
t = torch.linspace(0, np.pi, chunk_overlap_samples, device=converted_chunk.device)
fade_out = 0.5 * (1.0 + torch.cos(t))
fade_in = 1.0 - fade_out
overlap_part = prev_chunk[:, -chunk_overlap_samples:] * fade_out + converted_chunk[:, :chunk_overlap_samples] * fade_in
final_segments[-1] = prev_chunk[:, :-chunk_overlap_samples]
final_segments.append(overlap_part)
final_segments.append(converted_chunk[:, chunk_overlap_samples:])
@@ -189,4 +166,4 @@ class AIIA_CosyVoice_VoiceConversion:
if os.path.exists(source_path): os.unlink(source_path)
NODE_CLASS_MAPPINGS = {"AIIA_CosyVoice_VoiceConversion": AIIA_CosyVoice_VoiceConversion}
NODE_DISPLAY_NAME_MAPPINGS = {"AIIA_CosyVoice_VoiceConversion": "Voice Conversion (AIIA Unlimited)"}
NODE_DISPLAY_NAME_MAPPINGS = {"AIIA_CosyVoice_VoiceConversion": "Voice Conversion (AIIA Unlimited)"}