Design Space Exploration¶

Evaluate multiple hardware configurations to find optimal designs.

Brainsmith uses segment-based DSE to efficiently explore large design spaces by reusing computation across similar configurations.

explore_design_space ¶

explore_design_space(model_path: str, blueprint_path: str, output_dir: str | None = None, start_step_override: str | None = None, stop_step_override: str | None = None, verbosity: str = 'normal') -> TreeExecutionResult

Explore design space and synthesize FPGA accelerator from neural network model.

Parameters:

Name	Type	Description	Default
`model_path`	`str`	Path to ONNX model file	required
`blueprint_path`	`str`	Path to Blueprint YAML file	required
`output_dir`	`str \| None`	Output directory (defaults to $BSMITH_BUILD_DIR/dfc_YYYYMMDD_HHMMSS)	`None`
`start_step_override`	`str \| None`	Override blueprint start_step	`None`
`stop_step_override`	`str \| None`	Override blueprint stop_step	`None`
`verbosity`	`str`	Logging verbosity (quiet \| normal \| verbose \| debug)	`'normal'`

Returns:

Type	Description
`TreeExecutionResult`	TreeExecutionResult containing build artifacts and statistics

Raises:

Type	Description
`FileNotFoundError`	If model or blueprint file doesn't exist
`ValueError`	If blueprint is invalid or tree exceeds size limits
`ExecutionError`	If no successful builds were produced

Source code in brainsmith/dse/api.py

def explore_design_space(
    model_path: str,
    blueprint_path: str,
    output_dir: str | None = None,
    start_step_override: str | None = None,
    stop_step_override: str | None = None,
    verbosity: str = "normal",
) -> TreeExecutionResult:
    """
    Explore design space and synthesize FPGA accelerator from neural network model.

    Args:
        model_path: Path to ONNX model file
        blueprint_path: Path to Blueprint YAML file
        output_dir: Output directory (defaults to $BSMITH_BUILD_DIR/dfc_YYYYMMDD_HHMMSS)
        start_step_override: Override blueprint start_step
        stop_step_override: Override blueprint stop_step
        verbosity: Logging verbosity (quiet | normal | verbose | debug)

    Returns:
        TreeExecutionResult containing build artifacts and statistics

    Raises:
        FileNotFoundError: If model or blueprint file doesn't exist
        ValueError: If blueprint is invalid or tree exceeds size limits
        ExecutionError: If no successful builds were produced
    """
    # Verify files exist
    model_path_obj = Path(model_path)
    if not model_path_obj.exists():
        raise FileNotFoundError(f"Model file not found: {model_path_obj}")
    blueprint_path_obj = Path(blueprint_path)
    if not blueprint_path_obj.exists():
        raise FileNotFoundError(f"Blueprint file not found: {blueprint_path_obj}")

    # Load config early to ensure BSMITH_* vars are exported for YAML expansion
    from brainsmith.settings import get_config

    get_config()  # Exports BSMITH_DIR and other env vars

    # Determine output directory
    if not output_dir:
        build_dir = get_config().build_dir
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_dir = str(build_dir / f"dfc_{timestamp}")

    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    # Configure logging with output directory for file logging
    from brainsmith._internal.logging import setup_logging

    config = get_config()
    setup_logging(level=verbosity, output_dir=output_path, config=config.logging)

    logger.info("Exploring design space for dataflow accelerator:")
    logger.info(f"    Model: {model_path}")
    logger.info(f"    Blueprint: {blueprint_path}")
    logger.info(f"    Output: {output_dir}")

    # Parse blueprint (with optional step slicing)
    design_space, blueprint_config = parse_blueprint(
        blueprint_path,
        str(Path(model_path).absolute()),
        start_step=start_step_override,
        stop_step=stop_step_override,
    )

    # Build DSE tree
    tree_builder = DSETreeBuilder()
    tree = tree_builder.build_tree(design_space, blueprint_config)

    logger.debug(
        f"Design space: {len(design_space.steps)} steps, "
        f"{len(design_space.kernel_backends)} kernels"
    )

    # Log tree statistics
    stats = tree.get_statistics()
    logger.debug("DSE tree:")
    logger.debug(f"  - Total paths: {stats['total_paths']:,}")
    logger.debug(f"  - Total segments: {stats['total_segments']:,}")
    logger.debug(f"  - Segment efficiency: {stats['segment_efficiency']}%")

    # Create runner and execute
    finn_adapter = FINNAdapter()
    runner = SegmentRunner(finn_adapter, tree.root.finn_config)
    results = runner.run_tree(
        tree=tree, initial_model=Path(model_path), output_dir=Path(output_dir)
    )

    # Validate results
    results.validate_success(Path(output_dir))

    # Log success
    result_stats = results.compute_stats()
    logger.info("Design space exploration completed successfully")
    logger.info(f"  Successful builds: {result_stats['successful']}/{result_stats['total']}")
    logger.info(f"  Total time: {results.total_time:.2f}s")
    logger.info(f"  Output directory: {output_dir}")

    return TreeExecutionResult(
        segment_results=results.segment_results,
        total_time=results.total_time,
        design_space=design_space,
        dse_tree=tree,
    )

