Add Step Audio EditX CosyVoice model validation Technical improvements: - Add CosyVoice folder and file existence checks - Validate all required CosyVoice files before loading - Add overrides parameter to load_hyperpyyaml for version compatibility - Improve error messages with clear fix instructions Address issue #196 - provide better diagnostics for model loading failures
248 lines
9.0 KiB
Python
248 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Fix relative imports to prevent shadowing by other custom nodes.
|
|
|
|
Issue #191: Other custom nodes with utils.py files shadow our utils/ directory.
|
|
Solution: Convert implicit relative imports to explicit relative imports using dots.
|
|
|
|
Example conversions:
|
|
nodes/engines/file.py: from utils.X import Y → from ...utils.X import Y
|
|
engines/adapters/file.py: from utils.X import Y → from ...utils.X import Y
|
|
utils/models/file.py: from utils.X import Y → from ..X import Y
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import List, Tuple, Dict
|
|
|
|
# Directories to exclude from conversion
|
|
EXCLUDE_DIRS = {
|
|
"engines/step_audio_editx/step_audio_editx_impl",
|
|
"engines/rvc/impl/lib",
|
|
"engines/higgs_audio/boson_multimodal",
|
|
"engines/chatterbox/models",
|
|
"engines/chatterbox_official_23lang/models",
|
|
"IgnoredForGitHubDocs",
|
|
"engines/index_tts/indextts",
|
|
}
|
|
|
|
class RelativeImportFixer:
|
|
def __init__(self, project_root: Path, dry_run: bool = True):
|
|
self.project_root = project_root
|
|
self.dry_run = dry_run
|
|
self.changes: List[Tuple[str, int, str, str]] = []
|
|
|
|
# Regex patterns for utils imports
|
|
self.from_import_pattern = re.compile(r'^(\s*)from utils\.([a-zA-Z0-9_.]+) import (.+)$')
|
|
self.direct_import_pattern = re.compile(r'^(\s*)import utils\.([a-zA-Z0-9_.]+)(.*)$')
|
|
|
|
def calculate_relative_depth(self, file_path: Path) -> int:
|
|
"""Calculate how many levels up to go to reach project root"""
|
|
relative_path = file_path.relative_to(self.project_root)
|
|
|
|
# Get the parent directory path
|
|
parent = relative_path.parent
|
|
|
|
# Count directory depth (how many folders deep from root)
|
|
# nodes/engines/file.py -> depth 2 -> need ...utils (3 dots)
|
|
# engines/adapters/file.py -> depth 2 -> need ...utils (3 dots)
|
|
# utils/models/file.py -> depth 2 -> need ..X (2 dots to go to utils/)
|
|
# nodes.py (root) -> depth 0 -> need .utils (1 dot)
|
|
|
|
depth = len(parent.parts)
|
|
|
|
# Special case: files in utils/ directory
|
|
if parent.parts and parent.parts[0] == 'utils':
|
|
# From utils/models/file.py importing utils.audio -> ..audio
|
|
# We're in utils/, so we just need .. to stay in utils/
|
|
return 1
|
|
|
|
# For all other files, we need to go up to root, then into utils/
|
|
# depth 0 (root): .utils
|
|
# depth 1 (nodes/): ..utils
|
|
# depth 2 (nodes/engines/): ...utils
|
|
return depth + 1
|
|
|
|
def convert_line(self, line: str, depth: int, is_utils_file: bool) -> Tuple[str, bool]:
|
|
"""Convert a single import line"""
|
|
# Handle "from utils.X import Y"
|
|
match = self.from_import_pattern.match(line)
|
|
if match:
|
|
indent, module_path, imports = match.groups()
|
|
|
|
if is_utils_file:
|
|
# From utils/models/file.py: from utils.audio import X -> from ..audio import X
|
|
dots = '.' * depth
|
|
new_line = f"{indent}from {dots}{module_path} import {imports}\n"
|
|
else:
|
|
# From other files: need to reach utils/ from root
|
|
dots = '.' * depth
|
|
new_line = f"{indent}from {dots}utils.{module_path} import {imports}\n"
|
|
|
|
return new_line, True
|
|
|
|
# Handle "import utils.X"
|
|
match = self.direct_import_pattern.match(line)
|
|
if match:
|
|
indent, module_path, rest = match.groups()
|
|
|
|
# Cannot easily convert "import utils.X" to relative form
|
|
# Would need to be "from .utils import X" which changes usage
|
|
# Skip these for now - they're rare
|
|
return line, False
|
|
|
|
return line, False
|
|
|
|
def should_skip_directory(self, file_path: Path) -> bool:
|
|
"""Check if file is in an excluded directory"""
|
|
relative_path = file_path.relative_to(self.project_root)
|
|
path_str = str(relative_path.parent).replace('\\', '/')
|
|
|
|
for exclude_dir in EXCLUDE_DIRS:
|
|
if path_str.startswith(exclude_dir) or exclude_dir in path_str:
|
|
return True
|
|
return False
|
|
|
|
def convert_file(self, file_path: Path) -> int:
|
|
"""Convert all imports in a single file. Returns number of changes."""
|
|
if self.should_skip_directory(file_path):
|
|
return 0
|
|
|
|
relative_path = file_path.relative_to(self.project_root)
|
|
|
|
# Check if file is in utils/ directory
|
|
is_utils_file = relative_path.parts and relative_path.parts[0] == 'utils'
|
|
|
|
# Calculate relative depth
|
|
depth = self.calculate_relative_depth(file_path)
|
|
|
|
try:
|
|
with open(file_path, 'r', encoding='utf-8') as f:
|
|
lines = f.readlines()
|
|
except Exception as e:
|
|
print(f"⚠️ Error reading {file_path}: {e}")
|
|
return 0
|
|
|
|
new_lines = []
|
|
changes_in_file = 0
|
|
|
|
for line_num, line in enumerate(lines, 1):
|
|
new_line, changed = self.convert_line(line, depth, is_utils_file)
|
|
new_lines.append(new_line)
|
|
|
|
if changed:
|
|
changes_in_file += 1
|
|
self.changes.append((str(relative_path), line_num, line.strip(), new_line.strip()))
|
|
|
|
# Write changes if not dry run
|
|
if changes_in_file > 0 and not self.dry_run:
|
|
try:
|
|
with open(file_path, 'w', encoding='utf-8') as f:
|
|
f.writelines(new_lines)
|
|
except Exception as e:
|
|
print(f"⚠️ Error writing {file_path}: {e}")
|
|
return 0
|
|
|
|
return changes_in_file
|
|
|
|
def find_python_files(self) -> List[Path]:
|
|
"""Find all Python files in the project"""
|
|
python_files = []
|
|
for root, dirs, files in os.walk(self.project_root):
|
|
dirs[:] = [d for d in dirs if d not in ('__pycache__', '.git', '.venv', 'venv')]
|
|
|
|
for file in files:
|
|
if file.endswith('.py'):
|
|
file_path = Path(root) / file
|
|
python_files.append(file_path)
|
|
|
|
return python_files
|
|
|
|
def run(self):
|
|
"""Run the conversion"""
|
|
print(f"{'='*80}")
|
|
print(f"Fixing relative imports to prevent utils shadowing")
|
|
print(f"Mode: {'DRY RUN (no files changed)' if self.dry_run else 'LIVE (files will be modified)'}")
|
|
print(f"{'='*80}\n")
|
|
|
|
python_files = self.find_python_files()
|
|
print(f"Found {len(python_files)} Python files\n")
|
|
|
|
total_changes = 0
|
|
files_changed = 0
|
|
|
|
for file_path in python_files:
|
|
changes = self.convert_file(file_path)
|
|
if changes > 0:
|
|
total_changes += changes
|
|
files_changed += 1
|
|
|
|
# Report results
|
|
print(f"\n{'='*80}")
|
|
print(f"CONVERSION SUMMARY")
|
|
print(f"{'='*80}")
|
|
print(f"Files scanned: {len(python_files)}")
|
|
print(f"Files with changes: {files_changed}")
|
|
print(f"Total import lines changed: {total_changes}")
|
|
print(f"{'='*80}\n")
|
|
|
|
if self.changes:
|
|
print("DETAILED CHANGES:\n")
|
|
|
|
# Group by file
|
|
changes_by_file: Dict[str, List[Tuple[int, str, str]]] = {}
|
|
for file_path, line_num, old_line, new_line in self.changes:
|
|
if file_path not in changes_by_file:
|
|
changes_by_file[file_path] = []
|
|
changes_by_file[file_path].append((line_num, old_line, new_line))
|
|
|
|
# Print first 20 files as examples
|
|
for idx, file_path in enumerate(sorted(changes_by_file.keys())):
|
|
if idx >= 20:
|
|
print(f"\n... and {len(changes_by_file) - 20} more files")
|
|
break
|
|
|
|
print(f"\n📄 {file_path}")
|
|
for line_num, old_line, new_line in changes_by_file[file_path][:5]:
|
|
print(f" Line {line_num}:")
|
|
print(f" - {old_line}")
|
|
print(f" + {new_line}")
|
|
if len(changes_by_file[file_path]) > 5:
|
|
print(f" ... and {len(changes_by_file[file_path]) - 5} more changes in this file")
|
|
|
|
if self.dry_run:
|
|
print(f"\n{'='*80}")
|
|
print("This was a DRY RUN - no files were modified")
|
|
print("To apply these changes, run with --apply flag")
|
|
print(f"{'='*80}\n")
|
|
else:
|
|
print(f"\n{'='*80}")
|
|
print("✅ Changes have been applied to all files")
|
|
print(f"{'='*80}\n")
|
|
|
|
return files_changed, total_changes
|
|
|
|
|
|
def main():
|
|
import argparse
|
|
import sys
|
|
|
|
parser = argparse.ArgumentParser(description="Fix relative imports to prevent shadowing")
|
|
parser.add_argument("--apply", action="store_true", help="Apply changes (default is dry-run)")
|
|
args = parser.parse_args()
|
|
|
|
# Get project root
|
|
script_dir = Path(__file__).parent
|
|
project_root = script_dir.parent
|
|
|
|
# Run conversion
|
|
fixer = RelativeImportFixer(project_root, dry_run=not args.apply)
|
|
files_changed, total_changes = fixer.run()
|
|
|
|
sys.exit(0 if files_changed >= 0 else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|