Initial commit
This commit is contained in:
@@ -0,0 +1,63 @@
|
||||
import re
|
||||
from typing import Dict
|
||||
import pandas as pd
|
||||
|
||||
def increment_count(d: Dict[str, int], category: str) -> None:
|
||||
"""Increment the count of a category in the dictionary.
|
||||
|
||||
Args:
|
||||
d (Dict[str, int]): Dictionary storing category counts.
|
||||
category (str): The category to increment.
|
||||
"""
|
||||
d[category] = d.get(category, 0) + 1
|
||||
|
||||
def process_import_lines(file_path: str) -> Dict[str, int]:
|
||||
"""Extract and categorize import lines from a Python file.
|
||||
|
||||
Args:
|
||||
file_path (str): Path to the Python file.
|
||||
|
||||
Returns:
|
||||
Dict[str, int]: Dictionary containing category counts.
|
||||
"""
|
||||
category_counts = {}
|
||||
|
||||
with open(file_path, "r") as f:
|
||||
for line in f:
|
||||
if "import" in line:
|
||||
module_name = re.sub(r"^.*import ", "", line).strip()
|
||||
if module_name.startswith("SNS"):
|
||||
increment_count(category_counts, "Seaborn")
|
||||
elif module_name.startswith("MPL"):
|
||||
increment_count(category_counts, "Matplotlib")
|
||||
elif module_name.startswith("Pt"):
|
||||
increment_count(category_counts, "PyTorch")
|
||||
elif module_name.startswith("Pandas"):
|
||||
increment_count(category_counts, "Pandas")
|
||||
elif module_name.startswith(("CDA", "Python")):
|
||||
increment_count(category_counts, "Misc")
|
||||
elif module_name.startswith("Numpy"):
|
||||
increment_count(category_counts, "NumPy")
|
||||
else:
|
||||
increment_count(category_counts, "Misc")
|
||||
|
||||
return category_counts
|
||||
|
||||
def save_category_counts_to_csv(category_counts: Dict[str, int], output_file: str) -> None:
|
||||
"""Save category counts as a sorted CSV file.
|
||||
|
||||
Args:
|
||||
category_counts (Dict[str, int]): Dictionary containing category counts.
|
||||
output_file (str): Path to the output CSV file.
|
||||
"""
|
||||
df = pd.DataFrame(list(category_counts.items()), columns=["Category", "Count"])
|
||||
df = df.sort_values(by=["Count"], ascending=False)
|
||||
df.to_csv(output_file, index=False)
|
||||
|
||||
def main() -> None:
|
||||
"""Main function to process the import categories and save results."""
|
||||
category_counts = process_import_lines("__init__.py")
|
||||
save_category_counts_to_csv(category_counts, "node_count_breakdown.csv")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,174 @@
|
||||
"""
|
||||
Generate top level __init__.py.
|
||||
|
||||
Run this at the project root if you add a new .py file.
|
||||
|
||||
To skip the python file from included, put
|
||||
```
|
||||
pragma: skip_doc
|
||||
```
|
||||
in the docstring for the class. This pragma is used for both node reference as well as __init__.py.
|
||||
```
|
||||
"""
|
||||
import os
|
||||
import ast
|
||||
import pathlib
|
||||
from typing import TypeVar
|
||||
|
||||
T = TypeVar("T")
|
||||
MODULES_DIR = "modules"
|
||||
|
||||
|
||||
def extract_classes_from_file(file_path):
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
tree = ast.parse(f.read(), filename=file_path)
|
||||
|
||||
classes = []
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.ClassDef):
|
||||
docstring = ast.get_docstring(node)
|
||||
if docstring and "pragma: skip_doc" in docstring:
|
||||
continue
|
||||
classes.append(node.name)
|
||||
return classes
|
||||
|
||||
|
||||
def format_display_name(class_name):
|
||||
result = []
|
||||
first_lower_encountered = False
|
||||
|
||||
i = 0
|
||||
while i < len(class_name):
|
||||
if i > 0 and class_name[i].isupper():
|
||||
if first_lower_encountered or (i + 1 < len(class_name) and class_name[i + 1].islower()):
|
||||
result.append(" ")
|
||||
elif not first_lower_encountered and (i + 1 < len(class_name) and class_name[i + 1].isupper()):
|
||||
if i + 2 < len(class_name) and class_name[i + 2].islower():
|
||||
result.append(" ")
|
||||
|
||||
if class_name[i].islower():
|
||||
first_lower_encountered = True
|
||||
|
||||
# Special case for JSON and CSV
|
||||
if class_name[i:i+4] == "JSON":
|
||||
if result and result[-1] != " ":
|
||||
result.append(" ")
|
||||
result.append("JSON")
|
||||
if i + 4 < len(class_name) and class_name[i + 4].isupper():
|
||||
result.append(" ")
|
||||
i += 3 # Skip processed letters
|
||||
elif class_name[i:i+3] == "CSV":
|
||||
if result and result[-1] != " ":
|
||||
result.append(" ")
|
||||
result.append("CSV")
|
||||
if i + 3 < len(class_name) and class_name[i + 3].isupper():
|
||||
result.append(" ")
|
||||
i += 2 # Skip processed letters
|
||||
else:
|
||||
result.append(class_name[i])
|
||||
i += 1
|
||||
|
||||
# Ensure acronyms at the start are not incorrectly split (like "CD A" instead of "CDA")
|
||||
final_result = "".join(result)
|
||||
|
||||
# TODO: Reduce hack below
|
||||
replace_pairs = [
|
||||
("CD A", "CDA"),
|
||||
("H T M L", "HTML"),
|
||||
("MP L", "MPL"),
|
||||
("SN S", "SNS"),
|
||||
(" N A", " NA"),
|
||||
("Data Frame", "DataFrame"),
|
||||
("Multi Index", "MultiIndex"),
|
||||
("MPL Bar", "MPL Bar Chart"),
|
||||
("MPL Line", "MPL Line Plot"),
|
||||
("MPL Scatter", "MPL Scatter Plot"),
|
||||
("SNS Bar", "SNS Bar Chart"),
|
||||
("SNS Line", "SNS Line Plot"),
|
||||
("SNS Scatter", "SNS Scatter Plot")
|
||||
]
|
||||
for p in replace_pairs:
|
||||
final_result = final_result.replace(p[0], p[1])
|
||||
|
||||
return " ".join(final_result.split()) # Remove unintended extra spaces
|
||||
|
||||
|
||||
def discover_modules():
|
||||
module_entries = []
|
||||
for py_file in pathlib.Path(MODULES_DIR).rglob("*.py"):
|
||||
relative_path = py_file.relative_to(MODULES_DIR).with_suffix("")
|
||||
module_import_path = ".modules." + str(relative_path).replace(os.sep, ".")
|
||||
class_names = extract_classes_from_file(py_file)
|
||||
|
||||
for class_name in class_names:
|
||||
display_name = format_display_name(class_name)
|
||||
module_entries.append((module_import_path, class_name, display_name))
|
||||
|
||||
return module_entries
|
||||
|
||||
|
||||
def get_class_mappings_doc():
|
||||
return """
|
||||
\"\"\"
|
||||
NODE_CLASS_MAPPINGS (Dict[str, Type[T]]):
|
||||
A dictionary mapping node names to their corresponding class implementations.
|
||||
\"\"\"
|
||||
"""
|
||||
|
||||
|
||||
def get_display_name_mappings_doc():
|
||||
return """
|
||||
\"\"\"
|
||||
NODE_DISPLAY_NAME_MAPPINGS (Dict[str, str]):
|
||||
A dictionary mapping node names to user-friendly display names.
|
||||
\"\"\"
|
||||
"""
|
||||
|
||||
|
||||
def get_license_doc():
|
||||
return """
|
||||
\"\"\"
|
||||
Below two lines were taken from:
|
||||
https://github.com/pythongosssss/ComfyUI-Custom-Scripts/blob/main/__init__.py
|
||||
See credit/credit.md for the full license.
|
||||
\"\"\"
|
||||
"""
|
||||
|
||||
|
||||
def generate_init_content(import_statements, node_class_mappings, node_display_name_mappings):
|
||||
return f"""# This file is auto-generated. Do not edit manually.
|
||||
from typing import Dict, Type, TypeVar
|
||||
{os.linesep.join(import_statements)}
|
||||
T = TypeVar("T")
|
||||
|
||||
{get_class_mappings_doc()}
|
||||
NODE_CLASS_MAPPINGS: Dict[str, Type[T]] = {{
|
||||
{os.linesep.join(node_class_mappings)}
|
||||
}}
|
||||
|
||||
{get_display_name_mappings_doc()}
|
||||
NODE_DISPLAY_NAME_MAPPINGS: Dict[str, str] = {{
|
||||
{os.linesep.join(node_display_name_mappings)}
|
||||
}}
|
||||
|
||||
{get_license_doc()}
|
||||
WEB_DIRECTORY = "./web"
|
||||
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]
|
||||
"""
|
||||
|
||||
|
||||
def generate_init():
|
||||
module_entries = discover_modules()
|
||||
|
||||
import_statements = sorted([f"from {entry[0]} import {entry[1]}" for entry in module_entries])
|
||||
node_class_mappings = sorted([f' "{entry[1]}": {entry[1]},' for entry in module_entries])
|
||||
node_display_name_mappings = sorted([f' "{entry[1]}": "{entry[2]}",' for entry in module_entries])
|
||||
|
||||
template = generate_init_content(import_statements, node_class_mappings, node_display_name_mappings)
|
||||
|
||||
with open("__init__.py", "w", encoding="utf-8") as f:
|
||||
f.write(template)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
generate_init()
|
||||
@@ -0,0 +1,159 @@
|
||||
"""
|
||||
Generate node reference documents.
|
||||
|
||||
Run this at the project root if you update a docstring of the node source file.
|
||||
|
||||
To skip the python file from included in the reference, put
|
||||
```
|
||||
pragma: skip_doc
|
||||
```
|
||||
in the docstring for the class.
|
||||
|
||||
To use a custom node name, put the node name on the first line of the docstring for the class with ":".
|
||||
e.g.
|
||||
```
|
||||
CDA JSON Create:
|
||||
Creates a serialized JSON object using values entered in the text field.
|
||||
```
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import ast
|
||||
from collections import defaultdict
|
||||
from typing import Dict, List
|
||||
|
||||
# Mapping for data type conversion
|
||||
data_type_mapping = {
|
||||
"DATAFRAME": "DataFrame",
|
||||
"PDSERIES": "Series",
|
||||
"PDINDEX": "Index"
|
||||
}
|
||||
|
||||
def convert_data_type(data_type: str) -> str:
|
||||
"""Convert type to the required format."""
|
||||
if data_type in data_type_mapping:
|
||||
return data_type_mapping[data_type]
|
||||
return data_type.lower().capitalize()
|
||||
|
||||
def extract_class_info(class_def: ast.ClassDef) -> Dict[str, str]:
|
||||
"""Extract class name, title, category, and docstring."""
|
||||
class_name = class_def.name
|
||||
docstring = ast.get_docstring(class_def) or ""
|
||||
|
||||
if "pragma: skip_doc" in docstring:
|
||||
return None
|
||||
|
||||
lines = docstring.split("\n")
|
||||
|
||||
title = None
|
||||
category = "Miscellaneous"
|
||||
|
||||
if len(lines) > 1 and lines[0].strip().endswith(":") and lines[0].strip()[0].isupper():
|
||||
title = lines.pop(0).strip().rstrip(":")
|
||||
|
||||
if not title:
|
||||
title = "# " + " ".join(re.findall(r'[A-Z][a-z]*', class_name))
|
||||
else:
|
||||
title = f"# {title}"
|
||||
|
||||
description_lines = []
|
||||
for line in lines:
|
||||
if line.strip().startswith("category:"):
|
||||
category = line.split(":", 1)[-1].strip()
|
||||
else:
|
||||
description_lines.append(line)
|
||||
|
||||
description = "\n".join(description_lines).strip()
|
||||
|
||||
return {"title": title, "description": description, "class_name": class_name, "first_sentence": description_lines[0] if description_lines else "", "category": category}
|
||||
|
||||
def extract_input_types(class_def: ast.ClassDef) -> str:
|
||||
"""Extract INPUT_TYPES definition."""
|
||||
input_table = ""
|
||||
for node in class_def.body:
|
||||
if isinstance(node, ast.FunctionDef) and node.name == "INPUT_TYPES":
|
||||
for stmt in ast.walk(node):
|
||||
if isinstance(stmt, ast.Dict):
|
||||
for key, value in zip(stmt.keys, stmt.values):
|
||||
if isinstance(key, ast.Constant) and isinstance(value, ast.Tuple) and len(value.elts) > 0:
|
||||
param_name = key.value
|
||||
data_type = value.elts[0].value if isinstance(value.elts[0], ast.Constant) else ""
|
||||
input_table += f"| {param_name} | {convert_data_type(data_type)} |\n"
|
||||
return "## Input\n| Name | Data type |\n|---|---|\n" + input_table if input_table else ""
|
||||
|
||||
def extract_return_types(class_def: ast.ClassDef) -> str:
|
||||
"""Extract RETURN_TYPES value."""
|
||||
for node in class_def.body:
|
||||
# Look for assignment or annotated assignment (type hint) statement.
|
||||
# Node target is the RHS.
|
||||
if (
|
||||
isinstance(node, ast.Assign) and
|
||||
any(target.id == "RETURN_TYPES" for target in node.targets if isinstance(target, ast.Name))
|
||||
) or (
|
||||
isinstance(node, ast.AnnAssign) and
|
||||
isinstance(node.target, ast.Name) and
|
||||
node.target.id == "RETURN_TYPES"
|
||||
):
|
||||
if isinstance(node.value, ast.Tuple):
|
||||
return_types = [convert_data_type(elt.value) for elt in node.value.elts if isinstance(elt, ast.Constant)]
|
||||
return "## Output\n| Data type |\n|---|\n" + "".join(f"| {rtype} |\n" for rtype in return_types)
|
||||
return ""
|
||||
|
||||
def process_python_file(file_path: str, output_dir: str, node_list: Dict[str, List[str]]):
|
||||
"""Process a single Python file and generate corresponding Markdown."""
|
||||
with open(file_path, "r", encoding="utf-8") as f:
|
||||
tree = ast.parse(f.read(), filename=file_path)
|
||||
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.ClassDef):
|
||||
class_info = extract_class_info(node)
|
||||
if not class_info:
|
||||
return
|
||||
|
||||
input_section = extract_input_types(node)
|
||||
output_section = extract_return_types(node)
|
||||
category = class_info["category"]
|
||||
|
||||
file_name = os.path.splitext(os.path.basename(file_path))[0] + ".md"
|
||||
output_path = os.path.join(output_dir, file_name)
|
||||
|
||||
with open(output_path, "w", encoding="utf-8") as md_file:
|
||||
md_file.write(f"{class_info['title']}\n{class_info['description']}\n\n")
|
||||
if input_section:
|
||||
md_file.write(input_section + "\n")
|
||||
if output_section:
|
||||
md_file.write(output_section + "\n")
|
||||
md_file.write("<HR>\n")
|
||||
md_file.write(f"Category: {category}\n\n")
|
||||
md_file.write(f"ComfyUI Data Analysis Node Reference. © 2025 Hide Inada (HowToSD.com). All rights reserved.\n")
|
||||
|
||||
node_list[category].append((class_info['title'][2:], file_name, class_info['first_sentence']))
|
||||
|
||||
def generate_node_list(output_dir: str, node_list: Dict[str, List[str]]):
|
||||
"""Generate node_reference.md file."""
|
||||
node_list_path = os.path.join(output_dir, "node_reference.md")
|
||||
with open(node_list_path, "w", encoding="utf-8") as md_file:
|
||||
md_file.write("# Node Reference\n")
|
||||
for category in sorted(node_list.keys()):
|
||||
md_file.write(f"## {category}\n| Node | Description |\n| --- | --- |\n")
|
||||
for title, file_name, description in sorted(node_list[category]):
|
||||
md_file.write(f"| [{title}]({file_name}) | {description} |\n")
|
||||
|
||||
def main():
|
||||
"""Main function to process all Python files under modules/**.py."""
|
||||
source_dir = "modules"
|
||||
output_dir = "docs/reference"
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
|
||||
node_list = defaultdict(list)
|
||||
|
||||
for root, _, files in os.walk(source_dir):
|
||||
for file in files:
|
||||
if file.endswith(".py"):
|
||||
file_path = os.path.join(root, file)
|
||||
process_python_file(file_path, output_dir, node_list)
|
||||
|
||||
generate_node_list(output_dir, node_list)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Executable
+12
@@ -0,0 +1,12 @@
|
||||
#!/bin/bash
|
||||
# Run this command on project root
|
||||
# ```
|
||||
# dev/scripts/run_tests.sh
|
||||
# ````
|
||||
# If you add a new test file, make sure that the directory contains an empty
|
||||
# __init__.py. Otherwise, unittest won't be able to find the directory and
|
||||
# the tests in the directory will be skipped.
|
||||
export PYTHONPATH=modules:$PYTHONPATH
|
||||
echo $PYTHONPATH
|
||||
RUN_SKIPPED_TESTS=0 python -m unittest discover -s tests -p "*_test.py" -t .
|
||||
|
||||
Reference in New Issue
Block a user