NodeBlock Docs CI

.dockerignore

@@ -5,4 +5,6 @@ cpac_runs
 .git
 .github
 !.github/scripts
+.github/scripts/config_extractor.py
+.github/scripts/nodeblock_docs.py
 *.tar.gz

.github/scripts/config_extractor.py

@@ -0,0 +1,139 @@
+import json
+import pathlib as pl
+import sys
+from typing import Literal
+import yaml
+import re
+from typing import Any, Sequence
+
+
+def filesafe(s: str, replacement: str = "-") -> str:
+    """
+    Converts a string to a file safe string.
+    Removes all non-alphanumeric characters and
+    replaces them with the replacement string.
+    """
+    return re.sub(r"[^\w\d-]", replacement, s).lower()
+
+
+def multi_get(obj: dict, index: Sequence) -> Any | None:  # noqa: ANN401
+    """
+    Gets a value from a nested dictionary.
+    Returns None if the path does not exist.
+    """
+    for i in index:
+        if not isinstance(obj, dict) or i not in obj:
+            return None
+        obj = obj[i]
+    return obj
+
+
+def fetch_and_expand_cpac_configs(
+    cpac_dir: pl.Path,
+    output_dir: pl.Path,
+    config_names_ids: dict[str, str],
+) -> None:
+    """
+    Fetches C-PAC configs from github, fully expands them (FROM: parent),
+    and then saves them to the specified directory.
+    """
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    cpac_module_path = str(cpac_dir.absolute())
+
+    if cpac_module_path not in sys.path:
+        sys.path.append(cpac_module_path)
+
+    from CPAC.utils.configuration.configuration import Preconfiguration  # noqa
+    from CPAC.utils.configuration.yaml_template import create_yaml_from_template  # noqa
+
+    for config_name, config_id in config_names_ids.items():
+        conf = Preconfiguration(config_id)
+        config_yaml_string = create_yaml_from_template(conf.dict(), "blank")
+
+        with open(
+            output_dir / (filesafe(config_name) + ".yml"), "w", encoding="utf-8"
+        ) as handle:
+            handle.write(config_yaml_string)
+
+
+def get_cpac_config_ids() -> list[str]:
+    from CPAC.pipeline import ALL_PIPELINE_CONFIGS
+
+    return ALL_PIPELINE_CONFIGS
+
+
+def fetch_and_expand_all_cpac_configs(
+    cpac_dir: pl.Path,
+    output_dir: pl.Path,
+):
+    config_names_ids = {i: i for i in get_cpac_config_ids()}
+    fetch_and_expand_cpac_configs(
+        cpac_dir=cpac_dir,
+        output_dir=output_dir,
+        config_names_ids=config_names_ids,
+    )
+
+
+def _normalize_index_union(indices) -> list[list[str]]:
+    if not indices:
+        return []
+    if isinstance(indices[0], list):
+        return indices
+    return [indices]
+
+
+def normalize_index_union(indices) -> list[list[str]]:
+    re = _normalize_index_union(indices)
+    assert all(isinstance(item, list) for item in re)
+    assert all(isinstance(i, str) for item in re for i in item)
+    return re
+
+
+if __name__ == "__main__":
+
+    from CPAC.pipeline import ALL_PIPELINE_CONFIGS
+    from CPAC.utils.configuration.configuration import Configuration, Preconfiguration
+
+    with open("nodeblock_index.json") as f:
+        nbs = json.load(f)
+    configs: dict[str, Configuration] = {config: Preconfiguration(config, skip_env_check=True) for config in ALL_PIPELINE_CONFIGS}
+
+    print(f"Found {len(configs)} pre-configs!")
+
+    def _any_true_in_config(config, multi_index_union):
+        for path in multi_index_union:
+            if multi_get(config, path):
+                return True
+        return False
+
+    for nb in nbs:
+        nb_configs = normalize_index_union(nb["decorator_args"].get("config"))
+        nb_switchs = normalize_index_union(nb["decorator_args"].get("switch"))
+
+        # multiply
+
+        if not nb_configs and not nb_switchs:
+            continue
+
+        paths: list[list[str]] = []
+        if not nb_configs:
+            paths = nb_switchs
+        else:
+            for nb_config in nb_configs:
+                for nb_switch in nb_switchs:
+                    paths.append(nb_config + nb_switch)
+
+        assert all(isinstance(item, list) for item in paths), paths
+        assert all(isinstance(i, str) for item in paths for i in item), paths
+
+        configs_with_this_enabled = []
+        for config_name, config in configs.items():
+            if all(config.switch_is_on(switch) for switch in paths):
+                configs_with_this_enabled.append(config_name)
+
+        nb["workflows"] = configs_with_this_enabled
+
+    with open("nodeblock_index.json", "w", encoding="utf8") as handle:
+        json.dump(nbs, handle, indent=2)

.github/scripts/nodeblock_docs.py

