Robin Collins
d2871d75fd
- Introduced `__init__.py` for package initialization, dependency checking, and node management. - Added `image_nodes.py` with nodes for model name extraction and image saving with metadata. - Created `install.py` for dependency installation and ComfyUI Manager integration setup. - Included configuration files: `node_config.json`, `.manager_info.json`, and `CHANGELOG.md`. - Established `.gitignore` to exclude auto-generated files and unnecessary directories. - Added `README.md` for comprehensive project documentation and usage instructions. - Implemented `DependencyChecker` and `NodeManager` classes for enhanced functionality. - Updated `FILETREE.md` to reflect project structure and organization.
419 lines
14 KiB
Python
419 lines
14 KiB
Python
"""
|
|
ComfyUI-TechsToolz: A modular collection of ComfyUI custom nodes.
|
|
|
|
This package provides various utility nodes for ComfyUI with consistent TechsToolz branding.
|
|
All nodes are organized into modules for scalability and maintainability.
|
|
|
|
Features:
|
|
- Automatic dependency checking and reporting
|
|
- ComfyUI Manager integration
|
|
- Modular node architecture with enable/disable capabilities
|
|
- Runtime dependency validation
|
|
"""
|
|
|
|
import ast
|
|
import importlib
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from typing import Dict, List, Set, Tuple, Any
|
|
import json
|
|
import logging
|
|
|
|
# Set up logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ComfyUI Manager Integration Configuration
|
|
MANAGER_CONFIG = {
|
|
"package_name": "ComfyUI-TechsToolz",
|
|
"version": "0.0.2",
|
|
"author": "Robin Collins",
|
|
"description": "A collection of image processing and utility nodes for ComfyUI",
|
|
"supports_manager": True,
|
|
"configurable_nodes": True,
|
|
}
|
|
|
|
|
|
class DependencyChecker:
|
|
"""Scans Python files and checks for missing dependencies."""
|
|
|
|
def __init__(self, root_path: Path):
|
|
self.root_path = root_path
|
|
self.python_files = []
|
|
self.dependencies = set()
|
|
self.missing_dependencies = set()
|
|
self.available_dependencies = set()
|
|
|
|
def scan_python_files(self) -> List[Path]:
|
|
"""Find all Python files in the repository."""
|
|
python_files = []
|
|
for file_path in self.root_path.rglob("*.py"):
|
|
# Skip hidden directories and __pycache__
|
|
if any(part.startswith(".") or part == "__pycache__" for part in file_path.parts):
|
|
continue
|
|
python_files.append(file_path)
|
|
|
|
self.python_files = python_files
|
|
return python_files
|
|
|
|
def extract_imports(self, file_path: Path) -> Set[str]:
|
|
"""Extract import statements from a Python file."""
|
|
imports = set()
|
|
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
|
|
tree = ast.parse(content)
|
|
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
# Get the top-level package name
|
|
package = alias.name.split(".")[0]
|
|
imports.add(package)
|
|
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if node.module:
|
|
# Get the top-level package name
|
|
package = node.module.split(".")[0]
|
|
imports.add(package)
|
|
|
|
except (SyntaxError, UnicodeDecodeError) as e:
|
|
logger.warning(f"Could not parse {file_path}: {e}")
|
|
|
|
return imports
|
|
|
|
def check_dependency_availability(self, package_name: str) -> bool:
|
|
"""Check if a package is available for import."""
|
|
try:
|
|
importlib.import_module(package_name)
|
|
return True
|
|
except ImportError:
|
|
return False
|
|
|
|
def scan_all_dependencies(self) -> Dict[str, Any]:
|
|
"""Scan all Python files and collect dependency information."""
|
|
self.scan_python_files()
|
|
|
|
file_dependencies = {}
|
|
all_imports = set()
|
|
|
|
for file_path in self.python_files:
|
|
imports = self.extract_imports(file_path)
|
|
# Filter out built-in modules and relative imports
|
|
filtered_imports = {imp for imp in imports if not self._is_builtin_module(imp) and not imp.startswith(".")}
|
|
file_dependencies[str(file_path.relative_to(self.root_path))] = filtered_imports
|
|
all_imports.update(filtered_imports)
|
|
|
|
self.dependencies = all_imports
|
|
|
|
# Check availability of each dependency
|
|
for dep in all_imports:
|
|
if self.check_dependency_availability(dep):
|
|
self.available_dependencies.add(dep)
|
|
else:
|
|
self.missing_dependencies.add(dep)
|
|
|
|
return {
|
|
"file_dependencies": file_dependencies,
|
|
"all_dependencies": list(all_imports),
|
|
"available_dependencies": list(self.available_dependencies),
|
|
"missing_dependencies": list(self.missing_dependencies),
|
|
"python_files_scanned": len(self.python_files),
|
|
}
|
|
|
|
def _is_builtin_module(self, module_name: str) -> bool:
|
|
"""Check if a module is a built-in Python module."""
|
|
builtin_modules = {
|
|
"os",
|
|
"sys",
|
|
"json",
|
|
"ast",
|
|
"importlib",
|
|
"pathlib",
|
|
"typing",
|
|
"logging",
|
|
"datetime",
|
|
"hashlib",
|
|
"re",
|
|
"collections",
|
|
"itertools",
|
|
"functools",
|
|
"math",
|
|
"random",
|
|
"string",
|
|
"io",
|
|
"copy",
|
|
"pickle",
|
|
"time",
|
|
"threading",
|
|
"multiprocessing",
|
|
"subprocess",
|
|
"tempfile",
|
|
"shutil",
|
|
"glob",
|
|
"fnmatch",
|
|
"linecache",
|
|
"operator",
|
|
"keyword",
|
|
"heapq",
|
|
"bisect",
|
|
"array",
|
|
"struct",
|
|
"codecs",
|
|
"locale",
|
|
"gettext",
|
|
"argparse",
|
|
"configparser",
|
|
"urllib",
|
|
"http",
|
|
"socketserver",
|
|
"xmlrpc",
|
|
"email",
|
|
"mailbox",
|
|
"mimetypes",
|
|
"base64",
|
|
"binascii",
|
|
"quopri",
|
|
"uu",
|
|
"html",
|
|
"xml",
|
|
"webbrowser",
|
|
"cgi",
|
|
"cgitb",
|
|
"wsgiref",
|
|
"uuid",
|
|
"zlib",
|
|
"gzip",
|
|
"bz2",
|
|
"lzma",
|
|
"zipfile",
|
|
"tarfile",
|
|
"csv",
|
|
"sqlite3",
|
|
"dbm",
|
|
"pickle",
|
|
"copyreg",
|
|
"shelve",
|
|
"marshal",
|
|
"warnings",
|
|
"contextlib",
|
|
"abc",
|
|
"atexit",
|
|
"traceback",
|
|
"inspect",
|
|
"dis",
|
|
"pdb",
|
|
"profile",
|
|
"pstats",
|
|
"timeit",
|
|
"trace",
|
|
"faulthandler",
|
|
"tracemalloc",
|
|
}
|
|
return module_name in builtin_modules
|
|
|
|
def generate_report(self) -> str:
|
|
"""Generate a human-readable dependency report."""
|
|
report_lines = [
|
|
"=" * 60,
|
|
"ComfyUI-TechsToolz Dependency Report",
|
|
"=" * 60,
|
|
f"Python files scanned: {len(self.python_files)}",
|
|
f"Total unique dependencies: {len(self.dependencies)}",
|
|
f"Available dependencies: {len(self.available_dependencies)}",
|
|
f"Missing dependencies: {len(self.missing_dependencies)}",
|
|
"",
|
|
]
|
|
|
|
if self.available_dependencies:
|
|
report_lines.extend(["✅ Available Dependencies:", "-" * 30])
|
|
for dep in sorted(self.available_dependencies):
|
|
report_lines.append(f" ✓ {dep}")
|
|
report_lines.append("")
|
|
|
|
if self.missing_dependencies:
|
|
report_lines.extend(["❌ Missing Dependencies:", "-" * 30])
|
|
for dep in sorted(self.missing_dependencies):
|
|
report_lines.append(f" ✗ {dep}")
|
|
|
|
report_lines.extend(["", "Installation suggestions:", f"pip install {' '.join(sorted(self.missing_dependencies))}"])
|
|
|
|
report_lines.append("=" * 60)
|
|
return "\n".join(report_lines)
|
|
|
|
|
|
class NodeManager:
|
|
"""Manages node loading and ComfyUI Manager integration."""
|
|
|
|
def __init__(self):
|
|
self.enabled_modules = set()
|
|
self.disabled_modules = set()
|
|
self.node_classes = {}
|
|
self.node_display_names = {}
|
|
self.config_file = Path(__file__).parent / "node_config.json"
|
|
self.load_config()
|
|
|
|
def load_config(self):
|
|
"""Load node configuration from file."""
|
|
if self.config_file.exists():
|
|
try:
|
|
with open(self.config_file, "r") as f:
|
|
config = json.load(f)
|
|
self.enabled_modules = set(config.get("enabled_modules", []))
|
|
self.disabled_modules = set(config.get("disabled_modules", []))
|
|
except (json.JSONDecodeError, FileNotFoundError):
|
|
logger.warning("Could not load node configuration, using defaults")
|
|
self.enabled_modules = {"image_nodes"} # Default enabled
|
|
|
|
def save_config(self):
|
|
"""Save current configuration to file."""
|
|
config = {
|
|
"enabled_modules": list(self.enabled_modules),
|
|
"disabled_modules": list(self.disabled_modules),
|
|
"package_info": MANAGER_CONFIG,
|
|
}
|
|
|
|
try:
|
|
with open(self.config_file, "w") as f:
|
|
json.dump(config, f, indent=2)
|
|
except Exception as e:
|
|
logger.error(f"Could not save configuration: {e}")
|
|
|
|
def enable_module(self, module_name: str):
|
|
"""Enable a specific node module."""
|
|
self.enabled_modules.add(module_name)
|
|
self.disabled_modules.discard(module_name)
|
|
self.save_config()
|
|
logger.info(f"Enabled module: {module_name}")
|
|
|
|
def disable_module(self, module_name: str):
|
|
"""Disable a specific node module."""
|
|
self.disabled_modules.add(module_name)
|
|
self.enabled_modules.discard(module_name)
|
|
self.save_config()
|
|
logger.info(f"Disabled module: {module_name}")
|
|
|
|
def is_module_enabled(self, module_name: str) -> bool:
|
|
"""Check if a module is enabled."""
|
|
# If no specific configuration, default to enabled
|
|
if not self.enabled_modules and not self.disabled_modules:
|
|
return True
|
|
return module_name in self.enabled_modules and module_name not in self.disabled_modules
|
|
|
|
def load_module_nodes(self, module_name: str, module_path: str) -> Tuple[Dict, Dict]:
|
|
"""Load nodes from a specific module if enabled."""
|
|
if not self.is_module_enabled(module_name):
|
|
logger.info(f"Module {module_name} is disabled, skipping...")
|
|
return {}, {}
|
|
|
|
try:
|
|
# Import the module
|
|
spec = importlib.util.spec_from_file_location(module_name, module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
# Get node classes and display names
|
|
node_classes = getattr(module, "NODE_CLASSES", {})
|
|
node_display_names = getattr(module, "NODE_DISPLAY_NAMES", {})
|
|
|
|
logger.info(f"Loaded {len(node_classes)} nodes from {module_name}")
|
|
return node_classes, node_display_names
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load module {module_name}: {e}")
|
|
return {}, {}
|
|
|
|
def register_all_nodes(self):
|
|
"""Register all enabled node modules."""
|
|
base_path = Path(__file__).parent
|
|
|
|
# Define available modules
|
|
available_modules = {
|
|
"image_nodes": base_path / "image_nodes.py",
|
|
# Future modules can be added here:
|
|
# 'utility_nodes': base_path / 'utility_nodes.py',
|
|
# 'workflow_nodes': base_path / 'workflow_nodes.py',
|
|
}
|
|
|
|
for module_name, module_path in available_modules.items():
|
|
if module_path.exists():
|
|
node_classes, node_display_names = self.load_module_nodes(module_name, str(module_path))
|
|
self.node_classes.update(node_classes)
|
|
self.node_display_names.update(node_display_names)
|
|
|
|
return self.node_classes, self.node_display_names
|
|
|
|
|
|
# Initialize systems
|
|
def initialize_techstoolz():
|
|
"""Initialize ComfyUI-TechsToolz with dependency checking and node management."""
|
|
print("\n🖥️🛠 Initializing ComfyUI-TechsToolz...")
|
|
|
|
# Run dependency check
|
|
root_path = Path(__file__).parent
|
|
dependency_checker = DependencyChecker(root_path)
|
|
dependency_info = dependency_checker.scan_all_dependencies()
|
|
|
|
# Print dependency report
|
|
report = dependency_checker.generate_report()
|
|
print(report)
|
|
|
|
# Check for critical missing dependencies
|
|
if dependency_checker.missing_dependencies:
|
|
logger.warning("Some dependencies are missing. The affected nodes may not work correctly.")
|
|
missing_deps = list(dependency_checker.missing_dependencies)
|
|
logger.warning(f"Missing: {', '.join(missing_deps)}")
|
|
|
|
# Initialize node manager
|
|
node_manager = NodeManager()
|
|
node_classes, node_display_names = node_manager.register_all_nodes()
|
|
|
|
print(f"✅ Successfully loaded {len(node_classes)} nodes from ComfyUI-TechsToolz")
|
|
|
|
# ComfyUI Manager integration info
|
|
if MANAGER_CONFIG["supports_manager"]:
|
|
print("🔧 ComfyUI Manager integration: ENABLED")
|
|
print(" - Node enable/disable: Available")
|
|
print(" - Configuration management: Available")
|
|
|
|
return node_classes, node_display_names, dependency_info
|
|
|
|
|
|
# Initialize everything
|
|
NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS, DEPENDENCY_INFO = initialize_techstoolz()
|
|
|
|
# Export for ComfyUI Manager
|
|
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "MANAGER_CONFIG", "DEPENDENCY_INFO"]
|
|
|
|
|
|
# ComfyUI Manager API functions
|
|
def get_node_list():
|
|
"""Get list of available nodes for ComfyUI Manager."""
|
|
return list(NODE_CLASS_MAPPINGS.keys())
|
|
|
|
|
|
def get_package_info():
|
|
"""Get package information for ComfyUI Manager."""
|
|
return MANAGER_CONFIG
|
|
|
|
|
|
def enable_node_module(module_name: str):
|
|
"""Enable a node module (ComfyUI Manager integration)."""
|
|
node_manager = NodeManager()
|
|
node_manager.enable_module(module_name)
|
|
return f"Module {module_name} enabled"
|
|
|
|
|
|
def disable_node_module(module_name: str):
|
|
"""Disable a node module (ComfyUI Manager integration)."""
|
|
node_manager = NodeManager()
|
|
node_manager.disable_module(module_name)
|
|
return f"Module {module_name} disabled"
|
|
|
|
|
|
def get_dependency_report():
|
|
"""Get current dependency status report."""
|
|
return DEPENDENCY_INFO
|