Function _flowsep_process_chunk

app.py:190–223

(model, preprocessor, chunk_wav, text)

def _flowsep_process_chunk(model, preprocessor, chunk_wav, text):
    device = get_runtime_device()
    chunk_wav = preprocessor.preprocess_chunk(chunk_wav)
    if len(chunk_wav) < FLOWSEP_CHUNK_IN:
        pad = np.zeros(FLOWSEP_CHUNK_IN - len(chunk_wav), dtype=np.float32)
        chunk_wav = np.concatenate([chunk_wav, pad])
    chunk_wav = chunk_wav[:FLOWSEP_CHUNK_IN]
    mixed_mel, stft = preprocessor.wav_feature_extraction(chunk_wav.reshape(1, -1))
    batch = {
        "fname": ["temp"],
        "text": [text],
        "caption": [text],
        "waveform": torch.rand(1, 1, FLOWSEP_CHUNK_IN).to(device),
        "log_mel_spec": torch.rand(1, 1024, 64).to(device),
        "sampling_rate": torch.tensor([FLOWSEP_SR]).to(device),
        "label_vector": torch.rand(1, 527).to(device),
        "stft": torch.rand(1, 1024, 512).to(device),
        "mixed_waveform": torch.from_numpy(chunk_wav.reshape(1, 1, FLOWSEP_CHUNK_IN)).to(device),
        "mixed_mel": mixed_mel.reshape(1, mixed_mel.shape[0], mixed_mel.shape[1]).to(device),
    }
    result = model.generate_sample(
        [batch],
        name="temp_result",
        unconditional_guidance_scale=1.0,
        ddim_steps=20,
        n_gen=1,
        save=False,
        save_mixed=False,
    )
    if isinstance(result, np.ndarray):
        out = result.squeeze()
    else:
        out = result.squeeze().cpu().numpy()
    return out[:FLOWSEP_CHUNK_OUT]


def separate_flowsep(audio_path, text):