Example:

from brainsmith import explore_design_space
from brainsmith.dse import SegmentStatus

# Run complete DSE pipeline
results = explore_design_space(
    model_path="model.onnx",
    blueprint_path="blueprint.yaml",
    output_dir="./build"
)

# Analyze results
stats = results.compute_stats()
print(f"Successful: {stats['successful']}/{stats['total']}")
print(f"Total time: {results.total_time:.2f}s")

# Access successful outputs
for seg_id, result in results.segment_results.items():
    if result.status == SegmentStatus.COMPLETED:
        print(f"Output: {result.output_model}")

parse_blueprint ¶

parse_blueprint(blueprint_path: str, model_path: str, start_step: str | None = None, stop_step: str | None = None) -> tuple[GlobalDesignSpace, DSEConfig]

Parse blueprint YAML to design space and configuration.

Supports blueprint inheritance via the 'extends' field.

Parameters:

Name	Type	Description	Default
`blueprint_path`	`str`	Path to blueprint YAML file	required
`model_path`	`str`	Path to model file	required
`start_step`	`str \| None`	Optional start step override	`None`
`stop_step`	`str \| None`	Optional stop step override	`None`

Returns:

Type	Description
`tuple[GlobalDesignSpace, DSEConfig]`	Tuple of (GlobalDesignSpace, DSEConfig)

Source code in brainsmith/dse/_parser/__init__.py

def parse_blueprint(
    blueprint_path: str,
    model_path: str,
    start_step: str | None = None,
    stop_step: str | None = None,
) -> tuple[GlobalDesignSpace, DSEConfig]:
    """Parse blueprint YAML to design space and configuration.

    Supports blueprint inheritance via the 'extends' field.

    Args:
        blueprint_path: Path to blueprint YAML file
        model_path: Path to model file
        start_step: Optional start step override
        stop_step: Optional stop step override

    Returns:
        Tuple of (GlobalDesignSpace, DSEConfig)
    """
    # Load blueprint data and check for inheritance
    raw_data, merged_data, parent_path = load_blueprint_with_inheritance(blueprint_path)

    parent_steps = None

    # If this blueprint extends another, first parse the parent
    if parent_path:
        # Recursively parse parent to get its fully resolved steps
        parent_design_space, _ = parse_blueprint(parent_path, model_path)
        parent_steps = parent_design_space.steps

    # Extract config from merged data
    blueprint_config = extract_config(merged_data)

    # Override start_step/stop_step from parameters if provided
    if start_step is not None:
        blueprint_config.start_step = start_step
    if stop_step is not None:
        blueprint_config.stop_step = stop_step

    # Log step range if specified
    if blueprint_config.start_step or blueprint_config.stop_step:
        logger.info(
            f"Step range: start={blueprint_config.start_step or 'beginning'}, "
            f"stop={blueprint_config.stop_step or 'end'}"
        )

    # Parse steps from THIS blueprint only (not inherited steps)
    # Use raw_data to get only the steps defined in this file
    steps_data = raw_data.get("design_space", {}).get("steps", [])
    steps = parse_steps(steps_data, parent_steps=parent_steps)

    # Parse kernels (use merged data to inherit kernels)
    kernel_backends = parse_kernels(merged_data.get("design_space", {}).get("kernels", []))

    # Get max_combinations from environment or use default
    max_combinations = int(os.environ.get("BRAINSMITH_MAX_COMBINATIONS", "100000"))

    design_space = GlobalDesignSpace(
        model_path=model_path,
        steps=steps,
        kernel_backends=kernel_backends,
        max_combinations=max_combinations,
    )
    return design_space, blueprint_config

Example:

from brainsmith.dse import parse_blueprint

design_space, config = parse_blueprint(
    blueprint_path="blueprint.yaml",
    model_path="model.onnx"
)

print(f"Steps: {len(design_space.steps)}")
print(f"Kernels: {design_space.kernel_backends}")

build_tree ¶

build_tree(design_space: GlobalDesignSpace, config: DSEConfig) -> DSETree

Build execution tree from design space.

Separates tree construction from execution for inspection and validation.

Parameters:

