nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
4dbc553b0412c889446fb859eb552b1a4e0373f0
ComfyAssets-ComfyUI_PromptM…/utils/diagnostics.py
T
Vito Sansevero 22593e8514 docs: add comprehensive docstrings to all Python modules 
- Add Google-style docstrings to main node files (prompt_manager.py, prompt_manager_text.py)
- Enhance database module docstrings with detailed parameter and return documentation
- Add comprehensive docstrings to all utils modules with usage examples
- Document API endpoints and configuration classes thoroughly
- Update module-level docstrings in all __init__.py files
- Ensure consistent naming (ComfyUI_PromptManager) across all documentation
- Follow Python best practices for docstring formatting
2025-08-12 06:51:36 -07:00

453 lines
17 KiB
Python
Raw Blame History

"""Diagnostic utilities for troubleshooting the PromptManager gallery system.
This module provides comprehensive diagnostic tools to help identify and resolve
issues with the PromptManager system, particularly the gallery and image monitoring
functionality. It performs systematic checks of all system components and provides
clear feedback on potential problems.
The diagnostic system checks:
- Database connectivity and schema integrity
- Image table structure and data
- File system permissions and access
- ComfyUI output directory detection
- Required Python dependencies
- System integration points
Typical usage:
from utils.diagnostics import GalleryDiagnostics, run_diagnostics
# Run full diagnostic suite
results = run_diagnostics()
# Or create custom diagnostic instance
diagnostics = GalleryDiagnostics("custom_db.db")
database_status = diagnostics.check_database()
The diagnostics provide:
- Clear pass/fail status for each component
- Detailed error messages and remediation suggestions
- Statistics and system information
- Test utilities for verifying functionality
"""
import os
import sqlite3
from pathlib import Path
from typing import Dict, Any, List
from .logging_config import get_logger
class GalleryDiagnostics:
"""Diagnostics for the gallery system.
This class provides comprehensive diagnostic capabilities for the PromptManager
gallery system. It systematically checks all components and dependencies to
identify potential issues and provide actionable feedback.
The diagnostics cover:
- Database structure and connectivity
- Image tracking table integrity
- File system access and permissions
- ComfyUI integration points
- Python dependency availability
Each diagnostic method returns a standardized result dictionary with:
- status: 'ok', 'warning', or 'error'
- message: Descriptive message (for warnings/errors)
- Additional data specific to the diagnostic
"""
def __init__(self, db_path: str = "prompts.db"):
"""Initialize the diagnostics system.
Args:
db_path: Path to the SQLite database file to diagnose
"""
self.db_path = db_path
self.logger = get_logger('prompt_manager.diagnostics')
def run_full_diagnostic(self) -> Dict[str, Any]:
"""Run a complete diagnostic check.
Executes all diagnostic checks in sequence and provides a comprehensive
report of system status. Logs detailed information during the process.
Returns:
Dictionary mapping diagnostic categories to their results:
- database: Database connectivity and structure check
- images_table: Image tracking table specific check
- file_system: File system access and permissions check
- comfyui_output: ComfyUI output directory detection
- dependencies: Python dependency availability check
"""
self.logger.info("\n" + "="*60)
self.logger.info("[DIAG] PROMPTMANAGER GALLERY DIAGNOSTICS")
self.logger.info("="*60)
results = {
'database': self.check_database(),
'images_table': self.check_images_table(),
'file_system': self.check_file_system(),
'comfyui_output': self.check_comfyui_output(),
'dependencies': self.check_dependencies()
}
self.logger.info("\n" + "="*60)
self.logger.info("[SUMMARY] DIAGNOSTIC SUMMARY")
self.logger.info("="*60)
for category, result in results.items():
status = "[PASS] PASS" if result['status'] == 'ok' else "[FAIL] FAIL"
self.logger.info(f"{category.upper():<20} {status}")
if result['status'] != 'ok':
self.logger.warning(f" Issue: {result['message']}")
self.logger.info("\n" + "="*60)
return results
def check_database(self) -> Dict[str, Any]:
"""Check database connection and structure.
Verifies that the database file exists, is accessible, and contains
the expected prompt table structure.
Returns:
Dictionary containing:
- status: 'ok' or 'error'
- message: Error description (if status is 'error')
- prompt_count: Number of prompts in database (if successful)
- has_images_table: Whether generated_images table exists (if successful)
"""
self.logger.info("\n[DB] Checking Database...")
try:
if not os.path.exists(self.db_path):
return {
'status': 'error',
'message': f'Database file not found: {self.db_path}'
}
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
# Check prompts table
cursor = conn.execute("SELECT COUNT(*) as count FROM prompts")
prompt_count = cursor.fetchone()['count']
self.logger.info(f" [NOTE] Prompts in database: {prompt_count}")
# Check if generated_images table exists
cursor = conn.execute("""
SELECT name FROM sqlite_master
WHERE type='table' AND name='generated_images'
""")
has_images_table = cursor.fetchone() is not None
self.logger.info(f" [IMG] Images table exists: {has_images_table}")
return {
'status': 'ok',
'prompt_count': prompt_count,
'has_images_table': has_images_table
}
except Exception as e:
return {
'status': 'error',
'message': f'Database error: {str(e)}'
}
def check_images_table(self) -> Dict[str, Any]:
"""Check the generated_images table specifically.
Examines the generated_images table structure and content to verify
the gallery system can function properly.
Returns:
Dictionary containing:
- status: 'ok' or 'error'
- message: Error description (if table missing or inaccessible)
- image_count: Number of images in table (if successful)
- recent_images: List of recent image records (if successful)
"""
self.logger.info("\n[IMG] Checking Images Table...")
try:
with sqlite3.connect(self.db_path) as conn:
conn.row_factory = sqlite3.Row
# Check if table exists
cursor = conn.execute("""
SELECT name FROM sqlite_master
WHERE type='table' AND name='generated_images'
""")
if not cursor.fetchone():
return {
'status': 'error',
'message': 'generated_images table does not exist - run the updated code to create it'
}
# Check image records
cursor = conn.execute("SELECT COUNT(*) as count FROM generated_images")
image_count = cursor.fetchone()['count']
self.logger.info(f" [STATS] Images in database: {image_count}")
# Get recent images
cursor = conn.execute("""
SELECT gi.*, p.text
FROM generated_images gi
LEFT JOIN prompts p ON gi.prompt_id = p.id
ORDER BY gi.generation_time DESC
LIMIT 5
""")
recent_images = [dict(row) for row in cursor.fetchall()]
self.logger.info(f" [TIME] Recent images: {len(recent_images)}")
for img in recent_images:
self.logger.info(f" - {img['filename']} -> Prompt {img['prompt_id']}")
return {
'status': 'ok',
'image_count': image_count,
'recent_images': recent_images
}
except Exception as e:
return {
'status': 'error',
'message': f'Images table error: {str(e)}'
}
def check_file_system(self) -> Dict[str, Any]:
"""Check file system and permissions.
Verifies that the application has appropriate file system access
for reading images and writing database files.
Returns:
Dictionary containing:
- status: 'ok' or 'error'
- message: Error description (if issues detected)
- current_dir: Current working directory path
- can_write: Whether write access is available
"""
self.logger.info("\n[DIR] Checking File System...")
try:
# Check current directory
current_dir = os.getcwd()
self.logger.info(f" [FOLDER] Current directory: {current_dir}")
# Check if we can write to current directory
test_file = "test_write.tmp"
try:
with open(test_file, 'w') as f:
f.write("test")
os.remove(test_file)
can_write = True
except:
can_write = False
self.logger.info(f" [EDIT] Can write to directory: {can_write}")
return {
'status': 'ok',
'current_dir': current_dir,
'can_write': can_write
}
except Exception as e:
return {
'status': 'error',
'message': f'File system error: {str(e)}'
}
def check_comfyui_output(self) -> Dict[str, Any]:
"""Check ComfyUI output directories.
Attempts to locate ComfyUI output directories where generated images
would be stored. Checks both common locations and ComfyUI's configured paths.
Returns:
Dictionary containing:
- status: 'ok' if directories found, 'warning' if none found
- message: Warning message (if no directories found)
- output_dirs: List of detected output directory paths
"""
self.logger.info("\n[STYLE] Checking ComfyUI Output...")
output_dirs = []
# Try to detect ComfyUI output directories
potential_dirs = [
"output",
"../output",
"../../output",
"ComfyUI/output",
"../ComfyUI/output",
"../../ComfyUI/output"
]
for dir_path in potential_dirs:
abs_path = os.path.abspath(dir_path)
if os.path.exists(abs_path):
output_dirs.append(abs_path)
self.logger.info(f" [DIR] Found output dir: {abs_path}")
# Count images in this directory
try:
image_files = []
for ext in ['.png', '.jpg', '.jpeg', '.webp']:
image_files.extend(Path(abs_path).rglob(f'*{ext}'))
self.logger.info(f" [IMG] Images found: {len(image_files)}")
except Exception as e:
self.logger.error(f" [FAIL] Error scanning: {e}")
# Try ComfyUI's folder_paths
try:
import folder_paths
comfyui_output = folder_paths.get_output_directory()
if comfyui_output and comfyui_output not in output_dirs:
output_dirs.append(comfyui_output)
self.logger.info(f" [DIR] ComfyUI output dir: {comfyui_output}")
except ImportError:
self.logger.warning(" [WARN] ComfyUI folder_paths not available")
return {
'status': 'ok' if output_dirs else 'warning',
'message': 'No output directories found' if not output_dirs else None,
'output_dirs': output_dirs
}
def check_dependencies(self) -> Dict[str, Any]:
"""Check required dependencies.
Verifies that all required Python packages are available for the
PromptManager system to function properly.
Returns:
Dictionary containing:
- status: 'ok' if all dependencies available, 'error' if any missing
- message: Error description (if dependencies missing)
- dependencies: Dictionary mapping package names to availability status
"""
self.logger.info("\n[PKG] Checking Dependencies...")
dependencies = {
'watchdog': False,
'PIL': False,
'sqlite3': False
}
# Check watchdog
try:
import watchdog
dependencies['watchdog'] = True
self.logger.info(f" [PASS] watchdog: {watchdog.__version__}")
except ImportError:
self.logger.error(" [FAIL] watchdog: NOT INSTALLED")
# Check PIL
try:
from PIL import Image
dependencies['PIL'] = True
self.logger.info(f" [PASS] PIL (Pillow): Available")
except ImportError:
self.logger.error(" [FAIL] PIL (Pillow): NOT AVAILABLE")
# Check sqlite3
try:
import sqlite3
dependencies['sqlite3'] = True
self.logger.info(f" [PASS] sqlite3: {sqlite3.sqlite_version}")
except ImportError:
self.logger.error(" [FAIL] sqlite3: NOT AVAILABLE")
all_deps_ok = all(dependencies.values())
return {
'status': 'ok' if all_deps_ok else 'error',
'message': 'Missing dependencies' if not all_deps_ok else None,
'dependencies': dependencies
}
def create_test_image_link(self, prompt_id: int, test_image_path: str = None) -> Dict[str, Any]:
"""Create a test image link to verify the system works.
Creates a test entry in the generated_images table to verify that
the image linking functionality is working correctly.
Args:
prompt_id: ID of an existing prompt to link the test image to
test_image_path: Optional path for the test image (uses fake path if None)
Returns:
Dictionary containing:
- status: 'ok' if test link created successfully, 'error' otherwise
- message: Success or error message
- image_id: ID of created test image record (if successful)
"""
self.logger.info(f"\n[TEST] Creating test image link for prompt {prompt_id}...")
try:
# Import database operations
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from database.operations import PromptDatabase
db = PromptDatabase()
# Create a fake image path if none provided
if not test_image_path:
test_image_path = "/fake/test/image.png"
# Create test metadata
test_metadata = {
'file_info': {
'size': 1024000,
'dimensions': [512, 512],
'format': 'PNG'
},
'workflow': {'test': True},
'prompt': {'test_prompt': 'This is a test'}
}
# Link the test image
image_id = db.link_image_to_prompt(
prompt_id=str(prompt_id),
image_path=test_image_path,
metadata=test_metadata
)
self.logger.info(f" [PASS] Test image linked with ID: {image_id}")
return {
'status': 'ok',
'image_id': image_id,
'message': f'Test image linked successfully with ID {image_id}'
}
except Exception as e:
return {
'status': 'error',
'message': f'Failed to create test link: {str(e)}'
}
def run_diagnostics() -> Dict[str, Any]:
"""Run diagnostics from command line.
Convenience function to create a GalleryDiagnostics instance and run
the full diagnostic suite with default settings.
Returns:
Complete diagnostic results dictionary
"""
diagnostics = GalleryDiagnostics()
return diagnostics.run_full_diagnostic()
if __name__ == "__main__":
run_diagnostics()
Reference in New Issue View Git Blame Copy Permalink