323 lines
13 KiB
Python
323 lines
13 KiB
Python
import uuid
|
|
import requests
|
|
import torch
|
|
import numpy as np
|
|
import librosa
|
|
import tempfile
|
|
import os
|
|
from .utils import runwareUtils as rwUtils
|
|
|
|
|
|
class RunwareAudioInference:
|
|
"""Runware Audio Inference node for generating audio using Runware API"""
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"model": ("RUNWAREAUDIOMODEL", {
|
|
"tooltip": "AI model to use for audio generation"
|
|
}),
|
|
"usePositivePrompt": ("BOOLEAN", {
|
|
"default": True,
|
|
"tooltip": "Enable/disable positivePrompt parameter in API request"
|
|
}),
|
|
"positivePrompt": ("STRING", {
|
|
"multiline": True,
|
|
"default": "classical piano piece, gentle and melodic",
|
|
"tooltip": "Text description that guides the audio generation process"
|
|
}),
|
|
"negativePrompt": ("STRING", {
|
|
"multiline": True,
|
|
"default": "",
|
|
"tooltip": "Text description of what you don't want in the audio"
|
|
}),
|
|
"useDuration": ("BOOLEAN", {
|
|
"default": True,
|
|
"tooltip": "Enable/disable duration parameter in API request"
|
|
}),
|
|
"duration": ("INT", {
|
|
"default": 10,
|
|
"min": 10,
|
|
"max": 300,
|
|
"step": 1,
|
|
"tooltip": "Length of generated audio in seconds (10-300)"
|
|
}),
|
|
"useSampleRate": ("BOOLEAN", {
|
|
"default": True,
|
|
"tooltip": "Enable/disable sampleRate parameter in API request"
|
|
}),
|
|
"sampleRate": ("INT", {
|
|
"default": 22050,
|
|
"min": 8000,
|
|
"max": 48000,
|
|
"step": 1,
|
|
"tooltip": "Sample rate of the generated audio in Hz (8000-48000)"
|
|
}),
|
|
"useBitrate": ("BOOLEAN", {
|
|
"default": True,
|
|
"tooltip": "Enable/disable bitrate parameter in API request"
|
|
}),
|
|
"bitrate": ("INT", {
|
|
"default": 32,
|
|
"min": 32,
|
|
"max": 320,
|
|
"step": 32,
|
|
"tooltip": "Bitrate of the generated audio in kbps (32-320)"
|
|
}),
|
|
"outputType": (["URL", "dataURI", "base64Data"], {
|
|
"default": "URL",
|
|
"tooltip": "Output type for the generated audio"
|
|
}),
|
|
"outputFormat": (["MP3"], {
|
|
"default": "MP3",
|
|
"tooltip": "Format of the output audio"
|
|
}),
|
|
"numberResults": ("INT", {
|
|
"default": 1,
|
|
"min": 1,
|
|
"max": 3,
|
|
"tooltip": "Number of audio files to generate"
|
|
}),
|
|
},
|
|
"optional": {
|
|
"inputs": ("RUNWAREAUDIOINFERENCEINPUTS", {
|
|
"tooltip": "Custom inputs for audio generation (e.g., video URL for audio extraction)"
|
|
}),
|
|
"providerSettings": ("RUNWAREPROVIDERSETTINGS", {
|
|
"tooltip": "Provider-specific configuration settings"
|
|
}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("audio",)
|
|
FUNCTION = "generateAudio"
|
|
CATEGORY = "Runware/Audio"
|
|
|
|
def generateAudio(self, **kwargs):
|
|
"""Main function to generate audio using Runware API"""
|
|
params = self._extractParameters(kwargs)
|
|
genConfig = self._buildGenConfig(params)
|
|
genResult = rwUtils.inferenecRequest(genConfig)
|
|
|
|
self._logResponse(genResult)
|
|
self._validateResponse(genResult)
|
|
|
|
audioUrl = self._extractAudioUrl(genResult)
|
|
audioObj = self._downloadAndProcessAudio(audioUrl, params["sampleRate"])
|
|
|
|
return (audioObj,)
|
|
|
|
def _extractParameters(self, kwargs):
|
|
"""Extract and validate parameters from kwargs"""
|
|
return {
|
|
"positivePrompt": kwargs.get("positivePrompt", ""),
|
|
"usePositivePrompt": kwargs.get("usePositivePrompt", True),
|
|
"model": kwargs.get("model", ""),
|
|
"duration": kwargs.get("duration", 30),
|
|
"useDuration": kwargs.get("useDuration", True),
|
|
"sampleRate": kwargs.get("sampleRate", 22050),
|
|
"useSampleRate": kwargs.get("useSampleRate", True),
|
|
"bitrate": kwargs.get("bitrate", 32),
|
|
"useBitrate": kwargs.get("useBitrate", True),
|
|
"outputType": kwargs.get("outputType", "URL"),
|
|
"outputFormat": kwargs.get("outputFormat", "MP3"),
|
|
"negativePrompt": kwargs.get("negativePrompt", ""),
|
|
"numberResults": kwargs.get("numberResults", 1),
|
|
"inputs": kwargs.get("inputs", None),
|
|
"providerSettings": kwargs.get("providerSettings", None),
|
|
}
|
|
|
|
def _hasSections(self, providerSettings):
|
|
"""Check if provider settings contain sections"""
|
|
if not providerSettings or not isinstance(providerSettings, dict):
|
|
return False
|
|
|
|
# Provider settings is now flat (e.g., {"music": {...}})
|
|
# Check if it has music.compositionPlan.sections
|
|
musicSettings = providerSettings.get("music", {})
|
|
compositionPlan = musicSettings.get("compositionPlan", {})
|
|
sections = compositionPlan.get("sections", [])
|
|
|
|
return len(sections) > 0
|
|
|
|
def _buildGenConfig(self, params):
|
|
"""Build the generation configuration for API request"""
|
|
taskUuid = str(uuid.uuid4())
|
|
|
|
genConfig = [{
|
|
"taskType": "audioInference",
|
|
"taskUUID": taskUuid,
|
|
"model": params["model"],
|
|
"outputType": params["outputType"],
|
|
"outputFormat": params["outputFormat"],
|
|
"deliveryMethod": "sync",
|
|
"includeCost": True,
|
|
"numberResults": params["numberResults"],
|
|
}]
|
|
|
|
# Build audioSettings conditionally based on use flags
|
|
audioSettings = {}
|
|
if params["useSampleRate"]:
|
|
audioSettings["sampleRate"] = params["sampleRate"]
|
|
if params["useBitrate"]:
|
|
audioSettings["bitrate"] = params["bitrate"]
|
|
|
|
# Only add audioSettings if at least one setting is enabled
|
|
if audioSettings:
|
|
genConfig[0]["audioSettings"] = audioSettings
|
|
|
|
# Handle sections - disable duration if sections are provided
|
|
hasSections = self._hasSections(params["providerSettings"])
|
|
if hasSections:
|
|
params["useDuration"] = False
|
|
print(f"[DEBUG] Disabled duration because sections are provided")
|
|
|
|
# Add prompts conditionally
|
|
if params["usePositivePrompt"]:
|
|
genConfig[0]["positivePrompt"] = params["positivePrompt"]
|
|
print(f"[DEBUG] Added positivePrompt: '{params['positivePrompt']}'")
|
|
else:
|
|
print(f"[DEBUG] Skipped positivePrompt")
|
|
|
|
if params["useDuration"]:
|
|
genConfig[0]["duration"] = params["duration"]
|
|
print(f"[DEBUG] Added duration: {params['duration']}")
|
|
else:
|
|
print(f"[DEBUG] Skipped duration")
|
|
|
|
# Add optional parameters
|
|
if params["negativePrompt"]:
|
|
genConfig[0]["negativePrompt"] = params["negativePrompt"]
|
|
|
|
# Handle inputs - merge custom inputs from Audio Inference Inputs node
|
|
if params["inputs"] is not None:
|
|
# Merge inputs from audio inference inputs node
|
|
if "inputs" not in genConfig[0]:
|
|
genConfig[0]["inputs"] = {}
|
|
|
|
# Merge each input from inputs
|
|
for key, value in params["inputs"].items():
|
|
genConfig[0]["inputs"][key] = value
|
|
|
|
print(f"[DEBUG] Audio inference inputs merged: {rwUtils.sanitize_for_logging(params['inputs'])}")
|
|
print(f"[DEBUG] Final genConfig inputs: {rwUtils.sanitize_for_logging(genConfig[0].get('inputs', {}))}")
|
|
|
|
# Handle providerSettings - extract provider name from model and wrap with provider name (same pattern as video inference)
|
|
if params["providerSettings"] is not None:
|
|
# Extract provider name from model (e.g., "klingai:8@1" -> "klingai", "elevenlabs:1@1" -> "elevenlabs")
|
|
provider_name = params["model"].split(":")[0] if ":" in params["model"] else params["model"]
|
|
|
|
# If providerSettings is a dictionary, create the correct API format
|
|
if isinstance(params["providerSettings"], dict):
|
|
# Create the providerSettings object with provider name as key
|
|
final_provider_settings = {
|
|
provider_name: params["providerSettings"]
|
|
}
|
|
genConfig[0]["providerSettings"] = final_provider_settings
|
|
print(f"[DEBUG] Provider settings wrapped with provider name: {rwUtils.sanitize_for_logging(final_provider_settings)}")
|
|
else:
|
|
# If it's just a string, use it directly
|
|
genConfig[0]["providerSettings"] = params["providerSettings"]
|
|
|
|
print(f"[DEBUG] Sending Audio Inference Request:")
|
|
print(f"[DEBUG] Request Payload: {rwUtils.safe_json_dumps(genConfig, indent=2)}")
|
|
|
|
return genConfig
|
|
|
|
def _logResponse(self, genResult):
|
|
"""Log the API response for debugging"""
|
|
print(f"[DEBUG] Received Audio Inference Response:")
|
|
print(f"[DEBUG] Response: {rwUtils.safe_json_dumps(genResult, indent=2)}")
|
|
|
|
def _validateResponse(self, genResult):
|
|
"""Validate API response and raise errors if needed"""
|
|
if "errors" in genResult:
|
|
errorMessage = genResult["errors"][0]["message"]
|
|
raise Exception(f"Audio generation failed: {errorMessage}")
|
|
|
|
if "data" not in genResult or len(genResult["data"]) == 0:
|
|
raise Exception("No audio data received from API")
|
|
|
|
def _extractAudioUrl(self, genResult):
|
|
"""Extract audio URL from API response"""
|
|
audioData = genResult["data"][0]
|
|
|
|
audioUrl = audioData.get("audioURL", "")
|
|
if not audioUrl:
|
|
audioUrl = audioData.get("audioDataURI", "")
|
|
if not audioUrl:
|
|
audioUrl = audioData.get("audioBase64Data", "")
|
|
|
|
if not audioUrl:
|
|
raise Exception("No audio URL received from API")
|
|
|
|
return audioUrl
|
|
|
|
def _downloadAndProcessAudio(self, audioUrl, targetSampleRate):
|
|
"""Download audio file and process it for ComfyUI"""
|
|
try:
|
|
response = requests.get(audioUrl, timeout=30)
|
|
response.raise_for_status()
|
|
|
|
with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as tempFile:
|
|
tempFile.write(response.content)
|
|
tempFilePath = tempFile.name
|
|
|
|
try:
|
|
waveformNp, originalSampleRate = self._loadAudioFile(tempFilePath)
|
|
waveformNp = self._resampleAudio(waveformNp, originalSampleRate, targetSampleRate)
|
|
waveformTensor = self._convertToTensor(waveformNp)
|
|
|
|
return {
|
|
"waveform": waveformTensor,
|
|
"sample_rate": targetSampleRate
|
|
}
|
|
finally:
|
|
if os.path.exists(tempFilePath):
|
|
os.unlink(tempFilePath)
|
|
except Exception as e:
|
|
raise Exception(f"Failed to download audio: {e}")
|
|
|
|
def _loadAudioFile(self, filePath):
|
|
"""Load audio file using librosa"""
|
|
waveformNp, originalSampleRate = librosa.load(filePath, sr=None, mono=False)
|
|
|
|
# Convert to [samples, channels] format
|
|
if waveformNp.ndim == 1:
|
|
waveformNp = waveformNp.reshape(-1, 1)
|
|
else:
|
|
waveformNp = waveformNp.T # [channels, samples] -> [samples, channels]
|
|
|
|
return waveformNp, originalSampleRate
|
|
|
|
def _resampleAudio(self, waveformNp, originalSampleRate, targetSampleRate):
|
|
"""Resample audio to target sample rate"""
|
|
if originalSampleRate == targetSampleRate:
|
|
return waveformNp
|
|
|
|
if waveformNp.ndim == 1:
|
|
return librosa.resample(waveformNp, orig_sr=originalSampleRate, target_sr=targetSampleRate)
|
|
else:
|
|
# Multi-channel: resample each channel
|
|
return librosa.resample(waveformNp.T, orig_sr=originalSampleRate, target_sr=targetSampleRate).T
|
|
|
|
def _convertToTensor(self, waveformNp):
|
|
"""Convert numpy array to PyTorch tensor with correct format"""
|
|
# Ensure 2D shape [samples, channels]
|
|
if waveformNp.ndim == 1:
|
|
waveformNp = waveformNp.reshape(-1, 1)
|
|
|
|
# Convert to [channels, samples] for ComfyUI
|
|
waveformTensor = torch.from_numpy(waveformNp.T).float()
|
|
|
|
# Add batch dimension if needed: [batch, channels, samples]
|
|
if waveformTensor.dim() == 2:
|
|
waveformTensor = waveformTensor.unsqueeze(0)
|
|
|
|
return waveformTensor
|