Name	Type	Description	Default
`design_space`	`GlobalDesignSpace`	GlobalDesignSpace with steps and kernel backends	required
`config`	`DSEConfig`	DSEConfig with FINN configuration	required

Returns:

Type	Description
`DSETree`	DSETree ready for execution

Raises:

Type	Description
`ValueError`	If tree exceeds max_combinations limit

Source code in brainsmith/dse/api.py

def build_tree(design_space: GlobalDesignSpace, config: DSEConfig) -> DSETree:
    """Build execution tree from design space.

    Separates tree construction from execution for inspection and validation.

    Args:
        design_space: GlobalDesignSpace with steps and kernel backends
        config: DSEConfig with FINN configuration

    Returns:
        DSETree ready for execution

    Raises:
        ValueError: If tree exceeds max_combinations limit
    """
    builder = DSETreeBuilder()
    return builder.build_tree(design_space, config)

Example:

from brainsmith.dse import parse_blueprint, build_tree

design_space, config = parse_blueprint(
    blueprint_path="blueprint.yaml",
    model_path="model.onnx"
)

tree = build_tree(design_space, config)

# Inspect before execution
stats = tree.get_statistics()
print(f"Paths: {stats['total_paths']:,}")
print(f"Segments: {stats['total_segments']:,}")

execute_tree ¶

execute_tree(tree: DSETree, model_path: str, config: DSEConfig, output_dir: str, runner: SegmentRunner | None = None) -> TreeExecutionResult

Execute a pre-built DSE tree.

Separates execution from tree construction for custom execution strategies.

Parameters:

Name	Type	Description	Default
`tree`	`DSETree`	Pre-built DSETree to execute	required
`model_path`	`str`	Path to initial ONNX model	required
`config`	`DSEConfig`	DSEConfig (used for validation)	required
`output_dir`	`str`	Base output directory	required
`runner`	`SegmentRunner \| None`	Custom segment runner (uses default if None)	`None`

Returns:

Type	Description
`TreeExecutionResult`	TreeExecutionResult with segment results

Raises:

Type	Description
`ExecutionError`	If no successful builds were produced

Source code in brainsmith/dse/api.py

def execute_tree(
    tree: DSETree,
    model_path: str,
    config: DSEConfig,
    output_dir: str,
    runner: SegmentRunner | None = None,
) -> TreeExecutionResult:
    """Execute a pre-built DSE tree.

    Separates execution from tree construction for custom execution strategies.

    Args:
        tree: Pre-built DSETree to execute
        model_path: Path to initial ONNX model
        config: DSEConfig (used for validation)
        output_dir: Base output directory
        runner: Custom segment runner (uses default if None)

    Returns:
        TreeExecutionResult with segment results

    Raises:
        ExecutionError: If no successful builds were produced
    """
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    # Create default runner if not provided
    if runner is None:
        finn_adapter = FINNAdapter()
        runner = SegmentRunner(finn_adapter, tree.root.finn_config)

    # Execute tree
    results = runner.run_tree(tree=tree, initial_model=Path(model_path), output_dir=output_path)

    # Validate results
    results.validate_success(output_path)

    # Return result with all fields
    return TreeExecutionResult(
        segment_results=results.segment_results,
        total_time=results.total_time,
        design_space=None,  # Not available in this path
        dse_tree=tree,
    )

Example:

from brainsmith.dse import build_tree, execute_tree

tree = build_tree(design_space, config)

result = execute_tree(
    tree=tree,
    model_path="model.onnx",
    config=config,
    output_dir="./build"
)

SegmentRunner ¶

SegmentRunner(finn_adapter: FINNAdapter, base_config: dict[str, Any])

Runs DSE segments using FINN.

Handles both tree traversal and individual segment execution using FINNAdapter for all FINN interactions.

Initialize runner.

Parameters:

Name	Type	Description	Default
`finn_adapter`	`FINNAdapter`	Adapter for FINN-specific operations	required
`base_config`	`dict[str, Any]`	FINN configuration from blueprint	required

Source code in brainsmith/dse/runner.py

def __init__(self, finn_adapter: FINNAdapter, base_config: dict[str, Any]) -> None:
    """Initialize runner.

    Args:
        finn_adapter: Adapter for FINN-specific operations
        base_config: FINN configuration from blueprint
    """
    self.finn_adapter = finn_adapter
    self.base_config = base_config

    # Extract settings from FINN config
    self.fail_fast = False  # TODO: Add more robust tree exit options
    output_products = base_config.get("output_products", ["estimates"])
    output_product = output_products[0] if output_products else "estimates"
    self.output_type = OutputType.from_finn_product(output_product)

run_segment ¶

run_segment(segment: DSESegment, input_model: Path, base_output_dir: Path) -> SegmentResult

Run a single DSE segment.

Parameters:

