Release standalone FromListIdx node pack
This commit is contained in:
Hopping Mad Games
@@ -0,0 +1,232 @@
|
||||
"""Workflow-compatible implementations of the ETK and SWAIN list selectors."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from copy import deepcopy
|
||||
|
||||
|
||||
class ImageFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"images": ("LIST", {"default": []}),
|
||||
"index": ("INT", {"default": 0, "min": 0, "max": 10e10}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("IMAGE",)
|
||||
FUNCTION = "get_image"
|
||||
CATEGORY = "ETK/Image"
|
||||
|
||||
def get_image(self, images, index):
|
||||
return (images[index],)
|
||||
|
||||
|
||||
class LatentFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"latents": ("LIST", {"default": []}),
|
||||
"index": ("INT", {"default": 0, "min": -10e10, "max": 10e10}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("LATENT",)
|
||||
FUNCTION = "get_latent"
|
||||
CATEGORY = "ETK/Image"
|
||||
|
||||
def get_latent(self, latents, index):
|
||||
return (latents[index],)
|
||||
|
||||
|
||||
class ConditioningFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"conditionings": ("LIST", {"default": []}),
|
||||
"index": ("INT", {"default": 0, "min": 0, "max": 10e10}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("CONDITIONING",)
|
||||
FUNCTION = "get_conditioning"
|
||||
CATEGORY = "ETK/Image"
|
||||
|
||||
def get_conditioning(self, conditionings, index):
|
||||
try:
|
||||
return (conditionings[index],)
|
||||
except IndexError as exc:
|
||||
raise IndexError(
|
||||
f"ConditioningFromListIdx index {index} out of range for list length {len(conditionings)}"
|
||||
) from exc
|
||||
|
||||
|
||||
def _normalize_comfy_audio(audio):
|
||||
import torch
|
||||
|
||||
if audio is None:
|
||||
raise ValueError("audio input must be provided")
|
||||
if not isinstance(audio, Mapping):
|
||||
raise ValueError("AUDIO input must be a mapping with 'waveform' and 'sample_rate' keys")
|
||||
|
||||
try:
|
||||
waveform = audio["waveform"]
|
||||
except KeyError as exc:
|
||||
raise ValueError("AUDIO input must contain a 'waveform' key") from exc
|
||||
|
||||
if not torch.is_tensor(waveform):
|
||||
waveform = torch.tensor(waveform)
|
||||
if waveform.dim() == 1:
|
||||
waveform = waveform.unsqueeze(0).unsqueeze(0)
|
||||
if waveform.dim() == 2:
|
||||
waveform = waveform.unsqueeze(0)
|
||||
if waveform.dim() != 3:
|
||||
raise ValueError(f"AUDIO waveform must have shape [B, C, T], got {tuple(waveform.shape)}")
|
||||
|
||||
sample_rate = int(audio.get("sample_rate", 44100) or 44100)
|
||||
if sample_rate <= 0:
|
||||
raise ValueError(f"AUDIO sample_rate must be positive, got {sample_rate}")
|
||||
|
||||
audio_dict = dict(audio)
|
||||
audio_dict["waveform"] = waveform
|
||||
audio_dict["sample_rate"] = sample_rate
|
||||
return audio_dict
|
||||
|
||||
|
||||
class AudioFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"audios": ("LIST", {"default": []}),
|
||||
"index": ("INT", {"default": 0, "min": -10e10, "max": 10e10}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("AUDIO",)
|
||||
FUNCTION = "get_audio"
|
||||
CATEGORY = "ETK/audio"
|
||||
|
||||
def get_audio(self, audios, index):
|
||||
return (_normalize_comfy_audio(audios[int(index)]),)
|
||||
|
||||
|
||||
def _swain_input_types(*, bounded_index: bool):
|
||||
index_config = {"default": 0}
|
||||
if bounded_index:
|
||||
index_config.update({"min": -1000, "max": 1000})
|
||||
return {
|
||||
"required": {"list": ("LIST", {"default": []})},
|
||||
"optional": {"int": ("INT", index_config)},
|
||||
}
|
||||
|
||||
|
||||
def _selected(kwargs):
|
||||
values = kwargs.get("list", None)
|
||||
index = kwargs.get("int", None)
|
||||
if values is None:
|
||||
raise ValueError("Must provide an input")
|
||||
if not isinstance(values, list):
|
||||
raise ValueError("input must be a list")
|
||||
if not isinstance(index, int):
|
||||
raise ValueError("index must be an int")
|
||||
if index >= len(values):
|
||||
raise ValueError("index must be in the range of the list")
|
||||
return values, values[index]
|
||||
|
||||
|
||||
class DictFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return _swain_input_types(bounded_index=True)
|
||||
|
||||
RETURN_TYPES = ("DICT",)
|
||||
RETURN_NAMES = ("dictionary",)
|
||||
FUNCTION = "handler"
|
||||
CATEGORY = "SWAIN/text"
|
||||
|
||||
def handler(self, **kwargs):
|
||||
_, value = _selected(deepcopy(kwargs))
|
||||
return (value,)
|
||||
|
||||
|
||||
class StrFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return _swain_input_types(bounded_index=True)
|
||||
|
||||
RETURN_TYPES = ("STRING", "LIST")
|
||||
RETURN_NAMES = ("string", "list")
|
||||
FUNCTION = "handler"
|
||||
CATEGORY = "SWAIN/text"
|
||||
|
||||
def handler(self, **kwargs):
|
||||
values, value = _selected(deepcopy(kwargs))
|
||||
return value, values
|
||||
|
||||
|
||||
class BytesFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return _swain_input_types(bounded_index=False)
|
||||
|
||||
RETURN_TYPES = ("BYTES",)
|
||||
RETURN_NAMES = ("bytes",)
|
||||
FUNCTION = "handler"
|
||||
CATEGORY = "SWAIN/text"
|
||||
|
||||
def handler(self, **kwargs):
|
||||
_, value = _selected(deepcopy(kwargs))
|
||||
if isinstance(value, list):
|
||||
if len(value) > 1:
|
||||
raise ValueError("bytes must be a list of length 1")
|
||||
value = value[0]
|
||||
return (value,)
|
||||
|
||||
|
||||
class IntFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return _swain_input_types(bounded_index=False)
|
||||
|
||||
RETURN_TYPES = ("INT", "LIST")
|
||||
RETURN_NAMES = ("int", "list")
|
||||
FUNCTION = "handler"
|
||||
CATEGORY = "SWAIN/text"
|
||||
|
||||
def handler(self, **kwargs):
|
||||
values, value = _selected(deepcopy(kwargs))
|
||||
if not isinstance(value, int):
|
||||
try:
|
||||
value = int(value)
|
||||
except Exception as exc:
|
||||
raise ValueError(
|
||||
"list at idx must be an int or a string that can be converted to an int"
|
||||
) from exc
|
||||
return value, values
|
||||
|
||||
|
||||
class FloatFromListIdx:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return _swain_input_types(bounded_index=False)
|
||||
|
||||
RETURN_TYPES = ("FLOAT", "LIST")
|
||||
RETURN_NAMES = ("float", "list")
|
||||
FUNCTION = "handler"
|
||||
CATEGORY = "SWAIN/text"
|
||||
|
||||
def handler(self, **kwargs):
|
||||
values, value = _selected(deepcopy(kwargs))
|
||||
if not isinstance(value, float):
|
||||
try:
|
||||
value = float(value)
|
||||
except Exception as exc:
|
||||
raise ValueError(
|
||||
"list at idx must be an int or a string that can be converted to an int"
|
||||
) from exc
|
||||
return value, values
|
||||
Reference in New Issue
Block a user