Blame for app.py

"""CPU-capable, renderer-constrained speech-to-piano demonstration."""
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import shutil
import time

os.environ.setdefault('OMP_NUM_THREADS', '2')
os.environ.setdefault('MPLBACKEND', 'Agg')
import gradio as gr
import numpy as np
import soundfile as sf
from player import player_component, player_payload

ROOT = Path(__file__).resolve().parent
VOICE = '一位温柔自信的年轻女性，声音清晰，语气亲切，表达轻快而富有感染力。'
MODEL = 'Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign'
DEVICE = os.getenv('INFERENCE_DEVICE', 'cpu')
if DEVICE not in {'cpu', 'cuda', 'auto', 'zerogpu'}:
    raise ValueError('INFERENCE_DEVICE must be cpu, cuda, auto or zerogpu')
if DEVICE == 'zerogpu':
    from zero_tts import generate_speech as generate_zero_speech

RUNS = ROOT / 'outputs' / 'runs'
RUNS.mkdir(parents=True, exist_ok=True)


def worker(command, log, timeout=1800):
    """Bound each job and release all model memory when the child exits."""
    env = os.environ.copy()
    if DEVICE in {'cpu', 'zerogpu'}:
        env['CUDA_VISIBLE_DEVICES'] = ''
        env['SPACES_ZERO_GPU'] = '0'
    with log.open('w') as stream:
        process = subprocess.Popen([sys.executable, *command], cwd=ROOT, env=env,
                                   stdout=stream, stderr=subprocess.STDOUT)
        try:
            process.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            process.kill()
            process.wait()
            raise gr.Error('This job exceeded 30 minutes. Try a shorter clip or fewer fitting iterations.') from None
        except BaseException:
            process.kill()
            process.wait()
            raise
    if process.returncode:
        print(log.read_text()[-6000:], flush=True)
        raise gr.Error('Generation failed. Please try a shorter clip; server logs contain details.')


def prepare_audio(path, out, seconds):
    if not path:
        raise gr.Error('Upload or record an audio clip first.')
    info = sf.info(path)
    if not 8000 <= info.samplerate <= 192000 or info.channels > 8:
        raise gr.Error('Use audio with 1–8 channels and an 8–192 kHz sample rate.')
    # Read only the selected prefix, bounding memory even for long uploads.
    audio, rate = sf.read(path, frames=int(float(seconds) * info.samplerate), always_2d=True)
    audio = audio.mean(axis=1)
    if len(audio) < rate * .2 or not np.isfinite(audio).all() or np.max(np.abs(audio)) < 1e-7:
        raise gr.Error('Audio must contain at least 0.2 seconds of finite, non-silent sound.')
    sf.write(out, audio, rate, subtype='FLOAT')
    return len(audio) / rate


def result(directory, status, stem='dense'):
    return (status, str(directory/'target.wav'), player_payload(directory, stem=stem),
            str(directory/f'{stem}.mid'))


def run(mode, audio, text, language, description, seconds, steps, fitting_style='Speech (dense)', progress=gr.Progress()):
    if mode not in {'Audio → MIDI', 'Text → speech → MIDI'} or not .5 <= float(seconds) <= 30 or not 200 <= int(steps) <= 1200:
        raise gr.Error('Choose a valid mode, duration (0.5–30 s), and iteration count (200–1200).')
    if fitting_style not in {'Speech (dense)', 'Instrumental (sparse)'}:
        raise gr.Error('Choose Speech (dense) or Instrumental (sparse) fitting.')
    stem = 'instrumental' if fitting_style == 'Instrumental (sparse)' else 'dense'
    if language not in {'Auto', 'English', 'Chinese', 'Japanese', 'Korean', 'French', 'German', 'Spanish', 'Portuguese', 'Russian'}:
        raise gr.Error('Choose a supported language.')
    for old in RUNS.glob('job-*'):
        if old.is_dir() and time.time() - old.stat().st_mtime > 86400:
            shutil.rmtree(old)
    directory = Path(tempfile.mkdtemp(prefix='job-', dir=RUNS))
    empty = (None, '', None)
    started = time.monotonic()
    if mode == 'Text → speech → MIDI':
        if not text.strip() or len(text) > 300 or not description.strip() or len(description) > 600:
            raise gr.Error('Enter 1–300 text characters and a voice description of 1–600 characters.')
        yield ('Generating speech. '+('Waiting for a shared GPU.' if DEVICE == 'zerogpu' else 'CPU VoiceDesign can take several minutes on first use.'), *empty)
        progress(.05, desc='Loading VoiceDesign and generating speech')
        prompts = directory/'prompts.json'
        prompts.write_text(json.dumps([dict(name='speech', text=text.strip(), language=language, instruct=description.strip())]))
        command = ['scripts/generate_speech.py', '--device', DEVICE, '--model', MODEL,
                   '--samples', str(prompts), '--out', str(directory)]
        if os.getenv('QWEN_MODEL_PATH'):
            command += ['--model-path', os.environ['QWEN_MODEL_PATH']]
        if DEVICE == 'zerogpu':
            wav, rate = generate_zero_speech(text.strip(), language, description.strip())
            sf.write(directory/'speech.wav', wav, rate, subtype='FLOAT')
        else:
            worker(command, directory/'tts.log')
        audio = str(directory/'speech.wav')
    duration = prepare_audio(audio, directory/'input.wav', seconds)
    yield (f'Fitting the first {duration:.2f} s · {fitting_style}.',
           str(directory/'input.wav'), '', None)
    progress(.3, desc=f'Fitting {stem} piano notes and rendering the MIDI')
    module = 'speaking_midi.instrumental_demo' if stem == 'instrumental' else 'speaking_midi.dense'
    command = ['-m', module, str(directory/'input.wav'), '--out', str(directory),
               '--soundfont', str(ROOT/'assets/gm.sf3'), '--steps', str(int(steps)), '--device', 'cpu']
    worker(command, directory/'fit.log')
    progress(.95, desc='Preparing the MIDI player')
    report = json.loads((directory/'report.json').read_text())
    yield result(directory, f"Complete in {time.monotonic()-started:.1f} s · {report['notes']:,} piano notes · {fitting_style}.", stem=stem)


