First commit with core code

Core developed code and functionality of LevelPixel nodes
This commit is contained in:
Level Pixel Dev
2024-11-05 01:35:36 +06:00
parent 675353dcdb
commit 418db2c60d
21 changed files with 13123 additions and 12 deletions
+22 -12
View File
@@ -172,21 +172,31 @@
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
10. Additional information and licenses (but not excluding all of
the above):
The developers and contributors to this product and repository
do not bear any responsibility for any damage or loss of your data,
valuables, passwords, accounts and any other things that could be
damaged, distorted, damaged, lost, stolen due to the use of this
software and information files. The developers and contributors
to this product do not bear any responsibility for the violation
of your privacy and anonymity.
Use the code, scripts and programs from this repository at your
own risk, all responsibility for the use and launch of the program
code is entirely yours.
Level Pixel has no malicious intent in developing its products,
any problems and violations that you experience are either your
fault or the fault of third parties (the owners of those software
packages that you automatically download along with this product
and program code). Level Pixel does not sell this code, and the
original materials on which these files and program code were
created belong to their rightful owners, as indicated in their
licenses.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Copyright Level Pixel
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
+28
View File
@@ -0,0 +1,28 @@
In this Level Pixel node pack you will find:
LLM nodes - A node that generates text using the LLM model with subsequent unloading of the model from memory. Useful in those workflows where there is constant switching between different models and technologies under conditions of insufficient RAM of the video processor.
Our LLM nodes support the latest LLM and CLIP models, and should support future ones (please let us know if any models stop working).
Most of the functionality is taken from this package - https://github.com/gokayfem/ComfyUI_VLM_nodes
LLaVa nodes - A node that generates text using the LLM model and CLIP by image and prompt with subsequent unloading of the model from memory.
Our LLava nodes support the latest LLM models, and should support future ones (please let us know if any models stop working).
Most of the functionality is taken from this package - https://github.com/gokayfem/ComfyUI_VLM_nodes
Tag Category Filter nodes - a set of nodes that allow you to filter tags by category. There is an option to remove or leave certain categories of tags, there is a function for defining categories of all tags, there is a function for removing certain tags.
Nodes are very convenient because you can use them to remove unnecessary tags by certain categories, for example, to clean up tags and prepare them for use. You can use this to get certain prompts from an image (for example, if you need a description of only the background from an image - you can get this category of tags if you set the "background" category in Tag Category Keeper).
Most of the functionality is taken from this package - https://github.com/sugarkwork/comfyui_tag_fillter
Model Unloader nodes - a node that automatically unloads all checkpoints from memory. It must be added to a sequential chain of nodes in the workflow. There are three versions of this node: Hard (complete unloading of all checkpoints from memory, except for GGUF (not supported yet)), Middle (the same as Hard, but in the future I plan to add widgets with the ability to select a mode), Soft (without unloading checkpoints from memory, just soft cleaning of memory from garbage).
File Counter - A simple counter of files in a given folder. Convenient when you need to count the number of files in a certain format (for example, for subsequent use in loops or conditions).
Image Loader From Path - Loads images from a specific folder or path. It is convenient because you can specify both absolute paths and local paths (from the input folder), as well as the ability to sequentially receive images into the workflow at each step of the cycle by number. In addition, you can load images in batches with a certain number of loaded images in a batch. And the cherry on the cake - you can get a list of image file names at the output.
Load Image - This is a new image loading node that can retrieve the name of the files you load into your workflow.
Fast Checker Pattern - Quickly creates a background image with a checkerboard pattern according to the specified parameters for subsequent testing of images with a transparent background. You need to combine the resulting background image with your image with a transparent background in other ComfyUI nodes (at the moment there is no universal node, but perhaps we will make one in the future).
Simple Float Slider - Simple Float Slider is a handy slider from 0.0001 to 1.0000 to conveniently manage variables in your workflow. The min and max values ​​cannot be changed on the interface (but you can change these values ​​inside Python if you really need to).
(C) Level Pixel
+69
View File
@@ -0,0 +1,69 @@
import os
import pkg_resources
import sys
import subprocess
import folder_paths
import importlib
supported_LLava_extensions = set(['.gguf'])
try:
folder_paths.folder_names_and_paths["LLavacheckpoints"] = (folder_paths.folder_names_and_paths["LLavacheckpoints"][0], supported_LLava_extensions)
except:
if not os.path.isdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints")):
os.mkdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints"))
folder_paths.folder_names_and_paths["LLavacheckpoints"] = ([os.path.join(folder_paths.models_dir, "LLavacheckpoints")], supported_LLava_extensions)
def check_requirements_installed(requirements_path):
with open(requirements_path, 'r') as f:
requirements = [pkg_resources.Requirement.parse(line.strip()) for line in f if line.strip()]
installed_packages = {pkg.key: pkg for pkg in pkg_resources.working_set}
installed_packages_set = set(installed_packages.keys())
missing_packages = []
for requirement in requirements:
if requirement.key not in installed_packages_set or not installed_packages[requirement.key] in requirement:
missing_packages.append(str(requirement))
if missing_packages:
print(f"Missing or outdated packages: {', '.join(missing_packages)}")
print("Installing/Updating missing packages...")
subprocess.check_call([sys.executable, '-s', '-m', 'pip', 'install', *missing_packages])
requirements_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "requirements.txt")
check_requirements_installed(requirements_path)
from .install_init import init, get_system_info, install_llama
system_info = get_system_info()
install_llama(system_info)
llama_cpp_agent_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "cpp_agent_req.txt")
check_requirements_installed(llama_cpp_agent_path)
init()
node_list = [
"image.image_utils_LP",
"io.numbers_utils_LP",
"io.folder_workers_LP",
"io.image_loaders_LP",
"io.text_utils_LP",
"llm.llm_LP",
"tags.tags_utils_LP",
"text.text_utils_LP",
"unloaders.model_unloaders_LP",
"vlm.llava_LP",
]
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
for module_name in node_list:
imported_module = importlib.import_module(f".nodes.{module_name}", __name__)
NODE_CLASS_MAPPINGS = {**NODE_CLASS_MAPPINGS, **imported_module.NODE_CLASS_MAPPINGS}
NODE_DISPLAY_NAME_MAPPINGS = {**NODE_DISPLAY_NAME_MAPPINGS, **imported_module.NODE_DISPLAY_NAME_MAPPINGS}
WEB_DIRECTORY = "./web"
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
+5
View File
@@ -0,0 +1,5 @@
llama-cpp-agent
mkdocs
mkdocs-material
mkdocstrings[python]
docstring-parser
+230
View File
@@ -0,0 +1,230 @@
import os
import json
import shutil
import platform
import subprocess
import sys
import importlib.util
import re
import torch
import inspect
import packaging.tags
from requests import get
from server import PromptServer
def get_python_version():
version_match = re.match(r"3\.(\d+)", platform.python_version())
if version_match:
return "3" + version_match.group(1)
else:
return None
def get_system_info():
system_info = {
'gpu': False,
'cuda_version': None,
'python_version': get_python_version(),
'os': platform.system(),
'os_bit': platform.architecture()[0].replace("bit", ""),
'platform_tag': None,
}
# Check for NVIDIA GPU and CUDA version
if importlib.util.find_spec('torch'):
system_info['gpu'] = torch.cuda.is_available()
if system_info['gpu']:
system_info['cuda_version'] = "cu" + torch.version.cuda.replace(".", "").strip()
# Determine the platform tag
if importlib.util.find_spec('packaging.tags'):
system_info['platform_tag'] = next(packaging.tags.sys_tags()).platform
return system_info
def latest_lamacpp():
try:
response = get("https://api.github.com/repos/abetlen/llama-cpp-python/releases/latest")
return response.json()["tag_name"].replace("v", "")
except Exception:
return "0.2.20"
def install_package(package_name, custom_command=None):
if not package_is_installed(package_name):
print(f"Installing {package_name}...")
command = [sys.executable, "-m", "pip", "install", package_name, "--no-cache-dir"]
if custom_command:
command += custom_command.split()
subprocess.check_call(command)
else:
print(f"{package_name} is already installed.")
def package_is_installed(package_name):
return importlib.util.find_spec(package_name) is not None
def install_llama(system_info):
imported = package_is_installed("llama-cpp-python") or package_is_installed("llama_cpp")
if not imported:
lcpp_version = latest_lamacpp()
base_url = "https://github.com/abetlen/llama-cpp-python/releases/download/v"
if system_info['gpu']:
cuda_version = system_info['cuda_version']
custom_command = f"--force-reinstall --no-deps --index-url=https://abetlen.github.io/llama-cpp-python/whl/{cuda_version}"
else:
custom_command = f"{base_url}{lcpp_version}/llama_cpp_python-{lcpp_version}-{system_info['platform_tag']}.whl"
install_package("llama-cpp-python", custom_command=custom_command)
config = None
def is_logging_enabled():
config = get_extension_config()
if "logging" not in config:
return False
return config["logging"]
def log(message, type=None, always=False, name=None):
if not always and not is_logging_enabled():
return
if type is not None:
message = f"[{type}] {message}"
if name is None:
name = get_extension_config()["name"]
print(f"(levelpixel-nodes:{name}) {message}")
def get_ext_dir(subpath=None, mkdir=False):
dir = os.path.dirname(__file__)
if subpath is not None:
dir = os.path.join(dir, subpath)
dir = os.path.abspath(dir)
if mkdir and not os.path.exists(dir):
os.makedirs(dir)
return dir
def get_extension_config(reload=False):
global config
if reload == False and config is not None:
return config
config_path = get_ext_dir("levelpixel.json")
default_config_path = get_ext_dir("levelpixel.default.json")
if not os.path.exists(config_path):
if os.path.exists(default_config_path):
shutil.copy(default_config_path, config_path)
if not os.path.exists(config_path):
log(f"Failed to create config at {config_path}", type="ERROR", always=True, name="???")
print(f"Extension path: {get_ext_dir()}")
return {"name": "Unknown", "version": -1}
else:
log("Missing levelpixel.default.json, this extension may not work correctly. Please reinstall the extension.",
type="ERROR", always=True, name="???")
print(f"Extension path: {get_ext_dir()}")
return {"name": "Unknown", "version": -1}
with open(config_path, "r") as f:
config = json.loads(f.read())
return config
def link_js(src, dst):
src = os.path.abspath(src)
dst = os.path.abspath(dst)
if os.name == "nt":
try:
import _winapi
_winapi.CreateJunction(src, dst)
return True
except:
pass
try:
os.symlink(src, dst)
return True
except:
import logging
logging.exception('')
return False
def is_junction(path):
if os.name != "nt":
return False
try:
return bool(os.readlink(path))
except OSError:
return False
def install_js():
src_dir = get_ext_dir("web/js")
if not os.path.exists(src_dir):
log("No JS")
return
should_install = should_install_js()
if should_install:
log("it looks like you're running an old version of ComfyUI that requires manual setup of web files, it is recommended you update your installation.", "warning", True)
dst_dir = get_web_ext_dir()
linked = os.path.islink(dst_dir) or is_junction(dst_dir)
if linked or os.path.exists(dst_dir):
if linked:
if should_install:
log("JS already linked")
else:
os.unlink(dst_dir)
log("JS unlinked, PromptServer will serve extension")
elif not should_install:
shutil.rmtree(dst_dir)
log("JS deleted, PromptServer will serve extension")
return
if not should_install:
log("JS skipped, PromptServer will serve extension")
return
if link_js(src_dir, dst_dir):
log("JS linked")
return
log("Copying JS files")
shutil.copytree(src_dir, dst_dir, dirs_exist_ok=True)
def get_web_ext_dir():
config = get_extension_config()
name = config["name"]
dir = get_comfy_dir("web/extensions/levelpixel")
if not os.path.exists(dir):
os.makedirs(dir)
dir = os.path.join(dir, name)
return dir
def get_comfy_dir(subpath=None, mkdir=False):
dir = os.path.dirname(inspect.getfile(PromptServer))
if subpath is not None:
dir = os.path.join(dir, subpath)
dir = os.path.abspath(dir)
if mkdir and not os.path.exists(dir):
os.makedirs(dir)
return dir
def should_install_js():
return not hasattr(PromptServer.instance, "supports") or "custom_nodes_from_web" not in PromptServer.instance.supports
def init(check_imports=None):
log("Init")
if check_imports is not None:
import importlib.util
for imp in check_imports:
spec = importlib.util.find_spec(imp)
if spec is None:
log(f"{imp} is required, please check requirements are installed.",
type="ERROR", always=True)
return False
install_js()
return True
+4
View File
@@ -0,0 +1,4 @@
{
"name": "CustomScripts",
"logging": false
}
+4
View File
@@ -0,0 +1,4 @@
{
"name": "CustomScripts",
"logging": false
}
+128
View File
@@ -0,0 +1,128 @@
import numpy as np
import io
import torch
from PIL import Image
import matplotlib.pyplot as plt
color_mapping = {
"white": (255, 255, 255),
"black": (0, 0, 0),
"red": (255, 0, 0),
"green": (0, 255, 0),
"blue": (0, 0, 255),
"yellow": (255, 255, 0),
"cyan": (0, 255, 255),
"magenta": (255, 0, 255),
"orange": (255, 165, 0),
"purple": (128, 0, 128),
"pink": (255, 192, 203),
"brown": (165, 42, 42),
"gray": (128, 128, 128),
"lightgray": (211, 211, 211),
"darkgray": (169, 169, 169),
"olive": (128, 128, 0),
"lime": (0, 128, 0),
"teal": (0, 128, 128),
"navy": (0, 0, 128),
"maroon": (128, 0, 0),
"fuchsia": (255, 0, 128),
"aqua": (0, 255, 128),
"silver": (192, 192, 192),
"gold": (255, 215, 0),
"turquoise": (64, 224, 208),
"lavender": (230, 230, 250),
"violet": (238, 130, 238),
"coral": (255, 127, 80),
"indigo": (75, 0, 130),
}
COLORS = ["custom", "white", "black", "red", "green", "blue", "yellow",
"cyan", "magenta", "orange", "purple", "pink", "brown", "gray",
"lightgray", "darkgray", "olive", "lime", "teal", "navy", "maroon",
"fuchsia", "aqua", "silver", "gold", "turquoise", "lavender",
"violet", "coral", "indigo"]
def pil2tensor(image):
return torch.from_numpy(np.array(image).astype(np.float32) / 255.0).unsqueeze(0)
def hex_to_rgb(hex_color):
hex_color = hex_color.lstrip('#')
r = int(hex_color[0:2], 16)
g = int(hex_color[2:4], 16)
b = int(hex_color[4:6], 16)
return (r, g, b)
class FastCheckerPattern:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"width": ("INT", {"default": 512, "min": 64, "max": 4096}),
"height": ("INT", {"default": 512, "min": 64, "max": 4096}),
"color_1": (COLORS,),
"color_2": (COLORS,),
"grid_frequency": ("INT", {"default": 50, "min": 1, "max": 200, "step": 1}),
},
"optional": {
"color1_hex": ("STRING", {"multiline": False, "default": "#C0C0C0"}),
"color2_hex": ("STRING", {"multiline": False, "default": "#808080"}),
}
}
RETURN_TYPES = ("IMAGE", )
RETURN_NAMES = ("IMAGE", )
FUNCTION = "draw"
CATEGORY = "LevelPixel/Image"
def draw(self, width, height, color_1, color_2,
grid_frequency, color1_hex='#C0C0C0', color2_hex='#808080'):
if color_1 == "custom":
color1_rgb = hex_to_rgb(color1_hex)
else:
color1_rgb = color_mapping.get(color_1, (255, 255, 255))
if color_2 == "custom":
color2_rgb = hex_to_rgb(color2_hex)
else:
color2_rgb = color_mapping.get(color_2, (0, 0, 0))
canvas = np.zeros((height, width, 3), dtype=np.uint8)
grid_size = width // grid_frequency
x_indices = np.arange(width) // grid_size
y_indices = np.arange(height) // grid_size
x_grid, y_grid = np.meshgrid(x_indices, y_indices)
checker_pattern = (x_grid + y_grid) % 2
canvas[checker_pattern == 0] = color1_rgb
canvas[checker_pattern == 1] = color2_rgb
fig, ax = plt.subplots(figsize=(width/100, height/100))
ax.imshow(canvas)
plt.axis('off')
plt.tight_layout(pad=0, w_pad=0, h_pad=0)
plt.autoscale(tight=True)
img_buf = io.BytesIO()
plt.savefig(img_buf, format='png')
img = Image.open(img_buf)
image_out = pil2tensor(img.convert("RGB"))
return (image_out,)
NODE_CLASS_MAPPINGS = {
"FastCheckerPattern|LP": FastCheckerPattern,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"FastCheckerPattern|LP": "Fast Checker Pattern [LP]",
}
+44
View File
@@ -0,0 +1,44 @@
import glob
import os
class FileCounter:
FUNCTION = "file_counter"
OUTPUT_NODE = True
CATEGORY = "LevelPixel/IO"
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"directory_path": ("STRING", {"default": '', "multiline": False}),
"patterns": ("STRING", {"default": '*.jpg|*.png|*.jpeg', "multiline": False}),
},
}
RETURN_TYPES = ("INT","STRING")
RETURN_NAMES = ("Total INT","Total STRING")
FUNCTION = "file_counter"
@classmethod
def IS_CHANGED(cls, *v):
return ALWAYS_CHANGED_FLAG
def file_counter(self, directory_path, patterns):
if not os.path.isdir(directory_path):
return (0,)
total_int = 0
for pattern in patterns.split("|"):
files = list(glob.glob(pattern, root_dir=directory_path))
total_int += len(files)
total_string = str(total_int)
print("total " + str(total_int))
return (total_int, total_string)
NODE_CLASS_MAPPINGS = {
"FileCounter|LP": FileCounter,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"FileCounter|LP": "File Counter [LP]",
}
+174
View File
@@ -0,0 +1,174 @@
import numpy as np
import os
from PIL import Image, ImageOps, ImageSequence
import folder_paths
import re
import torch
import node_helpers
import hashlib
class ImageLoaderFromPath:
@classmethod
def INPUT_TYPES(s):
input_dir = folder_paths.input_directory
image_folder = [name for name in os.listdir(input_dir) if os.path.isdir(os.path.join(input_dir,name))]
image_folder.append("")
return {"required": {"input_folder": (sorted(image_folder), ),
"start_index": ("INT", {"default": 0, "min": 0, "max": 9999}),
"max_images": ("INT", {"default": 1, "min": 1, "max": 9999}),
"white_bg": (["disable","enable"],),
"patterns": ("STRING", {"default": '*.jpg|*.png|*.jpeg', "multiline": False}),
},
"optional": {"input_path": ("STRING", {"default": '', "multiline": False}),
}
}
RETURN_TYPES = ("IMAGE", "MASK", "STRING" )
RETURN_NAMES = ("IMAGE", "MASK", "Filename STRING" )
OUTPUT_IS_LIST = (True, True, True)
FUNCTION = "loader_images"
CATEGORY = "LevelPixel/IO"
def loader_images(self, start_index, max_images, white_bg, patterns, input_folder="", input_path=None):
if input_path != '' and input_path is not None:
if not os.path.exists(input_path):
print(f"[Warning] Image Loader From Path: The input_path `{input_path}` does not exist")
return ("",)
in_path = input_path
else:
input_dir = folder_paths.input_directory
in_path = os.path.join(input_dir, input_folder)
if not os.listdir(in_path):
print(f"[Warning] Image Loader From Path: The folder `{in_path}` is empty")
return None
file_list = sorted(os.listdir(in_path), key=lambda s: sum(((s, int(n)) for s, n in re.findall(r'(\D+)(\d+)', 'a%s0' % s)), ()))
extensions = tuple(patterns.replace('*', '').split('|'))
file_list = [f for f in file_list if f.lower().endswith(extensions)]
image_list = []
mask_list = []
filename_list = []
start_index = max(0, start_index)
end_index = min(start_index + max_images, len(file_list))
for num in range(start_index, end_index):
img = Image.open(os.path.join(in_path, file_list[num]))
image = img.convert("RGB")
image = np.array(image).astype(np.float32) / 255.0
image = torch.from_numpy(image)[None,]
if 'A' in img.getbands():
mask = np.array(img.getchannel('A')).astype(np.float32) / 255.0
mask = 1. - torch.from_numpy(mask)
if white_bg=="enable":
nw = mask.unsqueeze(0).unsqueeze(-1).repeat(1, 1, 1, 3)
image[nw == 1] = 1.0
else:
mask = torch.zeros((64,64), dtype=torch.float32, device="cpu")
image_list.append(image)
mask_list.append(mask)
filename_list.append(file_list[num])
if not image_list:
print("Image Loader From Path: No images found.")
return None
return (image_list, mask_list, filename_list)
class LoadImage:
@classmethod
def INPUT_TYPES(s):
input_dir = folder_paths.get_input_directory()
files = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))]
return {"required":
{"image": (sorted(files), {"image_upload": True})},
}
RETURN_TYPES = ("IMAGE", "MASK", "STRING" )
RETURN_NAMES = ("IMAGE", "MASK", "Filename STRING")
FUNCTION = "load_image"
CATEGORY = "LevelPixel/IO"
def load_image(self, image):
image_path = folder_paths.get_annotated_filepath(image)
file_name = os.path.basename(image_path)
img = node_helpers.pillow(Image.open, image_path)
output_images = []
output_masks = []
output_filenames = []
w, h = None, None
excluded_formats = ['MPO']
for i in ImageSequence.Iterator(img):
i = node_helpers.pillow(ImageOps.exif_transpose, i)
if i.mode == 'I':
i = i.point(lambda i: i * (1 / 255))
image = i.convert("RGB")
if len(output_images) == 0:
w = image.size[0]
h = image.size[1]
if image.size[0] != w or image.size[1] != h:
continue
image = np.array(image).astype(np.float32) / 255.0
image = torch.from_numpy(image)[None,]
if 'A' in i.getbands():
mask = np.array(i.getchannel('A')).astype(np.float32) / 255.0
mask = 1. - torch.from_numpy(mask)
else:
mask = torch.zeros((64,64), dtype=torch.float32, device="cpu")
output_images.append(image)
output_masks.append(mask.unsqueeze(0))
output_filenames.append(file_name)
if len(output_images) > 1 and img.format not in excluded_formats:
output_image = torch.cat(output_images, dim=0)
output_mask = torch.cat(output_masks, dim=0)
else:
output_image = output_images[0]
output_mask = output_masks[0]
output_filename = output_filenames[0]
return (output_image, output_mask, output_filename)
@classmethod
def IS_CHANGED(s, image):
image_path = folder_paths.get_annotated_filepath(image)
m = hashlib.sha256()
with open(image_path, 'rb') as f:
m.update(f.read())
return m.digest().hex()
@classmethod
def VALIDATE_INPUTS(s, image):
if not folder_paths.exists_annotated_filepath(image):
return "Invalid image file: {}".format(image)
return True
NODE_CLASS_MAPPINGS = {
"ImageLoaderFromPath|LP": ImageLoaderFromPath,
"LoadImage|LP": LoadImage,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"ImageLoaderFromPath|LP": "Image Loader From Path [LP]",
"LoadImage|LP": "Load Image [LP]",
}
+37
View File
@@ -0,0 +1,37 @@
class FloatSlider:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"number":("FLOAT", {
"default": 0,
"min": 0.0001,
"max": 1.0000,
"step": 0.0001,
"display": "slider"
}),
},
}
RETURN_TYPES = ("FLOAT",)
RETURN_NAMES = ('FLOAT',)
FUNCTION = "run"
CATEGORY = "LevelPixel/IO"
INPUT_IS_LIST = False
OUTPUT_IS_LIST = (False,)
def run(self, number):
if number < 0.0001:
number = 0.0001
elif number > 1.0000:
number = 1.0000
return (number,)
NODE_CLASS_MAPPINGS = {
"SimpleFloatSlider|LP": FloatSlider,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"SimpleFloatSlider|LP": "Simple Float Slider [LP]",
}
+48
View File
@@ -0,0 +1,48 @@
class ShowTextLP:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"text": ("STRING", {"forceInput": True}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
"extra_pnginfo": "EXTRA_PNGINFO",
},
}
INPUT_IS_LIST = True
RETURN_TYPES = ("STRING",)
FUNCTION = "show_text_lp"
OUTPUT_NODE = True
OUTPUT_IS_LIST = (True,)
CATEGORY = "LevelPixel/IO"
def show_text_lp(self, text, unique_id=None, extra_pnginfo=None):
if unique_id is not None and extra_pnginfo is not None:
if not isinstance(extra_pnginfo, list):
print("Error: extra_pnginfo is not a list")
elif (
not isinstance(extra_pnginfo[0], dict)
or "workflow" not in extra_pnginfo[0]
):
print("Error: extra_pnginfo[0] is not a dict or missing 'workflow' key")
else:
workflow = extra_pnginfo[0]["workflow"]
node = next(
(x for x in workflow["nodes"] if str(x["id"]) == str(unique_id[0])),
None,
)
if node:
node["widgets_values"] = [text]
return {"ui": {"text": text}, "result": (text,)}
NODE_CLASS_MAPPINGS = {
"ShowText|LP": ShowTextLP,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"ShowText|LP": "Show Text [LP]",
}
+164
View File
@@ -0,0 +1,164 @@
import folder_paths
import os
from llama_cpp import Llama
import gc
import torch
supported_LLava_extensions = set(['.gguf'])
try:
folder_paths.folder_names_and_paths["LLavacheckpoints"] = (folder_paths.folder_names_and_paths["LLavacheckpoints"][0], supported_LLava_extensions)
except:
if not os.path.isdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints")):
os.mkdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints"))
folder_paths.folder_names_and_paths["LLavacheckpoints"] = ([os.path.join(folder_paths.models_dir, "LLavacheckpoints")], supported_LLava_extensions)
class LLMLoader:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
"max_ctx": ("INT", {"default": 2048, "min": 128, "max": 128000, "step": 64}),
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
}
}
RETURN_TYPES = ("CUSTOM",)
RETURN_NAMES = ("model",)
FUNCTION = "load_llm_checkpoint"
CATEGORY = "LevelPixel/LLM"
def load_llm_checkpoint(self, ckpt_name, max_ctx, gpu_layers, n_threads):
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
llm = Llama(model_path = ckpt_path, chat_format="chatml", offload_kqv=True,
f16_kv=True, use_mlock=False, embedding=False, n_batch=1024,
last_n_tokens_size=1024, verbose=True, seed=42, n_ctx = max_ctx,
n_gpu_layers=gpu_layers, n_threads=n_threads,)
return (llm, )
class LLMSampler:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": ("CUSTOM", {"default": ""}),
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
"temperature": ("FLOAT", {"default": 0.2, "min": 0.01, "max": 1.0, "step": 0.01}),
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
"top_k": ("INT", {"default": 40, "step": 1}),
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
"seed": ("INT", {"default": 42, "step": 1}),
"prompt": ("STRING",{"multiline": True, "default": ""}),
"system_msg": ("STRING",{ "multiline": True, "default" : "You are an assistant who perfectly describes images."}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "generate_text_sampler"
CATEGORY = "LevelPixel/LLM"
def generate_text_sampler(self, system_msg, prompt, model, max_tokens,
temperature, top_p, top_k, frequency_penalty,
presence_penalty, repeat_penalty, seed):
llm = model
response = llm.create_chat_completion(messages=[
{"role": "system", "content": system_msg},
{"role": "user", "content": prompt + " Assistant:"},
],
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=top_k,
frequency_penalty=frequency_penalty,
presence_penalty=presence_penalty,
repeat_penalty=repeat_penalty,
seed=seed
)
return (f"{response['choices'][0]['message']['content']}", )
class LLMAdvanced:
def __init__(self):
self.llm = None
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
"max_ctx": ("INT", {"default": 2048, "min": 300, "max": 100000, "step": 64}),
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
"top_k": ("INT", {"default": 40, "step": 1}),
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
"seed": ("INT", {"default": 42, "step": 1}),
"unload": ("BOOLEAN", {"default": False}),
"prompt": ("STRING", {"multiline": True, "default": ""}),
"system_msg": ("STRING", {"multiline": True, "default": "You are an assistant who perfectly describes images."}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "generate_text_advanced"
CATEGORY = "LevelPixel/LLM"
def generate_text_advanced(self, ckpt_name, max_ctx, gpu_layers, n_threads,
system_msg, prompt, max_tokens, temperature, top_p,
top_k, frequency_penalty, presence_penalty, repeat_penalty, seed, unload):
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
self.llm = Llama(model_path = ckpt_path, offload_kqv=True, f16_kv=True,
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers,
n_threads=n_threads, logits_all=True, echo=False)
response = self.llm.create_chat_completion(messages=[
{"role": "system", "content": system_msg},
{"role": "user", "content": prompt},
],
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=top_k,
frequency_penalty=frequency_penalty,
presence_penalty=presence_penalty,
repeat_penalty=repeat_penalty,
seed=seed,
)
if unload and self.llm is not None:
self.llm.close()
del self.llm
self.llm = None
gc.collect()
torch.cuda.empty_cache()
return (f"{response['choices'][0]['message']['content']}", )
NODE_CLASS_MAPPINGS = {
"LLMLoader|LP": LLMLoader,
"LLMSampler|LP": LLMSampler,
"LLMAdvanced|LP": LLMAdvanced,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"LLMLoader|LP": "LLM Loader [LP]",
"LLMSampler|LP": "LLM Sampler [LP]",
"LLMAdvanced|LP": "LLM Advanced [LP]"
}
File diff suppressed because it is too large Load Diff
+443
View File
@@ -0,0 +1,443 @@
import os
import json
tag_category = json.load(open(os.path.join(os.path.dirname(os.path.realpath(__file__)),"tag_category.json")))
def category_for_tags(tags):
tags = [tag.strip() for tag in tags.replace(".", ",").replace("\n", ",").split(",")]
tags2 = [tag.replace(" ", "_").lower() for tag in tags]
result = []
for i, tag2 in enumerate(tags2):
if tag2 in tag_category:
category_list = tag_category.get(tag2, [])
for category in category_list:
if category not in result:
result.append(category)
break
return ", ".join(result)
class TagCategoryFilter:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"tags": ("STRING", {"default": ""}),
"include_categories": ("STRING", {"default": ""}),
"exclude_categories": ("STRING", {"default": ""}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("result",)
FUNCTION = "tag_category"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def tag_category(self, tags, include_categories="", exclude_categories=""):
targets = []
exclude_targets = []
if not include_categories:
print("categories - ", category_for_tags(tags))
include_categories = category_for_tags(tags)
include_categories = include_categories.strip()
if include_categories:
targets += [category.strip() for category in include_categories.replace("\n",",").split(",")]
if exclude_categories:
exclude_targets = [category.strip() for category in exclude_categories.replace("\n",",").split(",")]
targets = [target for target in targets if target not in exclude_targets]
print("targets", targets)
tags = [tag.strip() for tag in tags.replace(".", ",").replace("\n", ",").split(",")]
tags2 = [tag.replace(" ", "_").lower() for tag in tags]
result = []
for i, tag2 in enumerate(tags2):
if tag2 in tag_category:
category_list = tag_category.get(tag2, [])
for category in category_list:
if category in exclude_targets:
break
else:
for category in category_list:
if '*' in include_categories or ( category in targets and tags[i] not in result ):
result.append(tags[i])
break
return (", ".join(result),)
class TagCategoryKeeper:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"tags": ("STRING", {"default": ""}),
"include_categories": ("STRING", {"default": ""}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("result",)
FUNCTION = "tag_keeper"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def tag_keeper(self, tags, include_categories=""):
targets = []
include_categories = include_categories.strip()
if include_categories:
targets += [category.strip() for category in include_categories.replace("\n",",").split(",")]
print("targets", targets)
tags = [tag.strip() for tag in tags.replace(".", ",").replace("\n", ",").split(",")]
tags2 = [tag.replace(" ", "_").lower() for tag in tags]
result = []
for i, tag2 in enumerate(tags2):
if tag2 in tag_category:
category_list = tag_category.get(tag2, [])
for category in category_list:
if '*' in include_categories or ( category in targets and tags[i] not in result ):
result.append(tags[i])
break
return (", ".join(result),)
class TagCategoryRemover:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"tags": ("STRING", {"default": ""}),
"exclude_categories": ("STRING", {"default": ""}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("result",)
FUNCTION = "tag_remover"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def tag_remover(self, tags, exclude_categories=""):
targets = []
exclude_targets = []
if exclude_categories:
exclude_targets = [category.strip() for category in exclude_categories.replace("\n",",").split(",")]
targets = [target for target in targets if target not in exclude_targets]
print("targets", targets)
tags = [tag.strip() for tag in tags.replace(".", ",").replace("\n", ",").split(",")]
tags2 = [tag.replace(" ", "_").lower() for tag in tags]
result = []
for i, tag2 in enumerate(tags2):
if tag2 in tag_category:
category_list = tag_category.get(tag2, [])
for category in category_list:
if category in exclude_targets:
break
else:
if tags[i] not in result:
result.append(tags[i])
break
return (", ".join(result),)
class TagSwitcher:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input_tags": ("STRING", {"default": ""}),
"default_image": ("IMAGE", {"default": ""}),
"tags1": ("STRING", {"default": ""}),
"image1": ("IMAGE", {"default": ""}),
"any1": ("BOOLEAN", {"default": True}),
},
"optional": {
"tags2": ("STRING", {"default": ""}),
"image2": ("IMAGE", {"default": ""}),
"any2": ("BOOLEAN", {"default": True}),
"tags3": ("STRING", {"default": ""}),
"image3": ("IMAGE", {"default": ""}),
"any3": ("BOOLEAN", {"default": True}),
"tags4": ("STRING", {"default": ""}),
"image4": ("IMAGE", {"default": ""}),
"any4": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("image",)
FUNCTION = "tag_switcher"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def _tag_split(self, tags: str) -> list:
return [tag.strip().replace("_", " ").lower().strip() for tag in tags.replace(".",",").replace("\n",",").split(",")]
def tag_switcher(self, input_tags="", default_image=None, tags1="", image1=None, any1=True, tags2="", image2=None, any2=True, tags3="", image3=None, any3=True, tags4="", image4=None, any4=True):
input_tags = self._tag_split(input_tags)
target_tags = []
tags1 = set(self._tag_split(tags1))
target_tags.append((tags1, image1, any1))
tags2 = set(self._tag_split(tags2))
target_tags.append((tags2, image2, any2))
tags3 = set(self._tag_split(tags3))
target_tags.append((tags3, image3, any3))
tags4 = set(self._tag_split(tags4))
target_tags.append((tags4, image4, any4))
for tags, image, any_flag in target_tags:
if any_flag:
if any(tag in tags for tag in input_tags):
return (image,)
else:
if all(tag in input_tags for tag in tags):
return (image,)
return (default_image,)
class TagMerger:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"tags1": ("STRING", {"default": ""}),
"tags2": ("STRING", {"default": ""}),
"under_score": ("BOOLEAN", {"default": True}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("result",)
FUNCTION = "tag_merger"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def tag_merger(self, tags1:str, tags2:str, under_score=True):
tags1 = [tag.strip().replace(" ", "_").lower() for tag in tags1.replace(".",",").replace("\n",",").split(",")]
tags2 = [tag.strip().replace(" ", "_").lower() for tag in tags2.replace(".",",").replace("\n",",").split(",")]
tags = tags1 + list(set(tags2) - set(tags1))
tags = [tag for tag in tags if tag]
if not under_score:
tags = [tag.replace("_", " ") for tag in tags]
return (", ".join(tags),)
class TagRemover:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"tags": ("STRING", {"default": ""}),
"exclude_tags": ("STRING", {"default": ""}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("result",)
FUNCTION = "tag_remover"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def tag_remover(self, tags:str, exclude_tags:str=""):
tags = [tag.strip() for tag in tags.replace("\n",",").split(",")]
tags2 = [tag.replace(" ", "_").lower().strip() for tag in tags]
exclude_tags = [tag.strip() for tag in exclude_tags.replace("\n",",").split(",")]
exclude_tags2 = [tag.replace(" ", "_").lower().strip() for tag in exclude_tags]
result = []
for i, tag2 in enumerate(tags2):
if tag2 not in exclude_tags2:
result.append(tags[i])
return (", ".join(result),)
class TagReplace:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"tags": ("STRING", {"default": ""}),
"replace_tags": ("STRING", {"default": ""}),
"match": ("FLOAT", {"default": 0.3}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("result",)
FUNCTION = "tag_replace"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def _get_categories(self, tag: str) -> set:
return set(tag_category.get(tag, []))
def _category_match_percentage(self, categories1: set, categories2: set) -> float:
if not categories1 or not categories2:
return 0
intersection = categories1.intersection(categories2)
union = categories1.union(categories2)
return len(intersection) / len(union)
def tag_replace(self, tags:str, replace_tags:str="", match:float=0.3):
tags = [tag.strip() for tag in tags.replace("\n",",").split(",")]
tags_normalized = [tag.replace(" ", "_").lower().strip() for tag in tags]
replace_tags = [tag.strip() for tag in replace_tags.replace("\n",",").split(",")]
replace_tags_normalized = [tag.replace(" ", "_").lower().strip() for tag in replace_tags]
replace_tags_used = {tag:False for tag in replace_tags_normalized}
result = []
for i, tag in enumerate(tags_normalized):
tag_categories = self._get_categories(tag)
best_match_tag = None
best_match_tag_id = None
best_match_percentage = 0
for k, replace_tag in enumerate(replace_tags_normalized):
replace_categories = self._get_categories(replace_tag)
match_percentage = self._category_match_percentage(tag_categories, replace_categories)
if match_percentage and match_percentage > best_match_percentage:
best_match_percentage = match_percentage
best_match_tag = replace_tag
replace_tags_used[replace_tag] = True
best_match_tag_id = k
if best_match_tag and best_match_percentage >= match:
result.append(replace_tags[best_match_tag_id])
else:
result.append(tags[i])
extra_tags = [replace_tag for replace_tag, used in replace_tags_used.items() if not used]
result.extend(extra_tags)
return (", ".join(result),)
class TagCategory:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"tags": ("STRING", {"default": ""}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("result",)
FUNCTION = "tag_category_info"
CATEGORY = "LevelPixel/Tags"
OUTPUT_NODE = True
def tag_category_info(self, tags):
tags = [tag.strip() for tag in tags.replace(".", ",").replace("\n", ",").split(",")]
tags2 = [tag.replace(" ", "_").lower() for tag in tags]
result = []
for i, tag2 in enumerate(tags2):
if tag2 in tag_category:
category_list = tag_category.get(tag2, [])
for category in category_list:
if category not in result:
result.append(category)
break
return (", ".join(result),)
NODE_CLASS_MAPPINGS = {
"TagCategory|LP": TagCategory,
"TagCategoryFilter|LP": TagCategoryFilter,
"TagCategoryKeeper|LP": TagCategoryKeeper,
"TagCategoryRemover|LP": TagCategoryRemover,
"TagSwitcher|LP": TagSwitcher,
"TagMerger|LP": TagMerger,
"TagReplace|LP": TagReplace,
"TagRemover|LP": TagRemover,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"TagCategory|LP": "Tag Category [LP]",
"TagCategoryFilter|LP": "Tag Category Filter [LP]",
"TagCategoryKeeper|LP": "Tag Category Keeper [LP]",
"TagCategoryRemover|LP": "Tag Category Remover [LP]",
"TagSwitcher|LP": "Tag Switcher [LP]",
"TagMerger|LP": "Tag Merger [LP]",
"TagReplace|LP": "Tag Replace [LP]",
"TagRemover|LP": "Tag Remover [LP]"
}
+61
View File
@@ -0,0 +1,61 @@
import re
import random
import time
class TextChoiceParser:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"text": ("STRING", {"multiline": True, "dynamicPrompts": False}),
},
"optional": {
"variables": ("STRING", {"multiline": True, "dynamicPrompts": False}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("text",)
FUNCTION = "text_choice_parser"
OUTPUT_NODE = True
CATEGORY = "LevelPixel/Text"
def text_choice_parser(self, text, variables="", seed=None):
if len(text) > 10000:
return ("Text too large to process at once",)
if seed is None or seed == 0:
seed = int(time.time() * 1000)
random.seed(seed)
var_dict = {}
for line in variables.split('\n'):
if '=' in line:
key, value = line.split('=', 1)
var_dict[key.strip()] = value.strip()
for key, value in var_dict.items():
text = text.replace(f"[{key}]", value)
pattern = r'\{([^}]+)\}'
def replace_random(match):
return random.choice(match.group(1).split('|'))
result = re.sub(pattern, replace_random, text)
return (result,)
@classmethod
def IS_CHANGED(s, text, variables="", seed=None):
return (text, variables, seed)
NODE_CLASS_MAPPINGS = {
"TextChoiceParser|LP": TextChoiceParser,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"TextChoiceParser|LP": "Text Choice Parser [LP]",
}
+116
View File
@@ -0,0 +1,116 @@
from comfy import model_management
import gc
import torch
class AnyType(str):
def __ne__(self, __value: object) -> bool:
return False
any = AnyType("*")
class ModelUnloader:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"source": (any, {}),
},
}
RETURN_TYPES = (any,)
FUNCTION = "unload_model"
OUTPUT_NODE = True
CATEGORY = "LevelPixel/Unloaders"
def unload_model(self, **kwargs):
loadedmodels=model_management.current_loaded_models
unloaded_model = False
for i in range(len(loadedmodels) -1, -1, -1):
m = loadedmodels.pop(i)
m.model_unload()
del m
unloaded_model = True
if unloaded_model:
model_management.soft_empty_cache()
gc.collect()
torch.cuda.empty_cache()
return (kwargs["source"],)
class SoftModelUnloader:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"source": (any, {}),
},
}
RETURN_TYPES = (any,)
FUNCTION = "soft_unload_model"
OUTPUT_NODE = True
CATEGORY = "LevelPixel/Unloaders"
def soft_unload_model(self, **kwargs):
model_management.soft_empty_cache()
gc.collect()
torch.cuda.empty_cache()
return (kwargs["source"],)
class HardModelUnloader:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"source": (any, {}),
},
}
RETURN_TYPES = (any,)
FUNCTION = "hard_unload_model"
OUTPUT_NODE = True
CATEGORY = "LevelPixel/Unloaders"
def hard_unload_model(self, **kwargs):
loadedmodels=model_management.current_loaded_models
unloaded_model = False
for i in range(len(loadedmodels) -1, -1, -1):
m = loadedmodels.pop(i)
m.model_unload()
del m
unloaded_model = True
if unloaded_model:
model_management.soft_empty_cache()
gc.collect()
torch.cuda.empty_cache()
return (kwargs["source"],)
NODE_CLASS_MAPPINGS = {
"ModelUnloader|LP": ModelUnloader,
"SoftModelUnloader|LP": SoftModelUnloader,
"HardModelUnloader|LP": HardModelUnloader,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"ModelUnloader|LP": "Model Unloader [LP]",
"SoftModelUnloader|LP": "Soft Model Unloader [LP]",
"HardModelUnloader|LP": "Hard Model Unloader [LP]",
}
+367
View File
@@ -0,0 +1,367 @@
import folder_paths
import os
from io import BytesIO
from llama_cpp import Llama
from llama_cpp.llama_chat_format import Llava16ChatHandler
import base64
from torchvision.transforms import ToPILImage
import gc
import torch
supported_LLava_extensions = set(['.gguf'])
try:
folder_paths.folder_names_and_paths["LLavacheckpoints"] = (folder_paths.folder_names_and_paths["LLavacheckpoints"][0], supported_LLava_extensions)
except:
if not os.path.isdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints")):
os.mkdir(os.path.join(folder_paths.models_dir, "LLavacheckpoints"))
folder_paths.folder_names_and_paths["LLavacheckpoints"] = ([os.path.join(folder_paths.models_dir, "LLavacheckpoints")], supported_LLava_extensions)
class LLavaLoader:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
"max_ctx": ("INT", {"default": 4096, "min": 128, "max": 8192, "step": 64}),
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
"clip": ("CUSTOM", {"default": ""}),
}
}
RETURN_TYPES = ("CUSTOM",)
RETURN_NAMES = ("model",)
FUNCTION = "load_llava_checkpoint"
CATEGORY = "LevelPixel/LLava"
def load_llava_checkpoint(self, ckpt_name, max_ctx, gpu_layers, n_threads, clip ):
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
llm = Llama(model_path = ckpt_path, chat_handler=clip,offload_kqv=True, f16_kv=True,
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers, n_threads=n_threads,
logits_all=True, echo=False)
return (llm, )
class LLavaClipLoader:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"clip_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
}
}
RETURN_TYPES = ("CUSTOM", )
RETURN_NAMES = ("clip", )
FUNCTION = "load_clip_checkpoint"
CATEGORY = "LevelPixel/LLava"
def load_clip_checkpoint(self, clip_name):
clip_path = folder_paths.get_full_path("LLavacheckpoints", clip_name)
clip = Llava16ChatHandler(clip_model_path = clip_path, verbose=False)
return (clip, )
class LLavaSamplerSimple:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
"model": ("CUSTOM", {"default": ""}),
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
"prompt": ("STRING",{"multiline": True} ),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "generate_text_simple"
CATEGORY = "LevelPixel/LLava"
def generate_text_simple(self, image, prompt, model, temperature):
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
buffer = BytesIO()
pil_image.save(buffer, format="PNG")
image_bytes = buffer.getvalue()
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
llm = model
response = llm.create_chat_completion(
messages = [
{"role": "system", "content": "You are an assistant who perfectly describes images."},
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url" : base64_string}},
{"type" : "text", "text": f"{prompt}"}
]
}
],
temperature = temperature,
)
return (f"{response['choices'][0]['message']['content']}", )
class LLavaSamplerAdvanced:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
"model": ("CUSTOM", {"default": ""}),
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
"top_k": ("INT", {"default": 40, "step": 1}),
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
"seed": ("INT", {"default": 42, "step":1}),
"prompt": ("STRING",{"multiline": True, "default": ""}),
"system_msg": ("STRING",{"multiline": True, "default" : "You are an assistant who perfectly describes images."}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "generate_text_advanced"
CATEGORY = "LevelPixel/LLava"
def generate_text_advanced(self, image, system_msg, prompt, model, max_tokens, temperature, top_p,
frequency_penalty, presence_penalty, repeat_penalty, top_k,seed):
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
buffer = BytesIO()
pil_image.save(buffer, format="PNG")
image_bytes = buffer.getvalue()
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
llm = model
response = llm.create_chat_completion(
messages = [
{"role": "system", "content": system_msg},
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url" : base64_string}},
{"type" : "text", "text": f"{prompt}"}
]
}
],
max_tokens = max_tokens,
temperature = temperature,
top_p = top_p,
top_k = top_k,
frequency_penalty = frequency_penalty,
presence_penalty = presence_penalty,
repeat_penalty = repeat_penalty,
seed=seed
)
return (f"{response['choices'][0]['message']['content']}", )
class LLavaSimple:
def __init__(self):
self.llm = None
self.clip = None
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
"clip_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
"max_ctx": ("INT", {"default": 4096, "min": 128, "max": 128000, "step": 64}),
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
"unload": ("BOOLEAN", {"default": False}),
"prompt": ("STRING", {"multiline": True, "forceInput": True}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "generate_text_full_simple"
CATEGORY = "LevelPixel/LLava"
def generate_text_full_simple(self, ckpt_name, clip_name, max_ctx, gpu_layers, n_threads, image, prompt, temperature, unload):
clip_path = folder_paths.get_full_path("LLavacheckpoints", clip_name)
self.clip = Llava16ChatHandler(clip_model_path=clip_path, verbose=False)
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
self.llm = Llama(model_path = ckpt_path, chat_handler=self.clip, offload_kqv=True, f16_kv=True,
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers, n_threads=n_threads,
logits_all=True, echo=False)
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
buffer = BytesIO()
pil_image.save(buffer, format="PNG")
image_bytes = buffer.getvalue()
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
response = self.llm.create_chat_completion(
messages=[
{"role": "system", "content": "You are an assistant who perfectly describes images."},
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": base64_string}},
{"type": "text", "text": f"{prompt}"}
]
}
],
temperature=temperature,
)
if unload and self.llm is not None:
self.llm.close()
del self.llm
self.llm = None
gc.collect()
torch.cuda.empty_cache()
if unload and self.clip is not None:
self.clip._exit_stack.close() # info https://github.com/abetlen/llama-cpp-python/issues/1746
del self.clip
self.clip = None
gc.collect()
torch.cuda.empty_cache()
return (f"{response['choices'][0]['message']['content']}", )
class LLavaAdvanced:
def __init__(self):
self.llm = None
self.clip = None
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
"ckpt_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
"clip_name": (folder_paths.get_filename_list("LLavacheckpoints"), ),
"max_ctx": ("INT", {"default": 4096, "min": 128, "max": 128000, "step": 64}),
"gpu_layers": ("INT", {"default": 27, "min": 0, "max": 100, "step": 1}),
"n_threads": ("INT", {"default": 8, "min": 1, "max": 100, "step": 1}),
"max_tokens": ("INT", {"default": 512, "min": 1, "max": 2048, "step": 1}),
"temperature": ("FLOAT", {"default": 0.1, "min": 0.01, "max": 1.0, "step": 0.01}),
"top_p": ("FLOAT", {"default": 0.95, "min": 0.1, "max": 1.0, "step": 0.01}),
"top_k": ("INT", {"default": 40, "step": 1}),
"frequency_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"presence_penalty": ("FLOAT", {"default": 0.0, "step": 0.01}),
"repeat_penalty": ("FLOAT", {"default": 1.1, "step": 0.01}),
"seed": ("INT", {"default": 42, "step": 1}),
"unload": ("BOOLEAN", {"default": False}),
"prompt": ("STRING", {"multiline": True, "default": ""}),
"system_msg": ("STRING", {"multiline": True, "default": "You are an assistant who perfectly describes images."}),
}
}
RETURN_TYPES = ("STRING",)
FUNCTION = "generate_text_full_advanced"
CATEGORY = "LevelPixel/LLava"
def generate_text_full_advanced(self, ckpt_name, clip_name, max_ctx, gpu_layers, n_threads, image,
system_msg, prompt, max_tokens, temperature, top_p, top_k, frequency_penalty,
presence_penalty, repeat_penalty, seed, unload):
clip_path = folder_paths.get_full_path("LLavacheckpoints", clip_name)
self.clip = Llava16ChatHandler(clip_model_path=clip_path, verbose=False)
ckpt_path = folder_paths.get_full_path("LLavacheckpoints", ckpt_name)
self.llm = Llama(model_path = ckpt_path, chat_handler=self.clip, offload_kqv=True, f16_kv=True,
use_mlock=False, embedding=False, n_batch=1024, last_n_tokens_size=1024,
verbose=True, seed=42, n_ctx = max_ctx, n_gpu_layers=gpu_layers, n_threads=n_threads,
logits_all=True, echo=False)
pil_image = ToPILImage()(image[0].permute(2, 0, 1))
buffer = BytesIO()
pil_image.save(buffer, format="PNG")
image_bytes = buffer.getvalue()
base64_string = f"data:image/jpeg;base64,{base64.b64encode(image_bytes).decode('utf-8')}"
response = self.llm.create_chat_completion(
messages=[
{"role": "system", "content": system_msg},
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": base64_string}},
{"type": "text", "text": f"{prompt}"}
]
}
],
max_tokens=max_tokens,
temperature=temperature,
top_p=top_p,
top_k=top_k,
frequency_penalty=frequency_penalty,
presence_penalty=presence_penalty,
repeat_penalty=repeat_penalty,
seed=seed,
)
if unload and self.llm is not None:
self.llm.close()
del self.llm
self.llm = None
gc.collect()
torch.cuda.empty_cache()
if unload and self.clip is not None:
self.clip._exit_stack.close() # info https://github.com/abetlen/llama-cpp-python/issues/1746
del self.clip
self.clip = None
gc.collect()
torch.cuda.empty_cache()
return (f"{response['choices'][0]['message']['content']}", )
NODE_CLASS_MAPPINGS = {
"LLavaLoader|LP": LLavaLoader,
"LLavaClipLoader|LP": LLavaClipLoader,
"LLavaSamplerSimple|LP": LLavaSamplerSimple,
"LLavaSamplerAdvanced|LP": LLavaSamplerAdvanced,
"LLavaSimple|LP": LLavaSimple,
"LLavaAdvanced|LP": LLavaAdvanced,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"LLavaLoader|LP": "LLava Loader [LP]",
"LLavaClipLoader|LP": "LLava Clip Loader [LP]",
"LLavaSamplerSimple|LP": "LLava Sampler Simple [LP]",
"LLavaSamplerAdvanced|LP": "LLava Sampler Advanced [LP]",
"LLavaSimple|LP": "LLava Simple [LP]",
"LLavaAdvanced|LP": "LLava Advanced [LP]",
}
+15
View File
@@ -0,0 +1,15 @@
[project]
name = "comfyui_level_pixel"
description = "Various nodes of the Level Pixel company. Includes convenient advanced nodes for working with images from folders; counting files in a folder; cleaning memory; tag filters. Model Unloader, LLM Unloader (GGUF unloaders), Free memory, Tag Filters, Tag Category Filters, Tag Choice Parser, File counter, Image Loader From Path (with counters)."
version = "1.0.5"
license = { file = "LICENSE" }
dependencies = ["torch>=2.0.1", "torchvision>=0.15.2", "numpy", "matplotlib"]
[project.urls]
Repository = "https://github.com/LevelPixel/ComfyUI-LevelPixel"
[tool.comfy]
PublisherId = "Level Pixel"
DisplayName = "ComfyUI_LevelPixel"
Icon = ""
Models = [{location = "/checkpoints/model.safetensor", model_url = "https://example.com/model.zip"}]
+5
View File
@@ -0,0 +1,5 @@
torch>=2.0.1
torchvision>=0.15.2
pillow>=9.4.0
numpy
matplotlib
+57
View File
@@ -0,0 +1,57 @@
import { app } from "../../../scripts/app.js";
import { ComfyWidgets } from "../../../scripts/widgets.js";
// Displays input text on a node
app.registerExtension({
name: "levelpixel.ShowText",
async beforeRegisterNodeDef(nodeType, nodeData, app) {
if (nodeData.name === "ShowText|LP") {
function populate(text) {
if (this.widgets) {
for (let i = 1; i < this.widgets.length; i++) {
this.widgets[i].onRemove?.();
}
this.widgets.length = 1;
}
const v = [...text];
if (!v[0]) {
v.shift();
}
for (const list of v) {
const w = ComfyWidgets["STRING"](this, "text2", ["STRING", { multiline: true }], app).widget;
w.inputEl.readOnly = true;
w.inputEl.style.opacity = 0.6;
w.value = list;
}
requestAnimationFrame(() => {
const sz = this.computeSize();
if (sz[0] < this.size[0]) {
sz[0] = this.size[0];
}
if (sz[1] < this.size[1]) {
sz[1] = this.size[1];
}
this.onResize?.(sz);
app.graph.setDirtyCanvas(true, false);
});
}
// When the node is executed we will be sent the input text, display this in the widget
const onExecuted = nodeType.prototype.onExecuted;
nodeType.prototype.onExecuted = function (message) {
onExecuted?.apply(this, arguments);
populate.call(this, message.text);
};
const onConfigure = nodeType.prototype.onConfigure;
nodeType.prototype.onConfigure = function () {
onConfigure?.apply(this, arguments);
if (this.widgets_values?.length) {
populate.call(this, this.widgets_values.slice(+this.widgets_values.length > 1));
}
};
}
},
});