frontend_app.py

from __future__ import annotations

import json
import os
import subprocess

import gradio as gr

from pyharp import *


_BACKEND_PYTHON = os.environ.get("BACKEND_PYTHON", "/opt/backend/bin/python")
_BACKEND_SCRIPT = os.environ.get("BACKEND_SCRIPT", "/app/backend_worker.py")
_BACKEND_TIMEOUT = float(os.environ.get("BACKEND_TIMEOUT", "900"))


def _call_backend(payload):
    # One-shot subprocess into the isolated backend venv. The worker prints
    # exactly one JSON object as its final stdout line; all model/library
    # chatter is redirected to stderr (captured in the Space logs).
    completed = subprocess.run(
        [_BACKEND_PYTHON, _BACKEND_SCRIPT],
        input=json.dumps(payload),
        capture_output=True,
        text=True,
        timeout=_BACKEND_TIMEOUT,
    )
    _lines = [ln for ln in completed.stdout.splitlines() if ln.strip()]
    try:
        response = json.loads(_lines[-1]) if _lines else {}
    except json.JSONDecodeError as exc:
        raise gr.Error(
            "The backend returned no valid result. Last stderr: "
            + (completed.stderr[-1500:] or "(empty)")
        ) from exc
    if not response.get("ok"):
        raise gr.Error(
            response.get("error") or completed.stderr[-1500:] or "Backend worker failed"
        )
    return response.get("outputs") or {}


model_card = ModelCard(
    name="DDSP Timbre Transfer",
    description="Differentiable DSP timbre transfer: recast your audio in the timbre of a pretrained instrument (Violin, Flute, Trumpet, ...). Runs the original, unmodified DDSP stack in an isolated Python 3.9 backend.",
    author="Magenta (Google)",
    tags=["audio-to-audio", "timbre-transfer", "ddsp"],
)


def process_fn(audio, model_name, threshold, adjust, quiet, autotune, pitch_shift, loudness_shift):
    _inputs = dict(zip(["audio", "model_name", "threshold", "adjust", "quiet", "autotune", "pitch_shift", "loudness_shift"], [audio, model_name, threshold, adjust, quiet, autotune, pitch_shift, loudness_shift]))
    _outputs = _call_backend({"inputs": _inputs})
    _out_out_audio = _outputs.get("out_audio")
    if not _out_out_audio:
        raise gr.Error("The backend produced no 'out_audio' output. Check the Space logs (the backend worker's stderr is captured there).")
    return _out_out_audio


with gr.Blocks() as demo:
    input_components = [
        gr.Audio(type="filepath", label="Input audio").harp_required(True),
        gr.Dropdown(choices=["Violin", "Flute", "Flute2", "Trumpet", "Tenor_Saxophone"], value="Violin", label="Instrument"),
        gr.Slider(minimum=0.0, maximum=2.0, step=0.01, value=1.0, label="Note-detection threshold"),
        gr.Checkbox(value=True, label="Auto-adjust to model range"),
        gr.Slider(minimum=0.0, maximum=60.0, step=1.0, value=20.0, label="Quiet parts (dB)"),
        gr.Slider(minimum=0.0, maximum=1.0, step=0.1, value=0.0, label="Autotune amount"),
        gr.Slider(minimum=-2.0, maximum=2.0, step=1.0, value=0.0, label="Pitch shift (octaves)"),
        gr.Slider(minimum=-20.0, maximum=20.0, step=1.0, value=0.0, label="Loudness shift (dB)"),
    ]
    output_components = [
        gr.Audio(type="filepath", label="Timbre-transferred audio"),
    ]
    build_endpoint(
        model_card=model_card,
        input_components=input_components,
        output_components=output_components,
        process_fn=process_fn,
    )

demo.queue().launch(server_name="0.0.0.0", server_port=int(os.environ.get("PORT", "7860")), show_error=True)