Name	Type	Description	Default
`segment`	`DSESegment`	Segment to execute	required
`input_model`	`Path`	Input ONNX model path	required
`base_output_dir`	`Path`	Base output directory	required

Returns:

Type	Description
`SegmentResult`	SegmentResult with execution details

Source code in brainsmith/dse/runner.py

def run_segment(
    self, segment: DSESegment, input_model: Path, base_output_dir: Path
) -> SegmentResult:
    """Run a single DSE segment.

    Args:
        segment: Segment to execute
        input_model: Input ONNX model path
        base_output_dir: Base output directory

    Returns:
        SegmentResult with execution details
    """
    segment_dir = base_output_dir / segment.segment_id
    output_model = segment_dir / "output.onnx"

    # Check cache validity
    if output_model.exists():
        try:
            onnx.load(str(output_model))
            # Valid cache - return immediately
            logger.debug(f"Cache hit: {segment.segment_id}")
            return SegmentResult(
                segment_id=segment.segment_id,
                status=SegmentStatus.COMPLETED,
                output_model=output_model,
                output_dir=segment_dir,
                cached=True,
            )
        except (OnnxValidationError, OnnxInferenceError) as e:
            # Invalid ONNX model - rebuild
            logger.warning(f"Invalid cache for {segment.segment_id}, rebuilding: {e}")
            output_model.unlink()
        except OSError as e:
            # File corruption (rare but possible)
            logger.warning(f"Corrupted cache for {segment.segment_id}, rebuilding: {e}")
            output_model.unlink()

    # Cache miss or invalid - execute build
    logger.debug(f"Building segment: {segment.segment_id}")

    # Create FINN config
    finn_config = self._make_finn_config(segment, segment_dir)

    # Prepare directory and model
    segment_dir.mkdir(parents=True, exist_ok=True)
    segment_input = segment_dir / "input.onnx"
    self.finn_adapter.prepare_model(input_model, segment_input)

    # Execute build
    start_time = time.time()

    try:
        # Use adapter for clean FINN interaction
        final_model = self.finn_adapter.build(segment_input, finn_config, segment_dir)

        if final_model:
            # Copy to expected location
            self.finn_adapter.prepare_model(final_model, output_model)
            logger.debug(
                f"Completed segment: {segment.segment_id} ({time.time() - start_time:.1f}s)"
            )
            return SegmentResult(
                segment_id=segment.segment_id,
                status=SegmentStatus.COMPLETED,
                output_model=output_model,
                output_dir=segment_dir,
                execution_time=time.time() - start_time,
            )
        else:
            raise RuntimeError("Build succeeded but no output model generated")

    except Exception as e:
        contextualized_error = self._add_segment_context(segment.segment_id, e)
        logger.error(f"Segment failed: {segment.segment_id}: {contextualized_error}")
        if not isinstance(e, ExecutionError):
            logger.exception("Unexpected error details:")
        raise contextualized_error

run_tree ¶

run_tree(tree: DSETree, initial_model: Path, output_dir: Path) -> TreeExecutionResult

Run all segments in the DSE tree.

Parameters:

Name	Type	Description	Default
`tree`	`DSETree`	DSE tree to execute	required
`initial_model`	`Path`	Path to initial ONNX model	required
`output_dir`	`Path`	Base output directory	required

Returns:

Type	Description
`TreeExecutionResult`	TreeExecutionResult with all segment results

Source code in brainsmith/dse/runner.py

