Add multi-file trace parsing and analysis pipeline

projects/composablekernel/script/analyze_build/notebooks/file_analysis_example.ipynb

@@ -17,6 +17,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
+    "%load_ext autoreload\n",
+    "%autoreload 2\n",
+    "\n",
     "import sys\n",
     "from pathlib import Path\n",
     "\n",
@@ -62,7 +65,8 @@
     "\n",
     "print(f\"Total events: {len(df):,}\")\n",
     "starting_timestamp = datetime.fromtimestamp(df.attrs[\"beginningOfTime\"] / 1e6)\n",
-    "print(f\"Starting timestamp: {starting_timestamp.strftime('%Y-%m-%d:%H:%M:%S')}\")"
+    "print(f\"Starting timestamp: {starting_timestamp.strftime('%Y-%m-%d:%H:%M:%S')}\")\n",
+    "print(f\"Source file: {df.attrs['sourceFile']}\")"
    ]
   },
   {

projects/composablekernel/script/analyze_build/requirements.txt

@@ -4,6 +4,9 @@
 pandas>=2.0.0
 orjson>=3.9.0
 
+# Statistical analysis
+statsmodels>=0.14.0
+
 # Jupyter notebook support
 nbformat>=4.2.0
 ipykernel>=6.0.0
@@ -16,3 +19,6 @@ kaleido>=0.2.0
 
 # Full Jupyter environment (if not using VSCode)
 jupyter>=1.0.0
+
+# Progress meter in notebook
+tqdm>=4.0.0

projects/composablekernel/script/analyze_build/trace_analysis/__init__.py

@@ -22,13 +22,32 @@
     PhaseBreakdown,
 )
 
+from .parse_build import (
+    find_trace_files,
+    read_trace_files,
+)
+
+from .pipeline import (
+    Pipeline,
+)
+
+from .build_helpers import (
+    get_trace_file,
+)
+
 __all__ = [
     # Core parsing and filtering
     "parse_file",
     "get_metadata",
+    "find_trace_files",
+    "read_trace_files",
+    # Pipeline processing
+    "Pipeline",
     # Template analysis
     "get_template_instantiation_events",
     # Phase breakdown
     "get_phase_breakdown",
     "PhaseBreakdown",
+    # Build helpers
+    "get_trace_file",
 ]

projects/composablekernel/script/analyze_build/trace_analysis/build_helpers.py

@@ -0,0 +1,139 @@
+# Copyright (c) Advanced Micro Devices, Inc., or its affiliates.
+# SPDX-License-Identifier: MIT
+
+"""
+Helper functions for full build analysis.
+"""
+
+from .parse_file import get_metadata
+from .phase_breakdown import get_phase_breakdown
+from .template_analysis import get_template_instantiation_events
+
+
+def extract_all_data(df):
+    """
+    Extract metadata, phase breakdown, and template events from a parsed DataFrame.
+
+    Args:
+        df: Parsed DataFrame from parse_file()
+
+    Returns:
+        Dictionary with keys:
+        - build_unit: Source file path starting from composable_kernel/
+        - trace_file_path: Path to the original trace JSON file
+        - metadata: Metadata dictionary
+        - phase: Phase breakdown DataFrame
+        - template: Template events DataFrame
+    """
+    return {
+        "build_unit": df.attrs["sourceFile"],
+        "trace_file_path": df.attrs.get("traceFilePath"),
+        "metadata": get_metadata(df).__dict__,
+        "phase": get_phase_breakdown(df).df,
+        "template": get_template_instantiation_events(df),
+    }
+
+
+def get_trace_file(metadata_df, build_unit):
+    """
+    Get the trace file path for a given build unit.
+
+    Args:
+        metadata_df: Metadata DataFrame with trace_file_mapping in .attrs
+        build_unit: Source file path (build unit name)
+
+    Returns:
+        Path to the trace JSON file, or None if not found
+
+    Examples:
+        >>> # Get trace file for a specific build unit
+        >>> trace_path = get_trace_file(metadata_df, "library/src/tensor/gemm.cpp")
+        >>> print(f"Trace file: {trace_path}")
+        >>>
+        >>> # Get trace files for slowest compilation units
+        >>> slowest = metadata_df.nlargest(5, "total_wall_time_s")
+        >>> for _, row in slowest.iterrows():
+        ...     trace_path = get_trace_file(metadata_df, row['build_unit'])
+        ...     print(f"{row['build_unit']}: {trace_path}")
+    """
+    mapping = metadata_df.attrs.get("trace_file_mapping", {})
+    return mapping.get(build_unit)
+
+
+def print_phase_hierarchy(phase_df):
+    """
+    Print cumulative phase times in a hierarchical tree structure.
+
+    Args:
+        phase_df: DataFrame with columns: name, parent, depth, duration, build_unit
+                  (as created by concatenating phase breakdown results)
+    """
+    # Aggregate by phase name, parent, and depth
+    phase_summary = (
+        phase_df.groupby(["name", "parent", "depth"])
+        .agg({"duration": "sum"})
+        .reset_index()
+    )
+
+    # Convert to seconds
+    phase_summary["duration_s"] = phase_summary["duration"] / 1_000_000
+
+    # Calculate total time from root node only (depth == 0)
+    # With branchvalues="total", parent nodes include their children's time,
+    # so summing all phases would double/triple count nested values
+    root_phases = phase_summary[
+        (phase_summary["parent"] == "")
+        | (phase_summary["parent"].isna())
+        | (phase_summary["depth"] == 0)
+    ].sort_values("duration_s", ascending=False)
+
+    if len(root_phases) == 0:
+        raise ValueError("No root phase found (depth == 0)")
+    if len(root_phases) > 1:
+        raise ValueError(f"Multiple root phases found: {root_phases['name'].tolist()}")
+
+    total_time_s = root_phases.iloc[0]["duration_s"]
+
+    print("=== Cumulative Phase Time Across Build ===")
+    print(f"\nTotal compilation time: {total_time_s:,.1f} s")
+    print("\nBreakdown by phase:")
+
+    # Track which phases we've printed to handle hierarchy
+    printed_phases = set()
+
+    def print_phase_tree(df, parent_name, depth=0):
+        """Recursively print phases in hierarchical order"""
+        # Get children of this parent at the next depth level
+        children = df[(df["parent"] == parent_name) & (df["depth"] == depth)]
+        # Sort by duration descending within each level
+        children = children.sort_values("duration_s", ascending=False)
+
+        for _, row in children.iterrows():
+            phase_name = row["name"]
+            if phase_name in printed_phases:
+                continue
+
+            time_s = row["duration_s"]
+            pct = 100 * time_s / total_time_s
+            indent = "  " * depth
+            # Create indented name and pad the whole thing to align colons
+            indented_name = f"{indent}{phase_name}"
+            print(f"{indented_name:32s}: {time_s:12,.1f} s ({pct:5.1f}%)")
+            printed_phases.add(phase_name)
+
+            # Recursively print children
+            print_phase_tree(df, phase_name, depth + 1)
+
+    for _, row in root_phases.iterrows():
+        phase_name = row["name"]
+        if phase_name in printed_phases:
+            continue
+
+        time_s = row["duration_s"]
+        pct = 100 * time_s / total_time_s
+        # Pad root phase name to align with children
+        print(f"{phase_name:32s}: {time_s:12,.1f} s ({pct:5.1f}%)")
+        printed_phases.add(phase_name)
+
+        # Print children recursively
+        print_phase_tree(phase_summary, phase_name, 1)

