NINI
362 lines
19 KiB
Python
362 lines
19 KiB
Python
"""Produce pinned local Qwen Korean voice assets; validation never loads a model."""
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import importlib.metadata
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
|
|
import numpy as np
|
|
import soundfile as sf
|
|
|
|
MODEL_ID = 'Qwen/Qwen3-TTS-12Hz-1.7B-CustomVoice'
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True,
|
|
separators=(',', ':')).encode('utf-8')).hexdigest()
|
|
|
|
|
|
def file_hash(path):
|
|
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
|
|
def load_json(path):
|
|
return json.loads(Path(path).read_text(encoding='utf-8'))
|
|
|
|
|
|
def write_json(path, value):
|
|
Path(path).write_text(json.dumps(value, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
|
|
|
|
|
def validate_inputs(catalog, profile):
|
|
if catalog.get('schemaVersion') != 1 or profile.get('schemaVersion') != 1:
|
|
raise ValueError('catalog/profile schemaVersion must be 1')
|
|
clips = catalog.get('clips')
|
|
if not isinstance(clips, list) or not clips:
|
|
raise ValueError('catalog must contain clips')
|
|
ids = set()
|
|
for clip in clips:
|
|
clip_id = clip.get('id', '')
|
|
if not re.fullmatch(r'[a-z][a-z0-9_]*', clip_id) or clip_id in ids:
|
|
raise ValueError(f'{clip_id}: invalid or duplicate ID')
|
|
ids.add(clip_id)
|
|
if not isinstance(clip.get('text'), str) or not clip['text'].strip():
|
|
raise ValueError(f'{clip_id}: text is empty')
|
|
if not re.search('[가-힣]', clip['text']) or re.search(r'[0-9]', clip['text']):
|
|
raise ValueError(f'{clip_id}: use a complete Korean utterance, not numeral digits')
|
|
if not isinstance(clip.get('required'), bool) or not clip.get('usage'):
|
|
raise ValueError(f'{clip_id}: required/usage missing')
|
|
if not isinstance(clip.get('stageIds'), list) or not clip['stageIds'] or any(
|
|
type(n) is not int or n < 1 or n > 30 for n in clip['stageIds']):
|
|
raise ValueError(f'{clip_id}: stageIds must be within 1..30')
|
|
for field in ('modelRevision', 'instruct', 'device', 'dtype', 'attention', 'generationParameters', 'seed'):
|
|
if field not in profile:
|
|
raise ValueError(f'profile: missing {field}')
|
|
if profile.get('modelId') != MODEL_ID or profile.get('language') != 'Korean' or profile.get('speaker') != 'Sohee':
|
|
raise ValueError('profile must retain approved model/Korean/Sohee')
|
|
if not re.fullmatch('[0-9a-f]{40}', profile['modelRevision']):
|
|
raise ValueError('profile must pin a full model revision')
|
|
if profile['device'] != 'cuda:0' or profile['dtype'] not in ('bfloat16', 'float16') or profile['attention'] not in ('sdpa', 'eager'):
|
|
raise ValueError('profile requires approved CUDA configuration')
|
|
if not isinstance(profile['generationParameters'], dict) or type(profile['seed']) is not int:
|
|
raise ValueError('invalid generation settings')
|
|
|
|
|
|
def select_clips(catalog, ids=None, required_only=False):
|
|
by_id = {c['id']: c for c in catalog['clips']}
|
|
if ids is None:
|
|
return [c for c in catalog['clips'] if not required_only or c['required']]
|
|
missing = set(ids) - by_id.keys()
|
|
if missing:
|
|
raise ValueError('unknown IDs: ' + ', '.join(sorted(missing)))
|
|
if not ids or len(ids) != len(set(ids)):
|
|
raise ValueError('requested IDs must be nonempty and unique')
|
|
return [by_id[key] for key in ids]
|
|
|
|
|
|
def waveform_metadata(path):
|
|
samples, sample_rate = sf.read(path, dtype='float32', always_2d=True)
|
|
if sample_rate <= 0 or samples.size == 0 or not np.isfinite(samples).all():
|
|
raise ValueError('empty or nonfinite waveform')
|
|
peak = float(np.max(np.abs(samples)))
|
|
return {'sha256': file_hash(path), 'sampleRate': sample_rate, 'channels': samples.shape[1],
|
|
'frames': samples.shape[0], 'duration': samples.shape[0] / sample_rate,
|
|
'peak': peak, 'rms': float(np.sqrt(np.mean(samples ** 2))),
|
|
'qualityWarnings': (['near silence'] if peak < .001 else []) +
|
|
(['clipping candidate'] if peak >= .999 else [])}
|
|
|
|
|
|
def load_run_inputs(manifest, root):
|
|
run_id = manifest.get('runId', '')
|
|
if not isinstance(run_id, str) or not re.fullmatch(r'[A-Za-z0-9_]+', run_id):
|
|
raise ValueError('invalid input snapshot runId')
|
|
references = manifest.get('inputSnapshots', {'catalog': 'catalog.json', 'profile': 'profile.json'})
|
|
if references != {'catalog': 'catalog.json', 'profile': 'profile.json'}:
|
|
raise ValueError('input snapshots must reside in their immutable run directory')
|
|
run_dir = Path(root) / 'generated' / run_id
|
|
catalog = load_json(run_dir / references['catalog'])
|
|
profile = load_json(run_dir / references['profile'])
|
|
if not isinstance(catalog, dict) or not isinstance(profile, dict):
|
|
raise ValueError('input snapshots must contain JSON objects')
|
|
if digest(catalog) != manifest.get('catalogHash') or digest(profile) != manifest.get('profileHash'):
|
|
raise ValueError('input snapshot hash differs from manifest')
|
|
validate_inputs(catalog, profile)
|
|
if profile['modelRevision'] != manifest.get('modelRevision'):
|
|
raise ValueError('input snapshot model revision differs from manifest')
|
|
migration_path = Path(root) / 'input-snapshot-migration.json'
|
|
if migration_path.exists():
|
|
migrated = [record for record in load_json(migration_path)['runs'] if record['runId'] == run_id]
|
|
if migrated and (len(migrated) != 1 or file_hash(run_dir / 'manifest.json') != migrated[0]['originalManifestSha256']):
|
|
raise ValueError(f'{run_id}: immutable legacy manifest SHA-256 mismatch')
|
|
source_ids = set()
|
|
for reference in manifest.get('environment', {}).get('sourceGenerations', []):
|
|
source_id = reference['runId']
|
|
if not isinstance(source_id, str) or not re.fullmatch(r'[A-Za-z0-9_]+', source_id):
|
|
raise ValueError('invalid source generation reference runId')
|
|
if source_id in source_ids:
|
|
raise ValueError('duplicate source generation reference runId')
|
|
source_ids.add(source_id)
|
|
if reference['manifest'] != f'generated/{source_id}/manifest.json':
|
|
raise ValueError('source generation reference must identify its own run manifest')
|
|
source_path = Path(root) / reference['manifest']
|
|
if file_hash(source_path) != reference['manifestSha256']:
|
|
raise ValueError(f'{source_id}: recorded source manifest SHA-256 mismatch')
|
|
source = load_json(source_path)
|
|
if source.get('runId') != source_id or source.get('environment') != reference.get('environment'):
|
|
raise ValueError(f'{source_id}: source generation environment reference differs')
|
|
clips = {clip['id']: clip for clip in catalog['clips']}
|
|
for entry in manifest.get('clips', []):
|
|
clip = clips.get(entry.get('id'))
|
|
if not clip or entry.get('textHash') != digest(clip['text']) or entry.get('settingsHash') != digest(profile):
|
|
raise ValueError(f"{entry.get('id')}: input snapshot differs from clip text/settings hashes")
|
|
return catalog, profile
|
|
|
|
|
|
def clip_errors(entry, clip, profile, root):
|
|
errors = []
|
|
if entry.get('result') not in ('generated', 'reused'):
|
|
return [f"{clip['id']}: no successful generation"]
|
|
if entry.get('textHash') != digest(clip['text']) or entry.get('settingsHash') != digest(profile):
|
|
errors.append(f"{clip['id']}: current text/profile differs")
|
|
try:
|
|
path = (root / entry['file']).resolve()
|
|
if not path.is_relative_to(root.resolve()):
|
|
raise ValueError('waveform escapes source directory')
|
|
actual = waveform_metadata(path)
|
|
for key in ('sha256', 'sampleRate', 'channels', 'frames', 'duration'):
|
|
if actual[key] != entry.get(key):
|
|
errors.append(f"{clip['id']}: {key} mismatch")
|
|
if actual['channels'] != 1:
|
|
errors.append(f"{clip['id']}: expected mono")
|
|
run_id = entry['originalRunId']
|
|
if not isinstance(run_id, str) or not re.fullmatch(r'[A-Za-z0-9_]+', run_id):
|
|
raise ValueError('invalid original generation runId')
|
|
origin = load_json(root / 'generated' / run_id / 'manifest.json')
|
|
if origin.get('runId') != run_id:
|
|
raise ValueError('original manifest runId mismatch')
|
|
load_run_inputs(origin, root)
|
|
original_entries = [c for c in origin['clips'] if c.get('id') == clip['id']]
|
|
if len(original_entries) != 1 or original_entries[0].get('result') != 'generated':
|
|
raise ValueError('missing successful original generation entry')
|
|
for key in ('file', 'sha256', 'textHash', 'settingsHash'):
|
|
if original_entries[0].get(key) != entry.get(key):
|
|
raise ValueError(f'original generation {key} differs')
|
|
except (KeyError, OSError, ValueError, sf.LibsndfileError) as error:
|
|
errors.append(f"{clip['id']}: invalid source artifact: {error}")
|
|
return errors
|
|
|
|
|
|
def validate_manifest(manifest, catalog, profile, root, ids=None):
|
|
validate_inputs(catalog, profile)
|
|
requested = select_clips(catalog, ids, required_only=ids is None)
|
|
errors = []
|
|
try:
|
|
load_run_inputs(manifest, root)
|
|
except (KeyError, OSError, ValueError, TypeError) as error:
|
|
errors.append(f'input snapshots invalid: {error}')
|
|
if manifest.get('schemaVersion') != 1 or manifest.get('modelRevision') != profile['modelRevision']:
|
|
errors.append('manifest schema/revision differs')
|
|
if manifest.get('catalogHash') != digest(catalog) or manifest.get('profileHash') != digest(profile):
|
|
errors.append('manifest does not describe current catalog/profile')
|
|
entries = manifest.get('clips', [])
|
|
by_id = {c.get('id'): c for c in entries}
|
|
if len(by_id) != len(entries):
|
|
errors.append('manifest contains duplicate IDs')
|
|
for clip in requested:
|
|
if clip['id'] not in by_id:
|
|
errors.append(f"{clip['id']}: missing required/requested ID")
|
|
else:
|
|
errors.extend(clip_errors(by_id[clip['id']], clip, profile, root))
|
|
return errors
|
|
|
|
|
|
def reusable_entries(root, catalog, profile):
|
|
result = {}
|
|
by_id = {c['id']: c for c in catalog['clips']}
|
|
for path in sorted((root / 'generated').glob('*/manifest.json'), reverse=True):
|
|
try:
|
|
manifest = load_json(path)
|
|
load_run_inputs(manifest, root)
|
|
for entry in manifest.get('clips', []):
|
|
clip = by_id.get(entry.get('id'))
|
|
if clip and clip['id'] not in result and not clip_errors(entry, clip, profile, root):
|
|
result[clip['id']] = entry
|
|
except (OSError, ValueError, TypeError):
|
|
continue
|
|
return result
|
|
|
|
|
|
def reuse_environment(profile, entries, root):
|
|
environment = {'python': sys.version, 'packages': {name: importlib.metadata.version(name) for name in
|
|
('qwen-tts', 'torch', 'torchaudio', 'transformers', 'soundfile', 'huggingface-hub')},
|
|
'modelLoad': 'not_needed', 'requestedDevice': profile['device'], 'sourceGenerations': []}
|
|
try:
|
|
probe = subprocess.run(['nvidia-smi', '-i', '0', '--query-gpu=name', '--format=csv,noheader'],
|
|
check=True, capture_output=True, text=True, timeout=5)
|
|
environment['device'] = probe.stdout.strip()
|
|
environment['deviceProbe'] = 'nvidia-smi; no model or CUDA inference executed'
|
|
except (OSError, subprocess.SubprocessError) as error:
|
|
environment['deviceProbeError'] = str(error)
|
|
for run_id in sorted({entry['originalRunId'] for entry in entries.values()}):
|
|
manifest_path = Path(root) / 'generated' / run_id / 'manifest.json'
|
|
source = load_json(manifest_path)
|
|
environment['sourceGenerations'].append({'runId': run_id,
|
|
'manifest': manifest_path.relative_to(root).as_posix(),
|
|
'manifestSha256': file_hash(manifest_path),
|
|
'environment': source['environment']})
|
|
return environment
|
|
|
|
|
|
def generate(catalog, profile, root, ids=None, synthesize=None, environment=None, regenerate=False):
|
|
validate_inputs(catalog, profile)
|
|
selected = select_clips(catalog, ids)
|
|
root = Path(root)
|
|
previous = {} if regenerate else reusable_entries(root, catalog, profile)
|
|
run_id = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ_') + uuid.uuid4().hex[:8]
|
|
run_dir = root / 'generated' / run_id
|
|
run_dir.mkdir(parents=True)
|
|
write_json(run_dir / 'catalog.json', catalog)
|
|
write_json(run_dir / 'profile.json', profile)
|
|
manifest = {'schemaVersion': 1, 'runId': run_id, 'createdUtc': datetime.now(timezone.utc).isoformat(),
|
|
'modelRevision': profile['modelRevision'], 'catalogHash': digest(catalog),
|
|
'profileHash': digest(profile), 'scope': 'selected' if ids else 'all',
|
|
'regenerateRequested': regenerate,
|
|
'inputSnapshots': {'catalog': 'catalog.json', 'profile': 'profile.json'},
|
|
'requestedIds': [c['id'] for c in selected], 'environment': environment or {}, 'clips': []}
|
|
try:
|
|
for clip in selected:
|
|
clip_id = clip['id']
|
|
if clip_id in previous:
|
|
entry = dict(previous[clip_id], result='reused')
|
|
else:
|
|
entry = {'id': clip_id, 'textHash': digest(clip['text']), 'settingsHash': digest(profile),
|
|
'originalRunId': run_id, 'file': f'generated/{run_id}/{clip_id}.wav',
|
|
'review': 'not_listened'}
|
|
started = time.perf_counter()
|
|
try:
|
|
if synthesize is None:
|
|
raise ValueError('no synthesizer available')
|
|
samples, rate = synthesize(clip)
|
|
samples = np.asarray(samples, dtype=np.float32)
|
|
if samples.ndim != 1 or not samples.size or not np.isfinite(samples).all() or rate <= 0:
|
|
raise ValueError('synthesizer returned empty/nonfinite/nonmono audio')
|
|
path = root / entry['file']
|
|
sf.write(path, samples, rate, subtype='FLOAT')
|
|
entry.update(waveform_metadata(path), result='generated', generationSeconds=time.perf_counter() - started)
|
|
except Exception as error:
|
|
entry.update(result='failed', error=f'{type(error).__name__}: {error}')
|
|
manifest['clips'].append(entry)
|
|
write_json(run_dir / 'manifest.json', manifest)
|
|
print(f"{clip_id}: {entry['result']}", flush=True)
|
|
finally:
|
|
write_json(run_dir / 'manifest.json', manifest)
|
|
return manifest
|
|
|
|
|
|
def load_synthesizer(profile):
|
|
import torch
|
|
from huggingface_hub import snapshot_download
|
|
from qwen_tts import Qwen3TTSModel
|
|
if not torch.cuda.is_available():
|
|
raise RuntimeError('CUDA unavailable; refusing an unapproved CPU fallback')
|
|
probe = torch.arange(8, device=profile['device'], dtype=torch.float32).square().sum().item()
|
|
if probe != 140:
|
|
raise RuntimeError('CUDA arithmetic probe failed')
|
|
snapshot = snapshot_download(repo_id=profile['modelId'], revision=profile['modelRevision'])
|
|
started = time.perf_counter()
|
|
torch.cuda.reset_peak_memory_stats()
|
|
model = Qwen3TTSModel.from_pretrained(snapshot, device_map=profile['device'],
|
|
dtype=getattr(torch, profile['dtype']), attn_implementation=profile['attention'])
|
|
model_device = str(next(model.model.parameters()).device)
|
|
if model_device != profile['device']:
|
|
raise RuntimeError(f'loaded model device {model_device} differs from requested CUDA device')
|
|
if 'sohee' not in [s.lower() for s in model.get_supported_speakers()] or 'korean' not in [s.lower() for s in model.get_supported_languages()]:
|
|
raise RuntimeError('Sohee/Korean not supported by loaded model')
|
|
environment = {'python': sys.version, 'packages': {name: importlib.metadata.version(name) for name in
|
|
('qwen-tts', 'torch', 'torchaudio', 'transformers', 'soundfile', 'huggingface-hub')},
|
|
'device': torch.cuda.get_device_name(0), 'cuda': torch.version.cuda,
|
|
'modelDevice': model_device,
|
|
'cudaProbe': probe, 'modelLoadSeconds': time.perf_counter() - started,
|
|
'modelPeakVramBytes': torch.cuda.max_memory_allocated(),
|
|
'modelDefaults': model.generate_defaults,
|
|
'effectiveGenerationParameters': model._merge_generate_kwargs(**profile['generationParameters'])}
|
|
|
|
def synthesize(clip):
|
|
torch.manual_seed(profile['seed'])
|
|
torch.cuda.manual_seed_all(profile['seed'])
|
|
waves, rate = model.generate_custom_voice(text=clip['text'], language=profile['language'],
|
|
speaker=profile['speaker'], instruct=profile['instruct'],
|
|
non_streaming_mode=True, **profile['generationParameters'])
|
|
torch.cuda.synchronize()
|
|
environment['peakVramBytes'] = torch.cuda.max_memory_allocated()
|
|
return waves[0], rate
|
|
return synthesize, environment
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--catalog', required=True, type=Path)
|
|
parser.add_argument('--profile', required=True, type=Path)
|
|
parser.add_argument('--ids', help='comma separated IDs; absent means all catalog clips')
|
|
parser.add_argument('--validate-only', action='store_true')
|
|
parser.add_argument('--manifest', type=Path)
|
|
parser.add_argument('--regenerate', action='store_true', help='create a new take for selected IDs, preserving original runs')
|
|
args = parser.parse_args()
|
|
ids = args.ids.split(',') if args.ids else None
|
|
try:
|
|
catalog, profile = load_json(args.catalog), load_json(args.profile)
|
|
validate_inputs(catalog, profile)
|
|
root = args.catalog.parent
|
|
if args.validate_only:
|
|
if not args.manifest:
|
|
raise ValueError('--validate-only requires --manifest')
|
|
manifest = load_json(args.manifest)
|
|
errors = validate_manifest(manifest, catalog, profile, root, ids)
|
|
else:
|
|
selected = select_clips(catalog, ids)
|
|
reusable = reusable_entries(root, catalog, profile)
|
|
if not args.regenerate and all(c['id'] in reusable for c in selected):
|
|
synthesize = None
|
|
environment = reuse_environment(profile, {c['id']: reusable[c['id']] for c in selected}, root)
|
|
else:
|
|
synthesize, environment = load_synthesizer(profile)
|
|
manifest = generate(catalog, profile, root, ids, synthesize, environment, args.regenerate)
|
|
print('Manifest: ' + str(root / 'generated' / manifest['runId'] / 'manifest.json'))
|
|
errors = validate_manifest(manifest, catalog, profile, root, ids)
|
|
print(json.dumps({'validationScope': 'selected' if ids else 'all_required', 'ids': ids,
|
|
'errors': errors}, ensure_ascii=False))
|
|
return 1 if errors else 0
|
|
except Exception as error:
|
|
print(f'{type(error).__name__}: {error}', file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|