def run_tree(self, tree: DSETree, initial_model: Path, output_dir: Path) -> TreeExecutionResult:
    """Run all segments in the DSE tree.

    Args:
        tree: DSE tree to execute
        initial_model: Path to initial ONNX model
        output_dir: Base output directory

    Returns:
        TreeExecutionResult with all segment results
    """
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)

    logger.debug(f"Executing tree with fail_fast={self.fail_fast}")
    logger.debug(f"Output directory: {output_dir}")

    results = {}
    skipped = set()
    start_time = time.time()

    # Use a stack for cleaner iteration
    stack = [(tree.root, initial_model, 0)]

    while stack:
        segment, input_model, depth = stack.pop()
        indent = "  " * depth

        # Skip if parent failed
        if segment.segment_id in skipped:
            logger.warning(f"{indent}Skipped: {segment.segment_id}")
            results[segment.segment_id] = SegmentResult(
                segment_id=segment.segment_id,
                status=SegmentStatus.SKIPPED,
                error="Parent segment failed",
            )
            continue

        # Execute segment
        logger.debug(f"{indent}Executing: {segment.segment_id}")

        # Skip empty segments (e.g., root with immediate branches)
        if not segment.steps:
            logger.debug(f"{indent}  (empty segment, passing through)")
            # Create a pass-through result
            results[segment.segment_id] = SegmentResult(
                segment_id=segment.segment_id,
                status=SegmentStatus.COMPLETED,
                output_model=input_model,  # Pass input as output
                output_dir=output_dir / segment.segment_id,
                execution_time=0,
            )
            # Add children to stack
            for child in reversed(list(segment.children.values())):
                stack.append((child, input_model, depth + 1))
            continue

        try:
            result = self.run_segment(segment, input_model, output_dir)
            results[segment.segment_id] = result
        except KeyboardInterrupt:
            logger.warning("Build cancelled by user")
            raise
        except Exception as e:
            wrapped_error = self._add_segment_context(segment.segment_id, e)
            logger.error(f"Segment failed: {segment.segment_id}: {wrapped_error}")
            if not isinstance(e, ExecutionError):
                logger.exception("Unexpected error details:")

            if self.fail_fast:
                raise wrapped_error

            # Create failure result
            results[segment.segment_id] = SegmentResult(
                segment_id=segment.segment_id,
                status=SegmentStatus.FAILED,
                error=str(wrapped_error),
                execution_time=0,
            )

            # Mark descendants for skipping
            self._mark_descendants_skipped(segment, skipped)
            for child in reversed(list(segment.children.values())):
                stack.append((child, None, depth + 1))
            continue

        # Share artifacts at branch points
        if segment.is_branch_point:
            _share_artifacts_at_branch(result, list(segment.children.values()), output_dir)

        # Add children to stack (reversed for correct order)
        for child in reversed(list(segment.children.values())):
            stack.append((child, result.output_model, depth + 1))

    # Create result and print summary
    total_time = time.time() - start_time
    result = TreeExecutionResult(results, total_time)
    self._print_summary(result)

    return result

DSEConfig `dataclass` ¶

DSEConfig(clock_ns: float, output: OutputType = OutputType.ESTIMATES, board: str | None = None, start_step: str | None = None, stop_step: str | None = None, finn_overrides: dict[str, Any] = dict())

Configuration for Design Space Exploration.

Attributes:

Name	Type	Description
`clock_ns`	`float`	Target clock period in nanoseconds
`output`	`OutputType`	Output type (estimates, rtl, or bitfile)
`board`	`str \| None`	Target board (required for rtl/bitfile)
`start_step`	`str \| None`	Optional pipeline starting step
`stop_step`	`str \| None`	Optional pipeline ending step
`finn_overrides`	`dict[str, Any]`	Direct FINN configuration overrides

__post_init__ ¶

__post_init__() -> None

Validate configuration invariants.

Source code in brainsmith/dse/config.py

def __post_init__(self) -> None:
    """Validate configuration invariants."""
    # Validate clock_ns is positive
    if self.clock_ns <= 0:
        raise ValueError(f"clock_ns must be positive, got {self.clock_ns}")

    # Validate output type dependencies
    if self.output != OutputType.ESTIMATES and not self.board:
        raise ValueError(f"{self.output.value} requires board specification")

GlobalDesignSpace `dataclass` ¶

GlobalDesignSpace(model_path: str, steps: list[str | list[str | None]], kernel_backends: list[tuple[str, list[type]]], max_combinations: int = 100000)

Design space ready for DSE tree construction.

Attributes:

Name	Type	Description
`model_path`	`str`	Path to ONNX model file
`steps`	`list[str \| list[str \| None]]`	Pipeline steps with optional variations (list = branch point)
`kernel_backends`	`list[tuple[str, list[type]]]`	Kernel names mapped to backend classes
`max_combinations`	`int`	Maximum allowed design space size (default: 100,000)

Validates size on initialization and provides kernel summary formatting.

__post_init__ ¶

__post_init__()

Validate design space after initialization.

Source code in brainsmith/dse/design_space.py

def __post_init__(self):
    """Validate design space after initialization."""
    self._validate_size()

str ¶

__str__() -> str

Human-readable representation.

Source code in brainsmith/dse/design_space.py

def __str__(self) -> str:
    """Human-readable representation."""
    return (
        f"GlobalDesignSpace(\n"
        f"  model: {self.model_path}\n"
        f"  steps: {len(self.steps)}\n"
        f"  kernels: {len(self.kernel_backends)}\n"
        f")"
    )

get_kernel_summary ¶

get_kernel_summary() -> str

Get human-readable summary of kernels and backends.

Source code in brainsmith/dse/design_space.py

def get_kernel_summary(self) -> str:
    """Get human-readable summary of kernels and backends."""
    lines = []
    for kernel_name, backend_classes in self.kernel_backends:
        backend_names = [cls.__name__ for cls in backend_classes]
        lines.append(f"  {kernel_name}: {', '.join(backend_names)}")
    return "\n".join(lines)

