92 lines
5.5 KiB
Python
92 lines
5.5 KiB
Python
"""TangoFlux enemy combat one-shots; keep source jobs, hashes and deterministic edits."""
|
|
import hashlib
|
|
import json
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import requests
|
|
from scipy.io import wavfile
|
|
from scipy.signal import butter, sosfilt
|
|
from generate_ui_sfx import wait_job, download
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DESIGN = ROOT / 'Doc/Design/Sound/EnemyCombat'
|
|
OUT = ROOT / 'Tools/output/enemy_combat_sfx'
|
|
BUNDLE = ROOT / 'AIClient/Assets/Bundles/Sound/SFX/EnemyCombat'
|
|
BASE = 'http://127.0.0.1:8012'
|
|
SOUNDS = [
|
|
('Enemy_HornletAttack', 'One small cartoon horn creature headbutt, quick airy swish and compact woody knock, dry close recording', .48),
|
|
('Enemy_DashletAttack', 'One fast cartoon creature dash strike, sharp short sweeping air whip and light punch, dry close recording', .42),
|
|
('Enemy_ShellhornAttack', 'One heavy shell creature shoulder bash, solid hollow stone knock with tiny gravel clicks, dry close recording', .60),
|
|
('Enemy_HornCaptainAttack', 'One forceful cartoon horn creature strike, broad air swoosh then firm wooden thud, dry close recording', .58),
|
|
('Enemy_GreatHornAttack', 'One large fantasy stone creature ground slam, deep rounded punch with brief gritty stone debris, dry close recording', .82),
|
|
('Enemy_HitSoft', 'One small soft cartoon body impact, short clean muted punch and tiny airy pop, dry close recording, no vocalization', .34),
|
|
('Enemy_HitArmor', 'One small rock armor impact, crisp stone clack with two tiny gravel ticks, dry close recording', .42),
|
|
('Enemy_HitShield', 'One small magical shield deflecting a hit, clear bright glass ping with short soft resonant tail, dry close recording', .55),
|
|
]
|
|
|
|
def submit_request(req):
|
|
deadline = time.monotonic()+120
|
|
while time.monotonic() < deadline:
|
|
response = requests.post(BASE+'/api/generate', json={**req, 'sync': False}, timeout=60)
|
|
if response.status_code == 429:
|
|
time.sleep(3)
|
|
continue
|
|
response.raise_for_status()
|
|
return response.json()
|
|
raise TimeoutError('TangoFlux queue stayed full for 120 seconds')
|
|
|
|
def generate(row, seed_base=62100):
|
|
index, (name, prompt, length) = row
|
|
req = dict(prompt=prompt + ', single isolated event at the beginning, silence afterwards, no music, no speech, no ambience', duration=2, steps=50, guidance_scale=4.5, seed=seed_base+index)
|
|
key = hashlib.sha256(json.dumps(req, sort_keys=True).encode()).hexdigest()
|
|
receipt = OUT / (name+'.json')
|
|
source = OUT / (name+'_source.wav')
|
|
old = json.loads(receipt.read_text()) if receipt.exists() else {}
|
|
if old.get('requestHash') == key and source.exists() and old.get('sourceSha256') == hashlib.sha256(source.read_bytes()).hexdigest():
|
|
job = old['job']
|
|
else:
|
|
job = wait_job(BASE, submit_request(req)['id'])
|
|
download(BASE, job, source)
|
|
rate, samples = wavfile.read(source)
|
|
assert rate == 44100 and samples.ndim == 2 and samples.shape[1] == 2
|
|
x = samples.astype(np.float64)
|
|
if np.issubdtype(samples.dtype, np.integer): x /= max(abs(np.iinfo(samples.dtype).min), np.iinfo(samples.dtype).max)
|
|
x = sosfilt(butter(2, 65, fs=rate, btype='highpass', output='sos'), x, axis=0)
|
|
# Locate the strongest 10ms onset; retain a 15ms lead and a bounded gameplay tail.
|
|
hop = rate//100
|
|
env = np.array([np.sqrt(np.mean(x[i:i+hop]**2)) for i in range(0, len(x), hop)])
|
|
peak = int(np.argmax(env))
|
|
start = peak
|
|
while start > 0 and env[start-1] > env[peak]*.2: start -= 1
|
|
start = max(0, start*hop-int(.015*rate))
|
|
n = int(length*rate)
|
|
y = x[start:start+n].copy()
|
|
if len(y) < n: y = np.pad(y, ((0,n-len(y)), (0,0)))
|
|
y[:220] *= np.linspace(0,1,220)[:,None]
|
|
fade = min(2205, len(y)//4)
|
|
y[-fade:] *= np.linspace(1,0,fade)[:,None]
|
|
y *= .80/max(np.max(np.abs(y)), 1e-8)
|
|
target = BUNDLE/(name+'.wav')
|
|
wavfile.write(target, rate, (y*32767).astype(np.int16))
|
|
meta = dict(id=name, request=req, requestHash=key, job=job, source=str(source.relative_to(ROOT)), sourceSha256=hashlib.sha256(source.read_bytes()).hexdigest(), output=str(target.relative_to(ROOT)), outputSha256=hashlib.sha256(target.read_bytes()).hexdigest(), seconds=length, trimStartSeconds=start/rate, sampleRate=rate, channels=2, peakDb=float(20*np.log10(np.max(np.abs(y)))), rmsDb=float(20*np.log10(np.sqrt(np.mean(y*y)))), postprocess='65Hz highpass; onset trim; 5ms fade in / 50ms fade out; peak -1.94 dBFS', license='declare-lab/TangoFlux: noncommercial research weights per project guide')
|
|
receipt.write_text(json.dumps(meta, indent=2), encoding='utf-8')
|
|
print(name+' complete', flush=True)
|
|
return meta
|
|
|
|
if __name__ == '__main__':
|
|
status = requests.get(BASE+'/api/status', timeout=15).json()
|
|
if not status.get('ready') or status.get('error'):
|
|
raise RuntimeError('TangoFlux is not ready; inspect status/warmup per guide: '+str(status))
|
|
for folder in (DESIGN, OUT, BUNDLE): folder.mkdir(parents=True, exist_ok=True)
|
|
# At most two in flight; service limit is four across all callers.
|
|
with ThreadPoolExecutor(max_workers=2) as pool: items = list(pool.map(generate, enumerate(SOUNDS)))
|
|
(DESIGN/'generation.json').write_text(json.dumps(dict(service=BASE, guide='D:/UD/AI/XModel/TangoFlux_AI调用指南.md', status='generated', items=items), ensure_ascii=False, indent=2), encoding='utf-8')
|
|
preview=[]
|
|
for item in items:
|
|
_, data=wavfile.read(ROOT/item['output'])
|
|
preview.extend((data, np.zeros((13230,2), dtype=np.int16)))
|
|
wavfile.write(DESIGN/'preview.wav', 44100, np.concatenate(preview))
|