Initial commit: ComfyUI-RMAutomation node pack

Features:
- RM Styles Full/Pipe: Load prompts from JSON with random/increment/decrement modes
- RM Power Lora Loader: Advanced LoRA loading with random weight support
- RM Lora Apply: Apply multiple LoRAs with weight ranges
- Image fallback, mask gate, latent nodes, and more

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Moser9815
2026-02-04 19:37:35 -05:00
co-authored by Claude Opus 4.5
commit ace5516265
21 changed files with 8200 additions and 0 deletions
+20
View File
@@ -0,0 +1,20 @@
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
*.egg-info/
.eggs/
dist/
build/
*.egg
.env
.venv
venv/
.idea/
.vscode/
*.swp
*.swo
*~
.DS_Store
Thumbs.db
+123
View File
@@ -0,0 +1,123 @@
# ComfyUI-RMAutomation
A collection of workflow automation nodes for ComfyUI, designed to streamline AI image and video generation workflows.
## Installation
### Via ComfyUI Manager (Recommended)
1. Open ComfyUI Manager
2. Search for "RMAutomation"
3. Click Install
### Manual Installation
```bash
cd ComfyUI/custom_nodes
git clone https://github.com/yourusername/ComfyUI-RMAutomation
```
## Nodes
### LoRA Nodes
#### RM Power LoRA Loader
A flexible node for loading multiple LoRAs with support for:
- Dynamic number of LoRA inputs (add as many as you need)
- Fixed or random strength per LoRA
- Style string import for automated LoRA loading
#### RM LoRA Collector
Combines multiple LoRA stacks into one. Accepts both LORA_STACK inputs and lora strings.
#### RM LoRA Apply
Applies all LoRAs from a collected stack to the model and CLIP.
### Styles Nodes
#### RM Styles Full
Load prompts from a JSON styles file with multiple selection modes:
- **Manual**: Use specific prompt number
- **Random**: Random selection within range
- **Increment**: Auto-increment after each generation
- **Decrement**: Auto-decrement after each generation
Features an **Edit Styles** button that opens a modal editor for managing your styles.
#### RM Styles Full (Display)
Same as RM Styles Full but displays the current prompt on the node.
#### RM Styles Pipe / RM Styles Pipe Out
Pack and unpack style data for easier workflow connections.
### Video Nodes
#### RM Video Combine
Combines images into video with:
- Multiple format support (H.264, H.265, VP9, GIF, WebP)
- Configurable output directory
- Metadata embedding
- Audio support
- Ping-pong looping
### Utility Nodes
#### RM Image Fallback
Returns the first non-empty image from multiple inputs. Useful for conditional workflows.
#### RM Mask Gate
Validates masks and passes through only valid ones. Outputs None for invalid masks, preventing crashes in downstream nodes.
#### RM Mask Gate (Guide)
Same as Mask Gate but for guide inputs. Disables guides when masks are invalid.
#### RM Set Latent Noise Mask
Improved latent noise mask setter with proper error handling for edge cases.
#### RM Positive/Negative Text Embed
CLIP text encoding nodes that combine optional string inputs with text fields.
## Styles JSON Format
The styles file (`data/styles.json`) uses this format:
```json
{
"styles": [
{
"number": 1,
"name": "Style Name",
"positive": "positive prompt text",
"negative": "negative prompt text",
"motion": "motion prompt for video",
"imageLoras": [
{"path": "lora_name.safetensors", "weight": 1.0}
],
"motionLoras": [],
"motionLorasHigh": [],
"motionLorasLow": []
}
]
}
```
## Custom Video Formats
Add custom video formats by placing JSON files in the `video_formats` directory. Example format:
```json
{
"extension": "mp4",
"main_pass": ["-c:v", "libx264", "-preset", "medium", "-crf", "19", "-pix_fmt", "yuv420p"],
"audio_pass": ["-c:a", "aac", "-b:a", "192k"],
"save_metadata": "True",
"dim_alignment": 2
}
```
## License
MIT License - Do whatever you want with this code.
## Credits
- Video combine functionality based on [ComfyUI-VideoHelperSuite](https://github.com/Kosinkadink/ComfyUI-VideoHelperSuite)
- Power LoRA loader concept inspired by [rgthree-comfy](https://github.com/rgthree/rgthree-comfy)
+52
View File
@@ -0,0 +1,52 @@
"""
ComfyUI-RMAutomation
A collection of workflow automation nodes for ComfyUI.
"""
from .nodes.power_lora_loader import NODE_CLASS_MAPPINGS as POWER_LORA_MAPPINGS
from .nodes.power_lora_loader import NODE_DISPLAY_NAME_MAPPINGS as POWER_LORA_DISPLAY
from .nodes.styles_full import NODE_CLASS_MAPPINGS as STYLES_FULL_MAPPINGS
from .nodes.styles_full import NODE_DISPLAY_NAME_MAPPINGS as STYLES_FULL_DISPLAY
from .nodes.styles_pipe import NODE_CLASS_MAPPINGS as STYLES_PIPE_MAPPINGS
from .nodes.styles_pipe import NODE_DISPLAY_NAME_MAPPINGS as STYLES_PIPE_DISPLAY
from .nodes.video_combine import NODE_CLASS_MAPPINGS as VIDEO_MAPPINGS
from .nodes.video_combine import NODE_DISPLAY_NAME_MAPPINGS as VIDEO_DISPLAY
from .nodes.image_fallback import NODE_CLASS_MAPPINGS as IMAGE_FALLBACK_MAPPINGS
from .nodes.image_fallback import NODE_DISPLAY_NAME_MAPPINGS as IMAGE_FALLBACK_DISPLAY
from .nodes.mask_gate import NODE_CLASS_MAPPINGS as MASK_GATE_MAPPINGS
from .nodes.mask_gate import NODE_DISPLAY_NAME_MAPPINGS as MASK_GATE_DISPLAY
from .nodes.latent_nodes import NODE_CLASS_MAPPINGS as LATENT_MAPPINGS
from .nodes.latent_nodes import NODE_DISPLAY_NAME_MAPPINGS as LATENT_DISPLAY
from .nodes.text_embed import NODE_CLASS_MAPPINGS as TEXT_EMBED_MAPPINGS
from .nodes.text_embed import NODE_DISPLAY_NAME_MAPPINGS as TEXT_EMBED_DISPLAY
from .nodes.save_image_with_metadata import NODE_CLASS_MAPPINGS as SAVE_IMAGE_MAPPINGS
from .nodes.save_image_with_metadata import NODE_DISPLAY_NAME_MAPPINGS as SAVE_IMAGE_DISPLAY
# Combine all mappings
NODE_CLASS_MAPPINGS = {
**POWER_LORA_MAPPINGS,
**STYLES_FULL_MAPPINGS,
**STYLES_PIPE_MAPPINGS,
**VIDEO_MAPPINGS,
**IMAGE_FALLBACK_MAPPINGS,
**MASK_GATE_MAPPINGS,
**LATENT_MAPPINGS,
**TEXT_EMBED_MAPPINGS,
**SAVE_IMAGE_MAPPINGS,
}
NODE_DISPLAY_NAME_MAPPINGS = {
**POWER_LORA_DISPLAY,
**STYLES_FULL_DISPLAY,
**STYLES_PIPE_DISPLAY,
**VIDEO_DISPLAY,
**IMAGE_FALLBACK_DISPLAY,
**MASK_GATE_DISPLAY,
**LATENT_DISPLAY,
**TEXT_EMBED_DISPLAY,
**SAVE_IMAGE_DISPLAY,
}
WEB_DIRECTORY = "./js"
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
+3556
View File
File diff suppressed because it is too large Load Diff
+115
View File
@@ -0,0 +1,115 @@
/**
* RM LoRA Apply
* Adds dynamic "Add Lora Stack" and "Add Lora String" button functionality
*/
import { app } from "../../scripts/app.js";
console.log("[RMLoraApply] Script loading...");
app.registerExtension({
name: "RMAutomation.LoraApply",
async beforeRegisterNodeDef(nodeType, nodeData, app) {
if (nodeData.name !== "RMLoraApply") return;
console.log("[RMLoraApply] Setting up node");
const onNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = function() {
const result = onNodeCreated?.apply(this, arguments);
this.stackCount = 1;
this.stringCount = 0;
this.addButtons();
return result;
};
const onConfigure = nodeType.prototype.onConfigure;
nodeType.prototype.onConfigure = function(info) {
const result = onConfigure?.apply(this, arguments);
// Count existing inputs from saved state
if (info.inputs) {
let maxStack = 1;
let maxString = 0;
for (const input of info.inputs) {
if (input.name && input.name.startsWith("lora_stack_")) {
const num = parseInt(input.name.replace("lora_stack_", ""));
if (num > maxStack) maxStack = num;
}
if (input.name && input.name.startsWith("lora_string")) {
if (input.name === "lora_string") {
if (maxString < 1) maxString = 1;
} else {
const num = parseInt(input.name.replace("lora_string_", ""));
if (num > maxString) maxString = num;
}
}
}
this.stackCount = maxStack;
this.stringCount = maxString;
}
// Ensure buttons exist
this.addButtons();
return result;
};
nodeType.prototype.addButtons = function() {
const node = this;
// Add Lora Stack button
if (!this.widgets?.find(w => w._btnType === "add_stack")) {
const stackButton = this.addWidget("button", "Add Lora Stack", null, () => {
node.addLoraStackInput();
});
stackButton._btnType = "add_stack";
stackButton.serialize = false;
}
// Add Lora String button
if (!this.widgets?.find(w => w._btnType === "add_string")) {
const stringButton = this.addWidget("button", "Add Lora String", null, () => {
node.addLoraStringInput();
});
stringButton._btnType = "add_string";
stringButton.serialize = false;
}
};
nodeType.prototype.addLoraStackInput = function() {
this.stackCount = this.stackCount || 1;
this.stackCount++;
const inputName = `lora_stack_${this.stackCount}`;
this.addInput(inputName, "LORA_STACK");
// Resize node
const newSize = this.computeSize();
this.setSize([Math.max(this.size[0], newSize[0]), newSize[1]]);
this.setDirtyCanvas(true, true);
console.log(`[RMLoraApply] Added input: ${inputName}`);
};
nodeType.prototype.addLoraStringInput = function() {
this.stringCount = this.stringCount || 0;
this.stringCount++;
const inputName = this.stringCount === 1 ? "lora_string" : `lora_string_${this.stringCount}`;
this.addInput(inputName, "STRING");
// Resize node
const newSize = this.computeSize();
this.setSize([Math.max(this.size[0], newSize[0]), newSize[1]]);
this.setDirtyCanvas(true, true);
console.log(`[RMLoraApply] Added input: ${inputName}`);
};
},
});
console.log("[RMLoraApply] Script loaded");
+839
View File
@@ -0,0 +1,839 @@
/**
* RM Power LoRA Loader
* Full UI implementation with random strength support and Civitai AIR mode
*/
import { app } from "../../scripts/app.js";
import { api } from "../../scripts/api.js";
console.log("[RMPowerLoraLoader] Script loading...");
const NODE_TYPES = ["RMPowerLoraLoader", "RMPowerLoraLoaderPipe"];
const DEFAULT_WIDTH = 400;
// ============================================================================
// Utility Functions
// ============================================================================
function fitString(ctx, str, maxWidth) {
let width = ctx.measureText(str).width;
if (width <= maxWidth) return str;
const ellipsis = "...";
const ellipsisWidth = ctx.measureText(ellipsis).width;
let len = str.length;
while (width > maxWidth - ellipsisWidth && len > 0) {
len--;
str = str.substring(0, len);
width = ctx.measureText(str).width;
}
return str + ellipsis;
}
// Cache for loras list
let lorasCache = null;
let lorasCachePromise = null;
async function getLoras(force = false) {
if (lorasCache && !force) {
return lorasCache;
}
if (lorasCachePromise && !force) {
return lorasCachePromise;
}
lorasCachePromise = api.fetchApi("/object_info/LoraLoader")
.then(response => response.json())
.then(data => {
const loraInput = data?.LoraLoader?.input?.required?.lora_name;
if (loraInput && Array.isArray(loraInput[0])) {
lorasCache = loraInput[0];
return lorasCache;
}
return [];
})
.catch(err => {
console.error("[RMPowerLoraLoader] Failed to fetch loras:", err);
return [];
});
return lorasCachePromise;
}
// ============================================================================
// Register Extension
// ============================================================================
app.registerExtension({
name: "RMAutomation.PowerLoraLoader",
async beforeRegisterNodeDef(nodeType, nodeData, app) {
if (!NODE_TYPES.includes(nodeData.name)) return;
console.log("[RMPowerLoraLoader] Setting up node:", nodeData.name);
// Store original methods
const onNodeCreated = nodeType.prototype.onNodeCreated;
const onConfigure = nodeType.prototype.onConfigure;
nodeType.prototype.onNodeCreated = function() {
const result = onNodeCreated?.apply(this, arguments);
this.loraWidgets = [];
this.serialize_widgets = true;
this._civitaiApiKey = ""; // Store API key on node
// Set default width
if (this.size && this.size[0] < DEFAULT_WIDTH) {
this.size[0] = DEFAULT_WIDTH;
}
// Add the "Add Lora" button
this.addLoraButton();
return result;
};
nodeType.prototype.onConfigure = function(info) {
const result = onConfigure?.apply(this, arguments);
// Restore lora widgets from saved data
if (info.widgets_values) {
// Remove existing lora widgets and API key widget first
this.widgets = this.widgets?.filter(w => !w.name?.startsWith("lora_") && w.name !== "civitai_api_key") || [];
this.loraWidgets = [];
// Restore API key
this._civitaiApiKey = "";
for (const value of info.widgets_values) {
if (value && typeof value === "object") {
if (value.lora !== undefined || value.airMode !== undefined) {
this.addLoraWidget(value);
}
} else if (typeof value === "string" && info.widgets_values.indexOf(value) === info.widgets_values.length - 1) {
// Last string value might be the API key
this._civitaiApiKey = value;
}
}
// Update API key widget visibility
this.updateApiKeyWidget();
}
return result;
};
nodeType.prototype.addLoraButton = function() {
// Check if button already exists
if (this.widgets?.find(w => w.name === "add_lora_btn")) return;
const node = this;
// Create custom button widget
const button = {
name: "add_lora_btn",
type: "custom",
value: "Add LoRA",
serialize: false,
draw: function(ctx, nodeRef, width, posY, height) {
const margin = 15;
const midY = posY + height / 2;
ctx.save();
// Button background
ctx.fillStyle = LiteGraph.WIDGET_BGCOLOR;
ctx.strokeStyle = LiteGraph.WIDGET_OUTLINE_COLOR;
ctx.beginPath();
ctx.roundRect(margin, posY, width - margin * 2, height, height * 0.4);
ctx.fill();
ctx.stroke();
// Button text
ctx.font = "11px sans-serif";
ctx.fillStyle = LiteGraph.WIDGET_TEXT_COLOR;
ctx.textAlign = "center";
ctx.textBaseline = "middle";
ctx.fillText("Add LoRA", width / 2, midY);
ctx.restore();
},
mouse: function(event, pos, nodeRef) {
if (event.type === "pointerdown" && event.button === 0) {
node.showLoraMenu();
return true;
}
return false;
},
computeSize: function() {
return [DEFAULT_WIDTH, 22];
}
};
this.widgets = this.widgets || [];
this.widgets.push(button);
};
nodeType.prototype.showLoraMenu = async function() {
const node = this;
const loras = await getLoras();
// Build menu with Civitai AIR option at top
const menuItems = [
{
content: "🌐 Civitai AIR (Download by ID)",
callback: () => {
const air = prompt("Enter Civitai Model ID (AIR):\n\nFormat: model_id or model_id@version_id\nExample: 109395 or 109395@84321");
if (air && air.trim()) {
node.addLoraWidget({ airMode: true, air: air.trim() });
node.updateApiKeyWidget();
}
}
},
null // separator
];
if (!loras || loras.length === 0) {
menuItems.push({ content: "(No local LoRAs found)", disabled: true });
} else {
// Add hierarchical lora menu
const loraMenuItems = this.buildLoraMenu(loras);
menuItems.push(...loraMenuItems);
}
new LiteGraph.ContextMenu(menuItems, {
event: window.event,
callback: null,
title: "Select LoRA"
});
};
nodeType.prototype.buildLoraMenu = function(loras, forWidget = null) {
// Build a nested tree structure from lora paths
const tree = {};
for (const lora of loras) {
const parts = lora.split(/[\/\\]/);
let current = tree;
// Navigate/create nested structure
for (let i = 0; i < parts.length - 1; i++) {
const folder = parts[i];
if (!current[folder]) {
current[folder] = { __files: [], __folders: {} };
}
current = current[folder].__folders;
}
// Add the file to the deepest folder
const filename = parts[parts.length - 1];
if (parts.length === 1) {
// Root level file
if (!tree.__root) tree.__root = [];
tree.__root.push({ name: filename, path: lora });
} else {
// Get the parent folder
let parent = tree;
for (let i = 0; i < parts.length - 2; i++) {
parent = parent[parts[i]].__folders;
}
const parentFolder = parts[parts.length - 2];
parent[parentFolder].__files.push({ name: filename, path: lora });
}
}
const node = this;
// Recursively build menu items from tree
const buildMenuFromTree = (treeNode, isRoot = false) => {
const items = [];
// Get folders and files at this level
const folders = [];
const files = isRoot && treeNode.__root ? [...treeNode.__root] : [];
for (const [key, value] of Object.entries(treeNode)) {
if (key === '__root') continue;
if (key === '__files' || key === '__folders') continue;
folders.push({ name: key, data: value });
}
// Sort folders alphabetically
folders.sort((a, b) => a.name.localeCompare(b.name));
// Add folders first
for (const folder of folders) {
const subItems = [];
// Add subfolders
if (folder.data.__folders) {
const subFolderItems = buildMenuFromTree(folder.data.__folders);
subItems.push(...subFolderItems);
}
// Add files in this folder
if (folder.data.__files && folder.data.__files.length > 0) {
// Sort files
folder.data.__files.sort((a, b) => a.name.localeCompare(b.name));
// Add separator if there are subfolders
if (subItems.length > 0) {
subItems.push(null); // separator
}
for (const file of folder.data.__files) {
const displayName = file.name.replace('.safetensors', '');
subItems.push({
content: displayName,
callback: forWidget
? () => {
forWidget.value.lora = file.path;
forWidget.value.airMode = false;
forWidget.value.air = "";
node.updateApiKeyWidget();
node.setDirtyCanvas(true, true);
}
: () => node.addLoraWidget({ lora: file.path })
});
}
}
items.push({
content: `📁 ${folder.name}`,
has_submenu: true,
callback: () => {},
submenu: {
options: subItems
}
});
}
// Add root files (only at root level)
if (isRoot && files.length > 0) {
files.sort((a, b) => a.name.localeCompare(b.name));
if (items.length > 0) {
items.push(null); // separator
}
for (const file of files) {
const displayName = file.name.replace('.safetensors', '');
items.push({
content: displayName,
callback: forWidget
? () => {
forWidget.value.lora = file.path;
forWidget.value.airMode = false;
forWidget.value.air = "";
node.updateApiKeyWidget();
node.setDirtyCanvas(true, true);
}
: () => node.addLoraWidget({ lora: file.path })
});
}
}
return items;
};
return buildMenuFromTree(tree, true);
};
nodeType.prototype.addLoraWidget = function(config = {}) {
const defaults = {
on: true,
lora: null,
strength: 1.0,
random: false,
strengthMin: 0.5,
strengthMax: 1.0,
airMode: false,
air: "",
};
const value = { ...defaults, ...config };
const index = this.loraWidgets?.length || 0;
const widgetName = `lora_${index}`;
// Create a custom widget
const widget = {
name: widgetName,
type: "custom",
value: value,
draw: (ctx, node, width, posY, height) => {
this.drawLoraWidget(ctx, widget, width, posY, height);
},
mouse: (event, pos, node) => {
return this.handleLoraWidgetMouse(event, pos, widget);
},
computeSize: () => {
return [DEFAULT_WIDTH, 22];
},
serializeValue: () => {
return { ...widget.value };
}
};
// Insert before the button (and before API key widget if present)
const buttonIndex = this.widgets?.findIndex(w => w.name === "add_lora_btn") ?? -1;
if (buttonIndex >= 0) {
this.widgets.splice(buttonIndex, 0, widget);
} else {
this.widgets = this.widgets || [];
this.widgets.push(widget);
}
this.loraWidgets = this.loraWidgets || [];
this.loraWidgets.push(widget);
// Resize node but maintain minimum width
const newSize = this.computeSize();
newSize[0] = Math.max(newSize[0], DEFAULT_WIDTH);
this.setSize(newSize);
this.setDirtyCanvas(true, true);
return widget;
};
// Check if any lora uses AIR mode
nodeType.prototype.hasAirModeLora = function() {
return this.loraWidgets?.some(w => w.value.airMode) || false;
};
// Add or remove the API key widget based on AIR mode usage
nodeType.prototype.updateApiKeyWidget = function() {
const hasAir = this.hasAirModeLora();
const existingWidget = this.widgets?.find(w => w.name === "civitai_api_key");
if (hasAir && !existingWidget) {
// Add API key widget at the end (after button)
this.addApiKeyWidget();
} else if (!hasAir && existingWidget) {
// Remove API key widget
const idx = this.widgets.indexOf(existingWidget);
if (idx >= 0) {
this.widgets.splice(idx, 1);
}
}
// Resize but maintain minimum width
const newSize = this.computeSize();
newSize[0] = Math.max(newSize[0], DEFAULT_WIDTH);
this.setSize(newSize);
this.setDirtyCanvas(true, true);
};
nodeType.prototype.addApiKeyWidget = function() {
const node = this;
const widget = {
name: "civitai_api_key",
type: "custom",
value: this._civitaiApiKey || "",
draw: function(ctx, nodeRef, width, posY, height) {
const margin = 15;
const midY = posY + height / 2;
ctx.save();
// Background
ctx.fillStyle = LiteGraph.WIDGET_BGCOLOR;
ctx.strokeStyle = LiteGraph.WIDGET_OUTLINE_COLOR;
ctx.beginPath();
ctx.roundRect(margin, posY, width - margin * 2, height, height * 0.4);
ctx.fill();
ctx.stroke();
// Label
ctx.font = "10px sans-serif";
ctx.textBaseline = "middle";
ctx.textAlign = "left";
ctx.fillStyle = "#888";
ctx.fillText("Civitai API Key:", margin + 8, midY);
// Value (masked)
const labelWidth = ctx.measureText("Civitai API Key:").width;
ctx.fillStyle = LiteGraph.WIDGET_TEXT_COLOR;
const apiKey = node._civitaiApiKey || "";
const maskedKey = apiKey ? "●".repeat(Math.min(apiKey.length, 20)) : "(click to set)";
ctx.fillText(maskedKey, margin + 8 + labelWidth + 8, midY);
ctx.restore();
},
mouse: function(event, pos, nodeRef) {
if (event.type === "pointerdown" && event.button === 0) {
const currentKey = node._civitaiApiKey || "";
const newKey = prompt(
"Enter your Civitai API Key:\n\n" +
"Get your key from: https://civitai.com/user/account (API Keys section)\n\n" +
"This is required to download some models.",
currentKey
);
if (newKey !== null) {
node._civitaiApiKey = newKey;
widget.value = newKey;
node.setDirtyCanvas(true, true);
}
return true;
}
return false;
},
computeSize: function() {
return [DEFAULT_WIDTH, 22];
},
serializeValue: function() {
return node._civitaiApiKey || "";
}
};
this.widgets = this.widgets || [];
this.widgets.push(widget);
};
nodeType.prototype.drawLoraWidget = function(ctx, widget, width, posY, height) {
const margin = 15;
const value = widget.value;
const midY = posY + height / 2;
ctx.save();
// Row background - with gradient for random mode or AIR mode
if (value.airMode) {
// AIR mode - blue-ish gradient
const grad = ctx.createLinearGradient(margin, 0, width - margin, 0);
grad.addColorStop(0, value.on ? "rgba(50, 100, 150, 0.3)" : "rgba(40, 60, 80, 0.2)");
grad.addColorStop(0.15, LiteGraph.WIDGET_BGCOLOR);
if (value.random) {
grad.addColorStop(0.5, LiteGraph.WIDGET_BGCOLOR);
grad.addColorStop(1, value.on ? "rgba(130, 100, 180, 0.35)" : "rgba(90, 80, 110, 0.25)");
}
ctx.fillStyle = grad;
} else if (value.random) {
const grad = ctx.createLinearGradient(margin, 0, width - margin, 0);
grad.addColorStop(0, LiteGraph.WIDGET_BGCOLOR);
grad.addColorStop(0.5, LiteGraph.WIDGET_BGCOLOR);
grad.addColorStop(1, value.on ? "rgba(130, 100, 180, 0.35)" : "rgba(90, 80, 110, 0.25)");
ctx.fillStyle = grad;
} else {
ctx.fillStyle = LiteGraph.WIDGET_BGCOLOR;
}
ctx.strokeStyle = LiteGraph.WIDGET_OUTLINE_COLOR;
ctx.beginPath();
ctx.roundRect(margin, posY, width - margin * 2, height, height * 0.4);
ctx.fill();
ctx.stroke();
// === TOGGLE SWITCH ===
const toggleHeight = height * 0.55;
const toggleWidth = height * 1.1;
const toggleX = margin + 5;
const toggleY = midY - toggleHeight / 2;
const toggleRadius = toggleHeight * 0.5;
// Toggle track background
ctx.beginPath();
ctx.roundRect(toggleX, toggleY, toggleWidth, toggleHeight, toggleRadius);
ctx.fillStyle = value.on ? "rgba(100, 160, 100, 0.4)" : "rgba(80, 80, 80, 0.4)";
ctx.fill();
// Toggle knob
const knobRadius = toggleHeight * 0.38;
const knobX = value.on ? toggleX + toggleWidth - knobRadius - 2 : toggleX + knobRadius + 2;
ctx.beginPath();
ctx.arc(knobX, midY, knobRadius, 0, Math.PI * 2);
ctx.fillStyle = value.on ? "#8B8" : "#888";
ctx.fill();
// === GEAR ICON (after toggle) ===
const gearSize = 10;
const gearX = toggleX + toggleWidth + 8;
this.drawGearIcon(ctx, gearX, midY, gearSize * 0.5, value.on ? "#777" : "#555");
let posX = gearX + gearSize + 6;
// === AIR BADGE (if in AIR mode) ===
if (value.airMode) {
ctx.font = "bold 8px sans-serif";
ctx.fillStyle = value.on ? "#6af" : "#568";
ctx.textBaseline = "middle";
ctx.textAlign = "left";
ctx.fillText("AIR", posX, midY);
posX += ctx.measureText("AIR").width + 4;
}
// === LAYOUT CALCULATIONS ===
const strengthWidth = 72;
const strengthX = width - margin - strengthWidth - 4;
const loraWidth = strengthX - posX - 8;
// === LORA NAME / AIR ID ===
ctx.font = "11px sans-serif";
ctx.textBaseline = "middle";
ctx.textAlign = "left";
ctx.fillStyle = value.on ? LiteGraph.WIDGET_TEXT_COLOR : "#666";
let displayName;
if (value.airMode) {
displayName = value.air || "(click to set AIR)";
} else {
displayName = value.lora ? value.lora.split(/[\/\\]/).pop().replace('.safetensors', '') : "None";
}
ctx.fillText(fitString(ctx, displayName, loraWidth), posX, midY);
// === STRENGTH DISPLAY ===
ctx.font = "10px monospace";
ctx.textAlign = "center";
if (value.random) {
ctx.fillStyle = value.on ? "#c8b8e8" : "#807090";
ctx.fillText(`${value.strengthMin.toFixed(2)} to ${value.strengthMax.toFixed(2)}`, strengthX + strengthWidth/2, midY);
} else {
ctx.fillStyle = value.on ? LiteGraph.WIDGET_TEXT_COLOR : "#666";
ctx.fillText(value.strength.toFixed(2), strengthX + strengthWidth/2, midY);
}
ctx.restore();
};
// Helper to draw a gear icon
nodeType.prototype.drawGearIcon = function(ctx, x, y, radius, color) {
ctx.save();
ctx.fillStyle = color;
ctx.strokeStyle = color;
ctx.lineWidth = 1.5;
const teeth = 6;
const innerRadius = radius * 0.5;
const outerRadius = radius;
ctx.beginPath();
for (let i = 0; i < teeth * 2; i++) {
const angle = (i * Math.PI) / teeth;
const r = i % 2 === 0 ? outerRadius : innerRadius * 1.3;
const px = x + Math.cos(angle) * r;
const py = y + Math.sin(angle) * r;
if (i === 0) ctx.moveTo(px, py);
else ctx.lineTo(px, py);
}
ctx.closePath();
ctx.fill();
// Center hole
ctx.beginPath();
ctx.arc(x, y, innerRadius * 0.5, 0, Math.PI * 2);
ctx.fillStyle = LiteGraph.WIDGET_BGCOLOR;
ctx.fill();
ctx.restore();
};
nodeType.prototype.handleLoraWidgetMouse = function(event, pos, widget) {
const margin = 15;
const width = this.size[0];
const height = 22;
const toggleWidth = height * 1.1;
const gearSize = 10;
const strengthWidth = 72;
const localX = pos[0];
const toggleX = margin + 5;
const toggleEnd = toggleX + toggleWidth;
const gearX = toggleEnd + 8; // This is the CENTER of the gear icon
const gearRadius = gearSize * 0.5; // Gear is drawn with this radius
const strengthX = width - margin - strengthWidth - 4;
// Handle right-click for context menu
if (event.type === "pointerdown" && event.button === 2) {
this.showLoraWidgetMenu(widget, event);
return true;
}
// Handle left-click
if (event.type !== "pointerdown" || event.button !== 0) return false;
// Check toggle click
if (localX >= toggleX && localX < toggleEnd) {
widget.value.on = !widget.value.on;
this.setDirtyCanvas(true, true);
return true;
}
// Check gear icon click - gear is centered at gearX with radius gearRadius
if (localX >= gearX - gearRadius - 3 && localX < gearX + gearRadius + 3) {
this.showLoraWidgetMenu(widget, event);
return true;
}
// Check strength area click - prompt for value
if (localX >= strengthX && localX < strengthX + strengthWidth) {
if (widget.value.random) {
const range = prompt("Enter min to max range (e.g., -4 to -1, or 0.5 to 1.0):",
`${widget.value.strengthMin} to ${widget.value.strengthMax}`);
if (range) {
let min, max;
if (range.includes(" to ")) {
const parts = range.split(" to ").map(s => parseFloat(s.trim()));
if (parts.length === 2) {
min = parts[0];
max = parts[1];
}
} else {
const match = range.match(/^(-?\d*\.?\d+)\s*-\s*(-?\d*\.?\d+)$/);
if (match) {
min = parseFloat(match[1]);
max = parseFloat(match[2]);
}
}
if (min !== undefined && max !== undefined && !isNaN(min) && !isNaN(max)) {
widget.value.strengthMin = Math.min(min, max);
widget.value.strengthMax = Math.max(min, max);
this.setDirtyCanvas(true, true);
}
}
} else {
const val = prompt("Enter strength:", widget.value.strength);
if (val !== null && !isNaN(parseFloat(val))) {
widget.value.strength = parseFloat(val);
this.setDirtyCanvas(true, true);
}
}
return true;
}
// Check lora name / AIR click - starts after gear area
if (localX >= gearX + gearRadius + 3 && localX < strengthX - 4) {
if (widget.value.airMode) {
// Prompt for AIR
const air = prompt(
"Enter Civitai Model ID (AIR):\n\nFormat: model_id or model_id@version_id\nExample: 109395 or 109395@84321",
widget.value.air || ""
);
if (air !== null) {
widget.value.air = air.trim();
this.setDirtyCanvas(true, true);
}
} else {
// Show lora menu
this.showLoraMenuForWidget(widget);
}
return true;
}
return false;
};
nodeType.prototype.showLoraWidgetMenu = function(widget, event) {
const node = this;
const menuItems = [
{
content: widget.value.on ? "⚫ Disable" : "🟢 Enable",
callback: () => {
widget.value.on = !widget.value.on;
this.setDirtyCanvas(true, true);
}
},
{
content: widget.value.random ? "📊 Use Fixed Strength" : "🎲 Use Random Strength",
callback: () => {
widget.value.random = !widget.value.random;
this.setDirtyCanvas(true, true);
}
},
null,
{
content: widget.value.airMode ? "📁 Switch to Local LoRA" : "🌐 Switch to Civitai AIR",
callback: () => {
widget.value.airMode = !widget.value.airMode;
if (widget.value.airMode) {
// Switching to AIR mode - prompt for AIR
const air = prompt(
"Enter Civitai Model ID (AIR):\n\nFormat: model_id or model_id@version_id\nExample: 109395 or 109395@84321"
);
if (air && air.trim()) {
widget.value.air = air.trim();
}
} else {
// Switching to local mode - clear AIR
widget.value.air = "";
}
node.updateApiKeyWidget();
this.setDirtyCanvas(true, true);
}
},
null,
{
content: "🗑️ Remove",
callback: () => {
const idx = this.widgets.indexOf(widget);
if (idx >= 0) {
this.widgets.splice(idx, 1);
}
const loraIdx = this.loraWidgets.indexOf(widget);
if (loraIdx >= 0) {
this.loraWidgets.splice(loraIdx, 1);
}
node.updateApiKeyWidget();
const newSize = this.computeSize();
newSize[0] = Math.max(newSize[0], DEFAULT_WIDTH);
this.setSize(newSize);
this.setDirtyCanvas(true, true);
}
}
];
new LiteGraph.ContextMenu(menuItems, {
event: event,
title: "LoRA Options"
});
};
nodeType.prototype.showLoraMenuForWidget = async function(widget) {
const node = this;
const loras = await getLoras();
// Add Civitai AIR option at top
const menuItems = [
{
content: "🌐 Civitai AIR (Download by ID)",
callback: () => {
const air = prompt(
"Enter Civitai Model ID (AIR):\n\nFormat: model_id or model_id@version_id\nExample: 109395 or 109395@84321"
);
if (air && air.trim()) {
widget.value.airMode = true;
widget.value.air = air.trim();
widget.value.lora = null;
node.updateApiKeyWidget();
node.setDirtyCanvas(true, true);
}
}
},
null
];
const loraMenuItems = this.buildLoraMenu(loras, widget);
menuItems.push(...loraMenuItems);
new LiteGraph.ContextMenu(menuItems, {
event: window.event,
title: "Change LoRA"
});
};
}
});
console.log("[RMPowerLoraLoader] Script loaded successfully!");
+802
View File
@@ -0,0 +1,802 @@
/**
* RM Styles Editor Modal
* Provides a modal dialog for editing the styles.json file
*/
import { app } from "../../scripts/app.js";
import { api } from "../../scripts/api.js";
// Create and inject CSS styles - ComfyUI theme
const styleSheet = document.createElement("style");
styleSheet.textContent = `
.rm-styles-modal-overlay {
position: fixed;
top: 0;
left: 0;
width: 100%;
height: 100%;
background: rgba(0, 0, 0, 0.8);
backdrop-filter: blur(4px);
z-index: 10000;
display: flex;
align-items: center;
justify-content: center;
}
.rm-styles-modal {
background: #202020;
border-radius: 8px;
width: 90%;
max-width: 900px;
max-height: 85vh;
display: flex;
flex-direction: column;
box-shadow: 0 8px 32px rgba(0, 0, 0, 0.6);
border: 1px solid #444;
}
.rm-styles-modal-header {
display: flex;
align-items: center;
justify-content: space-between;
padding: 0.75rem 1rem;
border-bottom: 1px solid #444;
background: #2a2a2a;
border-radius: 8px 8px 0 0;
}
.rm-styles-modal-header h2 {
margin: 0;
font-size: 1rem;
color: #ddd;
font-weight: 500;
}
.rm-styles-modal-close {
background: none;
border: none;
font-size: 1.25rem;
color: #888;
cursor: pointer;
padding: 0.25rem;
line-height: 1;
}
.rm-styles-modal-close:hover {
color: #fff;
}
.rm-styles-modal-content {
flex: 1;
overflow-y: auto;
padding: 1rem;
background: #1a1a1a;
}
.rm-styles-toolbar {
display: flex;
gap: 0.5rem;
margin-bottom: 1rem;
flex-wrap: wrap;
}
.rm-styles-toolbar button {
padding: 0.4rem 0.75rem;
border: 1px solid #555;
border-radius: 4px;
background: #353535;
color: #ddd;
cursor: pointer;
font-size: 0.8rem;
transition: all 0.15s ease;
}
.rm-styles-toolbar button:hover {
background: #454545;
border-color: #666;
}
.rm-styles-toolbar .rm-btn-import {
background: #2d4a3e;
border-color: #3d6a5e;
}
.rm-styles-toolbar .rm-btn-import:hover {
background: #3d5a4e;
border-color: #4d7a6e;
}
.rm-styles-list {
display: flex;
flex-direction: column;
gap: 2px;
}
.rm-style-item {
display: flex;
align-items: center;
gap: 0.75rem;
padding: 0.6rem 0.75rem;
background: #2a2a2a;
border-radius: 4px;
cursor: pointer;
border: 1px solid transparent;
transition: all 0.1s ease;
}
.rm-style-item:hover {
background: #353535;
border-color: #555;
}
.rm-style-item.selected {
background: #3a3a3a;
border-color: #6a9;
}
.rm-style-number {
font-weight: 600;
color: #6a9;
min-width: 35px;
font-size: 0.85rem;
}
.rm-style-name {
flex: 1;
color: #ddd;
font-size: 0.85rem;
}
.rm-style-preview {
color: #777;
font-size: 0.7rem;
max-width: 280px;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.rm-style-editor {
margin-top: 1rem;
padding: 1rem;
background: #252525;
border-radius: 6px;
border: 1px solid #444;
}
.rm-style-editor-row {
display: flex;
flex-direction: column;
gap: 0.25rem;
margin-bottom: 0.75rem;
}
.rm-style-editor-row label {
color: #999;
font-size: 0.7rem;
text-transform: uppercase;
letter-spacing: 0.5px;
}
.rm-style-editor-row input,
.rm-style-editor-row textarea {
padding: 0.5rem;
border: 1px solid #444;
border-radius: 4px;
background: #1a1a1a;
color: #ddd;
font-family: inherit;
font-size: 0.85rem;
transition: border-color 0.15s ease;
}
.rm-style-editor-row textarea {
min-height: 70px;
resize: vertical;
}
.rm-style-editor-row input:focus,
.rm-style-editor-row textarea:focus {
outline: none;
border-color: #6a9;
}
.rm-loras-section {
margin-top: 0.75rem;
padding: 0.75rem;
background: #1e1e1e;
border-radius: 4px;
border: 1px solid #333;
}
.rm-loras-section h4 {
margin: 0 0 0.5rem 0;
color: #888;
font-size: 0.7rem;
text-transform: uppercase;
letter-spacing: 0.5px;
}
.rm-lora-item {
display: flex;
gap: 0.5rem;
margin-bottom: 0.4rem;
align-items: center;
}
.rm-lora-item input {
flex: 1;
padding: 0.35rem;
border: 1px solid #444;
border-radius: 3px;
background: #2a2a2a;
color: #ddd;
font-size: 0.75rem;
}
.rm-lora-item input[type="number"] {
width: 55px;
flex: none;
}
.rm-lora-remove {
background: #6a3333;
border: 1px solid #8a4444;
color: #ddd;
padding: 0.2rem 0.4rem;
border-radius: 3px;
cursor: pointer;
font-size: 0.7rem;
transition: all 0.15s ease;
}
.rm-lora-remove:hover {
background: #8a4444;
}
.rm-lora-add {
background: #2d4a3e;
border: 1px solid #3d6a5e;
color: #ddd;
padding: 0.25rem 0.5rem;
border-radius: 3px;
cursor: pointer;
font-size: 0.7rem;
margin-top: 0.4rem;
transition: all 0.15s ease;
}
.rm-lora-add:hover {
background: #3d5a4e;
}
.rm-style-actions {
display: flex;
gap: 0.5rem;
margin-top: 1rem;
justify-content: flex-end;
}
.rm-style-actions button {
padding: 0.4rem 0.75rem;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 0.8rem;
transition: all 0.15s ease;
}
.rm-btn-save {
background: #2d4a3e;
border: 1px solid #3d6a5e;
color: #ddd;
}
.rm-btn-save:hover {
background: #3d5a4e;
}
.rm-btn-delete {
background: #6a3333;
border: 1px solid #8a4444;
color: #ddd;
}
.rm-btn-delete:hover {
background: #8a4444;
}
.rm-btn-cancel {
background: #353535;
border: 1px solid #555;
color: #ddd;
}
.rm-btn-cancel:hover {
background: #454545;
}
/* Import dialog styles */
.rm-import-dialog-overlay {
position: fixed;
top: 0;
left: 0;
width: 100%;
height: 100%;
background: rgba(0, 0, 0, 0.7);
z-index: 10001;
display: flex;
align-items: center;
justify-content: center;
}
.rm-import-dialog {
background: #252525;
border-radius: 8px;
padding: 1.25rem;
border: 1px solid #555;
min-width: 320px;
box-shadow: 0 8px 24px rgba(0, 0, 0, 0.5);
}
.rm-import-dialog h3 {
margin: 0 0 0.75rem 0;
color: #ddd;
font-size: 0.95rem;
font-weight: 500;
}
.rm-import-dialog p {
color: #999;
font-size: 0.8rem;
margin: 0 0 1rem 0;
}
.rm-import-dialog-buttons {
display: flex;
gap: 0.5rem;
justify-content: flex-end;
}
.rm-import-dialog-buttons button {
padding: 0.4rem 0.75rem;
border-radius: 4px;
cursor: pointer;
font-size: 0.8rem;
transition: all 0.15s ease;
}
.rm-import-replace {
background: #5a4a2a;
border: 1px solid #7a6a4a;
color: #ddd;
}
.rm-import-replace:hover {
background: #6a5a3a;
}
.rm-import-append {
background: #2d4a3e;
border: 1px solid #3d6a5e;
color: #ddd;
}
.rm-import-append:hover {
background: #3d5a4e;
}
.rm-import-cancel {
background: #353535;
border: 1px solid #555;
color: #ddd;
}
.rm-import-cancel:hover {
background: #454545;
}
/* Hidden file input */
.rm-file-input {
display: none;
}
`;
document.head.appendChild(styleSheet);
class RMStylesEditorModal {
constructor() {
this.styles = [];
this.selectedStyle = null;
this.overlay = null;
}
async open() {
await this.loadStyles();
this.render();
}
close() {
if (this.overlay) {
this.overlay.remove();
this.overlay = null;
}
}
async loadStyles() {
try {
const response = await api.fetchApi("/api/rmautomation/styles");
const data = await response.json();
this.styles = data.styles || [];
} catch (e) {
console.error("Failed to load styles:", e);
this.styles = [];
}
}
async saveStyle(style) {
try {
const response = await api.fetchApi(`/api/rmautomation/styles/${style.number}`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(style),
});
if (response.ok) {
await this.loadStyles();
this.renderStylesList();
}
} catch (e) {
console.error("Failed to save style:", e);
}
}
async deleteStyle(number) {
if (!confirm(`Delete style #${number}?`)) return;
try {
const response = await api.fetchApi(`/api/rmautomation/styles/${number}`, {
method: "DELETE",
});
if (response.ok) {
this.selectedStyle = null;
await this.loadStyles();
this.renderStylesList();
this.renderEditor();
}
} catch (e) {
console.error("Failed to delete style:", e);
}
}
render() {
this.overlay = document.createElement("div");
this.overlay.className = "rm-styles-modal-overlay";
this.overlay.onclick = (e) => {
if (e.target === this.overlay) this.close();
};
const modal = document.createElement("div");
modal.className = "rm-styles-modal";
modal.innerHTML = `
<div class="rm-styles-modal-header">
<h2>RM Styles Editor</h2>
<button class="rm-styles-modal-close">&times;</button>
</div>
<div class="rm-styles-modal-content">
<div class="rm-styles-toolbar">
<button id="rm-add-style">+ Add New Style</button>
<button id="rm-import-styles" class="rm-btn-import">Import JSON</button>
<input type="file" id="rm-file-input" class="rm-file-input" accept=".json">
</div>
<div class="rm-styles-list" id="rm-styles-list"></div>
<div id="rm-style-editor"></div>
</div>
`;
modal.querySelector(".rm-styles-modal-close").onclick = () => this.close();
modal.querySelector("#rm-add-style").onclick = () => this.addNewStyle();
modal.querySelector("#rm-import-styles").onclick = () => this.triggerFileImport();
modal.querySelector("#rm-file-input").onchange = (e) => this.handleFileSelect(e);
this.overlay.appendChild(modal);
document.body.appendChild(this.overlay);
this.renderStylesList();
}
renderStylesList() {
const list = document.getElementById("rm-styles-list");
if (!list) return;
list.innerHTML = this.styles
.map(
(style) => `
<div class="rm-style-item ${this.selectedStyle?.number === style.number ? "selected" : ""}"
data-number="${style.number}">
<span class="rm-style-number">#${style.number}</span>
<span class="rm-style-name">${style.name || "(unnamed)"}</span>
<span class="rm-style-preview">${style.positive?.substring(0, 50) || ""}</span>
</div>
`
)
.join("");
list.querySelectorAll(".rm-style-item").forEach((item) => {
item.onclick = () => {
const number = parseInt(item.dataset.number);
this.selectedStyle = this.styles.find((s) => s.number === number);
this.renderStylesList();
this.renderEditor();
};
});
}
renderEditor() {
const editor = document.getElementById("rm-style-editor");
if (!editor) return;
if (!this.selectedStyle) {
editor.innerHTML = "";
return;
}
const s = this.selectedStyle;
editor.innerHTML = `
<div class="rm-style-editor">
<div class="rm-style-editor-row">
<label>Number</label>
<input type="number" id="rm-edit-number" value="${s.number}" min="1">
</div>
<div class="rm-style-editor-row">
<label>Name</label>
<input type="text" id="rm-edit-name" value="${s.name || ""}">
</div>
<div class="rm-style-editor-row">
<label>Positive Prompt</label>
<textarea id="rm-edit-positive">${s.positive || ""}</textarea>
</div>
<div class="rm-style-editor-row">
<label>Negative Prompt</label>
<textarea id="rm-edit-negative">${s.negative || ""}</textarea>
</div>
<div class="rm-style-editor-row">
<label>Motion Prompt</label>
<textarea id="rm-edit-motion">${s.motion || ""}</textarea>
</div>
<div class="rm-loras-section">
<h4>Image LoRAs</h4>
<div id="rm-image-loras"></div>
<button class="rm-lora-add" data-target="imageLoras">+ Add LoRA</button>
</div>
<div class="rm-loras-section">
<h4>Motion LoRAs High</h4>
<div id="rm-motion-loras-high"></div>
<button class="rm-lora-add" data-target="motionLorasHigh">+ Add LoRA</button>
</div>
<div class="rm-loras-section">
<h4>Motion LoRAs Low</h4>
<div id="rm-motion-loras-low"></div>
<button class="rm-lora-add" data-target="motionLorasLow">+ Add LoRA</button>
</div>
<div class="rm-style-actions">
<button class="rm-btn-delete" id="rm-delete-style">Delete</button>
<button class="rm-btn-cancel" id="rm-cancel-edit">Cancel</button>
<button class="rm-btn-save" id="rm-save-style">Save</button>
</div>
</div>
`;
this.renderLoras("imageLoras", "rm-image-loras");
this.renderLoras("motionLorasHigh", "rm-motion-loras-high");
this.renderLoras("motionLorasLow", "rm-motion-loras-low");
editor.querySelectorAll(".rm-lora-add").forEach((btn) => {
btn.onclick = () => this.addLora(btn.dataset.target);
});
editor.querySelector("#rm-save-style").onclick = () => this.saveCurrentStyle();
editor.querySelector("#rm-delete-style").onclick = () =>
this.deleteStyle(this.selectedStyle.number);
editor.querySelector("#rm-cancel-edit").onclick = () => {
this.selectedStyle = null;
this.renderStylesList();
this.renderEditor();
};
}
renderLoras(field, containerId) {
const container = document.getElementById(containerId);
if (!container) return;
const loras = this.selectedStyle[field] || [];
container.innerHTML = loras
.map(
(lora, i) => `
<div class="rm-lora-item">
<input type="text" value="${lora.path || ""}" placeholder="lora.safetensors" data-field="${field}" data-index="${i}" data-prop="path">
<input type="number" value="${lora.weight || 1}" step="0.1" data-field="${field}" data-index="${i}" data-prop="weight">
<button class="rm-lora-remove" data-field="${field}" data-index="${i}">X</button>
</div>
`
)
.join("");
container.querySelectorAll("input").forEach((input) => {
input.onchange = () => {
const { field, index, prop } = input.dataset;
if (!this.selectedStyle[field]) this.selectedStyle[field] = [];
if (!this.selectedStyle[field][index]) this.selectedStyle[field][index] = {};
this.selectedStyle[field][index][prop] =
prop === "weight" ? parseFloat(input.value) : input.value;
};
});
container.querySelectorAll(".rm-lora-remove").forEach((btn) => {
btn.onclick = () => {
const { field, index } = btn.dataset;
this.selectedStyle[field].splice(parseInt(index), 1);
this.renderLoras(field, containerId);
};
});
}
addLora(field) {
if (!this.selectedStyle[field]) this.selectedStyle[field] = [];
this.selectedStyle[field].push({ path: "", weight: 1 });
this.renderEditor();
}
addNewStyle() {
const maxNumber = Math.max(0, ...this.styles.map((s) => s.number));
this.selectedStyle = {
number: maxNumber + 1,
name: "",
positive: "",
negative: "",
motion: "",
imageLoras: [],
motionLoras: [],
motionLorasHigh: [],
motionLorasLow: [],
};
this.renderStylesList();
this.renderEditor();
}
saveCurrentStyle() {
const number = parseInt(document.getElementById("rm-edit-number").value);
const style = {
...this.selectedStyle,
number: number,
name: document.getElementById("rm-edit-name").value,
positive: document.getElementById("rm-edit-positive").value,
negative: document.getElementById("rm-edit-negative").value,
motion: document.getElementById("rm-edit-motion").value,
};
this.saveStyle(style);
}
triggerFileImport() {
const fileInput = document.getElementById("rm-file-input");
if (fileInput) {
fileInput.value = "";
fileInput.click();
}
}
handleFileSelect(event) {
const file = event.target.files[0];
if (!file) return;
const reader = new FileReader();
reader.onload = (e) => {
try {
const importedData = JSON.parse(e.target.result);
let importedStyles = [];
// Handle both array format and object with styles property
if (Array.isArray(importedData)) {
importedStyles = importedData;
} else if (importedData.styles && Array.isArray(importedData.styles)) {
importedStyles = importedData.styles;
} else {
alert("Invalid JSON format. Expected an array of styles or an object with a 'styles' array.");
return;
}
if (importedStyles.length === 0) {
alert("No styles found in the imported file.");
return;
}
this.showImportDialog(importedStyles);
} catch (err) {
console.error("Failed to parse JSON:", err);
alert("Failed to parse JSON file. Please check the file format.");
}
};
reader.readAsText(file);
}
showImportDialog(importedStyles) {
const dialogOverlay = document.createElement("div");
dialogOverlay.className = "rm-import-dialog-overlay";
const dialog = document.createElement("div");
dialog.className = "rm-import-dialog";
dialog.innerHTML = `
<h3>Import Styles</h3>
<p>Found ${importedStyles.length} style(s) to import. How would you like to proceed?</p>
<div class="rm-import-dialog-buttons">
<button class="rm-import-cancel">Cancel</button>
<button class="rm-import-replace">Replace All</button>
<button class="rm-import-append">Append</button>
</div>
`;
dialog.querySelector(".rm-import-cancel").onclick = () => {
dialogOverlay.remove();
};
dialog.querySelector(".rm-import-replace").onclick = async () => {
dialogOverlay.remove();
await this.importStyles(importedStyles, "replace");
};
dialog.querySelector(".rm-import-append").onclick = async () => {
dialogOverlay.remove();
await this.importStyles(importedStyles, "append");
};
dialogOverlay.appendChild(dialog);
document.body.appendChild(dialogOverlay);
}
async importStyles(importedStyles, mode) {
try {
const response = await api.fetchApi("/api/rmautomation/styles/import", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
styles: importedStyles,
mode: mode
}),
});
if (response.ok) {
await this.loadStyles();
this.selectedStyle = null;
this.renderStylesList();
this.renderEditor();
} else {
const error = await response.text();
console.error("Import failed:", error);
alert("Failed to import styles. Check console for details.");
}
} catch (e) {
console.error("Failed to import styles:", e);
alert("Failed to import styles. Check console for details.");
}
}
}
// Register the button on RM Styles nodes
app.registerExtension({
name: "RMAutomation.StylesEditor",
async beforeRegisterNodeDef(nodeType, nodeData, app) {
if (nodeData.name === "RMStylesFull" || nodeData.name === "RMStylesFullDisplay" || nodeData.name === "RMStylesPipe") {
const onNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = function () {
onNodeCreated?.apply(this, arguments);
// Add "Edit Styles" button
const editBtn = this.addWidget("button", "Edit Styles", null, () => {
const modal = new RMStylesEditorModal();
modal.open();
});
};
}
},
});
+281
View File
@@ -0,0 +1,281 @@
/**
* RM Styles Full - Frontend Widget Controller
* Handles mode switching, random/increment/decrement logic
*/
import { app } from "../../scripts/app.js";
import { api } from "../../scripts/api.js";
console.log("[RMStyles] Script loading...");
const RANDOM_HISTORY_SIZE = 30;
const RM_STYLES_NODES = ["RMStylesFull", "RMStylesFullDisplay", "RMStylesPipe"];
/**
* Generate a random integer between min and max (inclusive),
* avoiding recently used values tracked in the history array.
*/
function randomInRange(min, max, history) {
if (min >= max) return min;
const range = max - min + 1;
if (history) {
const effectiveBuffer = Math.min(RANDOM_HISTORY_SIZE, range - 1);
const avoid = new Set(history.slice(-effectiveBuffer));
let result, attempts = 0;
do {
result = Math.floor(Math.random() * range) + min;
attempts++;
} while (avoid.has(result) && attempts < 100);
history.push(result);
if (history.length > RANDOM_HISTORY_SIZE) {
history.splice(0, history.length - RANDOM_HISTORY_SIZE);
}
return result;
}
return Math.floor(Math.random() * range) + min;
}
/**
* Calculate the next prompt number based on mode
*/
function calculateNextNumber(mode, current, min, max, history) {
let result;
switch (mode) {
case "Random":
result = randomInRange(min, max, history);
break;
case "Increment":
result = current >= max ? min : current + 1;
break;
case "Decrement":
result = current <= min ? max : current - 1;
break;
case "Manual":
default:
result = current;
}
return result;
}
/**
* Get all widget values from a node for debugging
*/
function getWidgetValues(node) {
if (!node?.widgets) return "no widgets";
const values = {};
for (const w of node.widgets) {
values[w.name] = w.value;
}
return values;
}
/**
* Set up widget callbacks for a node
*/
function setupNodeWidgets(node) {
if (!node.widgets) return;
// Check if already set up
if (node._rmSetupDone) return;
node._rmSetupDone = true;
// Find our widgets
const modeWidget = node.widgets.find((w) => w.name === "mode");
const prevWidget = node.widgets.find((w) => w.name === "previous_prompt");
const nextWidget = node.widgets.find((w) => w.name === "next_prompt");
const minWidget = node.widgets.find((w) => w.name === "minimum");
const maxWidget = node.widgets.find((w) => w.name === "maximum");
if (!modeWidget || !prevWidget || !nextWidget || !minWidget || !maxWidget) {
return;
}
// Store widget references on node for easy access
node._rmWidgets = { modeWidget, prevWidget, nextWidget, minWidget, maxWidget };
// History buffer to prevent random repeats
if (!node._rmHistory) {
node._rmHistory = [];
}
// Flag to prevent recursive callbacks
node._rmUpdating = false;
// Hook mode widget callback
const origModeCallback = modeWidget.callback;
modeWidget.callback = function (value) {
if (origModeCallback) {
origModeCallback.call(this, value);
}
if (node._rmUpdating) return;
node._rmUpdating = true;
// Calculate new next_prompt based on mode
if (value !== "Manual") {
const current = nextWidget.value;
const min = minWidget.value;
const max = maxWidget.value;
const newValue = calculateNextNumber(value, current, min, max, node._rmHistory);
nextWidget.value = newValue;
console.log(`[RMStyles] Mode changed to ${value}, next_prompt: ${newValue}`);
}
node._rmUpdating = false;
};
// Hook next_prompt widget callback to detect manual edits
const origNextCallback = nextWidget.callback;
nextWidget.callback = function (value) {
if (origNextCallback) {
origNextCallback.call(this, value);
}
if (node._rmUpdating) return;
// User manually edited - switch to Manual mode
if (modeWidget.value !== "Manual") {
node._rmUpdating = true;
modeWidget.value = "Manual";
node._rmUpdating = false;
}
};
}
/**
* Handle node execution complete
*/
function handleNodeExecuted(node) {
const w = node._rmWidgets;
if (!w) return;
const mode = w.modeWidget.value;
const usedPrompt = w.nextWidget.value;
const min = w.minWidget.value;
const max = w.maxWidget.value;
// Update previous_prompt to show what was just used
node._rmUpdating = true;
w.prevWidget.value = usedPrompt;
// Calculate new next_prompt for non-Manual modes
if (mode !== "Manual") {
// Add the USED prompt to history before generating new one
if (mode === "Random" && !node._rmHistory.includes(usedPrompt)) {
node._rmHistory.push(usedPrompt);
if (node._rmHistory.length > RANDOM_HISTORY_SIZE) {
node._rmHistory.splice(0, node._rmHistory.length - RANDOM_HISTORY_SIZE);
}
}
const newNext = calculateNextNumber(mode, usedPrompt, min, max, node._rmHistory);
w.nextWidget.value = newNext;
console.log(`[RMStyles] Execution complete: used=${usedPrompt}, next=${newNext}, mode=${mode}`);
}
node._rmUpdating = false;
}
// Register the extension
app.registerExtension({
name: "RMAutomation.StylesFull",
async beforeRegisterNodeDef(nodeType, nodeData, app) {
if (!RM_STYLES_NODES.includes(nodeData.name)) return;
console.log(`[RMStyles] Registering handlers for ${nodeData.name}`);
// Hook into node creation
const origOnNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = function () {
const result = origOnNodeCreated?.apply(this, arguments);
const node = this;
setTimeout(() => {
setupNodeWidgets(node);
}, 200);
return result;
};
// Also hook onConfigure for loaded workflows
const origOnConfigure = nodeType.prototype.onConfigure;
nodeType.prototype.onConfigure = function (data) {
const result = origOnConfigure?.apply(this, arguments);
const node = this;
setTimeout(() => {
setupNodeWidgets(node);
}, 200);
return result;
};
},
async setup() {
console.log("[RMStyles] Setting up API event listeners...");
// Track nodes that are part of the current execution
const pendingNodes = new Set();
// Helper to process pending nodes
const processPendingNodes = () => {
for (const node of pendingNodes) {
handleNodeExecuted(node);
}
pendingNodes.clear();
};
api.addEventListener("execution_start", (e) => {
pendingNodes.clear();
});
api.addEventListener("executing", (e) => {
// Handle both old and new ComfyUI event formats
const nodeId =
typeof e.detail === "object" && e.detail !== null
? e.detail.node
: e.detail;
if (nodeId) {
const node = app.graph?.getNodeById(Number(nodeId));
if (node && RM_STYLES_NODES.includes(node.type)) {
pendingNodes.add(node);
}
} else {
// null detail = execution finished - update all our nodes that ran
processPendingNodes();
}
});
// Handle cached executions
api.addEventListener("execution_cached", (e) => {
const cachedNodes = e.detail?.nodes ?? [];
for (const nodeId of cachedNodes) {
const node = app.graph?.getNodeById(Number(nodeId));
if (node && RM_STYLES_NODES.includes(node.type)) {
pendingNodes.add(node);
}
}
});
// Handle execution interrupts - still update our nodes
api.addEventListener("execution_interrupted", (e) => {
console.log("[RMStyles] Execution interrupted, updating pending nodes...");
processPendingNodes();
});
// Handle execution errors - still update our nodes
api.addEventListener("execution_error", (e) => {
console.log("[RMStyles] Execution error, updating pending nodes...");
processPendingNodes();
});
console.log("[RMStyles] Extension setup complete!");
},
});
console.log("[RMStyles] Script loaded");
+1
View File
@@ -0,0 +1 @@
# Node modules
+48
View File
@@ -0,0 +1,48 @@
"""
RM Image Fallback - Return the first available image from multiple inputs.
"""
import torch
class RMImageFallback:
"""Return the first non-empty image from multiple optional inputs."""
@classmethod
def INPUT_TYPES(cls):
return {
"optional": {
"image_1": ("IMAGE", {"default": None}),
"image_2": ("IMAGE", {"default": None}),
"image_3": ("IMAGE", {"default": None}),
}
}
RETURN_TYPES = ("IMAGE",)
FUNCTION = "process"
CATEGORY = "RMAutomation/Image"
def process(self, **kwargs):
image_1 = kwargs.get('image_1', None)
image_2 = kwargs.get('image_2', None)
image_3 = kwargs.get('image_3', None)
if image_1 is not None and len(image_1) > 0:
return (image_1,)
if image_2 is not None and len(image_2) > 0:
return (image_2,)
if image_3 is not None and len(image_3) > 0:
return (image_3,)
return (torch.zeros((1, 3, 512, 512)),)
NODE_CLASS_MAPPINGS = {
"RMImageFallback": RMImageFallback,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMImageFallback": "RM Image Fallback",
}
+57
View File
@@ -0,0 +1,57 @@
"""
RM Latent Nodes - Improved latent manipulation nodes.
"""
import torch
class RMSetLatentNoiseMask:
"""Set a noise mask on latent samples with improved error handling."""
@classmethod
def INPUT_TYPES(s):
return {
"required": {"samples": ("LATENT",)},
"optional": {"mask": ("MASK",)}
}
RETURN_TYPES = ("LATENT", "STRING")
RETURN_NAMES = ("LATENT", "STATUS")
FUNCTION = "set_mask"
CATEGORY = "RMAutomation/Latent"
def set_mask(self, samples, mask=None):
s = samples.copy()
status_message = "Noise mask not added"
if mask is None:
return (s, status_message)
if mask.numel() == 0 or 0 in mask.shape:
print("Warning: Empty or zero-dimensional mask provided. Returning original samples without modification.")
return (s, status_message)
try:
if mask.dim() < 2:
mask = mask.unsqueeze(0).unsqueeze(0)
elif mask.dim() == 2:
mask = mask.unsqueeze(0)
reshaped_mask = mask.reshape((-1, 1, mask.shape[-2], mask.shape[-1]))
s["noise_mask"] = reshaped_mask
status_message = "Noise mask added"
except RuntimeError as e:
print(f"Error reshaping mask: {e}")
print("Returning original samples without modification.")
return (s, status_message)
NODE_CLASS_MAPPINGS = {
"RMSetLatentNoiseMask": RMSetLatentNoiseMask,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMSetLatentNoiseMask": "RM Set Latent Noise Mask",
}
+99
View File
@@ -0,0 +1,99 @@
"""
RM Mask Gate - Passes mask through only if it's valid (non-empty).
Outputs None if the mask is empty or has zero dimensions.
"""
class RMMaskGate:
"""
Checks if a mask is valid (non-empty, non-zero dimensions).
If valid, passes it through. If not, outputs None.
Use this before nodes that crash on empty masks.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"mask": ("MASK",),
}
}
CATEGORY = "RMAutomation/Mask"
RETURN_TYPES = ("MASK",)
RETURN_NAMES = ("mask",)
FUNCTION = "gate"
def gate(self, mask=None):
if mask is None:
print("[RMMaskGate] No mask input, outputting None")
return (None,)
if mask.numel() == 0:
print("[RMMaskGate] Empty mask (0 elements), outputting None")
return (None,)
if any(d == 0 for d in mask.shape):
print(f"[RMMaskGate] Zero-dimensional mask {mask.shape}, outputting None")
return (None,)
if mask.sum() == 0:
print(f"[RMMaskGate] All-zero mask {mask.shape}, outputting None")
return (None,)
print(f"[RMMaskGate] Valid mask {mask.shape}, passing through")
return (mask,)
class RMMaskGateGuide:
"""
Takes a mask and a guide input. If the mask is invalid, outputs None for the guide.
This completely disables the guide when no valid mask is detected.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"mask": ("MASK",),
"guide": ("GUIDE",),
}
}
CATEGORY = "RMAutomation/Mask"
RETURN_TYPES = ("GUIDE",)
RETURN_NAMES = ("guide",)
FUNCTION = "gate"
def _is_valid_mask(self, mask):
if mask is None:
return False
if mask.numel() == 0:
return False
if any(d == 0 for d in mask.shape):
return False
if mask.sum() == 0:
return False
return True
def gate(self, mask=None, guide=None):
if not self._is_valid_mask(mask):
print("[RMMaskGateGuide] Invalid/empty mask, disabling guide (outputting None)")
return (None,)
print(f"[RMMaskGateGuide] Valid mask {mask.shape}, passing guide through")
return (guide,)
NODE_CLASS_MAPPINGS = {
"RMMaskGate": RMMaskGate,
"RMMaskGateGuide": RMMaskGateGuide,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMMaskGate": "RM Mask Gate",
"RMMaskGateGuide": "RM Mask Gate (Guide)",
}
+741
View File
@@ -0,0 +1,741 @@
"""
RM Power LoRA Loader
Standalone implementation with dynamic inputs and random strength support.
Includes Civitai AIR mode for downloading loras by model ID.
"""
import random
import re
import folder_paths
from nodes import LoraLoader
from typing import Optional, Tuple, Any, List, Dict
import os
import json
import hashlib
import requests
import concurrent.futures
from tqdm import tqdm
import comfy.utils
# Type alias for our lora stack
LORA_STACK_TYPE = List[Dict[str, Any]]
# Civitai API constants
CIVITAI_API_BASE = "https://civitai.com/api/v1"
DOWNLOAD_HISTORY_FILE = os.path.join(os.path.dirname(os.path.dirname(__file__)), "civitai_download_history.json")
class CivitaiDownloader:
"""Handles downloading loras from Civitai using AIR (model IDs)."""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key
self.download_history = self._load_history()
def _load_history(self) -> dict:
"""Load download history from file."""
if os.path.exists(DOWNLOAD_HISTORY_FILE):
try:
with open(DOWNLOAD_HISTORY_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except Exception as e:
print(f"[CivitaiDownloader] Failed to load history: {e}")
return {}
def _save_history(self):
"""Save download history to file."""
try:
with open(DOWNLOAD_HISTORY_FILE, 'w', encoding='utf-8') as f:
json.dump(self.download_history, f, indent=2, ensure_ascii=False)
except Exception as e:
print(f"[CivitaiDownloader] Failed to save history: {e}")
def parse_air(self, air: str) -> Tuple[Optional[int], Optional[int]]:
"""Parse AIR format: model_id or model_id@version_id."""
air = str(air).strip()
if '@' in air:
parts = air.split('@')
model_id = int(parts[0]) if parts[0] else None
version_id = int(parts[1]) if len(parts) > 1 and parts[1] else None
else:
model_id = int(air) if air else None
version_id = None
return model_id, version_id
def get_cached_filename(self, model_id: int, version_id: Optional[int] = None) -> Optional[str]:
"""Check if we've already downloaded this model."""
model_key = str(model_id)
if model_key in self.download_history:
versions = self.download_history[model_key]
for version_info in versions:
if version_id is None or version_info.get('id') == version_id:
files = version_info.get('files', [])
if files:
return files[0].get('name')
return None
def find_local_file(self, filename: str) -> Optional[str]:
"""Check if a file exists in the loras folder."""
if not filename:
return None
lora_paths = folder_paths.get_folder_paths("loras")
for path in lora_paths:
full_path = os.path.join(path, filename)
if os.path.exists(full_path) and os.path.getsize(full_path) > 0:
return filename
# Also check subdirectories
for root, dirs, files in os.walk(path):
if filename in files:
full_path = os.path.join(root, filename)
if os.path.getsize(full_path) > 0:
# Return relative path
return os.path.relpath(full_path, path)
return None
def fetch_model_info(self, model_id: int) -> Optional[dict]:
"""Fetch model information from Civitai API."""
try:
url = f"{CIVITAI_API_BASE}/models/{model_id}"
response = requests.get(url, timeout=30)
if response.status_code == 200:
return response.json()
else:
print(f"[CivitaiDownloader] API returned {response.status_code} for model {model_id}")
return None
except Exception as e:
print(f"[CivitaiDownloader] Failed to fetch model info: {e}")
return None
def download_lora(self, air: str) -> Optional[str]:
"""
Download a lora from Civitai by AIR.
Returns the local filename if successful, None otherwise.
"""
model_id, version_id = self.parse_air(air)
if model_id is None:
print(f"[CivitaiDownloader] Invalid AIR: {air}")
return None
# Check if already downloaded
cached_name = self.get_cached_filename(model_id, version_id)
if cached_name:
local_file = self.find_local_file(cached_name)
if local_file:
print(f"[CivitaiDownloader] Using cached: {cached_name}")
return local_file
# Fetch model info from API
print(f"[CivitaiDownloader] Fetching info for model {model_id}...")
model_info = self.fetch_model_info(model_id)
if not model_info:
return None
# Validate model type
model_type = model_info.get('type', '').upper()
if model_type not in ['LORA', 'LOCON']:
print(f"[CivitaiDownloader] Model {model_id} is not a LoRA (type: {model_type})")
return None
# Get version info
versions = model_info.get('modelVersions', [])
if not versions:
print(f"[CivitaiDownloader] No versions found for model {model_id}")
return None
# Find the requested version or use the first (latest)
version_info = None
if version_id:
for v in versions:
if v.get('id') == version_id:
version_info = v
break
if not version_info:
print(f"[CivitaiDownloader] Version {version_id} not found, using latest")
if not version_info:
version_info = versions[0]
# Get file info
files = version_info.get('files', [])
if not files:
print(f"[CivitaiDownloader] No files found for version")
return None
# Find the primary/safetensors file
file_info = None
for f in files:
if f.get('name', '').endswith('.safetensors'):
file_info = f
break
if not file_info:
file_info = files[0]
filename = file_info.get('name')
download_url = file_info.get('downloadUrl')
file_size = file_info.get('sizeKB', 0) * 1024
expected_sha256 = file_info.get('hashes', {}).get('SHA256', '').upper()
if not download_url:
print(f"[CivitaiDownloader] No download URL for file")
return None
# Check if file already exists locally
local_file = self.find_local_file(filename)
if local_file:
print(f"[CivitaiDownloader] File already exists: {filename}")
# Save to history
self._update_history(model_id, version_info.get('id'), file_info)
return local_file
# Add API key to URL if provided
if self.api_key:
download_url = f"{download_url}?token={self.api_key}"
# Get save path
lora_paths = folder_paths.get_folder_paths("loras")
save_path = os.path.join(lora_paths[0], filename)
# Download the file
print(f"[CivitaiDownloader] Downloading {filename}...")
try:
success = self._download_file(download_url, save_path, file_size, expected_sha256)
if success:
self._update_history(model_id, version_info.get('id'), file_info)
print(f"[CivitaiDownloader] Successfully downloaded: {filename}")
return filename
else:
return None
except Exception as e:
print(f"[CivitaiDownloader] Download failed: {e}")
if os.path.exists(save_path):
os.remove(save_path)
return None
def _download_file(self, url: str, save_path: str, expected_size: float, expected_sha256: str) -> bool:
"""Download a file with progress bar and SHA256 verification."""
try:
response = requests.get(url, stream=True, timeout=30)
response.raise_for_status()
total_size = int(response.headers.get('content-length', expected_size))
with open(save_path, 'wb') as f:
with tqdm(total=total_size, unit='B', unit_scale=True, desc="Downloading") as pbar:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
pbar.update(len(chunk))
# Verify SHA256 if provided
if expected_sha256:
actual_sha256 = self._calculate_sha256(save_path)
if actual_sha256.upper() != expected_sha256.upper():
print(f"[CivitaiDownloader] SHA256 mismatch! Expected {expected_sha256}, got {actual_sha256}")
os.remove(save_path)
return False
print(f"[CivitaiDownloader] SHA256 verified: {actual_sha256}")
return True
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
print(f"[CivitaiDownloader] Authentication required. Please provide a valid Civitai API key.")
else:
print(f"[CivitaiDownloader] HTTP error: {e}")
return False
except Exception as e:
print(f"[CivitaiDownloader] Download error: {e}")
return False
def _calculate_sha256(self, file_path: str) -> str:
"""Calculate SHA256 hash of a file."""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest().upper()
def _update_history(self, model_id: int, version_id: int, file_info: dict):
"""Update download history."""
model_key = str(model_id)
if model_key not in self.download_history:
self.download_history[model_key] = []
# Check if this version already exists
for v in self.download_history[model_key]:
if v.get('id') == version_id:
return
self.download_history[model_key].append({
'id': version_id,
'files': [{
'name': file_info.get('name'),
'sizeKB': file_info.get('sizeKB'),
'hashes': file_info.get('hashes', {})
}]
})
self._save_history()
class FlexibleOptionalInputType(dict):
"""
Allows dynamic inputs - ComfyUI will accept any input we provide at runtime.
This enables the Power LoRA Loader to accept unlimited lora inputs added via UI.
"""
def __init__(self, any_type_str="*"):
self.any_type_str = any_type_str
super().__init__({
"Style Import": ("STRING", {"forceInput": True}),
})
def __getitem__(self, key):
if key in dict.keys(self):
return dict.__getitem__(self, key)
return (self.any_type_str,)
def __contains__(self, key):
if key == "style_import":
return False
return True
def get_lora_by_filename(filename, log_node=None):
"""
Find a lora file by name. Returns the full relative path or None.
"""
if not filename or filename == "None":
return None
lora_path = folder_paths.get_full_path("loras", filename)
if lora_path and os.path.exists(lora_path):
return filename
all_loras = folder_paths.get_filename_list("loras")
for lora in all_loras:
if lora == filename or os.path.basename(lora) == filename:
return lora
return None
def parse_lora_string(lora_string: str) -> List[Dict[str, Any]]:
"""
Parse a lora string in the format: <lora:filename:weight>, <lora:filename:weight>
Returns a list of dicts with 'name' and 'weight' keys.
"""
if not lora_string or not lora_string.strip():
return []
results = []
pattern = r'<lora:([^:>]+):([^>]+)>'
matches = re.findall(pattern, lora_string)
for name, weight_str in matches:
try:
weight = float(weight_str)
results.append({
'name': name.strip(),
'weight': weight
})
except ValueError:
print(f"[parse_lora_string] Invalid weight '{weight_str}' for lora '{name}'")
continue
return results
class RMPowerLoraLoader:
"""
Powerful, flexible node to add multiple loras with optional random strength.
Features:
- Add unlimited loras via UI button
- Each lora can have fixed or random strength
- Right-click on lora for options (toggle, delete, move)
- Accept lora_string input (e.g., from RM Styles Full "Image Loras")
- Civitai AIR mode - download loras by model ID
- Outputs LORA_STACK for use with RMLoraApply
"""
NAME = "RM Power Lora Loader"
CATEGORY = "RMAutomation/LoRA"
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": FlexibleOptionalInputType("*"),
"hidden": {},
}
RETURN_TYPES = ("LORA_STACK",)
RETURN_NAMES = ("Lora Stack",)
FUNCTION = "build_lora_stack"
def build_lora_stack(self, **kwargs):
"""Build a lora stack from the dynamic inputs provided by the UI and Style Import input."""
stack = []
# Get Civitai API key if any lora uses AIR mode
civitai_api_key = kwargs.get("civitai_api_key", None)
civitai_downloader = None
# First, parse the Style Import input
style_import = kwargs.get("Style Import", None)
if style_import:
parsed_loras = parse_lora_string(style_import)
for lora_info in parsed_loras:
lora_name = lora_info['name']
weight = lora_info['weight']
lora_file = get_lora_by_filename(lora_name, log_node=self.NAME)
if lora_file is None:
print(f"[RMPowerLoraLoader] LoRA not found: {lora_name}")
continue
lora_config = {
"name": lora_file,
"filename": os.path.basename(lora_file),
"strength": round(weight, 3),
"source": "style",
"random": False,
}
stack.append(lora_config)
# Then process UI-added loras
for key, value in kwargs.items():
key_upper = key.upper()
if key_upper.startswith('LORA_') and isinstance(value, dict):
if not value.get('on', True):
continue
# Check if this is AIR mode
air_mode = value.get('airMode', False)
if air_mode:
# AIR mode - download from Civitai
air = value.get('air', '')
if not air:
continue
# Initialize downloader lazily
if civitai_downloader is None:
civitai_downloader = CivitaiDownloader(api_key=civitai_api_key)
# Download or find the lora
lora_file = civitai_downloader.download_lora(air)
if lora_file is None:
print(f"[RMPowerLoraLoader] Failed to get Civitai lora: {air}")
continue
else:
# Normal mode - local lora
lora_name = value.get('lora')
if not lora_name or lora_name == "None":
continue
lora_file = get_lora_by_filename(lora_name, log_node=self.NAME)
if lora_file is None:
print(f"[RMPowerLoraLoader] LoRA not found: {lora_name}")
continue
use_random = value.get('random', False)
if use_random:
strength_min = value.get('strengthMin', 0.5)
strength_max = value.get('strengthMax', 1.0)
actual_min = min(strength_min, strength_max)
actual_max = max(strength_min, strength_max)
final_strength = random.uniform(actual_min, actual_max)
else:
final_strength = value.get('strength', 1.0)
lora_config = {
"name": lora_file,
"filename": os.path.basename(lora_file),
"strength": round(final_strength, 3),
"source": "civitai" if air_mode else "local",
"random": use_random,
}
if air_mode:
lora_config["air"] = value.get('air', '')
if use_random:
lora_config["strength_min"] = actual_min
lora_config["strength_max"] = actual_max
stack.append(lora_config)
return (stack,)
@classmethod
def IS_CHANGED(cls, **kwargs):
for key, value in kwargs.items():
if key.upper().startswith('LORA_') and isinstance(value, dict):
if value.get('random', False):
return float("nan")
return None
class RMLoraCollector:
"""
Combine multiple lora stacks into one.
Chainable - can connect multiple collectors together.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": {
"lora_stack_1": ("LORA_STACK",),
"lora_stack_2": ("LORA_STACK",),
"lora_stack_3": ("LORA_STACK",),
"lora_stack_4": ("LORA_STACK",),
"lora_stack_5": ("LORA_STACK",),
"lora_string_1": ("STRING", {"forceInput": True}),
"lora_string_2": ("STRING", {"forceInput": True}),
"lora_string_3": ("STRING", {"forceInput": True}),
},
}
RETURN_TYPES = ("LORA_STACK",)
RETURN_NAMES = ("lora_stack",)
FUNCTION = "collect_loras"
CATEGORY = "RMAutomation/LoRA"
def collect_loras(
self,
lora_stack_1: Optional[LORA_STACK_TYPE] = None,
lora_stack_2: Optional[LORA_STACK_TYPE] = None,
lora_stack_3: Optional[LORA_STACK_TYPE] = None,
lora_stack_4: Optional[LORA_STACK_TYPE] = None,
lora_stack_5: Optional[LORA_STACK_TYPE] = None,
lora_string_1: Optional[str] = None,
lora_string_2: Optional[str] = None,
lora_string_3: Optional[str] = None,
) -> Tuple[LORA_STACK_TYPE]:
combined = []
for stack in [lora_stack_1, lora_stack_2, lora_stack_3, lora_stack_4, lora_stack_5]:
if stack:
combined.extend(stack)
for lora_string in [lora_string_1, lora_string_2, lora_string_3]:
if lora_string:
parsed = parse_lora_string(lora_string)
for lora_info in parsed:
lora_file = get_lora_by_filename(lora_info['name'])
if lora_file is None:
print(f"[RMLoraCollector] Warning: LoRA not found: {lora_info['name']}")
continue
combined.append({
"name": lora_file,
"filename": os.path.basename(lora_file),
"strength": round(lora_info['weight'], 3),
"source": "lora_string",
"random": False,
})
return (combined,)
class FlexibleLoraApplyInputType(dict):
"""
Allows dynamic inputs for RMLoraApply.
Accepts lora_stack, lora_stack_2, etc. and lora_string, lora_string_2, etc.
"""
def __init__(self):
super().__init__({
"clip": ("CLIP",),
"lora_stack": ("LORA_STACK",),
})
def __getitem__(self, key):
if key in dict.keys(self):
return dict.__getitem__(self, key)
# Accept lora_stack_N keys
if key.startswith("lora_stack"):
return ("LORA_STACK",)
# Accept lora_string_N keys
if key.startswith("lora_string"):
return ("STRING", {"forceInput": True})
return ("*",)
def __contains__(self, key):
if key in dict.keys(self):
return True
if key.startswith("lora_stack"):
return True
if key.startswith("lora_string"):
return True
return False
class RMLoraApply:
"""
Apply all LoRAs from lora stacks and/or lora strings to the model and clip.
Use "Add Lora Stack" for LORA_STACK inputs from Power Lora Loader.
Use "Add Lora String" for STRING inputs (e.g., Image Loras from Styles Pipe Out).
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": ("MODEL",),
},
"optional": FlexibleLoraApplyInputType(),
}
RETURN_TYPES = ("MODEL", "CLIP")
RETURN_NAMES = ("Model", "Clip")
FUNCTION = "apply_loras"
CATEGORY = "RMAutomation/LoRA"
def apply_loras(self, model, clip=None, **kwargs) -> Tuple[Any, Any]:
# Combine all lora stacks
combined_stack = []
# Get lora_stack (the first one)
first_stack = kwargs.get("lora_stack")
if first_stack:
combined_stack.extend(first_stack)
# Get any additional lora_stack_N inputs
for key in sorted(kwargs.keys()):
if key.startswith("lora_stack_") and kwargs[key]:
combined_stack.extend(kwargs[key])
# Parse lora_string inputs
for key in sorted(kwargs.keys()):
if key.startswith("lora_string") and kwargs[key]:
parsed = parse_lora_string(kwargs[key])
for lora_info in parsed:
lora_file = get_lora_by_filename(lora_info['name'])
if lora_file is None:
print(f"[RMLoraApply] LoRA not found: {lora_info['name']}")
continue
combined_stack.append({
"name": lora_file,
"filename": os.path.basename(lora_file),
"strength": round(lora_info['weight'], 3),
"source": "string",
"random": False,
})
if not combined_stack:
return (model, clip)
# Apply all loras
print(f"\n{'='*60}")
print(f"[RMLoraApply] Applying {len(combined_stack)} LoRA(s)...")
for lora_config in combined_stack:
lora_name = lora_config["name"]
strength = lora_config["strength"]
if strength == 0:
continue
lora_path = folder_paths.get_full_path("loras", lora_name)
if lora_path is None:
print(f" LoRA: {lora_config.get('filename', lora_name)} - NOT FOUND")
continue
clip_strength = strength if clip is not None else 0
model, clip = LoraLoader().load_lora(model, clip, lora_name, strength, clip_strength)
# Log the applied lora
weight_str = f"{strength:.3f}"
if lora_config.get("random"):
min_w = lora_config.get("strength_min", 0)
max_w = lora_config.get("strength_max", 1)
weight_str = f"{strength:.3f} (random {min_w:.2f}-{max_w:.2f})"
source_str = f" [Civitai:{lora_config['air']}]" if lora_config.get("air") else ""
print(f" LoRA: {lora_config['filename']} @ {weight_str}{source_str}")
print(f"{'='*60}\n")
return (model, clip)
class FlexibleOptionalInputTypeNoPipe(dict):
"""
Like FlexibleOptionalInputType but without the Style Import field.
Used for Pipe variants that get loras from the pipe instead.
This must be used as the ENTIRE optional dict, not spread into another dict.
"""
def __init__(self, any_type_str="*", pipe_type="RM_STYLES_PIPE"):
self.any_type_str = any_type_str
# Include the pipe-specific fields in the base dict
super().__init__({
"lora_source": (["Image Loras", "Motion Loras High", "Motion Loras Low"], {"default": "Image Loras"}),
"Styles Pipe": (pipe_type,),
})
def __getitem__(self, key):
if key in dict.keys(self):
return dict.__getitem__(self, key)
return (self.any_type_str,)
def __contains__(self, key):
if key == "style_import" or key == "Style Import":
return False
return True
class RMPowerLoraLoaderPipe(RMPowerLoraLoader):
"""
Power LoRA Loader variant that accepts a Style Pipe input.
Allows selecting which lora type to use from the pipe.
"""
NAME = "RM Power Lora Loader (Pipe)"
CATEGORY = "RMAutomation/LoRA"
@classmethod
def INPUT_TYPES(cls):
return {
"required": {},
"optional": FlexibleOptionalInputTypeNoPipe("*", "RM_STYLES_PIPE"),
"hidden": {},
}
def build_lora_stack(self, lora_source="Image Loras", **kwargs):
"""Build a lora stack from style pipe and UI inputs."""
# Extract the selected lora string from the pipe
# RM_STYLES_PIPE order: Positive, Negative, Motion, Image Loras, Motion High, Motion Low, Number, Name
styles_pipe = kwargs.pop("Styles Pipe", None)
style_import = None
if styles_pipe:
pipe_data = styles_pipe[0] if isinstance(styles_pipe, tuple) and len(styles_pipe) == 1 else styles_pipe
if isinstance(pipe_data, tuple):
if lora_source == "Image Loras" and len(pipe_data) > 3:
style_import = pipe_data[3]
elif lora_source == "Motion Loras High" and len(pipe_data) > 4:
style_import = pipe_data[4]
elif lora_source == "Motion Loras Low" and len(pipe_data) > 5:
style_import = pipe_data[5]
# Add to kwargs so parent class processes it
if style_import:
kwargs["Style Import"] = style_import
return super().build_lora_stack(**kwargs)
NODE_CLASS_MAPPINGS = {
"RMPowerLoraLoader": RMPowerLoraLoader,
"RMPowerLoraLoaderPipe": RMPowerLoraLoaderPipe,
"RMLoraCollector": RMLoraCollector,
"RMLoraApply": RMLoraApply,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMPowerLoraLoader": "RM Power LoRA Loader",
"RMPowerLoraLoaderPipe": "RM Power LoRA Loader (Pipe)",
"RMLoraCollector": "RM LoRA Collector",
"RMLoraApply": "RM LoRA Apply",
}
+89
View File
@@ -0,0 +1,89 @@
"""
RM Save Image With Metadata - Save images with workflow metadata using RM_STYLES_PIPE.
"""
import os
from PIL import Image
import numpy as np
import json
from PIL.PngImagePlugin import PngInfo
from datetime import datetime
class RMSaveImageWithMetadata:
"""
Save images with metadata using RM_STYLES_PIPE.
RM_STYLES_PIPE order: Positive, Negative, Motion, Image Loras, Motion High, Motion Low, Prompt Number, Prompt Name
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE",),
"styles_pipe": ("RM_STYLES_PIPE",),
"destination": (["Sketchbook", "Playground"],),
"plot": (["yes", "no", "both"],),
},
"hidden": {
"prompt": "PROMPT",
"extra_pnginfo": "EXTRA_PNGINFO",
},
}
RETURN_TYPES = ()
FUNCTION = "save_image"
OUTPUT_NODE = True
CATEGORY = "RMAutomation/Image"
def save_image(self, images, styles_pipe, destination, plot, prompt=None, extra_pnginfo=None):
if images is None or images.shape[0] == 0:
print("[RMSaveImage] No images provided")
return ()
# RM_STYLES_PIPE: prompt_number is at index 6
prompt_number = styles_pipe[6]
base_dirs = {
"Sketchbook": r"C:\Users\rober\OneDrive\Documents\Sketchbook",
"Playground": r"C:\Users\rober\Playground"
}
base_dir = base_dirs[destination]
save_dirs = []
if plot in ["yes", "both"]:
save_dirs.append(os.path.join(base_dir, "Contact Sheets"))
if plot in ["no", "both"]:
save_dirs.append(os.path.join(base_dir, "Images", str(prompt_number)))
for save_dir in save_dirs:
os.makedirs(save_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
for i in range(images.shape[0]):
filename = f"{prompt_number}_{timestamp}_{i:03d}.png" if images.shape[0] > 1 else f"{prompt_number}_{timestamp}.png"
img = Image.fromarray(np.clip(255. * images[i].cpu().numpy(), 0, 255).astype(np.uint8))
metadata = PngInfo()
if prompt is not None:
metadata.add_text("prompt", json.dumps(prompt))
if extra_pnginfo is not None:
for k, v in extra_pnginfo.items():
metadata.add_text(k, json.dumps(v))
for save_dir in save_dirs:
filepath = os.path.join(save_dir, filename)
img.save(filepath, pnginfo=metadata, optimize=True)
print(f"[RMSaveImage] Saved: {filepath}")
return ()
NODE_CLASS_MAPPINGS = {
"RMSaveImageWithMetadata": RMSaveImageWithMetadata,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMSaveImageWithMetadata": "RM Save Image With Metadata",
}
+435
View File
@@ -0,0 +1,435 @@
"""
RM Styles Full - Load prompts from JSON styles file with random/increment/decrement support.
This node loads prompt data from a JSON file and supports multiple selection modes:
- Manual: Use the exact next_prompt number specified
- Random: JS calculates random number within min/max range
- Increment: JS increments after each execution (wraps at max)
- Decrement: JS decrements after each execution (wraps at min)
The JSON file can be edited via a modal editor accessible from a button on the node.
"""
import json
import os
from pathlib import Path
from server import PromptServer
from aiohttp import web
class RMStylesFull:
"""
A ComfyUI node that loads style/prompt information from a JSON file.
JSON Format Expected:
{
"styles": [
{
"number": 1,
"name": "Style Name",
"positive": "positive prompt text",
"negative": "negative prompt text",
"motion": "motion prompt for video",
"imageLoras": [{"path": "lora.safetensors", "weight": 1.0}],
"motionLoras": [],
"motionLorasHigh": [{"path": "motion_lora.safetensors", "weight": 1.0}],
"motionLorasLow": [{"path": "motion_lora.safetensors", "weight": 1.0}]
},
...
]
}
"""
# Quality prefixes (configurable)
POSITIVE_PREFIX = ""
NEGATIVE_PREFIX = ""
# Get the data directory relative to this file
DATA_DIR = Path(__file__).parent.parent / "data"
STYLES_FILE = DATA_DIR / "styles.json"
def __init__(self):
self._styles_cache = None
self._styles_mtime = 0
def _load_styles(self) -> dict:
"""Load styles from JSON file with caching."""
try:
if not self.STYLES_FILE.exists():
print(f"[RMStylesFull] Styles file not found: {self.STYLES_FILE}")
return {}
current_mtime = self.STYLES_FILE.stat().st_mtime
if self._styles_cache is None or current_mtime > self._styles_mtime:
with open(self.STYLES_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
self._styles_cache = {}
for style in data.get('styles', []):
num = style.get('number')
if num and num > 0:
self._styles_cache[num] = style
self._styles_mtime = current_mtime
print(f"[RMStylesFull] Loaded {len(self._styles_cache)} styles from JSON")
return self._styles_cache
except Exception as e:
print(f"[RMStylesFull] Error loading styles: {e}")
return {}
@classmethod
def INPUT_TYPES(cls):
"""Define the node's input widgets."""
return {
"required": {
"mode": (["Manual", "Random", "Increment", "Decrement"],),
"previous_prompt": ("INT", {
"default": 1,
"min": 1,
"max": 9999,
"tooltip": "The prompt number that was used in the LAST generation (read-only, updated by JS)"
}),
"next_prompt": ("INT", {
"default": 1,
"min": 1,
"max": 9999,
"tooltip": "The prompt number that WILL be used in the NEXT generation"
}),
"minimum": ("INT", {
"default": 1,
"min": 1,
"max": 9999,
"tooltip": "Minimum prompt number for Random/Increment/Decrement modes"
}),
"maximum": ("INT", {
"default": 100,
"min": 1,
"max": 9999,
"tooltip": "Maximum prompt number for Random/Increment/Decrement modes"
}),
"use_prefix": ("BOOLEAN", {
"default": False,
"tooltip": "Add quality prefixes to positive/negative prompts"
}),
}
}
CATEGORY = "RMAutomation/Styles"
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING", "STRING", "STRING", "INT", "STRING")
RETURN_NAMES = ("Positive", "Negative", "Motion Prompt", "Image Loras", "Motion Loras High", "Motion Loras Low", "Prompt Number", "Prompt Name")
FUNCTION = "load_style"
@classmethod
def IS_CHANGED(cls, **kwargs):
"""Always return NaN to force re-execution."""
return float("NaN")
def _format_loras(self, lora_list: list) -> str:
"""Convert a list of lora entries to ComfyUI lora string format."""
if not lora_list:
return ""
formatted = []
for lora in lora_list:
if isinstance(lora, dict):
path = lora.get('path', '')
weight = lora.get('weight', 1.0)
else:
path = lora
weight = 1.0
if path:
formatted.append(f"<lora:{path}:{weight}>")
return ", ".join(formatted)
def load_style(self, mode: str, previous_prompt: int, next_prompt: int, minimum: int, maximum: int, use_prefix: bool = False):
"""Main node execution function."""
empty_result = ("", "", "", "", "", "", 0, "")
styles = self._load_styles()
if not styles:
return empty_result
selected_number = next_prompt
if selected_number not in styles:
available = sorted(styles.keys())
if available:
selected_number = min(available, key=lambda x: abs(x - selected_number))
print(f"[RMStylesFull] Prompt #{next_prompt} not found, using closest: #{selected_number}")
else:
return empty_result
style = styles[selected_number]
name = style.get('name', '')
positive = style.get('positive', '')
negative = style.get('negative', '')
# Log prompt info
print(f"\n============================================================")
print(f"[RMStylesFull] Prompt #{selected_number}: {name or '(unnamed)'}")
print(f" Mode: {mode} | Range: {minimum}-{maximum}")
print(f" Positive: {positive[:100]}{'...' if len(positive) > 100 else ''}")
print(f"============================================================")
motion = style.get('motion', '')
image_loras = style.get('imageLoras', [])
motion_loras_high = style.get('motionLorasHigh', [])
motion_loras_low = style.get('motionLorasLow', [])
# Apply prefixes if enabled
if use_prefix and self.POSITIVE_PREFIX:
positive = f"{self.POSITIVE_PREFIX} {positive}" if positive else self.POSITIVE_PREFIX
if use_prefix and self.NEGATIVE_PREFIX:
negative = f"{self.NEGATIVE_PREFIX}, {negative}" if negative else self.NEGATIVE_PREFIX
image_loras_str = self._format_loras(image_loras)
motion_loras_high_str = self._format_loras(motion_loras_high)
motion_loras_low_str = self._format_loras(motion_loras_low)
return (
positive,
negative,
motion,
image_loras_str,
motion_loras_high_str,
motion_loras_low_str,
selected_number,
name
)
class RMStylesFullDisplay(RMStylesFull):
"""Same as RMStylesFull but displays the positive prompt text on the node."""
OUTPUT_NODE = True
@classmethod
def INPUT_TYPES(cls):
"""Define the node's input widgets."""
return {
"required": {
"mode": (["Manual", "Random", "Increment", "Decrement"],),
"previous_prompt": ("INT", {
"default": 1,
"min": 1,
"max": 9999,
}),
"next_prompt": ("INT", {
"default": 1,
"min": 1,
"max": 9999,
}),
"minimum": ("INT", {
"default": 1,
"min": 1,
"max": 9999,
}),
"maximum": ("INT", {
"default": 100,
"min": 1,
"max": 9999,
}),
"use_prefix": ("BOOLEAN", {
"default": False,
}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
}
}
def load_style(self, mode: str, previous_prompt: int, next_prompt: int, minimum: int, maximum: int, use_prefix: bool = False, unique_id=None):
"""Same as parent but also returns UI data to display on the node."""
result = super().load_style(mode, previous_prompt, next_prompt, minimum, maximum, use_prefix)
positive = result[0] if result[0] else "(empty)"
prompt_num = result[6]
name = result[7]
display_lines = []
if name:
display_lines.append(f"#{prompt_num}: {name}")
else:
display_lines.append(f"#{prompt_num}")
display_lines.append("")
display_lines.append(positive[:500] + "..." if len(positive) > 500 else positive)
display_text = "\n".join(display_lines)
return {"ui": {"text": [display_text]}, "result": result}
# API routes for styles editor
def setup_styles_api():
"""Register API routes for the styles editor modal."""
try:
data_dir = Path(__file__).parent.parent / "data"
styles_file = data_dir / "styles.json"
@PromptServer.instance.routes.get("/api/rmautomation/styles")
async def get_styles(request):
try:
if not styles_file.exists():
return web.json_response({"styles": []})
with open(styles_file, 'r', encoding='utf-8') as f:
data = json.load(f)
return web.json_response(data)
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@PromptServer.instance.routes.post("/api/rmautomation/styles")
async def save_styles(request):
try:
data = await request.json()
os.makedirs(data_dir, exist_ok=True)
with open(styles_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
return web.json_response({"success": True})
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@PromptServer.instance.routes.get("/api/rmautomation/styles/{number}")
async def get_style(request):
try:
number = int(request.match_info['number'])
if not styles_file.exists():
return web.json_response({"error": "Styles file not found"}, status=404)
with open(styles_file, 'r', encoding='utf-8') as f:
data = json.load(f)
for style in data.get('styles', []):
if style.get('number') == number:
return web.json_response(style)
return web.json_response({"error": "Style not found"}, status=404)
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@PromptServer.instance.routes.put("/api/rmautomation/styles/{number}")
async def update_style(request):
try:
number = int(request.match_info['number'])
new_style = await request.json()
new_style['number'] = number
if not styles_file.exists():
data = {"styles": []}
else:
with open(styles_file, 'r', encoding='utf-8') as f:
data = json.load(f)
styles = data.get('styles', [])
found = False
for i, style in enumerate(styles):
if style.get('number') == number:
styles[i] = new_style
found = True
break
if not found:
styles.append(new_style)
styles.sort(key=lambda x: x.get('number', 0))
data['styles'] = styles
os.makedirs(data_dir, exist_ok=True)
with open(styles_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
return web.json_response({"success": True})
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@PromptServer.instance.routes.delete("/api/rmautomation/styles/{number}")
async def delete_style(request):
try:
number = int(request.match_info['number'])
if not styles_file.exists():
return web.json_response({"error": "Styles file not found"}, status=404)
with open(styles_file, 'r', encoding='utf-8') as f:
data = json.load(f)
styles = [s for s in data.get('styles', []) if s.get('number') != number]
data['styles'] = styles
with open(styles_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
return web.json_response({"success": True})
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@PromptServer.instance.routes.post("/api/rmautomation/styles/import")
async def import_styles(request):
"""Import styles from uploaded JSON with replace or append mode."""
try:
req_data = await request.json()
imported_styles = req_data.get('styles', [])
mode = req_data.get('mode', 'append')
if not imported_styles:
return web.json_response({"error": "No styles provided"}, status=400)
os.makedirs(data_dir, exist_ok=True)
if mode == 'replace':
# Replace all existing styles with imported ones
# Ensure proper numbering
for i, style in enumerate(imported_styles):
if 'number' not in style:
style['number'] = i + 1
imported_styles.sort(key=lambda x: x.get('number', 0))
data = {"styles": imported_styles}
else:
# Append mode: add imported styles after existing ones
if styles_file.exists():
with open(styles_file, 'r', encoding='utf-8') as f:
data = json.load(f)
else:
data = {"styles": []}
existing_styles = data.get('styles', [])
# Find the highest existing number
max_number = 0
for style in existing_styles:
num = style.get('number', 0)
if num > max_number:
max_number = num
# Renumber imported styles to start after the highest existing number
for i, style in enumerate(imported_styles):
style['number'] = max_number + i + 1
existing_styles.extend(imported_styles)
existing_styles.sort(key=lambda x: x.get('number', 0))
data['styles'] = existing_styles
with open(styles_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
return web.json_response({"success": True, "count": len(imported_styles)})
except Exception as e:
print(f"[RMAutomation] Import error: {e}")
return web.json_response({"error": str(e)}, status=500)
print("[RMAutomation] Styles API routes registered")
except Exception as e:
print(f"[RMAutomation] Could not register styles API routes: {e}")
# Initialize API routes
setup_styles_api()
NODE_CLASS_MAPPINGS = {
"RMStylesFull": RMStylesFull,
"RMStylesFullDisplay": RMStylesFullDisplay,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMStylesFull": "RM Styles Full",
"RMStylesFullDisplay": "RM Styles Full (Display)",
}
+104
View File
@@ -0,0 +1,104 @@
"""
RM Styles Pipe - Style pipe input/output nodes.
"""
from .styles_full import RMStylesFull
class RMStylesPipe(RMStylesFull):
"""
Same as RMStylesFull but outputs a pipe tuple for easy data passing.
The pipe contains all style data in a single output (without PDXL prefixes).
"""
RETURN_TYPES = ("RM_STYLES_PIPE",)
RETURN_NAMES = ("Styles Pipe",)
OUTPUT_NODE = True # Required for onExecuted JS callback to fire
@classmethod
def INPUT_TYPES(cls):
"""Override to remove use_prefix - that's handled at the PipeOut node."""
return {
"required": {
"mode": (["Manual", "Random", "Increment", "Decrement"],),
"previous_prompt": ("INT", {"default": 1, "min": 1, "max": 9999}),
"next_prompt": ("INT", {"default": 1, "min": 1, "max": 9999}),
"minimum": ("INT", {"default": 1, "min": 1, "max": 9999}),
"maximum": ("INT", {"default": 100, "min": 1, "max": 9999}),
}
}
def load_style(self, mode: str, previous_prompt: int, next_prompt: int, minimum: int, maximum: int):
"""Load style and return as a pipe tuple (without prefixes)."""
result = super().load_style(mode, previous_prompt, next_prompt, minimum, maximum, use_prefix=False)
return (result,)
class RMStylesPipeOut:
"""
Unpack an RM_STYLES_PIPE into individual outputs.
Optionally apply PDXL (Pony) quality tags to positive/negative prompts.
"""
# PDXL quality prefixes (Pony Diffusion XL style tags)
PDXL_POSITIVE_PREFIX = (
"score_9, score_8_up, score_7_up, score_6_up, score_5_up, score_4_up"
)
PDXL_NEGATIVE_PREFIX = (
"score_6, score_5, score_4, score_3, score_2, score_1, "
"source_pony, source_furry, low quality, bad quality, ugly, worst quality"
)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"styles_pipe": ("RM_STYLES_PIPE",),
},
"optional": {
"Toggle Pony Tags": ("BOOLEAN", {
"default": False,
"tooltip": "Add PDXL (Pony) quality tags to positive/negative prompts"
}),
}
}
CATEGORY = "RMAutomation/Styles"
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING", "STRING", "STRING", "INT", "STRING")
RETURN_NAMES = ("Positive", "Negative", "Motion Prompt", "Image Loras", "Motion Loras High", "Motion Loras Low", "Prompt Number", "Prompt Name")
FUNCTION = "unpack"
def unpack(self, styles_pipe, **kwargs):
"""Unpack the styles pipe tuple into individual outputs."""
toggle_pony_tags = kwargs.get("Toggle Pony Tags", False)
if styles_pipe is None:
return ("", "", "", "", "", "", 0, "")
# Unpack the tuple
positive, negative, motion, image_loras, motion_high, motion_low, prompt_num, prompt_name = styles_pipe
# Apply PDXL prefixes if requested
if toggle_pony_tags:
if positive:
positive = f"{self.PDXL_POSITIVE_PREFIX}, {positive}"
else:
positive = self.PDXL_POSITIVE_PREFIX
if negative:
negative = f"{self.PDXL_NEGATIVE_PREFIX}, {negative}"
else:
negative = self.PDXL_NEGATIVE_PREFIX
return (positive, negative, motion, image_loras, motion_high, motion_low, prompt_num, prompt_name)
NODE_CLASS_MAPPINGS = {
"RMStylesPipe": RMStylesPipe,
"RMStylesPipeOut": RMStylesPipeOut,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMStylesPipe": "RM Styles Pipe",
"RMStylesPipeOut": "RM Styles Pipe Out",
}
+103
View File
@@ -0,0 +1,103 @@
"""
RM Text Embed Nodes
Simple positive and negative text embedding nodes with string input combining.
"""
from typing import Tuple, Any, Optional
class RMPositiveTextEmbed:
"""
Combines an optional string input with a text field and encodes to positive conditioning.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"clip": ("CLIP",),
"text": ("STRING", {"default": "", "multiline": True, "dynamicPrompts": True}),
},
"optional": {
"string_input": ("STRING", {"forceInput": True}),
},
}
RETURN_TYPES = ("CONDITIONING",)
RETURN_NAMES = ("Positive",)
FUNCTION = "encode"
CATEGORY = "RMAutomation/Text"
def encode(
self,
clip,
text: str,
string_input: Optional[str] = None,
) -> Tuple[Any]:
if clip is None:
raise RuntimeError("ERROR: clip input is invalid: None")
parts = []
if string_input and string_input.strip():
parts.append(string_input.strip())
if text and text.strip():
parts.append(text.strip())
combined_text = ", ".join(parts) if parts else ""
tokens = clip.tokenize(combined_text)
return (clip.encode_from_tokens_scheduled(tokens),)
class RMNegativeTextEmbed:
"""
Combines an optional string input with a text field and encodes to negative conditioning.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"clip": ("CLIP",),
"text": ("STRING", {"default": "", "multiline": True, "dynamicPrompts": True}),
},
"optional": {
"string_input": ("STRING", {"forceInput": True}),
},
}
RETURN_TYPES = ("CONDITIONING",)
RETURN_NAMES = ("Negative",)
FUNCTION = "encode"
CATEGORY = "RMAutomation/Text"
def encode(
self,
clip,
text: str,
string_input: Optional[str] = None,
) -> Tuple[Any]:
if clip is None:
raise RuntimeError("ERROR: clip input is invalid: None")
parts = []
if string_input and string_input.strip():
parts.append(string_input.strip())
if text and text.strip():
parts.append(text.strip())
combined_text = ", ".join(parts) if parts else ""
tokens = clip.tokenize(combined_text)
return (clip.encode_from_tokens_scheduled(tokens),)
NODE_CLASS_MAPPINGS = {
"RMPositiveTextEmbed": RMPositiveTextEmbed,
"RMNegativeTextEmbed": RMNegativeTextEmbed,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMPositiveTextEmbed": "RM Positive Text Embed",
"RMNegativeTextEmbed": "RM Negative Text Embed",
}
+714
View File
@@ -0,0 +1,714 @@
"""
RM Video Combine - Video output node with metadata embedding.
Based on comfyui-videohelpersuite by Kosinkadink
"""
import os
import sys
import json
import subprocess
import numpy as np
import datetime
from typing import List
import torch
from PIL import Image, ExifTags
from PIL.PngImagePlugin import PngInfo
from pathlib import Path
import itertools
import shutil
import time
import copy
import logging
import folder_paths
from comfy.utils import ProgressBar
# Logger setup
class ColoredFormatter(logging.Formatter):
COLORS = {
"DEBUG": "\033[0;36m",
"INFO": "\033[0;32m",
"WARNING": "\033[0;33m",
"ERROR": "\033[0;31m",
"CRITICAL": "\033[0;37;41m",
"RESET": "\033[0m",
}
def format(self, record):
colored_record = copy.copy(record)
levelname = colored_record.levelname
seq = self.COLORS.get(levelname, self.COLORS["RESET"])
colored_record.levelname = f"{seq}{levelname}{self.COLORS['RESET']}"
return super().format(colored_record)
logger = logging.getLogger("RMVideoCombine")
logger.propagate = False
if not logger.handlers:
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(ColoredFormatter("[%(name)s] - %(levelname)s - %(message)s"))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# Utils
BIGMIN = -(2**53-1)
BIGMAX = (2**53-1)
DIMMAX = 8192
ENCODE_ARGS = ("utf-8", 'backslashreplace')
def ffmpeg_suitability(path):
try:
version = subprocess.run([path, "-version"], check=True,
capture_output=True).stdout.decode(*ENCODE_ARGS)
except:
return 0
score = 0
simple_criterion = [("libvpx", 20),("264",10), ("265",3),
("svtav1",5),("libopus", 1)]
for criterion in simple_criterion:
if version.find(criterion[0]) >= 0:
score += criterion[1]
copyright_index = version.find('2000-2')
if copyright_index >= 0:
copyright_year = version[copyright_index+6:copyright_index+9]
if copyright_year.isnumeric():
score += int(copyright_year)
return score
class MultiInput(str):
def __new__(cls, string, allowed_types="*"):
res = super().__new__(cls, string)
res.allowed_types=allowed_types
return res
def __ne__(self, other):
if self.allowed_types == "*" or other == "*":
return False
return other not in self.allowed_types
imageOrLatent = MultiInput("IMAGE", ["IMAGE", "LATENT"])
floatOrInt = MultiInput("FLOAT", ["FLOAT", "INT"])
class ContainsAll(dict):
def __contains__(self, other):
return True
def __getitem__(self, key):
return super().get(key, (None, {}))
# Find ffmpeg
if "VHS_FORCE_FFMPEG_PATH" in os.environ:
ffmpeg_path = os.environ.get("VHS_FORCE_FFMPEG_PATH")
else:
ffmpeg_paths = []
try:
from imageio_ffmpeg import get_ffmpeg_exe
imageio_ffmpeg_path = get_ffmpeg_exe()
ffmpeg_paths.append(imageio_ffmpeg_path)
except:
if "VHS_USE_IMAGEIO_FFMPEG" in os.environ:
raise
logger.warn("Failed to import imageio_ffmpeg")
if "VHS_USE_IMAGEIO_FFMPEG" in os.environ:
ffmpeg_path = imageio_ffmpeg_path
else:
system_ffmpeg = shutil.which("ffmpeg")
if system_ffmpeg is not None:
ffmpeg_paths.append(system_ffmpeg)
if os.path.isfile("ffmpeg"):
ffmpeg_paths.append(os.path.abspath("ffmpeg"))
if os.path.isfile("ffmpeg.exe"):
ffmpeg_paths.append(os.path.abspath("ffmpeg.exe"))
if len(ffmpeg_paths) == 0:
logger.error("No valid ffmpeg found.")
ffmpeg_path = None
elif len(ffmpeg_paths) == 1:
ffmpeg_path = ffmpeg_paths[0]
else:
ffmpeg_path = max(ffmpeg_paths, key=ffmpeg_suitability)
gifski_path = os.environ.get("VHS_GIFSKI", None)
if gifski_path is None:
gifski_path = os.environ.get("JOV_GIFSKI", None)
if gifski_path is None:
gifski_path = shutil.which("gifski")
def cached(duration):
def dec(f):
cached_ret = None
cache_time = 0
def cached_func():
nonlocal cache_time, cached_ret
if time.time() > cache_time + duration or cached_ret is None:
cache_time = time.time()
cached_ret = f()
return cached_ret
return cached_func
return dec
def merge_filter_args(args, ftype="-vf"):
try:
start_index = args.index(ftype)+1
index = start_index
while True:
index = args.index(ftype, index)
args[start_index] += ',' + args[index+1]
args.pop(index)
args.pop(index)
except ValueError:
pass
# Video format handling
if 'RM_video_formats' not in folder_paths.folder_names_and_paths:
folder_paths.folder_names_and_paths["RM_video_formats"] = ((),{".json"})
if len(folder_paths.folder_names_and_paths['RM_video_formats'][1]) == 0:
folder_paths.folder_names_and_paths["RM_video_formats"][1].add(".json")
base_formats_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "video_formats")
def flatten_list(l):
ret = []
for e in l:
if isinstance(e, list):
ret.extend(e)
else:
ret.append(e)
return ret
def iterate_format(video_format, for_widgets=True):
def indirector(cont, index):
if isinstance(cont[index], list) and (not for_widgets
or len(cont[index])> 1 and not isinstance(cont[index][1], dict)):
inp = yield cont[index]
if inp is not None:
cont[index] = inp
yield
for k in video_format:
if k == "extra_widgets":
if for_widgets:
yield from video_format["extra_widgets"]
elif k.endswith("_pass"):
for i in range(len(video_format[k])):
yield from indirector(video_format[k], i)
if not for_widgets:
video_format[k] = flatten_list(video_format[k])
else:
yield from indirector(video_format, k)
@cached(5)
def get_video_formats():
format_files = {}
for format_name in folder_paths.get_filename_list("RM_video_formats"):
format_files[format_name] = folder_paths.get_full_path("RM_video_formats", format_name)
if os.path.exists(base_formats_dir):
for item in os.scandir(base_formats_dir):
if not item.is_file() or not item.name.endswith('.json'):
continue
format_files[item.name[:-5]] = item.path
formats = []
format_widgets = {}
for format_name, path in format_files.items():
with open(path, 'r') as stream:
video_format = json.load(stream)
if "gifski_pass" in video_format and gifski_path is None:
continue
widgets = list(iterate_format(video_format))
formats.append("video/" + format_name)
if (len(widgets) > 0):
format_widgets["video/"+ format_name] = widgets
return formats, format_widgets
def apply_format_widgets(format_name, kwargs):
if os.path.exists(os.path.join(base_formats_dir, format_name + ".json")):
video_format_path = os.path.join(base_formats_dir, format_name + ".json")
else:
video_format_path = folder_paths.get_full_path("RM_video_formats", format_name)
with open(video_format_path, 'r') as stream:
video_format = json.load(stream)
from string import Template
for w in iterate_format(video_format):
if w[0] not in kwargs:
if len(w) > 2 and 'default' in w[2]:
default = w[2]['default']
else:
if type(w[1]) is list:
default = w[1][0]
else:
default = {"BOOLEAN": False, "INT": 0, "FLOAT": 0, "STRING": ""}[w[1]]
kwargs[w[0]] = default
logger.warn(f"Missing input for {w[0][0]} has been set to {default}")
wit = iterate_format(video_format, False)
for w in wit:
while isinstance(w, list):
if len(w) == 1:
w = [Template(x).substitute(**kwargs) for x in w[0]]
break
elif isinstance(w[1], dict):
w = w[1][str(kwargs[w[0]])]
elif len(w) > 3:
w = Template(w[3]).substitute(val=kwargs[w[0]])
else:
w = str(kwargs[w[0]])
wit.send(w)
return video_format
# Tensor conversion
def tensor_to_int(tensor, bits):
tensor = tensor.cpu().numpy() * (2**bits-1) + 0.5
return np.clip(tensor, 0, (2**bits-1))
def tensor_to_shorts(tensor):
return tensor_to_int(tensor, 16).astype(np.uint16)
def tensor_to_bytes(tensor):
return tensor_to_int(tensor, 8).astype(np.uint8)
# FFmpeg processes
def ffmpeg_process(args, video_format, video_metadata, file_path, env):
res = None
frame_data = yield
total_frames_output = 0
if video_format.get('save_metadata', 'False') != 'False':
os.makedirs(folder_paths.get_temp_directory(), exist_ok=True)
metadata = json.dumps(video_metadata)
metadata_path = os.path.join(folder_paths.get_temp_directory(), "metadata.txt")
metadata = metadata.replace("\\","\\\\")
metadata = metadata.replace(";","\\;")
metadata = metadata.replace("#","\\#")
metadata = metadata.replace("=","\\=")
metadata = metadata.replace("\n","\\\n")
metadata = "comment=" + metadata
with open(metadata_path, "w") as f:
f.write(";FFMETADATA1\n")
f.write(metadata)
m_args = args[:1] + ["-i", metadata_path] + args[1:] + ["-metadata", "creation_time=now"]
with subprocess.Popen(m_args + [file_path], stderr=subprocess.PIPE,
stdin=subprocess.PIPE, env=env) as proc:
try:
while frame_data is not None:
proc.stdin.write(frame_data)
frame_data = yield
total_frames_output+=1
proc.stdin.flush()
proc.stdin.close()
res = proc.stderr.read()
except BrokenPipeError as e:
err = proc.stderr.read()
if os.path.exists(file_path):
raise Exception("An error occurred in the ffmpeg subprocess:\n" \
+ err.decode(*ENCODE_ARGS))
print(err.decode(*ENCODE_ARGS), end="", file=sys.stderr)
logger.warn("An error occurred when saving with metadata")
if res != b'':
with subprocess.Popen(args + [file_path], stderr=subprocess.PIPE,
stdin=subprocess.PIPE, env=env) as proc:
try:
while frame_data is not None:
proc.stdin.write(frame_data)
frame_data = yield
total_frames_output+=1
proc.stdin.flush()
proc.stdin.close()
res = proc.stderr.read()
except BrokenPipeError as e:
res = proc.stderr.read()
raise Exception("An error occurred in the ffmpeg subprocess:\n" \
+ res.decode(*ENCODE_ARGS))
yield total_frames_output
if len(res) > 0:
print(res.decode(*ENCODE_ARGS), end="", file=sys.stderr)
def gifski_process(args, dimensions, video_format, file_path, env):
frame_data = yield
with subprocess.Popen(args + video_format['main_pass'] + ['-f', 'yuv4mpegpipe', '-'],
stderr=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, env=env) as procff:
with subprocess.Popen([gifski_path] + video_format['gifski_pass']
+ ['-W', f'{dimensions[0]}', '-H', f'{dimensions[1]}']
+ ['-q', '-o', file_path, '-'], stderr=subprocess.PIPE,
stdin=procff.stdout, stdout=subprocess.PIPE,
env=env) as procgs:
try:
while frame_data is not None:
procff.stdin.write(frame_data)
frame_data = yield
procff.stdin.flush()
procff.stdin.close()
resff = procff.stderr.read()
resgs = procgs.stderr.read()
outgs = procgs.stdout.read()
except BrokenPipeError as e:
procff.stdin.close()
resff = procff.stderr.read()
resgs = procgs.stderr.read()
raise Exception("An error occurred while creating gifski output\n" \
+ "Make sure you are using gifski --version >=1.32.0\nffmpeg: " \
+ resff.decode(*ENCODE_ARGS) + '\ngifski: ' + resgs.decode(*ENCODE_ARGS))
if len(resff) > 0:
print(resff.decode(*ENCODE_ARGS), end="", file=sys.stderr)
if len(resgs) > 0:
print(resgs.decode(*ENCODE_ARGS), end="", file=sys.stderr)
if len(outgs) > 0:
print(outgs.decode(*ENCODE_ARGS))
def to_pingpong(inp):
if not hasattr(inp, "__getitem__"):
inp = list(inp)
yield from inp
for i in range(len(inp)-2,0,-1):
yield inp[i]
class RMVideoCombine:
"""
RM Video Combine - Combines images into video with full format support and metadata embedding.
"""
@classmethod
def INPUT_TYPES(s):
ffmpeg_formats, format_widgets = get_video_formats()
format_widgets["image/webp"] = [['lossless', "BOOLEAN", {'default': True}]]
return {
"required": {
"images": (imageOrLatent,),
"filename_prefix": ("STRING", {"default": "video"}),
"frame_rate": (
floatOrInt,
{"default": 8, "min": 1, "step": 1},
),
"loop_count": ("INT", {"default": 0, "min": 0, "max": 100, "step": 1}),
"format": (["image/gif", "image/webp"] + ffmpeg_formats, {'formats': format_widgets}),
"pingpong": ("BOOLEAN", {"default": False}),
"save_output": ("BOOLEAN", {"default": True}),
},
"optional": {
"output_dir": ("STRING", {"default": "", "multiline": False}),
"audio": ("AUDIO",),
"vae": ("VAE",),
},
"hidden": ContainsAll({
"prompt": "PROMPT",
"extra_pnginfo": "EXTRA_PNGINFO",
"unique_id": "UNIQUE_ID"
}),
}
RETURN_TYPES = ("VHS_FILENAMES",)
RETURN_NAMES = ("Filenames",)
OUTPUT_NODE = True
CATEGORY = "RMAutomation/Video"
FUNCTION = "combine_video"
def combine_video(
self,
frame_rate: int,
loop_count: int,
filename_prefix: str = "video",
images=None,
latents=None,
format="image/gif",
pingpong=False,
save_output=True,
output_dir="",
prompt=None,
extra_pnginfo=None,
audio=None,
unique_id=None,
manual_format_widgets=None,
vae=None,
**kwargs
):
if latents is not None:
images = latents
if images is None:
return ((save_output, []),)
if vae is not None:
if isinstance(images, dict):
images = images['samples']
else:
vae = None
if isinstance(images, torch.Tensor) and images.size(0) == 0:
return ((save_output, []),)
num_frames = len(images)
pbar = ProgressBar(num_frames)
if vae is not None:
downscale_ratio = getattr(vae, "downscale_ratio", 8)
width = images.size(-1)*downscale_ratio
height = images.size(-2)*downscale_ratio
frames_per_batch = (1920 * 1080 * 16) // (width * height) or 1
def batched(it, n):
while batch := tuple(itertools.islice(it, n)):
yield batch
def batched_encode(images, vae, frames_per_batch):
for batch in batched(iter(images), frames_per_batch):
image_batch = torch.from_numpy(np.array(batch))
yield from vae.decode(image_batch)
images = batched_encode(images, vae, frames_per_batch)
first_image = next(images)
images = itertools.chain([first_image], images)
while len(first_image.shape) > 3:
first_image = first_image[0]
else:
first_image = images[0]
images = iter(images)
# Determine output directory
if save_output:
if output_dir and output_dir.strip():
full_output_folder = output_dir.strip()
else:
full_output_folder = folder_paths.get_output_directory()
os.makedirs(full_output_folder, exist_ok=True)
subfolder = ""
else:
full_output_folder = folder_paths.get_temp_directory()
subfolder = ""
# Generate filename with timestamp
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{filename_prefix}_{timestamp}"
output_files = []
metadata = PngInfo()
video_metadata = {}
if prompt is not None:
metadata.add_text("prompt", json.dumps(prompt))
video_metadata["prompt"] = json.dumps(prompt)
if extra_pnginfo is not None:
for x in extra_pnginfo:
metadata.add_text(x, json.dumps(extra_pnginfo[x]))
video_metadata[x] = extra_pnginfo[x]
extra_options = extra_pnginfo.get('workflow', {}).get('extra', {})
else:
extra_options = {}
metadata.add_text("CreationTime", datetime.datetime.now().isoformat(" ")[:19])
output_process = None
# save first frame as png to keep metadata
first_image_file = f"{filename}.png"
file_path = os.path.join(full_output_folder, first_image_file)
if extra_options.get('VHS_MetadataImage', True) != False:
Image.fromarray(tensor_to_bytes(first_image)).save(
file_path,
pnginfo=metadata,
compress_level=4,
)
output_files.append(file_path)
format_type, format_ext = format.split("/")
if format_type == "image":
image_kwargs = {}
if format_ext == "gif":
image_kwargs['disposal'] = 2
if format_ext == "webp":
exif = Image.Exif()
exif[ExifTags.IFD.Exif] = {36867: datetime.datetime.now().isoformat(" ")[:19]}
image_kwargs['exif'] = exif
image_kwargs['lossless'] = kwargs.get("lossless", True)
file = f"{filename}.{format_ext}"
file_path = os.path.join(full_output_folder, file)
if pingpong:
images = to_pingpong(images)
def frames_gen(images):
for i in images:
pbar.update(1)
yield Image.fromarray(tensor_to_bytes(i))
frames = frames_gen(images)
next(frames).save(
file_path,
format=format_ext.upper(),
save_all=True,
append_images=frames,
duration=round(1000 / frame_rate),
loop=loop_count,
compress_level=4,
**image_kwargs
)
output_files.append(file_path)
else:
# Use ffmpeg to save a video
if ffmpeg_path is None:
raise ProcessLookupError(f"ffmpeg is required for video outputs and could not be found.\nIn order to use video outputs, you must either:\n- Install imageio-ffmpeg with pip,\n- Place a ffmpeg executable in {os.path.abspath('')}, or\n- Install ffmpeg and add it to the system path.")
if manual_format_widgets is not None:
logger.warn("Format args can now be passed directly. The manual_format_widgets argument is now deprecated")
kwargs.update(manual_format_widgets)
has_alpha = first_image.shape[-1] == 4
kwargs["has_alpha"] = has_alpha
video_format = apply_format_widgets(format_ext, kwargs)
dim_alignment = video_format.get("dim_alignment", 2)
if (first_image.shape[1] % dim_alignment) or (first_image.shape[0] % dim_alignment):
to_pad = (-first_image.shape[1] % dim_alignment,
-first_image.shape[0] % dim_alignment)
padding = (to_pad[0]//2, to_pad[0] - to_pad[0]//2,
to_pad[1]//2, to_pad[1] - to_pad[1]//2)
padfunc = torch.nn.ReplicationPad2d(padding)
def pad(image):
image = image.permute((2,0,1))
padded = padfunc(image.to(dtype=torch.float32))
return padded.permute((1,2,0))
images = map(pad, images)
dimensions = (-first_image.shape[1] % dim_alignment + first_image.shape[1],
-first_image.shape[0] % dim_alignment + first_image.shape[0])
logger.warn("Output images were not of valid resolution and have had padding applied")
else:
dimensions = (first_image.shape[1], first_image.shape[0])
if pingpong:
images = to_pingpong(images)
if num_frames > 2:
num_frames += num_frames -2
pbar.total = num_frames
if loop_count > 0:
loop_args = ["-vf", "loop=loop=" + str(loop_count)+":size=" + str(num_frames)]
else:
loop_args = []
if video_format.get('input_color_depth', '8bit') == '16bit':
images = map(tensor_to_shorts, images)
if has_alpha:
i_pix_fmt = 'rgba64'
else:
i_pix_fmt = 'rgb48'
else:
images = map(tensor_to_bytes, images)
if has_alpha:
i_pix_fmt = 'rgba'
else:
i_pix_fmt = 'rgb24'
file = f"{filename}.{video_format['extension']}"
file_path = os.path.join(full_output_folder, file)
bitrate_arg = []
bitrate = video_format.get('bitrate')
if bitrate is not None:
bitrate_arg = ["-b:v", str(bitrate) + "M" if video_format.get('megabit') == 'True' else str(bitrate) + "K"]
args = [ffmpeg_path, "-v", "error", "-f", "rawvideo", "-pix_fmt", i_pix_fmt,
"-color_range", "pc", "-colorspace", "rgb", "-color_primaries", "bt709",
"-color_trc", video_format.get("fake_trc", "iec61966-2-1"),
"-s", f"{dimensions[0]}x{dimensions[1]}", "-r", str(frame_rate), "-i", "-"] \
+ loop_args
images = map(lambda x: x.tobytes(), images)
env=os.environ.copy()
if "environment" in video_format:
env.update(video_format["environment"])
if "pre_pass" in video_format:
images = [b''.join(images)]
os.makedirs(folder_paths.get_temp_directory(), exist_ok=True)
in_args_len = args.index("-i") + 2
pre_pass_args = args[:in_args_len] + video_format['pre_pass']
merge_filter_args(pre_pass_args)
try:
subprocess.run(pre_pass_args, input=images[0], env=env,
capture_output=True, check=True)
except subprocess.CalledProcessError as e:
raise Exception("An error occurred in the ffmpeg prepass:\n" \
+ e.stderr.decode(*ENCODE_ARGS))
if "inputs_main_pass" in video_format:
in_args_len = args.index("-i") + 2
args = args[:in_args_len] + video_format['inputs_main_pass'] + args[in_args_len:]
if output_process is None:
if 'gifski_pass' in video_format:
format = 'image/gif'
output_process = gifski_process(args, dimensions, video_format, file_path, env)
audio = None
else:
args += video_format['main_pass'] + bitrate_arg
merge_filter_args(args)
output_process = ffmpeg_process(args, video_format, video_metadata, file_path, env)
output_process.send(None)
for image in images:
pbar.update(1)
output_process.send(image)
try:
total_frames_output = output_process.send(None)
output_process.send(None)
except StopIteration:
pass
output_files.append(file_path)
a_waveform = None
if audio is not None:
try:
a_waveform = audio['waveform']
except:
pass
if a_waveform is not None:
output_file_with_audio = f"{filename}-audio.{video_format['extension']}"
output_file_with_audio_path = os.path.join(full_output_folder, output_file_with_audio)
if "audio_pass" not in video_format:
logger.warn("Selected video format does not have explicit audio support")
video_format["audio_pass"] = ["-c:a", "libopus"]
channels = audio['waveform'].size(1)
min_audio_dur = total_frames_output / frame_rate + 1
if video_format.get('trim_to_audio', 'False') != 'False':
apad = []
else:
apad = ["-af", "apad=whole_dur="+str(min_audio_dur)]
mux_args = [ffmpeg_path, "-v", "error", "-n", "-i", file_path,
"-ar", str(audio['sample_rate']), "-ac", str(channels),
"-f", "f32le", "-i", "-", "-c:v", "copy"] \
+ video_format["audio_pass"] \
+ apad + ["-shortest", output_file_with_audio_path]
audio_data = audio['waveform'].squeeze(0).transpose(0,1) \
.numpy().tobytes()
merge_filter_args(mux_args, '-af')
try:
res = subprocess.run(mux_args, input=audio_data,
env=env, capture_output=True, check=True)
except subprocess.CalledProcessError as e:
raise Exception("An error occured in the ffmpeg subprocess:\n" \
+ e.stderr.decode(*ENCODE_ARGS))
if res.stderr:
print(res.stderr.decode(*ENCODE_ARGS), end="", file=sys.stderr)
output_files.append(output_file_with_audio_path)
file = output_file_with_audio
if extra_options.get('VHS_KeepIntermediate', True) == False:
for intermediate in output_files[1:-1]:
if os.path.exists(intermediate):
os.remove(intermediate)
# Copy to temp for preview
preview_type = "temp"
preview_subfolder = ""
if save_output and output_files:
temp_dir = folder_paths.get_temp_directory()
final_file = output_files[-1]
temp_preview_path = os.path.join(temp_dir, os.path.basename(file))
try:
shutil.copy2(final_file, temp_preview_path)
logger.info(f"Copied video to temp for preview: {temp_preview_path}")
except Exception as e:
logger.warn(f"Could not copy video to temp for preview: {e}")
logger.info(f"Preview info - filename: {file}, subfolder: {preview_subfolder}, type: {preview_type}")
preview = {
"filename": file,
"subfolder": preview_subfolder,
"type": preview_type,
"format": format,
"frame_rate": frame_rate,
"workflow": first_image_file,
"fullpath": output_files[-1] if output_files else "",
}
if num_frames == 1 and 'png' in format and '%03d' in file:
preview['format'] = 'image/png'
preview['filename'] = file.replace('%03d', '001')
return {"ui": {"gifs": [preview]}, "result": ((save_output, output_files),)}
NODE_CLASS_MAPPINGS = {
"RMVideoCombine": RMVideoCombine,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"RMVideoCombine": "RM Video Combine",
}
+7
View File
@@ -0,0 +1,7 @@
{
"extension": "mp4",
"main_pass": ["-c:v", "libx264", "-preset", "medium", "-crf", "19", "-pix_fmt", "yuv420p"],
"audio_pass": ["-c:a", "aac", "-b:a", "192k"],
"save_metadata": "True",
"dim_alignment": 2
}
+7
View File
@@ -0,0 +1,7 @@
{
"extension": "mp4",
"main_pass": ["-c:v", "libx265", "-preset", "medium", "-crf", "22", "-pix_fmt", "yuv420p", "-tag:v", "hvc1"],
"audio_pass": ["-c:a", "aac", "-b:a", "192k"],
"save_metadata": "True",
"dim_alignment": 2
}
+7
View File
@@ -0,0 +1,7 @@
{
"extension": "webm",
"main_pass": ["-c:v", "libvpx-vp9", "-crf", "30", "-b:v", "0", "-pix_fmt", "yuv420p"],
"audio_pass": ["-c:a", "libopus", "-b:a", "128k"],
"save_metadata": "True",
"dim_alignment": 2
}