TreeExecutionResult `dataclass` ¶

TreeExecutionResult(segment_results: dict[str, SegmentResult], total_time: float, design_space: GlobalDesignSpace | None = None, dse_tree: DSETree | None = None)

Results from design space exploration execution.

Attributes:

Name	Type	Description
`segment_results`	`dict[str, SegmentResult]`	Execution results for each DSE segment
`total_time`	`float`	Total execution time in seconds
`design_space`	`GlobalDesignSpace \| None`	Original design space (if available)
`dse_tree`	`DSETree \| None`	Execution tree structure (if available)

compute_stats ¶

compute_stats() -> dict[str, int]

Compute execution statistics.

Returns:

Type	Description
`dict[str, int]`	Dict with counts: total, successful, failed, cached, skipped

Source code in brainsmith/dse/types.py

def compute_stats(self) -> dict[str, int]:
    """Compute execution statistics.

    Returns:
        Dict with counts: total, successful, failed, cached, skipped
    """
    total = successful = failed = cached = skipped = 0

    for r in self.segment_results.values():
        total += 1
        if r.status == SegmentStatus.COMPLETED:
            if r.cached:
                cached += 1
            else:
                successful += 1
        elif r.status == SegmentStatus.FAILED:
            failed += 1
        elif r.status == SegmentStatus.SKIPPED:
            skipped += 1

    return {
        "total": total,
        "successful": successful,
        "failed": failed,
        "cached": cached,
        "skipped": skipped,
    }

validate_success ¶

validate_success(output_dir: Path) -> None

Validate that results contain at least one successful build.

Parameters:

Name	Type	Description	Default
`output_dir`	`Path`	Output directory for error messages	required

Raises:

Type	Description
`ExecutionError`	If no valid builds exist

Source code in brainsmith/dse/types.py

def validate_success(self, output_dir: Path) -> None:
    """Validate that results contain at least one successful build.

    Args:
        output_dir: Output directory for error messages

    Raises:
        ExecutionError: If no valid builds exist
    """
    stats = self.compute_stats()
    valid_builds = stats["successful"] + stats["cached"]

    if valid_builds == 0:
        raise ExecutionError(
            f"DSE failed: No successful builds\n"
            f"  Failed: {stats['failed']}\n"
            f"  Skipped: {stats['skipped']}\n"
            f"  Check segment logs in: {output_dir}/*/\n"
            f"  Run with --log-level debug for detailed output"
        )

    if stats["successful"] == 0 and stats["cached"] > 0:
        logger.warning(
            f"All builds used cached results ({stats['cached']} cached). "
            f"No new builds were executed."
        )

Example:

results = explore_design_space(
    model_path="model.onnx",
    blueprint_path="blueprint.yaml",
    output_dir="./build"
)

# Compute statistics
stats = results.compute_stats()
print(stats['successful'], stats['failed'], stats['total'])
# Also available: stats['cached'], stats['skipped']

# Access individual segments
for seg_id, seg_result in results.segment_results.items():
    print(f"{seg_id}: {seg_result.status}")

SegmentResult `dataclass` ¶

SegmentResult(segment_id: str, status: SegmentStatus, output_model: Path | None = None, output_dir: Path | None = None, error: str | None = None, execution_time: float = 0, cached: bool = False)

Result from executing a single DSE segment.

Attributes:

Name	Type	Description
`segment_id`	`str`	Unique identifier for the segment
`status`	`SegmentStatus`	Execution status (completed, failed, skipped, etc.)
`output_model`	`Path \| None`	Path to output ONNX model (if successful)
`output_dir`	`Path \| None`	Directory containing build artifacts
`error`	`str \| None`	Error message (if failed)
`execution_time`	`float`	Execution time in seconds
`cached`	`bool`	Whether result was retrieved from cache

SegmentStatus ¶

Bases: Enum

Execution status for DSE segments.

Attributes:

Name	Type	Description
`PENDING`		Segment not yet executed
`RUNNING`		Segment currently executing
`COMPLETED`		Segment executed successfully
`FAILED`		Segment execution failed
`SKIPPED`		Segment skipped due to parent failure

OutputType ¶

Bases: Enum

Build output types for DSE execution.

Attributes:

Name	Type	Description
`ESTIMATES`		Performance estimates only (fastest)
`RTL`		RTL simulation and IP generation
`BITFILE`		Full bitstream generation (slowest)

from_finn_product `classmethod` ¶

from_finn_product(product: str) -> OutputType

Get OutputType from FINN product string.

Parameters:

Name	Type	Description	Default
`product`	`str`	FINN product name (e.g., 'estimates', 'bitfile', 'rtl_sim')	required

