from __future__ import annotations import os import time import urllib.request import gradio as gr from pyharp import * from gradio_client import Client, handle_file _BACKEND_SPACE = "ASLP-lab/DiffRhythm2" _BACKEND_API_NAME = "/infer_music" _BACKEND_TOKEN_ENV = "HF_TOKEN" _ACCEPT_USER_TOKEN = False # How many times to wake+retry a sleeping backend, and how long to wait for # it to boot (a free Space cold start can take a few minutes). _CALL_RETRIES = int(os.environ.get("BACKEND_CALL_RETRIES", "4")) _WAKE_TIMEOUT = float(os.environ.get("BACKEND_WAKE_TIMEOUT", "420")) _client = None def _backend_client(): # Lazily create and cache one warm connection using this Space's own # token (from the HF_TOKEN secret) or anonymous if none is set. User # tokens are NOT cached here -- they get a fresh per-call connection. global _client if _client is None: _token = os.environ.get(_BACKEND_TOKEN_ENV) or None _client = Client(_BACKEND_SPACE, hf_token=_token) return _client def _reset_client(): # Drop the cached connection so the next attempt reconnects to a Space # that has since finished waking. global _client _client = None def _make_conn(tok): tok = (tok or '').strip() if tok: return Client(_BACKEND_SPACE, hf_token=tok) return _backend_client() def _space_url(space): slug = space.strip().lower().replace('/', '-').replace('_', '-') return f'https://{slug}.hf.space/' def _is_cold_start(message): # Errors that mean 'the backend was asleep/booting', worth waking+retrying # (vs. a real application error, which we surface immediately). _low = (message or '').lower() return any(s in _low for s in ( 'read operation timed out', 'timed out', 'timeout', 'starting', 'building', 'not ready', 'no application', 'connection', '503', '502', )) def _wake_backend(): # A sleeping Space boots when its URL is hit; poll until it answers (or # the budget expires) so the retried call lands on a running backend. _url = _space_url(_BACKEND_SPACE) _deadline = time.time() + _WAKE_TIMEOUT _delay = 5.0 while time.time() < _deadline: try: _req = urllib.request.Request(_url, headers={'User-Agent': 'harp-frontend'}) with urllib.request.urlopen(_req, timeout=30) as _resp: if getattr(_resp, 'status', 200) < 500: return True except Exception: pass time.sleep(_delay) _delay = min(_delay * 1.5, 30.0) return False def _quota_hint(message): # Turn a backend error into an actionable message. # NOTE: 'message' is the backend's error text; it never contains our token. _low = (message or "").lower() if "quota" in _low or "zerogpu" in _low: if _ACCEPT_USER_TOKEN: return ( "The backend's ZeroGPU quota is exhausted for the identity making " "this call. Paste your own Hugging Face token in the token field " "(read scope) so usage is attributed to your account." ) return ( "The backend's ZeroGPU quota is exhausted. This Space's calls are " "anonymous unless an HF_TOKEN secret is set (Settings -> Variables " "and secrets); use a token from a PRO account or a ZeroGPU-enabled org." ) # Opaque backend failure: the Space raised an exception it refuses to # expose (it runs with show_error=False), so all we get is a generic # 'Internal Gradio error'. The frontend can't fix a server-side crash -- # point the user at where the real cause lives. if ( "internal gradio error" in _low or "internal server error" in _low or "apperror" in _low or _low.strip() in ("", "none") ): _hint = ( "The backend Space raised an error it did not expose (it runs with " "show_error disabled), so the real cause is only in the backend " "Space's Logs tab (" + _BACKEND_SPACE + ")." ) if _ACCEPT_USER_TOKEN: _hint += ( " If it is a ZeroGPU Space, an anonymous call can fail this way -- " "paste a Hugging Face token in the token field and retry." ) return _hint return message or "Backend call failed." model_card = ModelCard( name="Diffrhythm2", description="TODO: describe this model.", author="ASLP-lab", tags=[], ) def process_fn(lrc, audio_prompt, text_prompt, seed, randomize_seed, steps, cfg_strength, file_type): _tok = '' # Call the backend, waking it and retrying if it was asleep (a cold # start otherwise fails the first hit with 'read operation timed out'). _raw = None for _attempt in range(_CALL_RETRIES + 1): try: _conn = _make_conn(_tok) _raw = _conn.predict( lrc, (handle_file(audio_prompt) if audio_prompt else None), text_prompt, seed, randomize_seed, steps, cfg_strength, file_type, 'euler', api_name="/infer_music", ) break except Exception as _exc: # never surfaces the token if _attempt < _CALL_RETRIES and _is_cold_start(str(_exc)): _reset_client() _wake_backend() continue raise gr.Error(_quota_hint(str(_exc))) _values = list(_raw) if isinstance(_raw, (list, tuple)) else [_raw] _detail = " | ".join(str(_v) for _v in _values if isinstance(_v, str) and _v.strip()) _out_audio_result = _values[0] if len(_values) > 0 else None if not _out_audio_result: raise gr.Error(_detail or "The backend Space returned no 'audio_result' output. Check the backend Space's logs; if it uses ZeroGPU it may need a moment to warm up.") return _out_audio_result with gr.Blocks() as demo: input_components = [ gr.Textbox(label="Lyrics", value="[start]\n[intro]\n[verse]\nThought I heard your voice yesterday\nWhen I turned around to say\nThat I loved you baby\nI realize it was juss my mind\nPlayed tricks on me\nAnd it seems colder lately at night\nAnd I try to sleep with the lights on\nEvery time the phone rings\nI pray to God it's you\nAnd I just can't believe\nThat we're through\n[chorus]\nI miss you\nThere's no other way to say it\nAnd I can't deny it\nI miss you\nIt's so easy to see\nI miss you and me\n[verse]\nIs it turning over this time\nHave we really changed our minds about each other's love\nAll the feelings that we used to share\nI refuse to believe\nThat you don't care\n[chorus]\nI miss you\nThere's no other way to say it\nAnd I and I can't deny it\nI miss you\n[verse]\nIt's so easy to see\nI've got to gather myself as together\nI've been through worst kinds of weather\nIf it's over now\n[outro]"), gr.Audio(type="filepath", label="Audio Prompt"), gr.Textbox(label="Text Prompt", value="Pop, Piano, Bass, Drums, Happy"), gr.Slider(minimum=0.0, maximum=10.0, step=1.0, value=0, label="Seed"), gr.Checkbox(value=True, label="Randomize seed"), gr.Slider(minimum=0.0, maximum=32.0, step=1.0, value=16, label="Diffusion Steps"), gr.Slider(minimum=1.3, maximum=100, step=1, value=1.3, label="CFG Strength"), gr.Dropdown(choices=["mp3"], value="mp3", label="Output Format"), ] output_components = [ gr.Audio(type="filepath", label="Audio Result"), ] build_endpoint( model_card=model_card, input_components=input_components, output_components=output_components, process_fn=process_fn, ) demo.queue().launch(share=True, show_error=False, pwa=True)