"""Audio ingest, source-timed renderers, and isolated GPU workers.""" from __future__ import annotations import hashlib import json import os from pathlib import Path import shutil import subprocess import tempfile import time from typing import Iterator from datetime import datetime, timezone from uuid import uuid4 from logic import INSTRUMENTS, drum_hits, instrumental_abc, notes_to_abc, validate_duration ROOT = Path(__file__).resolve().parent SHEET_PYTHON = Path(os.environ.get("HUM_SHEET_PYTHON", "/opt/sheet/bin/python")) YUE_PYTHON = Path(os.environ.get("HUM_YUE_PYTHON", "/opt/yue/bin/python")) SOUNDFONT_DIR = Path(os.environ.get("HUM_SOUNDFONTS", "/opt/soundfonts")) GM_SOUNDFONT = Path(os.environ.get("HUM_GM_SOUNDFONT", "/usr/share/sounds/sf2/FluidR3_GM.sf2")) RECORDINGS_DIR = Path(os.environ.get("HUM_RECORDINGS_DIR", "/data/recordings")) # A pinned MIDI sweep found audible notes 26..45 in Finger Bass YR. Route # whole phrases by octaves so no individual note silently exceeds that range. BASS_SAMPLE_LOW, BASS_SAMPLE_HIGH = 26, 45 SAMPLER_VARIANTS = ( "instrument", "instrument_drums", "contour", "guitar_clean", "guitar_amp", "bass_clean", "bass_amp", ) YUE_VARIANTS = ("yue_solo", "yue_drums", "yue_arrangement") VARIANTS = SAMPLER_VARIANTS + YUE_VARIANTS def command(args: list[str], log: Path, timeout: int = 900) -> None: with log.open("a") as stream: process = subprocess.run(args, stdout=stream, stderr=subprocess.STDOUT, text=True, timeout=timeout, check=False) if process.returncode: tail = "\n".join(log.read_text(errors="replace").splitlines()[-18:]) raise RuntimeError(f"Шаг завершился с кодом {process.returncode}:\n{tail}") def probe(source: Path) -> float: if not source.is_file() or source.stat().st_size > 30 * 1024 * 1024: raise ValueError("Файл отсутствует или больше 30 МБ.") result = subprocess.run([ "ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=duration", "-of", "json", str(source), ], capture_output=True, text=True, check=True) streams = json.loads(result.stdout).get("streams", []) if not streams: raise ValueError("В файле нет звуковой дорожки.") duration = float(streams[0]["duration"]) validate_duration(duration) return duration def decode(source: Path, folder: Path, log: Path) -> tuple[Path, Path, float]: duration = probe(source) pcm48, pcm24 = folder / "source-48k.wav", folder / "source-24k.wav" for rate, target in ((48000, pcm48), (24000, pcm24)): command(["ffmpeg", "-nostdin", "-v", "error", "-y", "-i", str(source), "-vn", "-ac", "1", "-ar", str(rate), "-c:a", "pcm_f32le", str(target)], log, timeout=120) return pcm48, pcm24, duration def source_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for block in iter(lambda: stream.read(1 << 20), b""): digest.update(block) return digest.hexdigest() def archive_recording(source: Path, digest: str, duration: float) -> dict: """Keep the original upload in the private mounted bucket, with a completion record.""" if not RECORDINGS_DIR.is_dir(): raise OSError("Приватный архив записей не подключён к Space") now = datetime.now(timezone.utc) extension = source.suffix.lower() if extension not in {".wav", ".mp3", ".m4a", ".mp4", ".ogg", ".opus", ".flac", ".webm", ".aac", ".aif", ".aiff"}: extension = ".audio" key = f"{now:%Y/%m/%d}/{uuid4().hex}" folder = RECORDINGS_DIR / key folder.mkdir(parents=True, exist_ok=False) saved = folder / f"source{extension}" shutil.copyfile(source, saved) if saved.stat().st_size != source.stat().st_size or source_sha256(saved) != digest: raise OSError("Архивная копия не совпала с исходным файлом") receipt = {"saved_at_utc": now.isoformat(), "source_sha256": digest, "source_bytes": saved.stat().st_size, "source_seconds": round(duration, 3), "source_file": saved.name} (folder / "metadata.json").write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + "\n") return {"requested": True, "saved": True, "archive": "private Hugging Face bucket", "key": key, "source_file": saved.name} def transcribe(pcm24: Path, folder: Path, log: Path) -> tuple[Path, Path, Path, Path]: command([str(SHEET_PYTHON), str(ROOT / "sheet_worker.py"), str(pcm24), str(folder)], log, timeout=900) return (folder / "notes.mid", folder / "melody.abc", folder / "full.abc", folder / "beat.lab") def render_midi(midi_path: Path, output: Path, duration: float, log: Path, soundfont: Path | None = None) -> Path: import numpy as np import soundfile as sf soundfont = soundfont or GM_SOUNDFONT if not soundfont.is_file(): raise RuntimeError(f"SoundFont не найден: {soundfont}") raw = output.with_name(output.stem + "-tail.wav") command(["fluidsynth", "-ni", "-C", "0", "-R", "0", "-r", "48000", "-z", "64", "-g", ".2", "-o", "player.timing-source=sample", "-o", "synth.min-note-length=0", "-T", "wav", "-O", "float", "-F", str(raw), str(soundfont), str(midi_path)], log, timeout=120) audio, rate = sf.read(raw, dtype="float32", always_2d=True) if rate != 48000 or not np.isfinite(audio).all(): raise RuntimeError("Синтезатор вернул повреждённое аудио") frames = round(duration * rate) audio = np.pad(audio, ((0, max(0, frames - len(audio))), (0, 0)))[:frames] sf.write(output, audio, rate, subtype="FLOAT") return output def render_preset(midi_path: Path, folder: Path, name: str, program: int, duration: float, log: Path, *, soundfont: Path | None = None, transpose: int = 0) -> tuple[Path, Path]: import pretty_midi source = pretty_midi.PrettyMIDI(str(midi_path)) notes = [n for track in source.instruments for n in track.notes] if not notes: raise RuntimeError("SheetSage2 не нашёл нот в напеве") midi = pretty_midi.PrettyMIDI(initial_tempo=120, resolution=10000) track = pretty_midi.Instrument(program=program, name=name) for note in notes: pitch = note.pitch + transpose if not 0 <= pitch <= 127: raise ValueError("Транспонированная нота вне MIDI-диапазона") track.notes.append(pretty_midi.Note(velocity=90, pitch=pitch, start=note.start, end=note.end)) midi.instruments.append(track) rendered_midi = folder / f"{name}.mid" midi.write(str(rendered_midi)) return (render_midi(rendered_midi, folder / f"{name}.wav", duration, log, soundfont=soundfont), rendered_midi) def render_instrument(midi_path: Path, folder: Path, instrument: str, duration: float, log: Path) -> tuple[Path, Path]: if instrument not in INSTRUMENTS: raise ValueError("Неизвестный инструмент") return render_preset(midi_path, folder, "instrument", INSTRUMENTS[instrument][0], duration, log) def choose_bass_route(midi_path: Path) -> tuple[int, Path, int, str]: """Choose one octave for the whole phrase; use GM bass if samples cannot fit.""" import pretty_midi pitches = [note.pitch for track in pretty_midi.PrettyMIDI(str(midi_path)).instruments for note in track.notes] if not pitches: raise ValueError("В напеве нет нот для баса") low, high = min(pitches), max(pitches) shifts = range(-108, 61, 12) sampled = [shift for shift in shifts if BASS_SAMPLE_LOW <= low + shift and high + shift <= BASS_SAMPLE_HIGH] if sampled: target = (BASS_SAMPLE_LOW + BASS_SAMPLE_HIGH) / 2 shift = min(sampled, key=lambda value: abs((low + high) / 2 + value - target)) return shift, SOUNDFONT_DIR / "finger-bass.sf2", 0, "FreePats Finger Bass YR" available = [shift for shift in shifts if 0 <= low + shift and high + shift <= 127] shift = min(available, key=lambda value: abs((low + high) / 2 + value - 40)) return shift, GM_SOUNDFONT, 33, "FluidR3 Fingered Bass fallback" def apply_effect(source: Path, output: Path, kind: str, duration: float, log: Path) -> Path: # Both chains use the same sampled notes and retain the source duration. chains = { "guitar": ("highpass=f=75,acompressor=threshold=0.08:ratio=2.5:" "attack=8:release=90,volume=2.2," "asoftclip=type=tanh:threshold=0.55:oversample=4," "lowpass=f=6500,aecho=0.9:0.85:75:0.08"), "bass": ("highpass=f=32,lowpass=f=4500," "acompressor=threshold=0.09:ratio=3:attack=12:release=120," "volume=1.65,asoftclip=type=tanh:threshold=0.72:oversample=4"), } if kind not in chains: raise ValueError("Неизвестный эффект") command(["ffmpeg", "-nostdin", "-v", "error", "-y", "-i", str(source), "-af", chains[kind], "-t", f"{duration:.6f}", "-ar", "48000", "-c:a", "pcm_f32le", str(output)], log, timeout=120) import numpy as np import soundfile as sf audio, rate = sf.read(output, dtype="float32", always_2d=True) if rate != 48000 or abs(len(audio) - round(duration * rate)) > 1 or not np.isfinite(audio).all(): raise RuntimeError("Эффект нарушил длительность или повредил аудио") return output def render_drums(melody_midi: Path, beat_lab: Path, folder: Path, duration: float, log: Path) -> tuple[Path, Path, int]: import pretty_midi source = pretty_midi.PrettyMIDI(str(melody_midi)) spans = [(n.start, n.end) for track in source.instruments for n in track.notes] hits = drum_hits(beat_lab.read_text(), spans, duration) midi = pretty_midi.PrettyMIDI(initial_tempo=120, resolution=10000) kit = pretty_midi.Instrument(program=0, is_drum=True, name="Drum Kit") for start, pitch, velocity in hits: kit.notes.append(pretty_midi.Note(velocity=velocity, pitch=pitch, start=start, end=min(duration, start + .09))) midi.instruments.append(kit) path = folder / "drums.mid" midi.write(str(path)) return render_midi(path, folder / "drums.wav", duration, log), path, len(hits) def pair_with_drums(melody_path: Path, drums_path: Path, folder: Path) -> tuple[Path, Path]: import numpy as np import soundfile as sf melody, rate = sf.read(melody_path, dtype="float32", always_2d=True) drums, drum_rate = sf.read(drums_path, dtype="float32", always_2d=True) if rate != drum_rate or melody.shape != drums.shape: raise RuntimeError("Синтезатор вернул дорожки разной длины") melody_peak, drum_peak = np.max(np.abs(melody)), np.max(np.abs(drums)) if melody_peak <= 1e-8 or drum_peak <= 1e-8: raise RuntimeError("Синтезатор вернул пустую мелодию или барабаны") # Same melody gain in both previews; the peak budget leaves room for drums. melody = melody * (.64 / melody_peak) drums = drums * (.31 / drum_peak) solo, together = folder / "instrument-balanced.wav", folder / "instrument-drums.wav" sf.write(solo, melody, rate, subtype="FLOAT") sf.write(together, melody + drums, rate, subtype="FLOAT") return solo, together def render_contour(source: Path, folder: Path) -> Path: import librosa import numpy as np from scipy.ndimage import median_filter import soundfile as sf y, rate = sf.read(source, dtype="float32") if rate != 48000 or y.ndim != 1: raise ValueError("Expected mono 48 kHz input") reduced = librosa.resample(y, orig_sr=rate, target_sr=16000) f0, voiced, _ = librosa.pyin(reduced, sr=16000, fmin=55., fmax=1000., frame_length=1024, hop_length=160, center=True) times = np.arange(len(f0)) * .01 keep = times < len(y) / rate times, f0, voiced = times[keep], f0[keep], voiced[keep] squares = np.r_[0., np.cumsum(np.square(y, dtype=np.float64))] left = np.maximum(0, np.rint((times - .01) * rate).astype(int)) right = np.minimum(len(y), np.rint((times + .01) * rate).astype(int)) rms = np.sqrt((squares[right] - squares[left]) / np.maximum(1, right - left)) threshold = max(10 ** (-60 / 20), float(np.percentile(rms, 95)) * 10 ** (-35 / 20)) active = voiced & np.isfinite(f0) & (f0 > 0) & (rms >= threshold) boundaries = np.flatnonzero(np.diff(np.r_[False, active, False])) wave = np.zeros(len(y), dtype=np.float32) for first, last in zip(boundaries[::2], boundaries[1::2]): pitch = median_filter(69 + 12 * np.log2(f0[first:last] / 440), size=3, mode="nearest") start = max(0, round((times[first] - .005) * rate)) end = min(len(y), round((times[last - 1] + .005) * rate)) t = np.arange(start, end) / rate envelope = np.interp(t, times[first:last], rms[first:last]) fade = min(round(.005 * rate), len(t) // 2) if fade: envelope[:fade] *= np.linspace(0, 1, fade) envelope[-fade:] *= np.linspace(1, 0, fade) frequency = 440 * 2 ** ((np.interp(t, times[first:last], pitch) - 69) / 12) phase = 2 * np.pi * np.cumsum(frequency) / rate harmonics = np.sin(phase) + .25 * np.sin(2 * phase) + .08 * np.sin(3 * phase) wave[start:end] = envelope * harmonics * np.sqrt(2 / (1 + .25**2 + .08**2)) if not np.any(wave): raise RuntimeError("Не удалось выделить звучащую высоту напева") output = folder / "contour.wav" sf.write(output, wave, rate, subtype="FLOAT") return output def render_yue_batch(melody_abc: Path, full_abc: Path, folder: Path, styles: dict[str, str], seed: int, log: Path) -> Iterator[str]: if not 0 <= seed < 2**31: raise ValueError("Seed должен быть целым числом от 0 до 2³¹−1") for style in styles.values(): if not style.strip() or len(style) > 500: raise ValueError("Каждый промпт YuE2 должен содержать 1–500 символов") melody_path = folder / "instrumental-melody.abc" full_path = folder / "instrumental-full.abc" melody_path.write_text(instrumental_abc(melody_abc.read_text())) full_path.write_text(instrumental_abc(full_abc.read_text(), keep_chords=True)) request = {"runs": [ {"name": "yue_solo", "abc_path": str(melody_path), "style": styles["yue_solo"].strip(), "seed": seed, "cot": "melody"}, {"name": "yue_drums", "abc_path": str(melody_path), "style": styles["yue_drums"].strip(), "seed": seed, "cot": "melody"}, {"name": "yue_arrangement", "abc_path": str(full_path), "style": styles["yue_arrangement"].strip(), "seed": seed, "cot": "full"}, ]} request_path = folder / "yue-requests.json" request_path.write_text(json.dumps(request, ensure_ascii=False, indent=2) + "\n") pending = {run["name"] for run in request["runs"]} with log.open("a") as stream: process = subprocess.Popen([str(YUE_PYTHON), str(ROOT / "yue_worker.py"), str(request_path), str(folder)], stdout=stream, stderr=subprocess.STDOUT) deadline = time.monotonic() + 1800 try: while process.poll() is None: for name in tuple(pending): target = folder / name if (target / "yue.json").is_file() or (target / "error.txt").is_file(): pending.remove(name) yield name if time.monotonic() > deadline: process.kill() process.wait() raise TimeoutError("YuE2 превысил лимит 30 минут") time.sleep(1) for name in tuple(pending): target = folder / name if (target / "yue.json").is_file() or (target / "error.txt").is_file(): yield name if process.returncode: tail = "\n".join(log.read_text(errors="replace").splitlines()[-18:]) raise RuntimeError(f"YuE2 завершился с кодом {process.returncode}:\n{tail}") finally: if process.poll() is None: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait() def encode_mp3(source: Path, destination: Path, log: Path, normalize: bool = True, max_gain: float = 4.) -> None: import numpy as np import soundfile as sf audio, rate = sf.read(source, dtype="float32", always_2d=True) if not len(audio) or not np.isfinite(audio).all(): raise RuntimeError("Результат пустой или содержит повреждённые отсчёты") peak = float(np.max(np.abs(audio))) if peak <= 1e-8: raise RuntimeError("Результат слишком тихий") gain = min(max_gain, .8 / peak) if normalize else 1. safe = destination.with_suffix(".safe.wav") sf.write(safe, audio * gain, rate, subtype="PCM_24") command(["ffmpeg", "-nostdin", "-v", "error", "-y", "-i", str(safe), "-codec:a", "libmp3lame", "-b:a", "320k", "-ar", "48000", str(destination)], log, timeout=120) if not destination.is_file() or destination.stat().st_size < 1000: raise RuntimeError("MP3 не создан") def generate_all(source: str, instrument: str, styles: dict[str, str], seed: int, include_yue: bool = True, progress=None, *, save_recording: bool = False) -> Iterator[tuple]: if not source or instrument not in INSTRUMENTS: raise ValueError("Загрузите или запишите напев и выберите инструмент") folder = Path(tempfile.mkdtemp(prefix="hum-")) log = folder / "run.log" began = time.monotonic() uploaded = Path(source) outputs = {name: None for name in VARIANTS} errors = {} durations = {} yue_timing = {} drum_count = 0 drums_midi = None bass_shift = None bass_source = None transcription_source = "SheetSage2" archive = {"requested": save_recording, "saved": False} def deliver(name: str, raw: Path, *, normalize: bool = True, max_gain: float = 4.) -> None: destination = folder / f"{name}.mp3" encode_mp3(raw, destination, log, normalize=normalize, max_gain=max_gain) outputs[name] = str(destination) durations[name] = round(__import__("soundfile").info(raw).duration, 3) def snapshot(*, complete: bool = False) -> tuple: metadata = {"instrument": instrument, "source_sha256": digest, "source_seconds": round(duration, 3), "result_seconds": dict(durations), "drum_hits": drum_count, "yue_requested": include_yue, "yue_timing": dict(yue_timing), "errors": dict(errors), "complete": complete, "source_timing_preserved": [name for name in SAMPLER_VARIANTS if outputs[name]], "bass_transpose_semitones": bass_shift, "bass_source": bass_source, "transcription_source": transcription_source, "recording_archive": dict(archive), "yue_vocal_absence_guaranteed": False, "elapsed_seconds": round(time.monotonic() - began, 2)} return (*(outputs[name] for name in VARIANTS), str(melody_midi), str(drums_midi) if drums_midi else None, str(beat_lab) if beat_lab.is_file() else None, str(folder / "instrumental-melody.abc"), metadata) if progress: progress(.02, desc="Проверяю и декодирую напев") pcm48, pcm24, duration = decode(uploaded, folder, log) digest = source_sha256(uploaded) if save_recording: try: archive = archive_recording(uploaded, digest, duration) except OSError as error: errors["recording_archive"] = str(error) if progress: progress(.12, desc="SheetSage2 извлекает мелодию и сетку долей") try: midi, melody_abc, full_abc, beat_lab = transcribe(pcm24, folder, log) except RuntimeError as error: if "SheetSage2 could not export" not in str(error) or "ABC" not in str(error): raise errors["sheet_sage"] = "SheetSage2 не смог записать ноты в ABC" midi, melody_abc, full_abc, beat_lab = ( folder / "notes.mid", folder / "melody.abc", folder / "full.abc", folder / "beat.lab") import pretty_midi if not midi.is_file() or not any(track.notes for track in pretty_midi.PrettyMIDI(str(midi)).instruments): from f0_fallback import transcribe_f0 transcribe_f0(pcm24, midi) notes = [(note.start, note.end, note.pitch) for track in pretty_midi.PrettyMIDI(str(midi)).instruments for note in track.notes] abc = notes_to_abc(notes, duration) melody_abc.write_text(abc) full_abc.write_text(abc) transcription_source = "pYIN fallback" errors["sheet_sage"] = "SheetSage2 не нашёл нот; использован резервный F0" bass_shift, bass_font, bass_program, bass_source = choose_bass_route(midi) if progress: progress(.30, desc="Синтезирую инструмент и барабаны") melody_raw, melody_midi = render_instrument(midi, folder, instrument, duration, log) try: if not beat_lab.is_file(): raise ValueError("SheetSage2 не дал сетку долей; барабаны пропущены") drums_raw, drums_midi, drum_count = render_drums(midi, beat_lab, folder, duration, log) solo_raw, combined_raw = pair_with_drums(melody_raw, drums_raw, folder) deliver("instrument", solo_raw, normalize=False) deliver("instrument_drums", combined_raw, normalize=False) except (ValueError, RuntimeError, OSError) as error: errors["instrument_drums"] = str(error) deliver("instrument", melody_raw) drum_count = 0 if progress: progress(.42, desc="Синтезирую непрерывную высоту") try: deliver("contour", render_contour(pcm48, folder)) except (ValueError, RuntimeError, OSError) as error: errors["contour"] = str(error) (folder / "instrumental-melody.abc").write_text( instrumental_abc(melody_abc.read_text())) yield snapshot() if progress: progress(.48, desc="Сэмплирую электрогитару и бас") for name, font, program, transpose, effect in ( ("guitar_clean", SOUNDFONT_DIR / "electric-clean.sf2", 0, 0, "guitar"), ("bass_clean", bass_font, bass_program, bass_shift, "bass"), ): try: raw, _ = render_preset(midi, folder, name, program, duration, log, soundfont=font, transpose=transpose) deliver(name, raw, max_gain=20.) deliver(name.replace("clean", "amp"), apply_effect(raw, folder / f"{name}-amp.wav", effect, duration, log), max_gain=20.) except (ValueError, RuntimeError, OSError) as error: errors[name] = str(error) yield snapshot() if not include_yue: final = snapshot(complete=True) (folder / "result.json").write_text(json.dumps(final[-1], ensure_ascii=False, indent=2) + "\n") yield final return if progress: progress(.50, desc="YuE2 создаёт сразу три варианта; это займёт несколько минут") completed = set() def accept_yue(name: str) -> None: raw = folder / name / "raw.wav" if raw.is_file(): try: deliver(name, raw) receipt = folder / name / "yue.json" if receipt.is_file(): yue_timing[name] = json.loads(receipt.read_text())["timing"] except (ValueError, RuntimeError, OSError) as error: errors[name] = str(error) else: detail = folder / name / "error.txt" errors[name] = (detail.read_text(errors="replace").splitlines()[-1] if detail.is_file() else "YuE2 не вернул файл") completed.add(name) try: for name in render_yue_batch(melody_abc, full_abc, folder, styles, int(seed), log): accept_yue(name) if progress: progress(.50 + .15 * len(completed), desc=f"YuE2: {len(completed)} из 3 готовы") yield snapshot() except (ValueError, RuntimeError, OSError) as error: errors["yue_batch"] = str(error) for name in YUE_VARIANTS: if name not in completed: accept_yue(name) final = snapshot(complete=True) (folder / "result.json").write_text(json.dumps(final[-1], ensure_ascii=False, indent=2) + "\n") yield final