Returns:

Type	Description
`OutputType`	Matching OutputType

Raises:

Type	Description
`ValueError`	If product is unknown

Example

OutputType.from_finn_product('bitfile') OutputType.from_finn_product('rtl_sim')

Source code in brainsmith/dse/types.py

@classmethod
def from_finn_product(cls, product: str) -> OutputType:
    """Get OutputType from FINN product string.

    Args:
        product: FINN product name (e.g., 'estimates', 'bitfile', 'rtl_sim')

    Returns:
        Matching OutputType

    Raises:
        ValueError: If product is unknown

    Example:
        >>> OutputType.from_finn_product('bitfile')
        <OutputType.BITFILE: 'bitfile'>
        >>> OutputType.from_finn_product('rtl_sim')
        <OutputType.RTL: 'rtl'>
    """
    for output_type in cls:
        if product in output_type.to_finn_products():
            return output_type

    # Product not found - create helpful error message
    all_products = [p for ot in cls for p in ot.to_finn_products()]
    raise ValueError(
        f"Unknown FINN product '{product}'. " f"Valid products: {', '.join(all_products)}"
    )

to_finn_outputs ¶

to_finn_outputs() -> list[str]

Convert to FINN generate_outputs configuration.

Source code in brainsmith/dse/types.py

def to_finn_outputs(self) -> list[str]:
    """Convert to FINN generate_outputs configuration."""
    return {
        OutputType.ESTIMATES: ["estimate_reports"],
        OutputType.RTL: ["estimate_reports", "rtlsim_performance", "stitched_ip"],
        OutputType.BITFILE: [
            "estimate_reports",
            "rtlsim_performance",
            "stitched_ip",
            "bitfile",
            "deployment_package",
        ],
    }[self]

to_finn_products ¶

to_finn_products() -> list[str]

Convert to FINN output_products configuration.

Source code in brainsmith/dse/types.py

def to_finn_products(self) -> list[str]:
    """Convert to FINN output_products configuration."""
    return {
        OutputType.ESTIMATES: ["estimates"],
        OutputType.RTL: ["rtl_sim", "ip_gen"],
        OutputType.BITFILE: ["bitfile"],
    }[self]

ExecutionError ¶

Bases: Exception

Exception raised during DSE execution failures.

DSETree ¶

DSETree(root: DSESegment)

Design space exploration tree structure.

Represents the complete exploration space as a tree of segments, where each segment contains steps to execute between branch points.

Attributes:

Name	Type	Description
`root`		Root segment of the tree

Source code in brainsmith/dse/tree.py

def __init__(self, root: DSESegment):
    self.root = root

format_tree ¶

format_tree() -> str

Format tree as a string representation.

Returns:

Type	Description
`str`	Multi-line string with ASCII tree visualization

Example

tree = build_tree(design_space, config) print(tree.format_tree()) └── transform_step_1 (3 steps) ├── kernel_backend_A (2 steps) └── kernel_backend_B (2 steps)

Source code in brainsmith/dse/tree.py

def format_tree(self) -> str:
    """Format tree as a string representation.

    Returns:
        Multi-line string with ASCII tree visualization

    Example:
        >>> tree = build_tree(design_space, config)
        >>> print(tree.format_tree())
        └── transform_step_1 (3 steps)
            ├── kernel_backend_A (2 steps)
            └── kernel_backend_B (2 steps)
    """
    lines = []
    self._format_node(self.root, "", True, lines)
    return "\n".join(lines)

get_all_segments ¶

get_all_segments() -> list[DSESegment]

Get all segments in the tree.

Source code in brainsmith/dse/tree.py

def get_all_segments(self) -> list[DSESegment]:
    """Get all segments in the tree."""
    all_segments = []

    def collect_segments(node: DSESegment):
        all_segments.append(node)
        for child in node.children.values():
            collect_segments(child)

    collect_segments(self.root)
    return all_segments

get_execution_order ¶

get_execution_order() -> list[DSESegment]

Get breadth-first execution order for the tree.

Skips root if it has no steps (purely structural).

Source code in brainsmith/dse/tree.py

def get_execution_order(self) -> list[DSESegment]:
    """Get breadth-first execution order for the tree.

    Skips root if it has no steps (purely structural).
    """
    # Start from root's children if root is structural-only
    if not self.root.steps and self.root.children:
        queue = deque(self.root.children.values())
    else:
        queue = deque([self.root])

    order = []

    while queue:
        node = queue.popleft()  # O(1)
        order.append(node)
        queue.extend(node.children.values())

    return order

get_statistics ¶

get_statistics() -> dict[str, Any]

Get statistics about the DSE tree.

