Implementation of a basic Logging/Debugging System

This commit is contained in:
gmorks
2025-01-30 09:03:31 -06:00
parent 51612bdbfd
commit 998a2d277f
5 changed files with 281 additions and 23 deletions
+2 -1
View File
@@ -1,2 +1,3 @@
__pycache__/
To-Do.md
To-Do.md
logs/
+74 -8
View File
@@ -3,11 +3,10 @@ Main node implementation for Animagine Prompt
"""
import os
import time
from typing import Tuple, Dict, Any, List, Optional
import logging
from .csv_handler import CSVHandler
logger = logging.getLogger('AnimaginePrompt')
from .logger import logger
class AnimaginePromptNode:
"""
@@ -32,6 +31,7 @@ class AnimaginePromptNode:
def __init__(self):
self.csv_handler = CSVHandler()
self.last_csv_path = None
logger.logger.info("AnimaginePromptNode initialized")
@classmethod
def INPUT_TYPES(cls) -> Dict[str, Dict[str, Any]]:
@@ -72,7 +72,10 @@ class AnimaginePromptNode:
tags.append(quality_good)
if score_good:
tags.append(score_good)
return ", ".join(tags)
result = ", ".join(tags)
logger.logger.debug(f"Built good tags: {result}")
return result
def _build_bad_tags(self, quality_bad: str, score_bad: str) -> str:
"""Construye la sección de tags negativas"""
@@ -81,17 +84,35 @@ class AnimaginePromptNode:
tags.append(quality_bad)
if score_bad:
tags.append(score_bad)
return ", ".join(tags)
result = ", ".join(tags)
logger.logger.debug(f"Built bad tags: {result}")
return result
def _get_csv_entry(self, csv_path: str, index: int) -> Optional[str]:
"""Obtiene y formatea una entrada del CSV"""
start_time = time.time()
if self.last_csv_path != csv_path:
logger.logger.info(f"Loading new CSV file: {csv_path}")
self.csv_handler.load_csv(csv_path)
self.last_csv_path = csv_path
entry = self.csv_handler.get_entry(index)
end_time = time.time()
logger.log_performance(
"CSV Entry Retrieval",
start_time,
end_time,
{"path": csv_path, "index": index}
)
if entry:
return f"{entry['gender']}, {entry['character']}, {entry['copyright']}"
formatted_entry = f"{entry['gender']}, {entry['character']}, {entry['copyright']}"
logger.logger.debug(f"Formatted CSV entry: {formatted_entry}")
return formatted_entry
return None
def generate_prompt(
@@ -115,7 +136,22 @@ class AnimaginePromptNode:
"""
Genera los prompts finales basados en los parámetros proporcionados
"""
start_time = time.time()
try:
# Log initial state
logger.log_prompt_generation(
positive_prompt,
negative_prompt,
{
"use_good_tags": use_good_tags,
"use_bad_tags": use_bad_tags,
"use_year": use_year,
"use_rating": use_rating,
"use_csv": use_csv
}
)
# Lista para construir el prompt positivo
positive_parts = []
@@ -124,47 +160,77 @@ class AnimaginePromptNode:
good_tags = self._build_good_tags(quality_good, score_good)
if good_tags:
positive_parts.append(good_tags)
logger.logger.debug(f"Added good tags: {good_tags}")
# Agregar bad tags al prompt positivo si están activadas
if use_bad_tags:
bad_tags = self._build_bad_tags(quality_bad, score_bad)
if bad_tags:
positive_parts.append(bad_tags)
logger.logger.debug(f"Added bad tags: {bad_tags}")
# Agregar entrada del CSV si está activado
if use_csv and csv_path:
csv_entry = self._get_csv_entry(csv_path, csv_index)
if csv_entry:
positive_parts.append(csv_entry)
logger.logger.debug(f"Added CSV entry: {csv_entry}")
# Agregar el prompt base del usuario
if positive_prompt:
positive_parts.append(positive_prompt)
logger.logger.debug(f"Added base positive prompt: {positive_prompt}")
# Agregar year tag si está activado
if use_year:
positive_parts.append(f"year {year}")
year_tag = f"year {year}"
positive_parts.append(year_tag)
logger.logger.debug(f"Added year tag: {year_tag}")
# Agregar rating si está activado
if use_rating:
positive_parts.append(rating)
logger.logger.debug(f"Added rating: {rating}")
# Construir el prompt negativo
negative_parts = []
# Agregar tags negativas por defecto
negative_parts.extend(self.DEFAULT_NEGATIVE_TAGS)
logger.logger.debug("Added default negative tags")
# Agregar el prompt negativo del usuario
if negative_prompt:
negative_parts.append(negative_prompt)
logger.logger.debug(f"Added user negative prompt: {negative_prompt}")
# Unir todas las partes
final_positive = ", ".join(part for part in positive_parts if part)
final_negative = ", ".join(part for part in negative_parts if part)
end_time = time.time()
logger.log_performance(
"Prompt Generation",
start_time,
end_time,
{
"positive_parts": len(positive_parts),
"negative_parts": len(negative_parts)
}
)
logger.logger.info("Prompt generation completed successfully")
return final_positive, final_negative
except Exception as e:
logger.error(f"Error generating prompt: {str(e)}")
logger.log_error(
e,
"Prompt Generation",
{
"positive_prompt": positive_prompt,
"negative_prompt": negative_prompt,
"use_csv": use_csv,
"csv_path": csv_path
}
)
raise e
+65 -14
View File
@@ -3,18 +3,17 @@ CSV handling functionality for Animagine Prompt Node
"""
import pandas as pd
import logging
from typing import Optional, List, Dict
from typing import Optional, Dict
import os
logger = logging.getLogger('AnimaginePrompt')
import time
from .logger import logger
class CSVHandler:
"""
Handles loading and processing of CSV files containing character data
"""
REQUIRED_COLUMNS = ['GENDER', 'CHARACTER', 'COPYRIGHT'] # Notar las mayúsculas
REQUIRED_COLUMNS = ['GENDER', 'CHARACTER', 'COPYRIGHT']
def __init__(self):
self.current_data = None
@@ -26,21 +25,33 @@ class CSVHandler:
"""
try:
if not os.path.exists(filepath):
logger.error(f"CSV file not found: {filepath}")
logger.log_error(
FileNotFoundError(f"CSV file not found: {filepath}"),
"CSV Validation",
{"filepath": filepath}
)
return False
df = pd.read_csv(filepath)
logger.info(f"CSV columns found: {df.columns.tolist()}")
logger.log_csv_operation(
"Validation",
filepath,
{"columns_found": df.columns.tolist()}
)
missing_columns = [col for col in self.REQUIRED_COLUMNS if col not in df.columns]
if missing_columns:
logger.error(f"Missing required columns: {missing_columns}")
logger.log_error(
ValueError("Missing required columns"),
"CSV Validation",
{"missing_columns": missing_columns}
)
return False
return True
except Exception as e:
logger.error(f"Error validating CSV structure: {str(e)}")
logger.log_error(e, "CSV Validation", {"filepath": filepath})
return False
def load_csv(self, filepath: str) -> Optional[pd.DataFrame]:
@@ -48,20 +59,39 @@ class CSVHandler:
Loads CSV file with basic caching
"""
try:
start_time = time.time()
if filepath in self.cache:
logger.log_csv_operation(
"Cache Hit",
filepath,
{"cache_size": len(self.cache)}
)
return self.cache[filepath]
if not self.validate_csv_structure(filepath):
logger.error(f"Invalid CSV structure in file: {filepath}")
return None
df = pd.read_csv(filepath)
self.cache[filepath] = df
self.current_data = df
end_time = time.time()
logger.log_performance(
"CSV Load",
start_time,
end_time,
{
"filepath": filepath,
"rows": len(df),
"cached": True
}
)
return df
except Exception as e:
logger.error(f"Error loading CSV file: {str(e)}")
logger.log_error(e, "CSV Load", {"filepath": filepath})
return None
def get_entry(self, index: int) -> Optional[Dict[str, str]]:
@@ -70,18 +100,39 @@ class CSVHandler:
"""
try:
if self.current_data is None:
logger.log_error(
ValueError("No CSV data loaded"),
"Get Entry",
{"index": index}
)
return None
if index >= len(self.current_data):
logger.error(f"Index {index} out of bounds for CSV with {len(self.current_data)} entries")
logger.log_error(
IndexError(f"Index {index} out of bounds"),
"Get Entry",
{
"index": index,
"max_index": len(self.current_data) - 1
}
)
return None
row = self.current_data.iloc[index]
return {
entry = {
'gender': row['GENDER'],
'character': row['CHARACTER'],
'copyright': row['COPYRIGHT']
}
logger.log_csv_operation(
"Get Entry",
None,
{"index": index, "entry": entry}
)
return entry
except Exception as e:
logger.error(f"Error getting CSV entry: {str(e)}")
logger.log_error(e, "Get Entry", {"index": index})
return None
+132
View File
@@ -0,0 +1,132 @@
"""
logger.py - Logging system for ComfyUI Animagine Prompt Node
"""
import logging
import os
from datetime import datetime
from typing import Optional, Dict, Any
class AnimagineLogger:
"""
Custom logger for the Animagine Prompt Node with enhanced debugging capabilities
"""
def __init__(self, log_dir: str = "logs"):
self.log_dir = log_dir
self.logger = None
self.setup_logger()
def setup_logger(self) -> None:
"""
Initialize the logger with both file and console handlers
"""
# Create logs directory if it doesn't exist
if not os.path.exists(self.log_dir):
os.makedirs(self.log_dir)
# Create logger
self.logger = logging.getLogger('AnimaginePrompt')
self.logger.setLevel(logging.DEBUG)
# Remove existing handlers if any
if self.logger.handlers:
self.logger.handlers.clear()
# File handler - detailed logging
log_file = os.path.join(
self.log_dir,
f"animagine_prompt_{datetime.now().strftime('%Y%m%d')}.log"
)
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
file_formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(file_formatter)
self.logger.addHandler(file_handler)
# Console handler - important messages only
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_formatter = logging.Formatter(
'%(levelname)s - %(message)s'
)
console_handler.setFormatter(console_formatter)
self.logger.addHandler(console_handler)
def log_prompt_generation(
self,
positive_prompt: str,
negative_prompt: str,
params: Dict[str, Any]
) -> None:
"""
Log prompt generation details
"""
self.logger.debug("=== Prompt Generation Started ===")
self.logger.debug(f"Parameters used: {params}")
self.logger.debug(f"Base positive prompt: {positive_prompt}")
self.logger.debug(f"Base negative prompt: {negative_prompt}")
def log_csv_operation(
self,
operation: str,
filepath: Optional[str] = None,
details: Optional[Dict[str, Any]] = None
) -> None:
"""
Log CSV-related operations
"""
self.logger.debug(f"=== CSV Operation: {operation} ===")
if filepath:
self.logger.debug(f"File: {filepath}")
if details:
self.logger.debug(f"Details: {details}")
def log_error(
self,
error: Exception,
context: str,
details: Optional[Dict[str, Any]] = None
) -> None:
"""
Log errors with context and details
"""
self.logger.error(f"=== Error in {context} ===")
self.logger.error(f"Error type: {type(error).__name__}")
self.logger.error(f"Error message: {str(error)}")
if details:
self.logger.error(f"Additional details: {details}")
def log_performance(
self,
operation: str,
start_time: float,
end_time: float,
details: Optional[Dict[str, Any]] = None
) -> None:
"""
Log performance metrics
"""
duration = end_time - start_time
self.logger.debug(f"=== Performance: {operation} ===")
self.logger.debug(f"Duration: {duration:.4f} seconds")
if details:
self.logger.debug(f"Details: {details}")
def log_state_change(
self,
component: str,
old_state: Any,
new_state: Any
) -> None:
"""
Log state changes in the node
"""
self.logger.debug(f"=== State Change: {component} ===")
self.logger.debug(f"Old state: {old_state}")
self.logger.debug(f"New state: {new_state}")
# Singleton instance
logger = AnimagineLogger()
+8
View File
@@ -0,0 +1,8 @@
2025-01-30 08:55:54,686 - AnimaginePrompt - DEBUG - === CSV Operation: Validation ===
2025-01-30 08:55:54,687 - AnimaginePrompt - DEBUG - File: C:\Users\Alex\Desktop\anime_prompt\ComfyUI\custom_nodes\ComfyUI-Animagine-prompt\example_csv\example.csv
2025-01-30 08:55:54,687 - AnimaginePrompt - DEBUG - Details: {'columns_found': ['GENDER', 'CHARACTER', 'COPYRIGHT']}
2025-01-30 08:55:54,688 - AnimaginePrompt - DEBUG - === Performance: CSV Load ===
2025-01-30 08:55:54,689 - AnimaginePrompt - DEBUG - Duration: 0.0150 seconds
2025-01-30 08:55:54,689 - AnimaginePrompt - DEBUG - Details: {'filepath': 'C:\\Users\\Alex\\Desktop\\anime_prompt\\ComfyUI\\custom_nodes\\ComfyUI-Animagine-prompt\\example_csv\\example.csv', 'rows': 23, 'cached': True}
2025-01-30 08:55:54,689 - AnimaginePrompt - DEBUG - === CSV Operation: Get Entry ===
2025-01-30 08:55:54,690 - AnimaginePrompt - DEBUG - Details: {'index': 4, 'entry': {'gender': '1girl', 'character': 'mokota mememe', 'copyright': '.live'}}