AIGame/Tools/generate_enemy_sfx.py

"""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):
    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=62100+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))