nodes/diodiogod-TTS-Audio-Suite
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
codex/audio-cpp-integration
diodiogod-TTS-Audio-Suite/tests/diagnostics/audio_sample_generator.py
T

284 lines
10 KiB
Python
Raw Permalink Blame History

"""
CosyVoice3 Audio Sample Generator
Generates audio samples with different parameter combinations
to help identify quality issues and optimal settings.
Usage:
python tests/diagnostics/audio_sample_generator.py [--server-url URL]
Outputs:
tests/output/audio_samples/
├── cosyvoice_zero_shot_speed1.0.wav
├── cosyvoice_instruct_gentle.wav
├── cosyvoice_cross_lingual.wav
└── generation_log.csv
"""
import os
import sys
import json
import csv
import time
import argparse
from pathlib import Path
from datetime import datetime
# Add project root to path
PROJECT_ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
try:
import requests
except ImportError:
print("❌ requests not installed. Run: pip install requests")
sys.exit(1)
class AudioSampleGenerator:
"""Generates audio samples via ComfyUI API for parameter testing."""
# Voice sample to use for testing
DEFAULT_VOICE = "voices_examples/Sophie_Anderson CC3.wav"
# Test text samples (kept short for faster generation)
TEST_TEXTS = {
"english_short": "Hello, this is a test of the text to speech system.",
"english_medium": "The quick brown fox jumps over the lazy dog. This is a pangram that contains every letter of the alphabet.",
"chinese_short": "你好，这是一个测试。",
}
# Parameter combinations to test
TEST_CASES = [
# Mode tests
{"name": "zero_shot_default", "mode": "zero_shot", "speed": 1.0, "text": "english_short"},
{"name": "zero_shot_slow", "mode": "zero_shot", "speed": 0.7, "text": "english_short"},
{"name": "zero_shot_fast", "mode": "zero_shot", "speed": 1.3, "text": "english_short"},
{"name": "zero_shot_medium_text", "mode": "zero_shot", "speed": 1.0, "text": "english_medium"},
# Cross-lingual mode (no reference text needed)
{"name": "cross_lingual_english", "mode": "cross_lingual", "speed": 1.0, "text": "english_short"},
# Instruct mode tests (Chinese instructions)
{"name": "instruct_gentle", "mode": "instruct", "speed": 1.0, "text": "english_short",
"instruct_text": "请用温柔的语气说。"},
{"name": "instruct_fast", "mode": "instruct", "speed": 1.0, "text": "english_short",
"instruct_text": "请用尽可能快地语速说一句话。"},
]
def __init__(self, server_url: str = "http://127.0.0.1:8188"):
self.server_url = server_url
self.output_dir = PROJECT_ROOT / "tests" / "output" / "audio_samples"
self.output_dir.mkdir(parents=True, exist_ok=True)
self.results = []
def check_server(self) -> bool:
"""Check if ComfyUI server is running."""
try:
r = requests.get(f"{self.server_url}/system_stats", timeout=5)
return r.status_code == 200
except:
return False
def create_workflow(self, test_case: dict) -> dict:
"""Create workflow JSON for a test case."""
text = self.TEST_TEXTS[test_case["text"]]
mode = test_case["mode"]
speed = test_case.get("speed", 1.0)
instruct_text = test_case.get("instruct_text", "")
# Base workflow
workflow = {
"1": {
"class_type": "CosyVoiceEngineNode",
"inputs": {
"model_path": "local:Fun-CosyVoice3-0.5B",
"device": "auto",
"mode": mode,
"speed": speed,
"use_fp16": True,
}
},
"2": {
"class_type": "UnifiedTTSTextNode",
"inputs": {
"TTS_engine": ["1", 0],
"text": text,
"narrator_voice": self.DEFAULT_VOICE,
"seed": 42
}
},
"3": {
"class_type": "SaveAudio",
"inputs": {
"audio": ["2", 0],
"filename_prefix": f"cosyvoice_{test_case['name']}"
}
}
}
# Add optional parameters
if mode == "zero_shot":
# Reference text is loaded from .reference.txt by the system
workflow["1"]["inputs"]["reference_text"] = "" # Let system load from file
if mode == "instruct" and instruct_text:
workflow["1"]["inputs"]["instruct_text"] = instruct_text
return workflow
def execute_workflow(self, workflow: dict, timeout: int = 120) -> dict:
"""Execute workflow via ComfyUI API."""
# Queue the prompt
response = requests.post(
f"{self.server_url}/prompt",
json={"prompt": workflow},
timeout=10
)
if response.status_code != 200:
error = response.json().get("error", {})
return {"success": False, "error": error}
prompt_id = response.json().get("prompt_id")
if not prompt_id:
return {"success": False, "error": "No prompt_id returned"}
# Wait for completion
start_time = time.time()
while time.time() - start_time < timeout:
history = requests.get(f"{self.server_url}/history/{prompt_id}", timeout=10)
if history.status_code == 200:
data = history.json()
if prompt_id in data:
prompt_data = data[prompt_id]
if prompt_data.get("status", {}).get("completed"):
return {"success": True, "prompt_id": prompt_id, "data": prompt_data}
if prompt_data.get("status", {}).get("status_str") == "error":
return {"success": False, "error": prompt_data.get("status", {}).get("messages", [])}
time.sleep(2)
return {"success": False, "error": "Timeout waiting for completion"}
def run_test_case(self, test_case: dict) -> dict:
"""Run a single test case and return results."""
name = test_case["name"]
print(f"\n🎵 Testing: {name}")
print(f" Mode: {test_case['mode']}, Speed: {test_case.get('speed', 1.0)}")
start_time = time.time()
try:
workflow = self.create_workflow(test_case)
result = self.execute_workflow(workflow)
elapsed = time.time() - start_time
if result["success"]:
print(f" ✅ Generated in {elapsed:.1f}s")
return {
"name": name,
"success": True,
"duration_sec": elapsed,
"mode": test_case["mode"],
"speed": test_case.get("speed", 1.0),
"text": test_case["text"],
"error": ""
}
else:
print(f" ❌ Failed: {result.get('error', 'Unknown error')}")
return {
"name": name,
"success": False,
"duration_sec": elapsed,
"mode": test_case["mode"],
"speed": test_case.get("speed", 1.0),
"text": test_case["text"],
"error": str(result.get("error", "Unknown"))
}
except Exception as e:
elapsed = time.time() - start_time
print(f" ❌ Exception: {e}")
return {
"name": name,
"success": False,
"duration_sec": elapsed,
"mode": test_case["mode"],
"speed": test_case.get("speed", 1.0),
"text": test_case["text"],
"error": str(e)
}
def run_all_tests(self):
"""Run all test cases and save results."""
print("=" * 60)
print("CosyVoice3 Audio Sample Generator")
print("=" * 60)
if not self.check_server():
print(f"❌ ComfyUI server not running at {self.server_url}")
print(" Start ComfyUI first, then run this script again.")
return False
print(f"✅ Connected to ComfyUI at {self.server_url}")
print(f"📁 Output directory: {self.output_dir}")
print(f"🎤 Voice sample: {self.DEFAULT_VOICE}")
print(f"\n🧪 Running {len(self.TEST_CASES)} test cases...\n")
for test_case in self.TEST_CASES:
result = self.run_test_case(test_case)
self.results.append(result)
# Save results to CSV
self.save_results()
# Print summary
self.print_summary()
return True
def save_results(self):
"""Save results to CSV file."""
csv_path = self.output_dir / "generation_log.csv"
with open(csv_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=[
"name", "success", "duration_sec", "mode", "speed", "text", "error", "quality_rating"
])
writer.writeheader()
for result in self.results:
result["quality_rating"] = "" # User fills this in
writer.writerow(result)
print(f"\n📝 Results saved to: {csv_path}")
def print_summary(self):
"""Print test summary."""
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
successful = sum(1 for r in self.results if r["success"])
failed = len(self.results) - successful
print(f"✅ Successful: {successful}")
print(f"❌ Failed: {failed}")
print(f"\n📂 Audio files saved to: {self.output_dir}")
print(f" (Look in ComfyUI output folder for SaveAudio outputs)")
print(f"\n📝 Review log: {self.output_dir / 'generation_log.csv'}")
print("\n👂 Please listen to each audio file and rate quality in the CSV!")
def main():
parser = argparse.ArgumentParser(description="Generate CosyVoice3 audio samples")
parser.add_argument("--server-url", default="http://127.0.0.1:8188", help="ComfyUI server URL")
args = parser.parse_args()
generator = AudioSampleGenerator(server_url=args.server_url)
generator.run_all_tests()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink