Ethanfel
980f406573
Remote-aware LoRA Manager nodes that fetch metadata via HTTP from a remote Docker instance while loading LoRA files from local NFS/SMB mounts. Includes reverse-proxy middleware for transparent web UI access. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
114 lines
4.4 KiB
Python
114 lines
4.4 KiB
Python
"""Remote LoRA Randomizer — uses remote API instead of local ServiceRegistry."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
|
|
from .remote_utils import get_lora_info_remote
|
|
from .utils import extract_lora_name
|
|
from ..remote_client import RemoteLoraClient
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LoraRandomizerRemoteLM:
|
|
"""Node that randomly selects LoRAs from a pool (remote metadata)."""
|
|
|
|
NAME = "Lora Randomizer (Remote, LoraManager)"
|
|
CATEGORY = "Lora Manager/randomizer"
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"randomizer_config": ("RANDOMIZER_CONFIG", {}),
|
|
"loras": ("LORAS", {}),
|
|
},
|
|
"optional": {
|
|
"pool_config": ("POOL_CONFIG", {}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("LORA_STACK",)
|
|
RETURN_NAMES = ("LORA_STACK",)
|
|
FUNCTION = "randomize"
|
|
OUTPUT_NODE = False
|
|
|
|
def _preprocess_loras_input(self, loras):
|
|
if isinstance(loras, dict) and "__value__" in loras:
|
|
return loras["__value__"]
|
|
return loras
|
|
|
|
async def randomize(self, randomizer_config, loras, pool_config=None):
|
|
loras = self._preprocess_loras_input(loras)
|
|
|
|
roll_mode = randomizer_config.get("roll_mode", "always")
|
|
execution_seed = randomizer_config.get("execution_seed", None)
|
|
next_seed = randomizer_config.get("next_seed", None)
|
|
|
|
if roll_mode == "fixed":
|
|
ui_loras = loras
|
|
execution_loras = loras
|
|
else:
|
|
client = RemoteLoraClient.get_instance()
|
|
|
|
# Build common kwargs for remote API
|
|
api_kwargs = self._build_api_kwargs(randomizer_config, loras, pool_config)
|
|
|
|
if execution_seed is not None:
|
|
exec_kwargs = {**api_kwargs, "seed": execution_seed}
|
|
execution_loras = await client.get_random_loras(**exec_kwargs)
|
|
if not execution_loras:
|
|
execution_loras = loras
|
|
else:
|
|
execution_loras = loras
|
|
|
|
ui_kwargs = {**api_kwargs, "seed": next_seed}
|
|
ui_loras = await client.get_random_loras(**ui_kwargs)
|
|
if not ui_loras:
|
|
ui_loras = loras
|
|
|
|
execution_stack = self._build_execution_stack_from_input(execution_loras)
|
|
|
|
return {
|
|
"result": (execution_stack,),
|
|
"ui": {"loras": ui_loras, "last_used": execution_loras},
|
|
}
|
|
|
|
def _build_api_kwargs(self, randomizer_config, input_loras, pool_config):
|
|
locked_loras = [l for l in input_loras if l.get("locked", False)]
|
|
return {
|
|
"count": int(randomizer_config.get("count_fixed", 5)),
|
|
"count_mode": randomizer_config.get("count_mode", "range"),
|
|
"count_min": int(randomizer_config.get("count_min", 3)),
|
|
"count_max": int(randomizer_config.get("count_max", 7)),
|
|
"model_strength_min": float(randomizer_config.get("model_strength_min", 0.0)),
|
|
"model_strength_max": float(randomizer_config.get("model_strength_max", 1.0)),
|
|
"use_same_clip_strength": randomizer_config.get("use_same_clip_strength", True),
|
|
"clip_strength_min": float(randomizer_config.get("clip_strength_min", 0.0)),
|
|
"clip_strength_max": float(randomizer_config.get("clip_strength_max", 1.0)),
|
|
"use_recommended_strength": randomizer_config.get("use_recommended_strength", False),
|
|
"recommended_strength_scale_min": float(randomizer_config.get("recommended_strength_scale_min", 0.5)),
|
|
"recommended_strength_scale_max": float(randomizer_config.get("recommended_strength_scale_max", 1.0)),
|
|
"locked_loras": locked_loras,
|
|
"pool_config": pool_config,
|
|
}
|
|
|
|
def _build_execution_stack_from_input(self, loras):
|
|
lora_stack = []
|
|
for lora in loras:
|
|
if not lora.get("active", False):
|
|
continue
|
|
|
|
lora_path, _ = get_lora_info_remote(lora["name"])
|
|
if not lora_path:
|
|
logger.warning("[LoraRandomizerRemoteLM] Could not find path for LoRA: %s", lora["name"])
|
|
continue
|
|
|
|
lora_path = lora_path.replace("/", os.sep)
|
|
model_strength = float(lora.get("strength", 1.0))
|
|
clip_strength = float(lora.get("clipStrength", model_strength))
|
|
lora_stack.append((lora_path, model_strength, clip_strength))
|
|
|
|
return lora_stack
|