Introduce Phase Architecture for FINN Builder Pipeline

src/finn/analysis/fpgadataflow/validate_dataflow_conversion.py

@@ -0,0 +1,100 @@
+# Copyright Advanced Micro Devices, Inc.
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Analysis pass to validate that model has been properly converted to fpgadataflow layers."""
+
+from finn.util.fpgadataflow import is_fpgadataflow_node
+
+
+def validate_dataflow_conversion(model):
+    """Validate that model has been properly converted to dataflow layers.
+
+    Checks that either:
+    1. All layers are fpgadataflow layers (ideal case), OR
+    2. Fpgadataflow layers form a contiguous block in the middle of the model,
+       with only non-dataflow layers on the outside (partition case)
+
+    Returns a dictionary with validation results:
+    - 'valid': bool indicating if validation passed
+    - 'message': str with validation status message
+    - 'unconverted_layers': list of (index, name, op_type) tuples for non-dataflow layers
+    - 'dataflow_block': tuple (first_index, last_index) if dataflow forms a block, else None
+
+    Example usage in transformation:
+        result = model.analysis(validate_dataflow_conversion)
+        if not result['valid']:
+            raise AssertionError(result['message'])
+    """
+    nodes = model.graph.node
+    fpgadataflow_nodes = []
+    non_fpgadataflow_nodes = []
+
+    for i, node in enumerate(nodes):
+        if is_fpgadataflow_node(node):
+            fpgadataflow_nodes.append((i, node.name, node.op_type))
+        else:
+            non_fpgadataflow_nodes.append((i, node.name, node.op_type))
+
+    # Case 1: All nodes are fpgadataflow (ideal)
+    if len(non_fpgadataflow_nodes) == 0:
+        return {
+            "valid": True,
+            "message": "Dataflow conversion validation: All layers are fpgadataflow layers",
+            "unconverted_layers": [],
+            "dataflow_block": None,
+        }
+
+    # Case 2: Check if fpgadataflow nodes form contiguous block
+    if len(fpgadataflow_nodes) > 0:
+        dataflow_indices = [i for i, _, _ in fpgadataflow_nodes]
+        first_dataflow = min(dataflow_indices)
+        last_dataflow = max(dataflow_indices)
+
+        # Check all indices between first and last are dataflow
+        for i in range(first_dataflow, last_dataflow + 1):
+            node = nodes[i]
+            if not is_fpgadataflow_node(node):
+                # Found non-dataflow layer inside dataflow block
+                unconverted_str = "\n".join(
+                    [
+                        f"  [{idx}] {name} (op_type: {op})"
+                        for idx, name, op in non_fpgadataflow_nodes
+                    ]
+                )
+                return {
+                    "valid": False,
+                    "message": (
+                        "Non-contiguous dataflow block detected.\n"
+                        f"Layer '{node.name}' (op_type: {node.op_type}) at position {i} "
+                        "is not a fpgadataflow layer but is between dataflow layers.\n"
+                        f"Dataflow block spans positions {first_dataflow} to {last_dataflow}.\n"
+                        f"Unconverted layers:\n{unconverted_str}"
+                    ),
+                    "unconverted_layers": non_fpgadataflow_nodes,
+                    "dataflow_block": (first_dataflow, last_dataflow),
+                }
+
+        # Valid: fpgadataflow block in middle
+        return {
+            "valid": True,
+            "message": (
+                "Dataflow conversion validation: Fpgadataflow layers form contiguous block "
+                f"(positions {first_dataflow}-{last_dataflow})"
+            ),
+            "unconverted_layers": non_fpgadataflow_nodes,
+            "dataflow_block": (first_dataflow, last_dataflow),
+        }
+
+    # Case 3: No fpgadataflow layers at all
+    unconverted_str = "\n".join(
+        [f"  [{idx}] {name} (op_type: {op})" for idx, name, op in non_fpgadataflow_nodes]
+    )
+    return {
+        "valid": False,
+        "message": (
+            "No fpgadataflow layers found in model.\n"
+            f"All layers remain unconverted:\n{unconverted_str}"
+        ),
+        "unconverted_layers": non_fpgadataflow_nodes,
+        "dataflow_block": None,
+    }

src/finn/builder/build_dataflow.py

@@ -45,6 +45,7 @@
     DataflowBuildConfig,
     default_build_dataflow_steps,
 )
+from finn.builder.build_dataflow_phases import build_dataflow_phase_lookup
 from finn.builder.build_dataflow_steps import build_dataflow_step_lookup
 
 
@@ -68,19 +69,53 @@ def flush(self):
 
 
 def resolve_build_steps(cfg: DataflowBuildConfig, partial: bool = True):
+    """Resolve build steps from config, supporting both phases and fine-grained steps.
+
+    Note: When using phase-based builds with start_step/stop_step, specify phase names
+    (e.g., start_step="phase_build_hardware") rather than fine-grained step names.
+    Phases save intermediate models for each internal step, so checkpoints like
+    step_hw_ipgen.onnx will exist, but the build loop operates at the phase level.
+    """
     steps = cfg.steps
     if steps is None:
         steps = default_build_dataflow_steps
+
+    # Merge phase and step lookup dictionaries
+    all_steps = {
+        **build_dataflow_step_lookup,
+        **build_dataflow_phase_lookup,
+    }
+
     steps_as_fxns = []
     for transform_step in steps:
+        step_name = None
+
+        # Get step function and name
         if type(transform_step) is str:
-            # lookup step function from step name
-            steps_as_fxns.append(build_dataflow_step_lookup[transform_step])
+            step_name = transform_step
+            if transform_step in all_steps:
+                step_fn = all_steps[transform_step]
+            else:
+                raise ValueError(f"Unknown step or phase: {transform_step}")
         elif callable(transform_step):
-            # treat step as function to be called as-is
-            steps_as_fxns.append(transform_step)
+            step_fn = transform_step
+            step_name = getattr(transform_step, "__name__", None)
         else:
-            raise Exception("Could not resolve build step: " + str(transform_step))
+            raise ValueError(f"Invalid step type: {type(transform_step)}")
+
+        # Inject steps BEFORE this step
+        if step_name and step_name in cfg.inject_steps_before:
+            for injected_step in cfg.inject_steps_before[step_name]:
+                steps_as_fxns.append(injected_step)
+
+        # Add the main step
+        steps_as_fxns.append(step_fn)
+
+        # Inject steps AFTER this step
+        if step_name and step_name in cfg.inject_steps_after:
+            for injected_step in cfg.inject_steps_after[step_name]:
+                steps_as_fxns.append(injected_step)
+
     if partial:
         step_names = list(map(lambda x: x.__name__, steps_as_fxns))
         if cfg.start_step is None:

src/finn/builder/build_dataflow_config.py

@@ -29,10 +29,10 @@
 
 import numpy as np
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from dataclasses_json import dataclass_json
 from enum import Enum
-from typing import Any, List, Optional
+from typing import Any, Callable, Dict, List, Optional
 
 from finn.transformation.fpgadataflow.alveo_build import VitisOptStrategy
 from finn.util.basic import part_map, vitis_default_platform
@@ -108,39 +108,20 @@ class VerificationStepType(str, Enum):
 #: specified order. Use the `steps` as part of build config to restrict which
 #: steps will be run.
 default_build_dataflow_steps = [
-    "step_qonnx_to_finn",
-    "step_tidy_up",
-    "step_streamline",
-    "step_convert_to_hw",
-    "step_create_dataflow_partition",
-    "step_specialize_layers",
-    "step_target_fps_parallelization",
-    "step_apply_folding_config",
-    "step_minimize_bit_width",
-    "step_transpose_decomposition",
-    "step_generate_estimate_reports",
-    "step_hw_codegen",
-    "step_hw_ipgen",
-    "step_set_fifo_depths",
-    "step_create_stitched_ip",
-    "step_measure_rtlsim_performance",
-    "step_synthesize_bitfile",
-    "step_make_driver",
-    "step_deployment_package",
+    "phase_prepare_model",
+    "phase_optimize_model",
+    "phase_convert_to_hardware",
+    "phase_optimize_hardware",
+    "phase_build_hardware",
+    "phase_synthesize_hardware",
 ]
 
 #: List of steps to run for an estimate-only (no synthesis) dataflow build
 estimate_only_dataflow_steps = [
-    "step_qonnx_to_finn",
-    "step_tidy_up",
-    "step_streamline",
-    "step_convert_to_hw",
-    "step_create_dataflow_partition",
-    "step_specialize_layers",
-    "step_target_fps_parallelization",
-    "step_apply_folding_config",
-    "step_minimize_bit_width",
-    "step_generate_estimate_reports",
+    "phase_prepare_model",
+    "phase_optimize_model",
+    "phase_convert_to_hardware",
+    "phase_optimize_hardware",
 ]
 
 #: List of steps to run for a dataflow build including HW code generation, but
@@ -361,10 +342,14 @@ class DataflowBuildConfig:
     steps: Optional[List[Any]] = None
 
     #: If given, start from this step, loading the intermediate model generated
-    #: from the previous step (save_intermediate_models must be enabled)
+    #: from the previous step (save_intermediate_models must be enabled).
+    #: Note: When using phase-based builds (default), specify phase names
+    #: (e.g., "phase_build_hardware") rather than fine-grained step names.
     start_step: Optional[str] = None
 
     #: If given, stop at this step.
+    #: Note: When using phase-based builds (default), specify phase names
+    #: (e.g., "phase_build_hardware") rather than fine-grained step names.
     stop_step: Optional[str] = None
 
     #: The optional argument `max_multithreshold_bit_width` affects which Quant nodes
@@ -406,6 +391,16 @@ class DataflowBuildConfig:
     #: Warnings and info will still be printed but errors will not halt the build.
     mute_config_assertions: Optional[bool] = False
 
+    #: Inject custom steps after named steps/phases.
+    #: Dict mapping step/phase names to list of callable functions to run after that step.
+    #: Example: inject_steps_after={"phase_optimize_model": [my_custom_verification]}
+    inject_steps_after: Dict[str, List[Callable]] = field(default_factory=dict)
+
+    #: Inject custom steps before named steps/phases.
+    #: Dict mapping step/phase names to list of callable functions to run before that step.
+    #: Example: inject_steps_before={"phase_build_hardware": [my_custom_analysis]}
+    inject_steps_before: Dict[str, List[Callable]] = field(default_factory=dict)
+
     def _resolve_hls_clk_period(self):
         if self.hls_clk_period_ns is None:
             # use same clk for synth and hls if not explicitly specified