File size: 5,997 Bytes
b281264 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 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) |