Source code in brainsmith/dse/tree.py

def get_statistics(self) -> dict[str, Any]:
    """Get statistics about the DSE tree."""
    stats = {"nodes": 0, "leaves": 0, "max_depth": 0, "total_steps": 0, "leaf_steps": []}

    def traverse(node: DSESegment, depth: int = 0):
        stats["nodes"] += 1
        stats["total_steps"] += len(node.steps)
        stats["max_depth"] = max(stats["max_depth"], depth)

        if not node.children:
            # Leaf node
            stats["leaves"] += 1
            stats["leaf_steps"].append(len(node.get_all_steps()))
        else:
            for child in node.children.values():
                traverse(child, depth + 1)

    traverse(self.root)

    # Calculate efficiency
    steps_without_segments = sum(stats["leaf_steps"])
    segment_efficiency = (
        1 - stats["total_steps"] / steps_without_segments if steps_without_segments else 0
    )

    return {
        "total_paths": stats["leaves"],
        "total_segments": stats["nodes"],
        "max_depth": stats["max_depth"],
        "total_steps": stats["total_steps"],
        "steps_without_segments": steps_without_segments,
        "segment_efficiency": round(segment_efficiency * 100, 1),
        "avg_steps_per_segment": (
            round(stats["total_steps"] / stats["nodes"], 1) if stats["nodes"] > 0 else 0
        ),
    }

DSESegment `dataclass` ¶

DSESegment(steps: list[dict[str, Any]], branch_choice: str | None = None, parent: DSESegment | None = None, children: dict[str, DSESegment] = dict(), status: SegmentStatus = SegmentStatus.PENDING, output_dir: Path | None = None, error: str | None = None, execution_time: float | None = None, finn_config: dict[str, Any] = dict())

A segment in the design space exploration tree.

Each segment is executed as a single FINN build, containing all steps from the last branch point (or root) to the next branch point (or leaf).

segment_id `cached` `property` ¶

segment_id: str

Deterministic ID from branch path (cached for O(1) access).

add_child ¶

add_child(branch_id: str, steps: list[dict[str, Any]]) -> DSESegment

Create a child segment for a branch.

Source code in brainsmith/dse/segment.py

def add_child(self, branch_id: str, steps: list[dict[str, Any]]) -> DSESegment:
    """Create a child segment for a branch."""
    child = DSESegment(
        steps=steps, branch_choice=branch_id, parent=self, finn_config=self.finn_config.copy()
    )
    self.children[branch_id] = child
    return child

count_descendants ¶

count_descendants() -> int

Count total number of descendant nodes.

Source code in brainsmith/dse/segment.py

def count_descendants(self) -> int:
    """Count total number of descendant nodes."""
    count = len(self.children)
    for child in self.children.values():
        count += child.count_descendants()
    return count

get_all_steps ¶

get_all_steps() -> list[dict[str, Any]]

Get all steps from root to end of this segment.

Source code in brainsmith/dse/segment.py

def get_all_steps(self) -> list[dict[str, Any]]:
    """Get all steps from root to end of this segment."""
    steps = []
    for segment in self.get_path():
        steps.extend(segment.steps)
    return steps

get_path ¶

get_path() -> list[DSESegment]

Get all segments from root to here.

Source code in brainsmith/dse/segment.py

def get_path(self) -> list[DSESegment]:
    """Get all segments from root to here."""
    path = []
    node = self
    while node:
        path.append(node)
        node = node.parent
    path.reverse()
    return path

Design Space Exploration¶

explore_design_space ¶

parse_blueprint ¶

build_tree ¶

execute_tree ¶

SegmentRunner ¶

run_segment ¶

run_tree ¶

DSEConfig dataclass ¶

__post_init__ ¶

GlobalDesignSpace dataclass ¶

__post_init__ ¶

__str__ ¶

get_kernel_summary ¶

TreeExecutionResult dataclass ¶

compute_stats ¶

validate_success ¶

SegmentResult dataclass ¶

SegmentStatus ¶

OutputType ¶

from_finn_product classmethod ¶

to_finn_outputs ¶

to_finn_products ¶

ExecutionError ¶

DSETree ¶

format_tree ¶

get_all_segments ¶

get_execution_order ¶

get_statistics ¶

DSESegment dataclass ¶

segment_id cached property ¶

add_child ¶

count_descendants ¶

get_all_steps ¶

get_path ¶

See Also¶

DSEConfig `dataclass` ¶

GlobalDesignSpace `dataclass` ¶

str ¶

TreeExecutionResult `dataclass` ¶

SegmentResult `dataclass` ¶

from_finn_product `classmethod` ¶

DSESegment `dataclass` ¶

segment_id `cached` `property` ¶