projects/composablekernel/script/analyze_build/trace_analysis/parse_build.py

@@ -0,0 +1,96 @@
+# Copyright (c) Advanced Micro Devices, Inc., or its affiliates.
+# SPDX-License-Identifier: MIT
+
+"""
+Utility functions for trace analysis.
+
+Helper functions for file discovery, path handling, and other common operations.
+"""
+
+import subprocess
+import pandas as pd
+from pathlib import Path
+from typing import List
+
+
+def find_trace_files(trace_dir: Path) -> List[Path]:
+    """
+    Find all JSON trace files in a directory.
+
+    Uses Unix 'find' command when available (2-5x faster than Python),
+    with automatic fallback to Python's rglob for cross-platform compatibility.
+
+    Args:
+        trace_dir: Directory to search for trace files
+
+    Returns:
+        List of Path objects pointing to .json files
+
+    Example:
+        >>> from pathlib import Path
+        >>> from trace_analysis import find_trace_files
+        >>> trace_files = find_trace_files(Path("build/CMakeFiles"))
+        >>> print(f"Found {len(trace_files)} trace files")
+    """
+    try:
+        # Try Unix find (2-5x faster than Python)
+        result = subprocess.run(
+            ["find", str(trace_dir), "-name", "*.cpp.json", "-type", "f"],
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=True,
+        )
+        json_files = [Path(p) for p in result.stdout.strip().split("\n") if p]
+    except (subprocess.SubprocessError, FileNotFoundError, OSError):
+        # Fallback to Python (cross-platform)
+        print("Using Python to find trace files (this may be slower)...")
+        json_files = list(trace_dir.rglob("*.cpp.json"))
+
+    return json_files
+
+
+def read_trace_files(json_files: List[Path], workers: int = -1) -> List["pd.DataFrame"]:
+    """
+    Parse trace files in parallel and return list of DataFrames.
+
+    This is a convenience function that uses the Pipeline API to parse
+    multiple trace files in parallel with progress tracking.
+
+    Args:
+        json_files: List of paths to trace JSON files
+        workers: Number of parallel workers to use:
+            - -1: Use all available CPUs (default)
+            - None: Sequential processing (single-threaded)
+            - N > 0: Use N worker processes
+
+    Returns:
+        List of parsed DataFrames, one per file
+
+    Example:
+        >>> from pathlib import Path
+        >>> from trace_analysis import find_trace_files, read_trace_files
+        >>>
+        >>> # Find and parse all trace files
+        >>> trace_files = find_trace_files(Path("build/CMakeFiles"))
+        >>> dataframes = read_trace_files(trace_files, workers=8)
+        >>> print(f"Parsed {len(dataframes)} files")
+        >>>
+        >>> # Use Pipeline directly for more control
+        >>> from trace_analysis import Pipeline
+        >>> from trace_analysis.parse_file import parse_file
+        >>>
+        >>> pipeline = Pipeline(trace_files).map(parse_file, workers=8)
+        >>> all_events, metadata = pipeline.tee(
+        ...     lambda dfs: pd.concat(dfs, ignore_index=True),
+        ...     lambda dfs: [get_metadata(df) for df in dfs]
+        ... )
+    """
+    from trace_analysis.pipeline import Pipeline
+    from trace_analysis.parse_file import parse_file
+
+    return (
+        Pipeline(json_files)
+        .map(parse_file, workers=workers, desc="Parsing trace files")
+        .collect()
+    )