def example(language):
    name = {'en': 'the-tyger', 'zh': 'quiet-night'}[language]
    return result(ROOT/'examples/dense'/name, 'Precomputed poetry voice · dense piano fitting.')


def build_ui():
    with gr.Blocks(title='Speaking MIDI') as demo:
        gr.Markdown('# Speaking MIDI\nTurn speech or music into piano notes. Listen, follow the piano roll, and download the MIDI.\n'
                    'This is experimental sound approximation: words may become difficult to understand.')
        with gr.Row():
            english = gr.Button('Listen to The Tyger · English')
            chinese = gr.Button('Listen to Quiet Night · Chinese')
        with gr.Row():
            with gr.Column(scale=1):
                mode = gr.Radio(['Audio → MIDI', 'Text → speech → MIDI'], value='Audio → MIDI', label='Start with')
                audio = gr.Audio(sources=['upload', 'microphone'], type='filepath', label='Input audio')
                with gr.Group(visible=False) as speech_controls:
                    text = gr.Textbox(value='Hello, can you hear the piano speaking?', label='Words to speak', lines=3, max_length=300)
                    language = gr.Dropdown(['Auto', 'English', 'Chinese', 'Japanese', 'Korean', 'French', 'German', 'Spanish', 'Portuguese', 'Russian'], value='English', label='Language')
                    description = gr.Textbox(value=VOICE, label='Voice description', lines=3, max_length=600)
                fitting_style = gr.Radio(['Speech (dense)', 'Instrumental (sparse)'], value='Speech (dense)',
                                         label='Fitting style', info='Speech keeps dense short notes. Instrumental uses fewer, longer notes.')
                with gr.Accordion('Fitting options', open=False):
                    seconds = gr.Slider(.5, 30, value=30, step=.5, label='Fit the first N seconds')
                    steps = gr.Slider(200, 1200, value=600, step=200, label='Fitting iterations',
                                      info='600 is the tested setting. More iterations take longer and may not improve clarity.')
                submit = gr.Button('Generate piano MIDI', variant='primary')
                gr.Markdown('CPU inference is supported. Speech generation can take several minutes; piano fitting runs on CPU. '
                            'Try the ready-made examples for instant playback. Clips are processed up to the selected duration.')
            with gr.Column(scale=2):
                status = gr.Markdown('Choose an example or create a new fit.')
                source = gr.Audio(label='1 · Source used for fitting', interactive=False)
                player = player_component(gr)
                midi = gr.File(label='Download MIDI', interactive=False)
        poetry = json.loads((ROOT/'examples/poetry/prompts.json').read_text())
        gr.Examples(
            examples=[['Audio → MIDI', str(ROOT/'examples/poetry'/f"{row['name']}.wav"),
                       row['text'], row['language'], 12, 'Speech (dense)'] for row in poetry],
            inputs=[mode, audio, text, language, seconds, fitting_style], cache_examples=False,
            label='Poetry voices · select to listen, then generate piano MIDI',
            example_labels=[f"{row['title']} — {row['author']} ({row['language']})" for row in poetry],
        )
        selected_text = gr.Textbox(label='Selected text', interactive=False, lines=2)
        text.change(lambda value: value, text, selected_text, queue=False)
        gr.Markdown('**How it works:** render piano templates → fit dense short notes or sparse held notes → '
                    'convert strengths to MIDI velocities → render the MIDI with the same piano sound. '
                    '[Algorithm and experiments](https://github.com/k-l-lambda/speaking-midi) · '
                    '[Qwen3-TTS](https://huggingface.co/Qwen/Qwen3-TTS-12Hz-1.7B-VoiceDesign)')
        outputs = [status, source, player, midi]
        mode.change(lambda value: (gr.update(visible=value=='Audio → MIDI'), gr.update(visible=value!='Audio → MIDI')),
                    mode, [audio, speech_controls], queue=False)
        submit.click(run, [mode,audio,text,language,description,seconds,steps,fitting_style], outputs,
                     concurrency_limit=1, concurrency_id='compute', api_name='generate')
        english.click(lambda: example('en'), outputs=outputs, api_name='english_example')
        chinese.click(lambda: example('zh'), outputs=outputs, api_name='chinese_example')
    return demo.queue(max_size=8, default_concurrency_limit=1)


if __name__ == '__main__':
    build_ui().launch(server_name='0.0.0.0', server_port=int(os.getenv('PORT', '7860')),
                      max_file_size='25mb', allowed_paths=[str(RUNS), str(ROOT/'examples')])