@@ -0,0 +1,112 @@
+import ast
+import json
+import os
+from pathlib import Path
+
+class NodeBlockVisitor(ast.NodeVisitor):
+    def __init__(self):
+        self.nodeblocks = []
+        self.current_file = None
+
+    def safe_eval(self, node):
+        """Safely evaluate AST nodes, handling special cases."""
+        if isinstance(node, ast.Constant):
+            return node.value
+        elif isinstance(node, ast.List):
+            return [self.safe_eval(elt) for elt in node.elts]
+        elif isinstance(node, ast.Tuple):
+            return tuple(self.safe_eval(elt) for elt in node.elts)
+        elif isinstance(node, ast.Starred):
+            # For starred expressions, we'll return a placeholder
+            return f"*{self.get_starred_expr(node.value)}"
+        elif isinstance(node, ast.Name):
+            return f"@{node.id}"  # Return variable names with @ prefix to distinguish them
+        elif isinstance(node, ast.Attribute):
+            # Handle attribute access (e.g., module.attribute)
+            return f"@{self.get_attribute_chain(node)}"
+        else:
+            return f"!UNSUPPORTED_TYPE:{type(node).__name__}!"
+
+    def get_starred_expr(self, node):
+        """Get a string representation of what's being starred."""
+        if isinstance(node, ast.Name):
+            return node.id
+        elif isinstance(node, ast.Attribute):
+            return self.get_attribute_chain(node)
+        return f"unknown_starred_{type(node).__name__}"
+
+    def get_attribute_chain(self, node):
+        """Get the full chain of attributes (e.g., 'module.attribute')."""
+        parts = []
+        current = node
+        while isinstance(current, ast.Attribute):
+            parts.append(current.attr)
+            current = current.value
+        if isinstance(current, ast.Name):
+            parts.append(current.id)
+        return '.'.join(reversed(parts))
+
+    def visit_FunctionDef(self, node):
+        # Check if the function has a decorator that matches @nodeblock
+        for decorator in node.decorator_list:
+            if isinstance(decorator, ast.Call):
+                if isinstance(decorator.func, ast.Name) and decorator.func.id == 'nodeblock':
+                    # Extract decorator arguments
+                    args = {}
+                    for kw in decorator.keywords:
+                        # Use safe_eval instead of ast.literal_eval
+                        args[kw.arg] = self.safe_eval(kw.value)
+
+                    # Get function docstring
+                    docstring = ast.get_docstring(node)
+
+                    # Get source code lines
+                    source_lines = []
+                    for i in range(node.lineno - 1, node.end_lineno):
+                        source_lines.append(self.current_source[i])
+                    source_code = '\n'.join(source_lines)
+
+                    # Create nodeblock info
+                    nodeblock_info = {
+                        'name': node.name,
+                        'file': str(self.current_file),
+                        'line_number': node.lineno,
+                        'decorator_args': args,
+                        'docstring': docstring,
+                        'source_code': source_code,
+                    }
+                    self.nodeblocks.append(nodeblock_info)
+
+def find_nodeblocks(root_dir):
+    visitor = NodeBlockVisitor()
+    root_path = Path(root_dir)
+
+    # Walk through all Python files
+    for python_file in root_path.rglob('*.py'):
+        try:
+            with open(python_file, 'r', encoding='utf-8') as f:
+                source = f.read()
+                visitor.current_source = source.splitlines()
+                visitor.current_file = python_file.relative_to(root_path)
+                tree = ast.parse(source)
+                visitor.visit(tree)
+        except Exception as e:
+            print(f"Error processing {python_file}: {e}")
+
+    return visitor.nodeblocks
+
+def main():
+    # Assuming you're running this from the root of the C-PAC repository
+    root_dir = '.'  # or specify the full path to C-PAC repository
+    nodeblocks = find_nodeblocks(root_dir)
+
+    # Save to JSON file
+    output_file = 'nodeblock_index.json'
+    with open(output_file, 'w', encoding='utf-8') as f:
+        json.dump(nodeblocks, f, indent=2, ensure_ascii=False)
+
+    print(f"Found {len(nodeblocks)} nodeblocks")
+    print(f"Index saved to {output_file}")
+
+if __name__ == '__main__':
+    main()

.github/workflows/nodeblock_docs.yml

@@ -0,0 +1,61 @@
+name: Update NodeBlock documentation
+on:
+  push:
+    branches: [main]
+
+jobs:
+  update-nodeblock-docs:
+    name: Update NodeBlock Documentation
+    runs-on: ubuntu-latest
+    steps:
+      # Check out the source repo
+      - name: Check out source repository (C-PAC)
+        uses: actions/checkout@v3
+        with:
+          path: source-repo
+
+      # Set up Python
+      - name: Set up Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: '3.10'
+
+      # Run the documentation extraction script
+      - name: Generate documentation JSON
+        run: |
+          cd source-repo
+          python scripts/nodeblock_docs.py
+
+      # Check out the target repo where we want to commit the JSON file
+      - name: Check out target repository
+        uses: actions/checkout@v3
+        with:
+          repository: FCP-INDI/cpac-docs
+          path: target-repo
+          token: ${{ secrets.CPAC_DOC_REPO_ACCESS }}
+
+      # Install cpac dependencies
+      - name: Install dependencies
+        run: |
+          cd source-repo
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+
+      # Run the config extraction script
+      - name: Generate documentation JSON
+        run: |
+          cd source-repo
+          python scripts/config_extractor.py
+
+      # Copy the JSON file to the target repo and commit it
+      - name: Commit documentation to target repo
+        run: |
+          cp source-repo/nodeblock_index.json target-repo/src
+          cd target-repo
+          git config user.name "GitHub Actions Bot"
+          git config user.email "actions@github.com"
+          git add nodeblock_index.json
+          git diff --quiet && git diff --staged --quiet || git commit -m "Update NodeBlock documentation"
+          git push
+
+