Create yml/separate-audio.py

yml/separate-audio.py

@@ -0,0 +1,167 @@
+import os
+import sys
+import requests
+from gradio_client import Client, handle_file
+
+raw_prompt = os.environ.get('PROMPT', '')
+run_id = os.environ.get('RUN_ID', '')
+space_url = os.environ.get('SPACE_URL', '')
+github_run_id = os.environ.get('GITHUB_RUN_ID', '')
+
+def report_failure(error_msg):
+    try:
+        requests.post(
+            f"{space_url}/api/webhook/fail",
+            json={
+                "run_id": run_id,
+                "error": error_msg,
+                "event_type": "separate-audio",
+                "client_payload": {
+                    "prompt": raw_prompt,
+                    "run_id": run_id,
+                    "space_url": space_url
+                },
+                "github_run_id": github_run_id
+            },
+            timeout=15
+        )
+    except Exception as e:
+        print(f"Failed to report failure: {e}")
+
+print('1. Decoding configuration from separate payload...')
+if not raw_prompt.startswith("VOICECONFIG_SEPARATE_"):
+    err_str = "Error: Invalid separate configuration payload signature."
+    print(err_str)
+    report_failure(err_str)
+    sys.exit(1)
+
+config_str = raw_prompt[len("VOICECONFIG_SEPARATE_"):]
+parts = config_str.split("_")
+config = {}
+i = 0
+while i < len(parts) - 1:
+    key = parts[i]
+    val = parts[i+1]
+    config[key] = val
+    i += 2
+
+user_run_id = config.get("userRunId", run_id)
+ext = config.get("ext", "mp3")
+stem = config.get("stem", "vocal")
+
+# تبدیل مقادیر رشته‌ای به مقادیر و متغیرهای اصلی پایتون
+main = config.get("main", "false").lower() == "true"
+dereverb = config.get("dereverb", "false").lower() == "true"
+vocal_effects = config.get("vocEff", "false").lower() == "true"
+background_effects = config.get("bgEff", "false").lower() == "true"
+
+vocal_reverb_room_size = float(config.get("vRevRoom", "0.15"))
+vocal_reverb_damping = float(config.get("vRevDamp", "0.7"))
+vocal_reverb_dryness = float(config.get("vRevDry", "0.8"))
+vocal_reverb_wet_level = float(config.get("vRevWet", "0.2"))
+vocal_delay_seconds = float(config.get("vDelaySec", "0.0"))
+vocal_delay_mix = float(config.get("vDelayMix", "0.0"))
+vocal_compressor_threshold_db = float(config.get("vCompThresh", "-15"))
+vocal_compressor_ratio = float(config.get("vCompRatio", "4"))
+vocal_compressor_attack_ms = float(config.get("vCompAttack", "1"))
+vocal_compressor_release_ms = float(config.get("vCompRelease", "100"))
+vocal_gain_db = float(config.get("vGain", "0"))
+
+background_highpass_freq = float(config.get("bgHigh", "120"))
+background_lowpass_freq = float(config.get("bgLow", "11000"))
+background_reverb_room_size = float(config.get("bgRevRoom", "0.1"))
+background_reverb_damping = float(config.get("bgRevDamp", "0.5"))
+background_reverb_wet_level = float(config.get("bgRevWet", "0.25"))
+background_compressor_threshold_db = float(config.get("bgCompThresh", "-15"))
+background_compressor_ratio = float(config.get("bgCompRatio", "4"))
+background_compressor_attack_ms = float(config.get("bgCompAttack", "15"))
+background_compressor_release_ms = float(config.get("bgCompRelease", "60"))
+background_gain_db = float(config.get("bgGain", "0"))
+target_format = config.get("format", "WAV")
+
+input_audio_url = f"{space_url}/static/images/{user_run_id}_input.{ext}"
+local_input = f"input.{ext}"
+
+print("2. Downloading source audio from host...")
+try:
+    r_input = requests.get(input_audio_url, timeout=60)
+    if r_input.status_code != 200:
+        raise Exception(f"Input audio download failed. Status: {r_input.status_code}")
+    with open(local_input, 'wb') as f:
+        f.write(r_input.content)
+except Exception as download_err:
+    err_str = f"Error downloading source files: {download_err}"
+    print(err_str)
+    report_failure(err_str)
+    sys.exit(1)
+
+print("3. Connecting to Audio Separator Space...")
+try:
+    client = Client("https://r3gm-audio-separator.hf.space/")
+    
+    print("4. Executing separation engine...")
+    result = client.predict(
+        handle_file(local_input),
+        [stem],
+        main,
+        dereverb,
+        vocal_effects,
+        background_effects,
+        vocal_reverb_room_size,
+        vocal_reverb_damping,
+        vocal_reverb_dryness,
+        vocal_reverb_wet_level,
+        vocal_delay_seconds,
+        vocal_delay_mix,
+        vocal_compressor_threshold_db,
+        vocal_compressor_ratio,
+        vocal_compressor_attack_ms,
+        vocal_compressor_release_ms,
+        vocal_gain_db,
+        background_highpass_freq,
+        background_lowpass_freq,
+        background_reverb_room_size,
+        background_reverb_damping,
+        background_reverb_wet_level,
+        background_compressor_threshold_db,
+        background_compressor_ratio,
+        background_compressor_attack_ms,
+        background_compressor_release_ms,
+        background_gain_db,
+        target_format,
+        fn_index=3
+    )
+
+    def parse_file_response(f):
+        if not f: return None
+        if isinstance(f, (list, tuple)):
+            if len(f) > 0:
+                return parse_file_response(f[0])
+        if isinstance(f, dict):
+            return f.get('path') or f.get('name')
+        return str(f)
+
+    final_audio_path = parse_file_response(result)
+
+    if not final_audio_path or not os.path.exists(str(final_audio_path)):
+        raise Exception("Audio separation output file was not found or invalid.")
+
+    print("5. Uploading result file back...")
+    ext_out = target_format.lower()
+    with open(final_audio_path, 'rb') as f:
+        res_upload = requests.post(
+            f'{space_url}/api/webhook/upload', 
+            data={'run_id': run_id, 'github_run_id': github_run_id, 'ext': ext_out}, 
+            files={'file': f}
+        )
+
+    if res_upload.status_code == 200:
+        print('6. SUCCESS! Process complete.')
+    else:
+        raise Exception(f"Webhook upload failed. Status code: {res_upload.status_code}")
+
+except Exception as e:
+    err_str = str(e)
+    print(f"CRITICAL ERROR during separation: {err_str}")
+    report_failure(err_str)
+    sys.exit(1)