""" Unified demo: Higgs Audio v3 + FireRedTTS3 + ACE-Step XL-Turbo, CPU-only. Без Dockerfile: ace-step ставится в рантайме через pip --no-deps. Адаптивный интерфейс: десктоп / планшет / мобильный. """ import os import sys import subprocess import importlib.util # --------------------------------------------------------------------------- # 0. Рантайм-установка ace-step БЕЗ nano-vllm (он CUDA-only) # --------------------------------------------------------------------------- _ACESTEP_TARGET = "/tmp/acestep_install" def _ensure_acestep_installed(): if importlib.util.find_spec("acestep") is not None: return if _ACESTEP_TARGET in sys.path: return os.makedirs(_ACESTEP_TARGET, exist_ok=True) print("[acestep] runtime install (--no-deps)…", flush=True) subprocess.check_call([ sys.executable, "-m", "pip", "install", "--no-cache-dir", "--no-deps", "--target", _ACESTEP_TARGET, "git+https://github.com/ace-step/ACE-Step-1.5.git", ]) sys.path.insert(0, _ACESTEP_TARGET) print("[acestep] installed to", _ACESTEP_TARGET, flush=True) _ensure_acestep_installed() # --------------------------------------------------------------------------- # 1. Корректный детект CPU ДО импорта torch/numpy. # --------------------------------------------------------------------------- def _cgroup_cpu_quota() -> float | None: try: with open("/sys/fs/cgroup/cpu.max") as f: parts = f.read().split() if len(parts) == 2 and parts[0] != "max": quota, period = int(parts[0]), int(parts[1]) if period > 0: return quota / period except (FileNotFoundError, ValueError, OSError): pass try: with open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us") as f: quota = int(f.read().strip()) with open("/sys/fs/cgroup/cpu/cpu.cfs_period_us") as f: period = int(f.read().strip()) if quota > 0 and period > 0: return quota / period except (FileNotFoundError, ValueError, OSError): pass return None def _available_cpus() -> int: try: affinity = len(os.sched_getaffinity(0)) except AttributeError: affinity = os.cpu_count() or 1 quota = _cgroup_cpu_quota() if quota is not None: quota_int = max(1, int(round(quota))) return min(affinity, quota_int) return max(1, affinity) _CPU_COUNT = _available_cpus() os.environ.setdefault("OMP_NUM_THREADS", str(_CPU_COUNT)) os.environ.setdefault("MKL_NUM_THREADS", str(_CPU_COUNT)) os.environ.setdefault("OPENBLAS_NUM_THREADS", str(_CPU_COUNT)) os.environ.setdefault("NUMEXPR_NUM_THREADS", str(_CPU_COUNT)) os.environ.setdefault("VECLIB_MAXIMUM_THREADS", str(_CPU_COUNT)) os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") os.environ.setdefault("HF_HUB_DISABLE_TELEMETRY", "1") os.environ.setdefault("TRANSFORMERS_NO_ADVISORY_WARNINGS", "1") os.environ["CUDA_VISIBLE_DEVICES"] = "" os.environ["ACCELERATE_USE_CPU"] = "true" os.environ["TRANSFORMERS_ATTN_IMPLEMENTATION"] = "eager" os.environ.setdefault("ACESTEP_LM_BACKEND", "pt") os.environ.setdefault("ACESTEP_DTYPE", "float32") _cgroup_quota = _cgroup_cpu_quota() print( f"[cpu] cpu_count={os.cpu_count()} " f"sched_getaffinity={len(os.sched_getaffinity(0))} " f"cgroup_quota={_cgroup_quota} " f"-> using {_CPU_COUNT} threads " f"(CUDA_VISIBLE_DEVICES={os.environ.get('CUDA_VISIBLE_DEVICES')!r})" ) import contextlib import functools import re import tempfile import urllib.request import uuid import gc import asyncio import asyncio.base_events as base_events import spaces # noqa: F401 import torch torch.set_default_dtype(torch.float32) torch.cuda.is_available = lambda: False # type: ignore[assignment] torch.cuda.device_count = lambda: 0 # type: ignore[assignment] torch.cuda.current_device = lambda: 0 # type: ignore[assignment] torch.cuda.get_device_name = lambda *a, **k: "cpu" # type: ignore[assignment] def _to_cpu_device(d): if d is None: return None if isinstance(d, torch.device): return torch.device("cpu") if d.type == "cuda" else d if isinstance(d, str) and "cuda" in d.lower(): return "cpu" return d _orig_module_to = torch.nn.Module.to def _patched_module_to(self, *args, **kwargs): if args: new_args = list(args) new_args[0] = _to_cpu_device(new_args[0]) args = tuple(new_args) if "device" in kwargs: kwargs["device"] = _to_cpu_device(kwargs["device"]) return _orig_module_to(self, *args, **kwargs) torch.nn.Module.to = _patched_module_to # type: ignore[assignment] torch.nn.Module.cuda = lambda self, *a, **k: self # type: ignore[assignment] _orig_tensor_to = torch.Tensor.to def _patched_tensor_to(self, *args, **kwargs): if args: new_args = list(args) new_args[0] = _to_cpu_device(new_args[0]) args = tuple(new_args) if "device" in kwargs: kwargs["device"] = _to_cpu_device(kwargs["device"]) return _orig_tensor_to(self, *args, **kwargs) torch.Tensor.to = _patched_tensor_to # type: ignore[assignment] torch.Tensor.cuda = lambda self, *a, **k: self # type: ignore[assignment] torch.set_default_device("cpu") torch.set_float32_matmul_precision("highest") try: torch.set_num_threads(_CPU_COUNT) torch.set_num_interop_threads(max(1, _CPU_COUNT // 2)) except Exception: pass import numpy as np import soundfile as sf import gradio as gr from huggingface_hub import hf_hub_download, snapshot_download from safetensors import safe_open try: import torchaudio # noqa: F401 _HAS_TORCHAUDIO = True except ImportError: _HAS_TORCHAUDIO = False def _patch_asyncio_event_loop_del(): original_del = getattr(base_events.BaseEventLoop, "__del__", None) if original_del is None: return def patched_del(self): try: original_del(self) except ValueError as e: if "Invalid file descriptor" not in str(e): raise base_events.BaseEventLoop.__del__ = patched_del _patch_asyncio_event_loop_del() # --------- ОСТАВЛЕНО КАК ЕСТЬ --------- @spaces.GPU(duration=10) def _zerogpu_startup_probe(): return True # -------------------------------------- def _patch_spaces_gpu_noop(): if getattr(spaces.GPU, "_acestep_noop", False): return def _noop_gpu(*args, **kwargs): if len(args) == 1 and callable(args[0]) and not kwargs: return args[0] def _decorator(fn): return fn return _decorator _noop_gpu._acestep_noop = True spaces.GPU = _noop_gpu print("[acestep] spaces.GPU patched to no-op (CPU mode)", flush=True) HERE = os.path.dirname(os.path.abspath(__file__)) try: from pydub import AudioSegment _PYDUB_OK = True except Exception as _e: # pragma: no cover print(f"[WARN] pydub not available: {_e}", flush=True) _PYDUB_OK = False # =========================================================================== # # SHARED: N-GAN # =========================================================================== # _NGAN_AVAILABLE = False _ngan_model = None try: from ngan import NGANSuperResolver _NGAN_AVAILABLE = True except Exception as exc: # pragma: no cover print(f"[WARN] ngan not available: {exc}", flush=True) _NGAN_WEIGHT_CANDIDATES = [ os.path.join(HERE, "ngan", "n8.ckpt"), os.path.join(HERE, "ngan", "weights", "n8.ckpt"), os.path.join(HERE, "ngan_weights", "n8.ckpt"), os.path.join(HERE, "firered_tts_v3", "ngan_weights", "n8.ckpt"), ] def _find_ngan_weights(): for path in _NGAN_WEIGHT_CANDIDATES: if os.path.isfile(path): return path return None def _get_ngan(): global _ngan_model if _ngan_model is None: if not _NGAN_AVAILABLE: raise RuntimeError("ngan package is not installed") local_path = _find_ngan_weights() if local_path: print(f"[ngan] loading local weights: {local_path}", flush=True) _ngan_model = NGANSuperResolver.from_pretrained( local_path, n_channels=8, ) else: print("[ngan] local weights not found — loading 'n8'", flush=True) _ngan_model = NGANSuperResolver.from_pretrained("n8") print("[ngan] ready", flush=True) return _ngan_model def enhance_to_48k(audio, sr: int): if isinstance(audio, torch.Tensor): x = audio.detach().to(torch.float32).cpu().numpy() else: x = np.asarray(audio, dtype=np.float32) if x.ndim > 1: if x.shape[0] == 1: x = x[0] elif x.shape[-1] == 1: x = x[..., 0] elif x.shape[0] < x.shape[1]: x = x.mean(axis=0) else: x = x.mean(axis=1) x = np.nan_to_num(x, nan=0.0, posinf=0.0, neginf=0.0).astype(np.float32) model = _get_ngan() y = model.upsample(x, input_sample_rate=sr) y = np.nan_to_num(y, nan=0.0, posinf=0.0, neginf=0.0) y = np.clip(y, -1.0, 1.0).astype(np.float32).reshape(-1) return y, 48_000 # =========================================================================== # # SHARED: Whisper large-v3-turbo # =========================================================================== # ASR_MODEL = "openai/whisper-large-v3-turbo" _asr_pipe = None def _get_whisper(): global _asr_pipe if _asr_pipe is None: from transformers import ( AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline, ) print(f"[asr] loading {ASR_MODEL} (CPU, fp32)…", flush=True) proc = AutoProcessor.from_pretrained(ASR_MODEL) mdl = AutoModelForSpeechSeq2Seq.from_pretrained( ASR_MODEL, dtype=torch.float32, low_cpu_mem_usage=True, use_safetensors=True, ).eval().to("cpu") _asr_pipe = pipeline( "automatic-speech-recognition", model=mdl, tokenizer=proc.tokenizer, feature_extractor=proc.feature_extractor, dtype=torch.float32, device="cpu", chunk_length_s=30, stride_length_s=5, ) gc.collect() print("[asr] Whisper ready", flush=True) return _asr_pipe def transcribe_reference(audio_path): if not audio_path: return "" try: data, sr = sf.read(audio_path, dtype="float32", always_2d=True) except Exception as e: raise gr.Error(f"Не удалось прочитать аудио: {e}") wav = data.mean(axis=1) pipe = _get_whisper() result = pipe( {"array": wav, "sampling_rate": sr}, generate_kwargs={"task": "transcribe"}, return_timestamps=False, ) return (result.get("text") or "").strip() # =========================================================================== # # TAB 1: HIGGS AUDIO V3 # =========================================================================== # from transformers import AutoModelForCausalLM, AutoTokenizer HIGGS_REPO = "multimodalart/higgs-audio-v3-tts-4b-transformers" HIGGS_CKPT_KEY = "tied.head.modality_heads.0.weight" HIGGS_DTYPE = torch.bfloat16 print(f"[higgs] loading {HIGGS_REPO} (bf16)…", flush=True) higgs_tokenizer = AutoTokenizer.from_pretrained(HIGGS_REPO) higgs_model = AutoModelForCausalLM.from_pretrained( HIGGS_REPO, trust_remote_code=True, dtype=HIGGS_DTYPE, low_cpu_mem_usage=True, ).eval().to("cpu") higgs_model.get_audio_codec() HIGGS_SR = higgs_model.config.sample_rate _ckpt_path = hf_hub_download(HIGGS_REPO, "model.safetensors") with safe_open(_ckpt_path, framework="pt") as _f: _ckpt_w = _f.get_tensor(HIGGS_CKPT_KEY) _fixed = False for _n, _p in higgs_model.audio_head.named_parameters(): if _p.shape == _ckpt_w.shape: with torch.no_grad(): _p.copy_(_ckpt_w.to(_p.dtype)) print(f"[higgs] audio_head.{_n} <- {HIGGS_CKPT_KEY}", flush=True) _fixed = True break if not _fixed: print("[higgs] WARNING: audio_head target parameter not found", flush=True) del _ckpt_w gc.collect() print("[higgs] ready", flush=True) def synthesize_higgs( text, reference_audio, reference_text, temperature, top_p, top_k, max_new_tokens, seed, ): text = (text or "").strip() if not text: raise gr.Error("Введите текст для синтеза.") if seed is not None and int(seed) >= 0: torch.manual_seed(int(seed)) kwargs = dict( max_new_tokens=int(max_new_tokens), temperature=float(temperature), top_p=float(top_p) if float(top_p) < 1.0 else None, top_k=int(top_k) if int(top_k) > 0 else None, ) if reference_audio is not None: try: data, sr = sf.read(reference_audio, dtype="float32", always_2d=True) except Exception as e: raise gr.Error(f"Не удалось прочитать референсное аудио: {e}") wav = torch.from_numpy(data).mean(dim=1) kwargs["reference_audio"] = wav kwargs["reference_sample_rate"] = sr if reference_text and reference_text.strip(): kwargs["reference_text"] = reference_text.strip() with torch.inference_mode(): audio = higgs_model.generate_speech(text, higgs_tokenizer, **kwargs) if audio.numel() == 0: raise gr.Error("Генерация вернула пустое аудио.") audio_np = audio.detach().to(torch.float32).cpu().numpy() audio_np = np.clip(audio_np, -1.0, 1.0) audio_f, out_sr = enhance_to_48k(audio_np, HIGGS_SR) audio_f = np.asarray(audio_f, dtype=np.float32).reshape(-1) audio_int16 = (audio_f * 32767.0).astype(np.int16) return out_sr, audio_int16 # =========================================================================== # # TAB 2: FIREREDTTS3 # =========================================================================== # from transformers.modeling_utils import PreTrainedModel def _force_eager_attn(self, *args, **kwargs): # noqa: ANN001 return "eager" def _force_eager_get(self, requested_attn_implementation=None, *args, **kwargs): if requested_attn_implementation in (None, "flash_attention_2", "fa2"): return "eager" return requested_attn_implementation PreTrainedModel._check_and_adjust_attn_implementation = _force_eager_attn # type: ignore[assignment] PreTrainedModel.get_correct_attn_implementation = _force_eager_get # type: ignore[assignment] from transformers import PretrainedConfig _orig_config_init = PretrainedConfig.__init__ def _config_init_fp32(self, *args, **kwargs): _orig_config_init(self, *args, **kwargs) try: self.torch_dtype = torch.float32 self.dtype = torch.float32 self._attn_implementation_internal = "eager" except Exception: pass PretrainedConfig.__init__ = _config_init_fp32 # type: ignore[assignment] try: import fasttext # type: ignore _FASTTEXT_AVAILABLE = True except Exception as _e: # pragma: no cover print(f"[WARN] fasttext import failed: {_e}", flush=True) _FASTTEXT_AVAILABLE = False @contextlib.contextmanager def _numpy2_compat(): _orig = np.array def _patched(*args, **kwargs): if kwargs.get("copy") is False: kwargs["copy"] = None return _orig(*args, **kwargs) np.array = _patched try: yield finally: np.array = _orig _LID_URL = "https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.ftz" _LID_PATH = os.path.join(HERE, "fireredtts3", "utils", "llm_tn", "models", "lid.176.ftz") os.makedirs(os.path.dirname(_LID_PATH), exist_ok=True) if not os.path.exists(_LID_PATH): try: urllib.request.urlretrieve(_LID_URL, _LID_PATH) print(f"[INFO] fastText lid.176 downloaded to {_LID_PATH}", flush=True) except Exception as exc: # pragma: no cover print(f"[WARN] Could not fetch fastText lid.176: {exc}", flush=True) os.environ.setdefault("LLM_TN_API_URL", "http://127.0.0.1:1/unused") os.environ.setdefault("LLM_TN_API_KEY", "unused") FIRERED_REPO = "FireRedTeam/FireRedTTS3" FIRERED_MODEL_DIR = snapshot_download(FIRERED_REPO) print(f"[firered] weights at {FIRERED_MODEL_DIR}", flush=True) from fireredtts3.core import FireRedTTS3 # noqa: E402 from fireredtts3.redae.redae import RedAE # noqa: E402 from fireredtts3.utils.llm_tn.text_normalizer import TextNormalizer # noqa: E402 from fireredtts3.utils.text_tokenizer import ( # noqa: E402 MULTI_DIALECT_TAGS, MULTI_LANG_TAGS, ) _redae_real_from_pretrained = RedAE.from_pretrained _redae_singleton = None def _shared_redae(*args, **kwargs): global _redae_singleton if _redae_singleton is None: kwargs.setdefault("attn_implementation", "eager") kwargs.setdefault("torch_dtype", torch.float32) _redae_singleton = _redae_real_from_pretrained(*args, **kwargs) try: _redae_singleton.to(torch.float32) print("[firered] RedAE → float32", flush=True) except Exception as e: print(f"[WARN] RedAE fp32 conversion failed: {e}", flush=True) return _redae_singleton RedAE.from_pretrained = _shared_redae firered_tts = FireRedTTS3( FIRERED_MODEL_DIR, use_fasttext=True, use_llm_tn=False, use_wetext=True, ) for _pipe_obj in (firered_tts,): _norm = getattr(_pipe_obj, "_llm_tn", None) if _norm is None: continue _detect = getattr(TextNormalizer, "detect_locale", None) if _detect is None: print( "[INFO] TextNormalizer.detect_locale unavailable; " "skipping monkeypatch", flush=True, ) continue try: _norm.detect_locale = functools.partial( _detect, _norm, use_llm_fallback=False, ) print("[INFO] detect_locale monkeypatched (LLM fallback disabled)", flush=True) except Exception as e: print(f"[WARN] Could not patch detect_locale: {e}", flush=True) print( f"[firered] FireRedTTS3 ready (CPU, fp32, eager attn, threads={_CPU_COUNT})", flush=True, ) AUTO = "Auto-detect" LANGUAGES = [t.strip("<|>") for t in MULTI_LANG_TAGS] DIALECTS = [t.strip("<|>") for t in MULTI_DIALECT_TAGS] LANG_CHOICES = [AUTO] + LANGUAGES + [f"{d} (Chinese dialect)" for d in DIALECTS] def _resolve_language(choice: str): if not choice or choice == AUTO: return None return choice.split(" (")[0] _ISO_TO_LANG_NAME = { "en": "English", "zh": "Chinese", "ja": "Japanese", "ko": "Korean", "fr": "French", "de": "German", "es": "Spanish", "it": "Italian", "ru": "Russian", "pt": "Portuguese", "ar": "Arabic", "hi": "Hindi", "nl": "Dutch", "pl": "Polish", "tr": "Turkish", "vi": "Vietnamese", "th": "Thai", "id": "Indonesian", "ms": "Malay", "sv": "Swedish", "da": "Danish", "fi": "Finnish", "no": "Norwegian", "cs": "Czech", "el": "Greek", "he": "Hebrew", "uk": "Ukrainian", "ro": "Romanian", "hu": "Hungarian", "bg": "Bulgarian", "fa": "Persian", "ur": "Urdu", "bn": "Bengali", "ta": "Tamil", "te": "Telugu", "mr": "Marathi", "gu": "Gujarati", "kn": "Kannada", "ml": "Malayalam", "pa": "Punjabi", } _LANG_NAME_BY_ISO = { code: name for code, name in _ISO_TO_LANG_NAME.items() if name in LANGUAGES } _fasttext_model = None def _get_fasttext_model(): global _fasttext_model if _fasttext_model is None: if not _FASTTEXT_AVAILABLE: raise RuntimeError("fasttext is not installed") _fasttext_model = fasttext.load_model(_LID_PATH) print(f"[INFO] fastText lid.176 loaded from {_LID_PATH}", flush=True) return _fasttext_model def detect_language_from_text(text: str) -> str: text = (text or "").strip() if not text: return AUTO try: model = _get_fasttext_model() with _numpy2_compat(): labels, _ = model.predict(text.replace("\n", " "), k=1) except Exception as exc: print(f"[WARN] fastText language detection failed: {exc}", flush=True) return AUTO if not labels: return AUTO code = labels[0].replace("__label__", "").strip().lower() name = _LANG_NAME_BY_ISO.get(code, AUTO) if name == AUTO: print( f"[INFO] fastText detected '{code}' — not in FireRedTTS3 language list", flush=True, ) return name def _resolve_language_with_auto(choice: str, text: str): resolved = _resolve_language(choice) if resolved is not None: return resolved detected = detect_language_from_text(text) if detected != AUTO: print(f"[INFO] Auto-detected language (fastText): {detected}", flush=True) return detected print( "[INFO] fastText detection inconclusive — using FireRedTTS3 default", flush=True, ) return None MAX_PROMPT_SECONDS = 20.0 MAX_TEXT_CHARS = 4096 def _load_audio(path: str, max_seconds: float): if not path: raise gr.Error("Please provide an audio file first.") wav, sr = sf.read(path, always_2d=True, dtype="float32") wav = wav[:, 0] if wav.shape[0] > int(max_seconds * sr): wav = wav[: int(max_seconds * sr)] gr.Info(f"Audio truncated to the first {max_seconds:.0f}s.") peak = float(np.abs(wav).max()) if wav.size else 0.0 if peak > 0: wav = wav / peak * 0.95 return torch.from_numpy(np.ascontiguousarray(wav)[None, :]).to(torch.float32), sr def _check_text(text: str, what: str = "Text"): text = (text or "").strip() if not text: raise gr.Error(f"{what} must not be empty.") if len(text) > MAX_TEXT_CHARS: gr.Info(f"{what} truncated to {MAX_TEXT_CHARS} characters.") text = text[:MAX_TEXT_CHARS] return text def _safe_output_stem(raw: str | None, fallback: str = "output") -> str: if not raw: return fallback stem = os.path.splitext(os.path.basename(str(raw).strip()))[0] stem = re.sub(r"[^\w\-. ]+", "_", stem).strip() stem = stem.lstrip("._- ") if len(stem) > 100: stem = stem[:100].rstrip() return stem or fallback def _write_mp3(audio, sr: int, bitrate: str = "192k", output_stem: str | None = None) -> str: if isinstance(audio, torch.Tensor): x = audio.detach().float().cpu().numpy() else: x = np.asarray(audio, dtype=np.float32) if x.ndim > 1: if x.shape[0] == 1: x = x[0] elif x.shape[-1] == 1: x = x[..., 0] elif x.shape[0] < x.shape[1]: x = x.mean(axis=0) else: x = x.mean(axis=1) x = np.nan_to_num(x, nan=0.0, posinf=0.0, neginf=0.0) x = np.clip(x, -1.0, 1.0).reshape(-1) out_dir = tempfile.mkdtemp(prefix="fireredtts3_") stem = _safe_output_stem(output_stem) if output_stem \ else f"fireredtts3_{uuid.uuid4().hex[:8]}" if _PYDUB_OK: pcm = (x * 32767.0).astype(np.int16).tobytes() seg = AudioSegment(data=pcm, sample_width=2, frame_rate=sr, channels=1) path = os.path.join(out_dir, f"{stem}.mp3") seg.export(path, format="mp3", bitrate=bitrate) return path path = os.path.join(out_dir, f"{stem}.wav") sf.write(path, x, sr, subtype="PCM_16") return path def voice_clone_firered( prompt_audio, prompt_text, text, language=AUTO, inference_cfg=2.0, n_timesteps=10, seed=1234, do_tn=True, ): text = _check_text(text, "Text to synthesize") prompt_text = (prompt_text or "").strip() if not prompt_text: raise gr.Error("Please provide the transcript of the reference audio.") wav, sr = _load_audio(prompt_audio, MAX_PROMPT_SECONDS) resolved_lang = _resolve_language_with_auto(language, text) output_stem = _safe_output_stem(prompt_audio, fallback="output") print( f"[firered] lang={resolved_lang!r} steps={n_timesteps} cfg={inference_cfg} " f"seed={seed} prompt_sec={wav.shape[-1] / sr:.2f} " f"output_stem={output_stem!r}", flush=True, ) with torch.inference_mode(): gen_audio, gen_sr = firered_tts.generate( text=text, language=resolved_lang, prompt_text=prompt_text, prompt_audio=wav, prompt_audio_sr=sr, n_timesteps=int(n_timesteps), inference_cfg=float(inference_cfg), seed=int(seed), do_tn=bool(do_tn), ) a = gen_audio.detach().float().cpu() print( f"[firered] gen_audio: shape={tuple(a.shape)} sr={gen_sr} " f"dur={a.shape[-1] / gen_sr:.2f}s", flush=True, ) enhanced, out_sr = enhance_to_48k(a, gen_sr) print(f"[firered] after N-GAN: {enhanced.shape} @ {out_sr}", flush=True) path = _write_mp3(enhanced, out_sr, bitrate="192k", output_stem=output_stem) print(f"[firered] saved {path}", flush=True) return path # =========================================================================== # # TAB 3: ACE-STEP 1.5 XL-TURBO # =========================================================================== # ACESTEP_DIT_MODEL = os.environ.get("ACESTEP_DIT_MODEL", "acestep-v15-turbo") ACESTEP_LM_MODEL = os.environ.get("ACESTEP_LM_MODEL", "acestep-5Hz-lm-1.7B") ACESTEP_CHECKPOINT_DIR = os.environ.get("ACESTEP_CHECKPOINT_DIR", "./checkpoints") _acestep_dit = None _acestep_llm = None _acestep_status = {"available": True, "error": None} @contextlib.contextmanager def _no_meta_init(): """Убирает meta-устройство из контекста инициализации transformers 5.x.""" orig = PreTrainedModel.get_init_context.__func__ def patched(cls, *args, **kwargs): return [ c for c in orig(cls, *args, **kwargs) if not (isinstance(c, torch.device) and c.type == "meta") ] PreTrainedModel.get_init_context = classmethod(patched) try: yield finally: PreTrainedModel.get_init_context = classmethod(orig) def _initialize_acestep_dit(dit): """Загрузка DiT без meta-инициализации и с принудительным eager attention.""" with _no_meta_init(): try: dit.initialize_service( project_root=".", config_path=ACESTEP_DIT_MODEL, device="cpu", attn_implementation="eager", ) except TypeError: # Fallback для версий ACE-Step без параметра attn_implementation dit.initialize_service( project_root=".", config_path=ACESTEP_DIT_MODEL, device="cpu", ) def _get_acestep_handlers(): global _acestep_dit, _acestep_llm if _acestep_dit is not None and _acestep_llm is not None: return _acestep_dit, _acestep_llm if not _acestep_status["available"]: raise gr.Error(f"ACE-Step недоступен: {_acestep_status['error']}") _patch_spaces_gpu_noop() try: from acestep.handler import AceStepHandler from acestep.llm_inference import LLMHandler except ImportError as exc: _acestep_status["available"] = False _acestep_status["error"] = f"модуль не установлен ({exc})" raise gr.Error( "ACE-Step не установлен. Проверьте рантайм-установку в начале app.py." ) print( f"[acestep] loading DiT={ACESTEP_DIT_MODEL} " f"LM={ACESTEP_LM_MODEL} (CPU, fp32)…", flush=True, ) try: dit = AceStepHandler() _initialize_acestep_dit(dit) llm = LLMHandler() llm.initialize( checkpoint_dir=ACESTEP_CHECKPOINT_DIR, lm_model_path=ACESTEP_LM_MODEL, backend="pt", device="cpu", ) except Exception as exc: _acestep_status["available"] = False _acestep_status["error"] = str(exc) print(f"[acestep] load failed: {exc}", flush=True) raise gr.Error(f"Не удалось загрузить ACE-Step: {exc}") _acestep_dit, _acestep_llm = dit, llm gc.collect() print("[acestep] ready (CPU)", flush=True) return dit, llm def _acestep_lyrics(verse: str, chorus: str, bridge: str, instrumental: bool) -> str: if instrumental: return "[inst]" parts = [] if verse and verse.strip(): parts += ["[verse]", verse.strip()] if chorus and chorus.strip(): parts += ["[chorus]", chorus.strip()] if bridge and bridge.strip(): parts += ["[bridge]", bridge.strip()] return "\n".join(parts) if parts else "[inst]" def _run_acestep_generate(params, config, save_dir): """Multi-tier retry при NaN: shift=1 → steps=4 + dcw off.""" from acestep.inference import generate_music dit, llm = _get_acestep_handlers() def _attempt(p, d): with torch.inference_mode(): return generate_music( dit_handler=dit, llm_handler=llm, params=p, config=config, save_dir=d, ) try: result = _attempt(params, save_dir) except Exception as exc: raise gr.Error(f"Ошибка генерации ACE-Step: {exc}") if not result.success: err = str(getattr(result, "error", "") or "") is_nan = ("NaN" in err) or ("nan" in err) if is_nan: # --- Tier 1: shift=1.0 --- print("[acestep] NaN detected → retry tier 1 (shift=1.0)", flush=True) try: params.shift = 1.0 except Exception: pass d2 = tempfile.mkdtemp(prefix="acestep_retry1_") try: result = _attempt(params, d2) except Exception as exc: raise gr.Error(f"Retry tier 1 failed: {exc}") if not result.success: err = str(getattr(result, "error", "") or "") is_nan = ("NaN" in err) or ("nan" in err) else: is_nan = False if is_nan: # --- Tier 2: steps=4 + dcw off + shift=3.0 --- print("[acestep] NaN persisted → retry tier 2 (steps=4, dcw off)", flush=True) try: params.inference_steps = 4 params.shift = 3.0 params.dcw_enabled = False except Exception: pass d3 = tempfile.mkdtemp(prefix="acestep_retry2_") try: result = _attempt(params, d3) except Exception as exc: raise gr.Error(f"Retry tier 2 failed: {exc}") if not result.success: raise gr.Error(f"ACE-Step вернул ошибку: {result.error}") if not result.audios: raise gr.Error("ACE-Step не вернул аудио.") return result.audios[0]["path"] def acestep_text2music( caption, verse, chorus, bridge, instrumental, duration, steps, guidance, seed, shift, ): _patch_spaces_gpu_noop() from acestep.inference import GenerationParams, GenerationConfig lyrics = _acestep_lyrics(verse, chorus, bridge, instrumental) params = GenerationParams( caption=caption or "", lyrics=lyrics, duration=float(max(10.0, min(float(duration), 240.0))), inference_steps=int(max(1, int(steps))), guidance_scale=float(guidance), seed=int(seed) if (seed is not None and int(seed) >= 0) else -1, shift=float(shift) if shift is not None else 3.0, dcw_enabled=False, # DCW без pytorch_wavelets не работает → отключаем ) config = GenerationConfig( batch_size=1, audio_format="wav", use_random_seed=(seed is None or int(seed) < 0), ) save_dir = tempfile.mkdtemp(prefix="acestep_") print( f"[acestep] gen: steps={steps} guidance={guidance} " f"shift={shift} dur={duration}s dcw=off → {save_dir}", flush=True, ) return _run_acestep_generate(params, config, save_dir) def acestep_create_sample(query, instrumental, vocal_language): _patch_spaces_gpu_noop() from acestep.inference import create_sample _, llm = _get_acestep_handlers() result = create_sample( llm_handler=llm, query=query, instrumental=instrumental, vocal_language=vocal_language if vocal_language != "unknown" else None, ) if not result.success: raise gr.Error(result.error) return ( result.caption or "", result.lyrics or "", result.bpm or "", result.keyscale or "", result.duration or "", result.language or "", ) def acestep_format_sample(caption, lyrics, user_bpm, user_keyscale): _patch_spaces_gpu_noop() from acestep.inference import format_sample _, llm = _get_acestep_handlers() user_metadata = {} if user_bpm: user_metadata["bpm"] = int(user_bpm) if user_keyscale: user_metadata["keyscale"] = user_keyscale result = format_sample( llm_handler=llm, caption=caption or "", lyrics=lyrics or "", user_metadata=user_metadata or None, ) if not result.success: raise gr.Error(result.error) return ( result.caption or "", result.lyrics or "", result.bpm or "", result.keyscale or "", result.duration or "", result.language or "", ) # =========================================================================== # # UI # =========================================================================== # EN_PROMPT = os.path.join(HERE, "examples", "en_prompt.wav") ZH_PROMPT = os.path.join(HERE, "examples", "zh_prompt.wav") EN_PROMPT_TEXT = ( "Just by listening a few minutes a day, you'll be able to eliminate negative " "thoughts by conditioning your mind to be more positive." ) ZH_PROMPT_TEXT = "比如具体一点的，他觉得最大的一个跟他预想的不一样的是在什么地方。" CSS = """ .gradio-container { max-width: 1400px !important; margin: 0 auto !important; padding: 16px !important; } .gradio-container h1 { font-size: clamp(1.35rem, 2.5vw, 1.9rem) !important; line-height: 1.3 !important; } .gradio-container h2 { font-size: clamp(1.15rem, 2vw, 1.45rem) !important; } .gradio-container h3 { font-size: clamp(1.0rem, 1.7vw, 1.2rem) !important; } .gradio-container .prose, .gradio-container .markdown-text { font-size: clamp(0.9rem, 1.2vw, 1rem) !important; line-height: 1.55 !important; } .gradio-container .tabs > .tab-nav { display: flex !important; flex-wrap: nowrap !important; overflow-x: auto !important; overflow-y: hidden !important; -webkit-overflow-scrolling: touch !important; scrollbar-width: thin !important; gap: 4px !important; padding-bottom: 4px !important; } .gradio-container .tabs > .tab-nav > button { white-space: nowrap !important; flex: 0 0 auto !important; font-size: 0.92rem !important; padding: 8px 14px !important; } .gradio-container .tabs > .tab-nav::-webkit-scrollbar { height: 6px; } .gradio-container .tabs > .tab-nav::-webkit-scrollbar-thumb { background: rgba(128, 128, 128, 0.4); border-radius: 3px; } .gradio-container button.primary, .gradio-container button.lg { font-weight: 600 !important; border-radius: 8px !important; } .gradio-container button.primary { min-height: 44px !important; } .gradio-container input, .gradio-container textarea, .gradio-container select, .gradio-container .gr-dropdown input, .gradio-container .gr-number input, .gradio-container .gr-slider input[type="number"] { font-size: 16px !important; } .gradio-container .gr-audio, .gradio-container .gr-audio *, .gradio-container .gr-file, .gradio-container .gr-box { max-width: 100% !important; box-sizing: border-box !important; } .gradio-container .gr-form, .gradio-container .gr-panel { padding: 12px !important; border-radius: 10px !important; } @media (max-width: 1024px) { .gradio-container { padding: 12px !important; max-width: 100% !important; } } @media (max-width: 768px) { .gradio-container { padding: 8px !important; max-width: 100% !important; } .gradio-container h1 { font-size: 1.25rem !important; line-height: 1.35 !important; margin-bottom: 0.5rem !important; } .gradio-container h2, .gradio-container h3 { font-size: 1.05rem !important; } .gradio-container .prose, .gradio-container .markdown-text { font-size: 0.92rem !important; line-height: 1.5 !important; } .gradio-container button.primary, .gradio-container button.lg { width: 100% !important; font-size: 1rem !important; padding: 12px 16px !important; min-height: 48px !important; } .gradio-container button.sm { padding: 8px 12px !important; min-height: 40px !important; } .gradio-container .gr-audio, .gradio-container .gr-audio > div, .gradio-container .gr-audio audio, .gradio-container .gr-audio .waveform-container { max-width: 100% !important; width: 100% !important; box-sizing: border-box !important; } .gradio-container .gr-form, .gradio-container .gr-panel { padding: 8px !important; } .gradio-container .examples table, .gradio-container .examples td { font-size: 0.85rem !important; } .gradio-container .gr-row { flex-direction: column !important; } .gradio-container .gr-row > .gr-column { width: 100% !important; flex: 1 1 100% !important; min-width: 0 !important; } } @media (max-width: 480px) { .gradio-container { padding: 6px !important; } .gradio-container h1 { font-size: 1.1rem !important; } .gradio-container h2, .gradio-container h3 { font-size: 0.98rem !important; } .gradio-container .prose, .gradio-container .markdown-text { font-size: 0.87rem !important; } .gradio-container button.primary, .gradio-container button.lg { font-size: 0.95rem !important; padding: 11px 14px !important; } .gradio-container .tabs > .tab-nav > button { font-size: 0.82rem !important; padding: 6px 10px !important; } } @media (min-width: 1400px) { .gradio-container { max-width: 1400px !important; padding: 20px !important; } } """ with gr.Blocks(title="Higgs Audio v3 + FireRedTTS3 + ACE-Step (CPU)", fill_width=True) as demo: gr.Markdown( """ # 🍋 Higgs Audio v3 + 🔥 FireRedTTS3 + 🎵 ACE-Step 1.5 XL-Turbo — CPU Три генеративные модели в одном Space, все принудительно на **CPU**. Higgs и FireRedTTS3 используют общие **Whisper large-v3-turbo** (ASR) и **N-GAN** (24 кГц → 48 кГц). ACE-Step — собственная аудио-пайплайн, выход WAV. """ ) with gr.Tabs(): # ============ TAB 1 ============ with gr.Tab("🍋 Higgs Audio v3"): gr.Markdown( "Zero-shot TTS & voice cloning на **Higgs Audio v3 (4B)** " "(bf16). Финальный вокодер — **N-GAN** (24 кГц → 48 кГц). " "ASR: **Whisper large-v3-turbo**." ) with gr.Row(): with gr.Column(): higgs_text = gr.Textbox( label="Текст для синтеза", placeholder="Введите, что должен произнести голос…", lines=4, ) higgs_ref_audio = gr.Audio( label="Референсный голос (опционально)", type="filepath", ) higgs_ref_text = gr.Textbox( label="Транскрипт референса (заполняется автоматически)", lines=2, ) with gr.Accordion("Продвинутые настройки", open=False): higgs_temperature = gr.Slider( 0.0, 1.5, value=1.0, step=0.05, label="Temperature", ) higgs_top_p = gr.Slider( 0.1, 1.0, value=0.95, step=0.01, label="Top-p", ) higgs_top_k = gr.Slider( 0, 1026, value=513, step=1, label="Top-k (0 = off)", ) higgs_max_new = gr.Slider( 64, 4096, value=2048, step=64, label="Max new tokens", ) higgs_seed = gr.Number( value=-1, label="Seed (-1 = random)", precision=0, ) higgs_btn = gr.Button("Синтезировать", variant="primary", size="lg") with gr.Column(): higgs_out = gr.Audio( label="Сгенерированная речь (48 кГц, MP3)", type="numpy", format="mp3", ) higgs_ref_audio.change( transcribe_reference, inputs=[higgs_ref_audio], outputs=[higgs_ref_text], ) higgs_btn.click( synthesize_higgs, inputs=[ higgs_text, higgs_ref_audio, higgs_ref_text, higgs_temperature, higgs_top_p, higgs_top_k, higgs_max_new, higgs_seed, ], outputs=[higgs_out], api_name="higgs_synthesize", ) # ============ TAB 2 ============ with gr.Tab("🔥 FireRedTTS3"): gr.Markdown( "Zero-shot voice cloning на **FireRedTTS3** " "(**24 языка + 21 китайский диалект**). Генерация → **N-GAN** " "(24 кГц → 48 кГц) → **MP3**, имя файла совпадает с именем " "референсного аудио (расширение → `.mp3`)." ) with gr.Row(): with gr.Column(): firered_prompt_audio = gr.Audio( label="Reference audio (5–20 s)", sources=["upload", "microphone"], type="filepath", ) firered_prompt_text = gr.Textbox( label="Reference transcript (auto-filled by Whisper)", placeholder="Заполнится автоматически после загрузки…", lines=2, ) firered_text = gr.Textbox( label="Text to synthesize", placeholder="Введите текст, который должен произнести голос…", lines=4, ) firered_language = gr.Dropdown( LANG_CHOICES, value=AUTO, label="TTS language / dialect", info="Auto-detect — fastText по 'Text to synthesize'. " "На Whisper-транскрипцию не влияет.", ) firered_btn = gr.Button( "Generate speech", variant="primary", size="lg", ) with gr.Column(): firered_out = gr.Audio( label="Generated speech (48 kHz, MP3)", type="filepath", format="mp3", ) with gr.Accordion("Advanced options", open=False): firered_cfg = gr.Slider( 0.0, 4.0, value=1.8, step=0.1, label="CFG strength", info="Higher — ближе к тембру референса.", ) firered_steps = gr.Slider( 4, 30, value=18, step=1, label="Flow-matching timesteps", ) firered_seed = gr.Number(value=-1, precision=0, label="Seed") firered_tn = gr.Checkbox( value=True, label="Text normalization (numbers, dates, units → words)", ) gr.Examples( examples=[ [ EN_PROMPT, EN_PROMPT_TEXT, "FireRedTTS3 turns a handful of seconds of speech into a " "voice that can read anything you write.", "English", ], [ ZH_PROMPT, ZH_PROMPT_TEXT, "法院与不动产登记部门加强沟通，并督促银行提前办理抵押预约登记。", "Chinese", ], [ EN_PROMPT, EN_PROMPT_TEXT, "Le modèle peut aussi parler français avec la même voix " "de référence.", "French", ], ], inputs=[ firered_prompt_audio, firered_prompt_text, firered_text, firered_language, ], outputs=[firered_out], fn=voice_clone_firered, cache_examples=True, cache_mode="lazy", ) firered_btn.click( voice_clone_firered, inputs=[ firered_prompt_audio, firered_prompt_text, firered_text, firered_language, firered_cfg, firered_steps, firered_seed, firered_tn, ], outputs=[firered_out], api_name="firered_voice_clone", ) firered_prompt_audio.upload( transcribe_reference, inputs=[firered_prompt_audio], outputs=[firered_prompt_text], show_progress="full", ) # ============ TAB 3 ============ with gr.Tab("🎵 ACE-Step XL-Turbo"): gr.Markdown( "Генерация **музыки** по описанию + куплет/припев на " "**ACE-Step 1.5 XL-Turbo**, принудительно на CPU. " "⚠️ Первый запуск скачивает ~18 ГБ весов; генерация на CPU идёт " "медленно — начинайте с 20–30 секунд. При NaN — авто-retry: " "tier 1 `shift=1.0`, tier 2 `steps=4 + dcw off`." ) with gr.Tabs(): with gr.Tab("Text → Music"): with gr.Row(): with gr.Column(): as_caption = gr.Textbox( label="Описание трека (caption)", placeholder="upbeat indie pop, bright electric guitars, punchy drums, female lead vocal", lines=2, ) with gr.Row(): as_verse = gr.Textbox( label="Куплет [verse]", lines=4, placeholder="Текст куплета…", ) as_chorus = gr.Textbox( label="Припев [chorus]", lines=4, placeholder="Текст припева…", ) as_bridge = gr.Textbox( label="Бридж [bridge] (необязательно)", lines=2, ) as_instrumental = gr.Checkbox( label="Инструментал (без вокала)", value=False, ) as_duration = gr.Slider( 10, 120, value=30, step=5, label="Длительность (сек)", ) with gr.Column(): as_out = gr.Audio( label="Результат", type="filepath", format="wav", ) as_btn = gr.Button( "🎶 Сгенерировать", variant="primary", size="lg", ) with gr.Accordion("Продвинутые настройки", open=False): as_steps = gr.Slider( 4, 20, value=8, step=1, label="Inference steps (Turbo: 8)", ) as_guidance = gr.Slider( 0.0, 20.0, value=7.0, step=0.5, label="Guidance scale (Turbo ignores → 1.0)", ) as_shift = gr.Slider( 1.0, 5.0, value=3.0, step=0.1, label="Shift (Timestep shift)", info="3.0 — стабильный старт; retry-логика автоматическая", ) as_seed = gr.Number( value=-1, precision=0, label="Seed (-1 = random)", ) as_btn.click( acestep_text2music, inputs=[ as_caption, as_verse, as_chorus, as_bridge, as_instrumental, as_duration, as_steps, as_guidance, as_seed, as_shift, ], outputs=[as_out], api_name="acestep_text2music", ) with gr.Tab("🧠 LM Helpers"): gr.Markdown( "**Create Sample** — из свободного описания создаёт " "caption, lyrics и метаданные. " "**Format Sample** — улучшает и структурирует уже " "введённые caption и lyrics." ) with gr.Row(): with gr.Column(): gr.Markdown("### ✨ Create Sample") cs_query = gr.Textbox( label="Свободное описание", placeholder="a soft Bengali love song for a quiet evening", lines=2, ) cs_instrumental = gr.Checkbox( label="Инструментал", value=False, ) cs_lang = gr.Dropdown( choices=["unknown", "en", "ru", "zh", "ja", "es", "fr"], value="unknown", label="Язык вокала (необязательно)", ) cs_btn = gr.Button("Создать сэмпл", variant="secondary") with gr.Column(): gr.Markdown("### 📝 Format Sample") fs_caption = gr.Textbox(label="Caption", lines=2) fs_lyrics = gr.Textbox(label="Lyrics", lines=4) fs_bpm = gr.Number(label="BPM (ограничение)", value=None) fs_keyscale = gr.Textbox(label="Тональность (ограничение)") fs_btn = gr.Button("Форматировать", variant="secondary") with gr.Row(): lm_out_caption = gr.Textbox(label="Caption", interactive=False) lm_out_lyrics = gr.Textbox( label="Lyrics", interactive=False, lines=4, ) with gr.Row(): lm_out_bpm = gr.Number(label="BPM", interactive=False) lm_out_keyscale = gr.Textbox( label="Тональность", interactive=False, ) with gr.Row(): lm_out_duration = gr.Number( label="Длительность", interactive=False, ) lm_out_lang = gr.Textbox(label="Язык", interactive=False) cs_btn.click( acestep_create_sample, inputs=[cs_query, cs_instrumental, cs_lang], outputs=[ lm_out_caption, lm_out_lyrics, lm_out_bpm, lm_out_keyscale, lm_out_duration, lm_out_lang, ], ) fs_btn.click( acestep_format_sample, inputs=[fs_caption, fs_lyrics, fs_bpm, fs_keyscale], outputs=[ lm_out_caption, lm_out_lyrics, lm_out_bpm, lm_out_keyscale, lm_out_duration, lm_out_lang, ], ) gr.Markdown( """ --- **Higgs Audio v3:** [multimodalart/higgs-audio-v3-tts-4b-transformers](https://huggingface.co/multimodalart/higgs-audio-v3-tts-4b-transformers) **FireRedTTS3:** [FireRedTeam/FireRedTTS3](https://huggingface.co/FireRedTeam/FireRedTTS3) (Apache-2.0) **ACE-Step 1.5:** [ace-step/ACE-Step-1.5](https://github.com/ace-step/ACE-Step-1.5) **Whisper:** [openai/whisper-large-v3-turbo](https://huggingface.co/openai/whisper-large-v3-turbo) **N-GAN:** n8 (118K params, Zenodo). Всё работает на **CPU**. Клонирование голоса предназначено **только для академических исследований** — не использовать для имперсонации. """ ) if __name__ == "__main__": demo.queue().launch( theme=gr.themes.Soft(), css=False, mcp_server=True, )