Dataflow Modeling

Schema-driven kernel modeling for ONNX-to-hardware transformation with efficient design space exploration.

Two-phase construction separates expensive setup from fast configuration: Design Space is built once and defines valid parameter ranges, while Design Point is configured many times to represent specific hardware instances. This enables efficient exploration by avoiding redundant computation.

---

KernelOp

KernelOp(onnx_node, **kwargs)

Bases: HWCustomOp, ABC

Kernel operator base class.

Shapes extracted from ModelWrapper context, never stored in nodeattrs. Subclasses implement build_schema() to construct their KernelSchema.

Caching Strategy

• design_space: Cached (expensive to build, invalidated on structural changes)
• design_point: Regenerated from nodeattrs (guarantees consistency)

For execution compatibility notes, see module docstring.

Source code in brainsmith/dataflow/kernel_op.py

def __init__(self, onnx_node, **kwargs):
    super().__init__(onnx_node, **kwargs)
    # Build and freeze schema from node structure
    self.kernel_schema = self.build_schema(onnx_node, model=None)

    # Design space caching for DSE performance
    # Design points are regenerated on each access to ensure consistency with nodeattrs
    self._design_space: "KernelDesignSpace" | None = (
        None  # Cached, call invalidate() to rebuild
    )

design_point property

design_point: KernelDesignPoint

Current kernel configuration as design point (regenerated from nodeattrs).

This property regenerates on every access to ensure consistency with current nodeattrs. For better performance when accessing multiple properties, cache the design point in a local variable:

Example

GOOD: Cache locally for multiple accesses

point = self.design_point simd = point.inputs["input"].stream_shape[-1] width = point.inputs["input"].tensor_shape[-1] dtype = point.inputs["input"].datatype

AVOID: Multiple accesses trigger multiple rebuilds

simd = self.design_point.inputs["input"].stream_shape[-1] width = self.design_point.inputs["input"].tensor_shape[-1] dtype = self.design_point.inputs["input"].datatype

design_space property

design_space: KernelDesignSpace

Cached design space (call method with model_w first to initialize).

apply_design_point

apply_design_point(point: KernelDesignPoint) -> None

Apply chosen design point to nodeattrs (persist to ONNX).

Syncs design point configuration back to node attributes. The design_point property will regenerate from these nodeattrs on next access.

Parameters:

Name	Type	Description	Default
point	KernelDesignPoint	Design point to apply	required

Raises:

Type	Description
ValueError	If point from different design space
RuntimeError	If design space not initialized

Example

DSE exploration

best_point = None best_cycles = float('inf') for point in op.design_space.sweep_dimension("SIMD"): ... if point.initiation_interval < best_cycles: ... best_cycles = point.initiation_interval ... best_point = point

Apply winner to node

op.apply_design_point(best_point)

Source code in brainsmith/dataflow/kernel_op.py

def apply_design_point(self, point: "KernelDesignPoint") -> None:
    """Apply chosen design point to nodeattrs (persist to ONNX).

    Syncs design point configuration back to node attributes.
    The design_point property will regenerate from these nodeattrs on next access.

    Args:
        point: Design point to apply

    Raises:
        ValueError: If point from different design space
        RuntimeError: If design space not initialized

    Example:
        >>> # DSE exploration
        >>> best_point = None
        >>> best_cycles = float('inf')
        >>> for point in op.design_space.sweep_dimension("SIMD"):
        ...     if point.initiation_interval < best_cycles:
        ...         best_cycles = point.initiation_interval
        ...         best_point = point
        >>>
        >>> # Apply winner to node
        >>> op.apply_design_point(best_point)
    """
    if self._design_space is None:
        raise RuntimeError(
            f"{self.onnx_node.name}: Design space not initialized. "
            f"Call a method with model_w parameter first."
        )

    if point.design_space is not self._design_space:
        raise ValueError(
            f"{self.onnx_node.name}: DesignPoint from different DesignSpace. "
            f"Cannot apply to this node."
        )

    # Sync config → nodeattrs (bypass set_nodeattr to avoid invalidation)
    for dim_name, value in point.config.items():
        super().set_nodeattr(dim_name, value)

build_design_space

build_design_space(model_w: ModelWrapper) -> None

FINN API compatibility: Build design space.

This method provides compatibility with FINN's getHWCustomOp() utility, which detects KernelOp via kernel_schema attribute and calls this method.

Parameters:

Name	Type	Description	Default
model_w	ModelWrapper	ModelWrapper for graph context	required

Source code in brainsmith/dataflow/kernel_op.py

def build_design_space(self, model_w: ModelWrapper) -> None:
    """FINN API compatibility: Build design space.

    This method provides compatibility with FINN's getHWCustomOp() utility,
    which detects KernelOp via kernel_schema attribute and calls this method.

    Args:
        model_w: ModelWrapper for graph context
    """
    # Delegate to existing lazy initialization
    self._ensure_ready(model_w)

build_schema abstractmethod classmethod

build_schema(node: NodeProto, model: ModelWrapper | None) -> KernelSchema

Build kernel schema from ONNX node.

Polymorphic method that handles both static and dynamic schemas: - Static schemas: return constant, ignore parameters - Dynamic schemas: inspect node structure to build schema

Called in two contexts: 1. During init: model=None (schema built for instance) 2. During can_infer_from(): model provided (schema built for validation)

Parameters:

Name	Type	Description	Default
node	NodeProto	ONNX node (provides inputs, outputs, attributes)	required
model	ModelWrapper | None	Optional ModelWrapper (provides shapes, datatypes for validation context)	required

Returns:

Type	Description
KernelSchema	KernelSchema defining kernel structure

Example (static schema): @classmethod def build_schema(cls, node, model): return LAYERNORM_SCHEMA

Example (dynamic schema): @classmethod def build_schema(cls, node, model): num_inputs = len(node.input) inputs = [InputSchema(name=f"input{i}", ...) for i in range(num_inputs)] return KernelSchema(name="Concat", inputs=inputs, outputs=[...])

Source code in brainsmith/dataflow/kernel_op.py

@classmethod
@abstractmethod
def build_schema(cls, node: NodeProto, model: ModelWrapper | None) -> KernelSchema:
    """Build kernel schema from ONNX node.

    Polymorphic method that handles both static and dynamic schemas:
    - Static schemas: return constant, ignore parameters
    - Dynamic schemas: inspect node structure to build schema

    Called in two contexts:
    1. During __init__: model=None (schema built for instance)
    2. During can_infer_from(): model provided (schema built for validation)

    Args:
        node: ONNX node (provides inputs, outputs, attributes)
        model: Optional ModelWrapper (provides shapes, datatypes for validation context)

    Returns:
        KernelSchema defining kernel structure

    Example (static schema):
        @classmethod
        def build_schema(cls, node, model):
            return LAYERNORM_SCHEMA

    Example (dynamic schema):
        @classmethod
        def build_schema(cls, node, model):
            num_inputs = len(node.input)
            inputs = [InputSchema(name=f"input{i}", ...) for i in range(num_inputs)]
            return KernelSchema(name="Concat", inputs=inputs, outputs=[...])
    """
    raise NotImplementedError(f"{cls.__name__}.build_schema()")

can_infer_from classmethod

can_infer_from(node: NodeProto, model: ModelWrapper) -> bool

Check if this kernel can transform the given ONNX node (default: no).

Source code in brainsmith/dataflow/kernel_op.py

@classmethod
def can_infer_from(cls, node: NodeProto, model: ModelWrapper) -> bool:
    """Check if this kernel can transform the given ONNX node (default: no)."""
    return False

get_input_datatype

get_input_datatype(ind=0) -> DataType

Get input datatype.

Source code in brainsmith/dataflow/kernel_op.py

def get_input_datatype(self, ind=0) -> DataType:
    """Get input datatype."""
    return DataType[self.get_nodeattr(f"input{ind}Datatype")]

get_nodeattr_types

get_nodeattr_types()

Return nodeattr registry (datatypes + user params + kernel params).

Auto-delegates to kernel_schema.build_nodeattr_registry() which includes: - Interface datatypes (input0Datatype, output0Datatype, etc.) - Internal datatypes (accumulatorDatatype, etc.) - Template parameters (SIMD, PE, etc.) - Kernel-specific parameters (epsilon, algorithm, etc.)

Automatically sets FIFO depth defaults based on kernel schema interface counts.

Only override if build_schema() needs to read nodeattrs (circular dependency). In that case, define nodeattrs explicitly before calling build_schema().

Source code in brainsmith/dataflow/kernel_op.py

def get_nodeattr_types(self):
    """Return nodeattr registry (datatypes + user params + kernel params).

    Auto-delegates to kernel_schema.build_nodeattr_registry() which includes:
    - Interface datatypes (input0Datatype, output0Datatype, etc.)
    - Internal datatypes (accumulatorDatatype, etc.)
    - Template parameters (SIMD, PE, etc.)
    - Kernel-specific parameters (epsilon, algorithm, etc.)

    Automatically sets FIFO depth defaults based on kernel schema interface counts.

    Only override if build_schema() needs to read nodeattrs (circular dependency).
    In that case, define nodeattrs explicitly before calling build_schema().
    """
    base_attrs = super().get_nodeattr_types()

    # Add implementation attribute for backend selection
    # Replaces domain mutation used by FINN's SpecializeLayers
    base_attrs.update(
        {
            "implementation": (
                "s",
                False,
                "",
                {
                    "",  # Not yet specialized
                    "vitis_hls",  # Vitis HLS (2020.1+)
                    "verilog",  # Verilog RTL
                    "systemverilog",  # SystemVerilog RTL
                    "static_ip",  # Pre-generated IP core
                },
            ),
        }
    )

    try:
        base_attrs.update(self.kernel_schema.build_nodeattr_registry())
    except RecursionError as e:
        raise RuntimeError(
            f"{self.__class__.__name__}.kernel_schema property calls get_nodeattr(), "
            f"creating circular dependency. You must override get_nodeattr_types() "
            f"explicitly to define nodeattrs before schema construction. "
            f"See KernelOp docstring for mode-dependent schema pattern."
        ) from e

    # Auto-configure FIFO depths based on schema interface counts
    # This prevents IndexError in FINN's InsertFIFO transform for multi-input/output kernels
    num_inputs = len(self.kernel_schema.inputs)
    num_outputs = len(self.kernel_schema.outputs)

    if num_inputs > 0:
        base_attrs["inFIFODepths"] = ("ints", False, [2] * num_inputs)
    if num_outputs > 0:
        base_attrs["outFIFODepths"] = ("ints", False, [2] * num_outputs)

    return base_attrs

get_normal_input_shape

get_normal_input_shape(ind=0) -> tuple[int, ...]

Return normal (unfolded) input shape as immutable tuple (FINN convention).

Source code in brainsmith/dataflow/kernel_op.py

def get_normal_input_shape(self, ind=0) -> tuple[int, ...]:
    """Return normal (unfolded) input shape as immutable tuple (FINN convention)."""
    return self.design_point.input_list[ind].tensor_shape

get_normal_output_shape

get_normal_output_shape(ind=0) -> tuple[int, ...]

Return normal (unfolded) output shape as immutable tuple (FINN convention).

Source code in brainsmith/dataflow/kernel_op.py

def get_normal_output_shape(self, ind=0) -> tuple[int, ...]:
    """Return normal (unfolded) output shape as immutable tuple (FINN convention)."""
    return self.design_point.output_list[ind].tensor_shape

get_number_output_values

get_number_output_values()

Get iteration count(s) for output values.

Matches FINN API pattern: - Single-output kernels: Returns int (iteration count) - Multi-output kernels: Returns dict mapping output names → iteration counts

Returns:

Name	Type	Description
int		For single-output kernels (e.g., MVAU, Thresholding, AddStreams)
dict		For multi-output kernels (e.g., DuplicateStreams, Split)

Examples:

Single-output: 512 Multi-output: {'out0': 512, 'out1': 512}

Source code in brainsmith/dataflow/kernel_op.py

def get_number_output_values(self):
    """Get iteration count(s) for output values.

    Matches FINN API pattern:
    - Single-output kernels: Returns int (iteration count)
    - Multi-output kernels: Returns dict mapping output names → iteration counts

    Returns:
        int: For single-output kernels (e.g., MVAU, Thresholding, AddStreams)
        dict: For multi-output kernels (e.g., DuplicateStreams, Split)

    Examples:
        Single-output: 512
        Multi-output: {'out0': 512, 'out1': 512}
    """
    num_outputs = len(self.onnx_node.output)

    if num_outputs == 1:
        # Single-output: Return int (FINN pattern for MVAU, Thresholding, etc.)
        folded_shape = self.get_folded_output_shape(ind=0)
        return math.prod(folded_shape[:-1])
    else:
        # Multi-output: Return dict (FINN pattern for DuplicateStreams, Split)
        out_val = {}
        for i in range(num_outputs):
            folded_shape = self.get_folded_output_shape(ind=i)
            iteration_count = math.prod(folded_shape[:-1])
            out_val[f"out{i}"] = iteration_count
        return out_val

get_output_datatype

get_output_datatype(ind=0) -> DataType

Get output datatype.

Source code in brainsmith/dataflow/kernel_op.py

def get_output_datatype(self, ind=0) -> DataType:
    """Get output datatype."""
    return DataType[self.get_nodeattr(f"output{ind}Datatype")]

get_valid_ranges

get_valid_ranges(model_w: ModelWrapper) -> dict[str, Union[OrderedParameter, frozenset]]

Valid parameter values for DSE (tiling + resource).

Returns:

Type	Description
dict[str, Union[OrderedParameter, frozenset]]	Dict mapping parameter names to OrderedParameter (ordered sequences)
dict[str, Union[OrderedParameter, frozenset]]	or frozenset (discrete categories).

Source code in brainsmith/dataflow/kernel_op.py

def get_valid_ranges(
    self, model_w: ModelWrapper
) -> dict[str, Union["OrderedParameter", frozenset]]:
    """Valid parameter values for DSE (tiling + resource).

    Returns:
        Dict mapping parameter names to OrderedParameter (ordered sequences)
        or frozenset (discrete categories).
    """
    self._ensure_ready(model_w)
    return self.design_space.parameters

infer_from classmethod

infer_from(node: NodeProto, model: ModelWrapper, insert_index: int) -> TransformationResult

Transform ONNX node to hardware kernel node(s).

Source code in brainsmith/dataflow/kernel_op.py

@classmethod
def infer_from(
    cls, node: NodeProto, model: ModelWrapper, insert_index: int
) -> TransformationResult:
    """Transform ONNX node to hardware kernel node(s)."""
    raise NotImplementedError(f"{cls.__name__}.infer_from()")

infer_node_datatype

infer_node_datatype(model_w)

Sync datatypes: model → nodeattrs (inputs), nodeattrs → model (outputs).

Initializes design space which syncs input datatypes from model to nodeattrs. Then propagates output datatypes from nodeattrs back to model.

Source code in brainsmith/dataflow/kernel_op.py

def infer_node_datatype(self, model_w):
    """Sync datatypes: model → nodeattrs (inputs), nodeattrs → model (outputs).

    Initializes design space which syncs input datatypes from model to nodeattrs.
    Then propagates output datatypes from nodeattrs back to model.
    """
    # Initialize (syncs inputs: model → nodeattrs)
    self._ensure_ready(model_w)

    # Propagate output datatypes: nodeattrs → model
    for i, out_name in enumerate(self.onnx_node.output):
        if out_name:
            odt = self.get_output_datatype(i)
            model_w.set_tensor_datatype(out_name, odt)

invalidate

invalidate() -> None

Invalidate cached design space after external graph changes.

Call this after transforms that change: - Tensor shapes (padding, reshape) - Datatypes in graph metadata - Node rewiring (FIFO insertion)

Next method call with model_w will rebuild design space automatically. Design points regenerate on every access, so no explicit invalidation needed.

Example

After transform changes graph

model = ApplyPadding().apply(model) for node in model.graph.node: ... op = getCustomOp(node) ... if isinstance(op, KernelOp): ... op.invalidate()

Source code in brainsmith/dataflow/kernel_op.py

def invalidate(self) -> None:
    """Invalidate cached design space after external graph changes.

    Call this after transforms that change:
    - Tensor shapes (padding, reshape)
    - Datatypes in graph metadata
    - Node rewiring (FIFO insertion)

    Next method call with model_w will rebuild design space automatically.
    Design points regenerate on every access, so no explicit invalidation needed.

    Example:
        >>> # After transform changes graph
        >>> model = ApplyPadding().apply(model)
        >>> for node in model.graph.node:
        ...     op = getCustomOp(node)
        ...     if isinstance(op, KernelOp):
        ...         op.invalidate()
    """
    self._design_space = None

make_shape_compatible_op

make_shape_compatible_op(model_w)

Create standard ONNX op for shape inference (auto-detects pattern).

Source code in brainsmith/dataflow/kernel_op.py

def make_shape_compatible_op(self, model_w):
    """Create standard ONNX op for shape inference (auto-detects pattern)."""
    from onnx import helper

    num_out = len(self.onnx_node.output)
    num_in = len(self.onnx_node.input)

    if num_in == 1 and num_out > 1:
        return helper.make_node(
            "Split",
            inputs=[self.onnx_node.input[0]],
            outputs=list(self.onnx_node.output),
            axis=-1,
        )

    if num_out == 1:
        input_shapes = [tuple(model_w.get_tensor_shape(inp)) for inp in self.onnx_node.input]

        if len(set(input_shapes)) == 1:
            return super().make_const_shape_op(input_shapes[0])
        else:
            raise NotImplementedError(
                f"{self.__class__.__name__}: {num_in} inputs with different shapes "
                f"{input_shapes}. Override make_shape_compatible_op()."
            )

    raise NotImplementedError(
        f"{self.__class__.__name__}: {num_in} inputs → {num_out} outputs. "
        f"Override make_shape_compatible_op()."
    )

set_nodeattr

set_nodeattr(name: str, value: Any) -> None

Set nodeattr and auto-invalidate design space if needed.

Design points regenerate on each access, so no explicit invalidation needed.

Source code in brainsmith/dataflow/kernel_op.py

def set_nodeattr(self, name: str, value: Any) -> None:
    """Set nodeattr and auto-invalidate design space if needed.

    Design points regenerate on each access, so no explicit invalidation needed.
    """
    try:
        old_value = self.get_nodeattr(name)
    except (AttributeError, Exception):
        old_value = None

    if old_value != value:
        super().set_nodeattr(name, value)

        # Only invalidate design space for structural changes (datatypes)
        # Dimension changes (PE, SIMD, etc.) don't require invalidation since
        # design points regenerate from nodeattrs on each access
        if name in self.kernel_schema.get_structural_nodeattrs():
            self._design_space = None

Example:

import brainsmith.dataflow as df
from brainsmith.registry import kernel
from onnx import NodeProto, helper
from qonnx.core.modelwrapper import ModelWrapper

@kernel(description="Hardware LayerNorm", author="Your Name")
class LayerNorm(df.KernelOp):
    """Hardware LayerNorm kernel."""

    @classmethod
    def build_schema(cls, node: NodeProto, model: ModelWrapper) -> df.KernelSchema:
        """Define kernel structure."""
        return LAYERNORM_SCHEMA

    @classmethod
    def can_infer_from(cls, node: NodeProto, model: ModelWrapper) -> bool:
        """Check if node can be converted to this kernel."""
        return node.op_type == "FuncLayerNorm"

    @classmethod
    def infer_from(cls, node: NodeProto, model: ModelWrapper, insert_index: int):
        """Transform ONNX node to hardware kernel."""
        hw_node = helper.make_node(
            "LayerNorm",
            inputs=list(node.input),
            outputs=list(node.output),
            domain="brainsmith.kernels",
        )
        return df.TransformationResult(
            nodes_to_insert=[hw_node],
            nodes_to_remove=[node]
        )

---

KernelOpError

KernelOpError(node, message)

Bases: Exception

Exception raised by kernel operators with node context.

Attributes:

Name	Type	Description
node		ONNX node that caused the error
message		Error message

Source code in brainsmith/dataflow/kernel_op.py

def __init__(self, node, message):
    self.node = node
    super().__init__(f"{node.name}: {message}")

---

KernelSchema dataclass

KernelSchema(name: str, inputs: list[InputSchema] = list(), outputs: list[OutputSchema] = list(), internal_datatypes: dict[str, Any] = dict(), kernel_params: dict[str, tuple] = dict(), dse_parameters: dict[str, ParameterSpec] = dict(), constraints: list[Constraint] = list(), attribute_mapping: dict[str, str] = dict())

Kernel specification defining structure and validation.

Combines interface definitions, validation constraints, and design space parameters. Defines structure only - shapes come from ONNX context, execution logic lives in KernelOp.

Attributes:

Name	Type	Description
name	str	Kernel name
inputs	list[InputSchema]	Input interface schemas
outputs	list[OutputSchema]	Output interface schemas
internal_datatypes	dict[str, Any]	Internal datatype derivation specs (e.g., accumulator)
kernel_params	dict[str, tuple]	Kernel-specific parameters (e.g., epsilon, algorithm)
dse_parameters	dict[str, ParameterSpec]	Explorable resource/implementation parameters (e.g., ram_style)
constraints	list[Constraint]	Validation constraints (datatype, shape, ONNX requirements)
attribute_mapping	dict[str, str]	Map ONNX attributes to kernel parameters

attribute_mapping class-attribute instance-attribute

attribute_mapping: dict[str, str] = field(default_factory=dict)

Map ONNX attributes to kernel parameters.

Example: {"epsilon": "epsilon", "axis": "normalized_axis"}

dse_parameters class-attribute instance-attribute

dse_parameters: dict[str, ParameterSpec] = field(default_factory=dict)

Explorable resource/implementation parameters (ram_style, res_type, etc.).

Tiling parameters (PE, SIMD) NOT declared here - auto-extracted from stream_tiling templates with defaults computed from factoring.

Example: {"ram_style": ParameterSpec("ram_style", {"distributed", "block"}, "distributed")}

__post_init__

__post_init__()

Validate schema structure and transformation consistency.

Source code in brainsmith/dataflow/schemas.py

def __post_init__(self):
    """Validate schema structure and transformation consistency."""
    self.validate()

    # Validate transformation fields
    self._validate_transformation_fields()

build_nodeattr_registry

build_nodeattr_registry() -> dict[str, tuple]

Build nodeattr registry from schema definition.

Schemas define STRUCTURE, not STORAGE. Generates persistence layer from structural schema, returning only attributes that need persistence: - Datatypes (for interfaces and internals) - Tiling parameters (SIMD, PE, etc.) - auto-extracted from stream_tiling - DSE parameters (ram_style, res_type, etc.) - from dse_parameters - Kernel-specific parameters (epsilon, algorithm, etc.) - from kernel_params

Shapes are NEVER stored in nodeattrs. They are either: - Tensor shapes: extracted from ModelWrapper (ONNX graph) - Block/stream shapes: computed from schema templates

Returns:

Name	Type	Description
	dict[str, tuple]	Dict mapping nodeattr name to (type, required, default_value)
Format	dict[str, tuple]	{"attrName": ("i"|"s"|"f", True|False, default)}

Source code in brainsmith/dataflow/schemas.py

def build_nodeattr_registry(self) -> dict[str, tuple]:
    """Build nodeattr registry from schema definition.

    Schemas define STRUCTURE, not STORAGE. Generates persistence layer
    from structural schema, returning only attributes that need persistence:
    - Datatypes (for interfaces and internals)
    - Tiling parameters (SIMD, PE, etc.) - auto-extracted from stream_tiling
    - DSE parameters (ram_style, res_type, etc.) - from dse_parameters
    - Kernel-specific parameters (epsilon, algorithm, etc.) - from kernel_params

    Shapes are NEVER stored in nodeattrs. They are either:
    - Tensor shapes: extracted from ModelWrapper (ONNX graph)
    - Block/stream shapes: computed from schema templates

    Returns:
        Dict mapping nodeattr name to (type, required, default_value)
        Format: {"attrName": ("i"|"s"|"f", True|False, default)}
    """
    attrs = {}

    # Datatypes
    for i in range(len(self.inputs)):
        attrs[f"input{i}Datatype"] = ("s", False, "")

    for i in range(len(self.outputs)):
        attrs[f"output{i}Datatype"] = ("s", False, "")

    for internal_name in self.internal_datatypes.keys():
        attrs[f"{internal_name}Datatype"] = ("s", False, "")

    # Tiling parameters (PE, SIMD, etc.) - auto-extracted
    template_params = self._extract_template_params()
    for param in template_params:
        attrs[param] = ("i", False, 1)  # Default 1, will be computed from factoring

    # DSE parameters (resource parameters)
    for param_name, param_spec in self.dse_parameters.items():
        attrs[param_name] = _infer_nodeattr_type(param_spec)

    # Kernel-specific parameters (structural)
    attrs.update(self.kernel_params)

    return attrs

get_optimization_nodeattrs

get_optimization_nodeattrs() -> set

Get nodeattrs that affect optimization (re-explore if changed).

Optimization nodeattrs are those whose changes only require re-exploring the design space (trying different stream shapes), not rebuilding the entire design space.

These include: - Parallelization parameters (SIMD, PE, MW, MH, etc.): Appear in stream_tiling templates and determine stream shapes during DSE

Returns:

Type	Description
set	Set of optimization nodeattr names

Example

schema.get_optimization_nodeattrs()

Source code in brainsmith/dataflow/schemas.py

def get_optimization_nodeattrs(self) -> set:
    """Get nodeattrs that affect optimization (re-explore if changed).

    Optimization nodeattrs are those whose changes only require
    re-exploring the design space (trying different stream shapes),
    not rebuilding the entire design space.

    These include:
    - Parallelization parameters (SIMD, PE, MW, MH, etc.): Appear in
      stream_tiling templates and determine stream shapes during DSE

    Returns:
        Set of optimization nodeattr names

    Example:
        >>> schema.get_optimization_nodeattrs()
        {'SIMD', 'PE'}
    """
    # Parameters in stream_tiling affect optimization
    return self._extract_template_params()

get_structural_nodeattrs

get_structural_nodeattrs() -> set

Get nodeattrs that affect design space (rebuild if changed).

Structural nodeattrs are those whose changes require rebuilding the entire KernelDesignSpace (not just reconfiguration).

These include: - All datatypes (input, output, internal): Affect internal datatype derivation (e.g., accumulator width depends on input datatype) - Parameters in block_tiling (rare): Affect block shape computation

Returns:

Type	Description
set	Set of structural nodeattr names

Example

schema.get_structural_nodeattrs()

Source code in brainsmith/dataflow/schemas.py

def get_structural_nodeattrs(self) -> set:
    """Get nodeattrs that affect design space (rebuild if changed).

    Structural nodeattrs are those whose changes require rebuilding
    the entire KernelDesignSpace (not just reconfiguration).

    These include:
    - All datatypes (input, output, internal): Affect internal datatype
      derivation (e.g., accumulator width depends on input datatype)
    - Parameters in block_tiling (rare): Affect block shape computation

    Returns:
        Set of structural nodeattr names

    Example:
        >>> schema.get_structural_nodeattrs()
        {'input0Datatype', 'output0Datatype', 'accumulatorDatatype'}
    """
    structural = set()

    # All datatypes are structural
    for i in range(len(self.inputs)):
        structural.add(f"input{i}Datatype")
    for i in range(len(self.outputs)):
        structural.add(f"output{i}Datatype")
    for internal_name in self.internal_datatypes.keys():
        structural.add(f"{internal_name}Datatype")

    # Parameters in block_tiling are structural (rare)
    for inp in self.inputs:
        if inp.block_tiling and inp.block_tiling is not FULL_SHAPE:
            for elem in inp.block_tiling:
                if isinstance(elem, str):
                    structural.add(elem)
    for out in self.outputs:
        if out.block_tiling and out.block_tiling is not FULL_SHAPE:
            for elem in out.block_tiling:
                if isinstance(elem, str):
                    structural.add(elem)

    return structural

validate

validate() -> None

Validate the schema structure.

Source code in brainsmith/dataflow/schemas.py

def validate(self) -> None:
    """Validate the schema structure."""

    # Create sets directly (no intermediate lists)
    input_names = {inp.name for inp in self.inputs}
    output_names = {out.name for out in self.outputs}

    # Check for duplicates (comparing lengths to original counts)
    if len(input_names) != len(self.inputs):
        raise ValueError(f"Duplicate input names in kernel '{self.name}'")
    if len(output_names) != len(self.outputs):
        raise ValueError(f"Duplicate output names in kernel '{self.name}'")

    # Check for conflicts between inputs and outputs
    conflicts = input_names & output_names
    if conflicts:
        raise ValueError(
            f"Interface names must be unique across inputs and outputs in kernel '{self.name}'. "
            f"Duplicate names: {', '.join(sorted(conflicts))}"
        )

    # Check internal datatypes (use set operation instead of loop)
    all_interface_names = input_names | output_names
    internal_conflicts = set(self.internal_datatypes) & all_interface_names
    if internal_conflicts:
        raise ValueError(
            f"Internal datatypes conflict with interface names in kernel '{self.name}': "
            f"{', '.join(sorted(internal_conflicts))}"
        )

Example:

import brainsmith.dataflow as df
from brainsmith.dataflow import FULL_DIM

# Define kernel schema
LAYERNORM_SCHEMA = df.KernelSchema(
    name="LayerNorm",
    inputs=[
        df.InputSchema(
            name="input",
            block_tiling=[FULL_DIM],
            stream_tiling=["SIMD"],
            required_layout="NHWC",
        )
    ],
    outputs=[
        df.OutputSchema(
            name="output",
            block_tiling=[FULL_DIM],
            stream_tiling=[df.derive_dim("input", df.ShapeHierarchy.STREAM, -1)],
            required_layout="NHWC",
        )
    ],
    kernel_params={
        "epsilon": ("f", True, 1e-5),
    },
    constraints=[
        df.AttrCompare("epsilon", ">", 0),
    ],
)

---

InputSchema dataclass

InputSchema(name: str, block_tiling: TilingSpec | None = None, stream_tiling: TilingSpec | None = None, datatype: Any | None = None, required_layout: str | None = None)

Input interface specification.

Defines input structure (tiling) and requirements (layout, datatype).

Attributes:

Name	Type	Description
name	str	Interface name (e.g., "input", "input0")
block_tiling	TilingSpec | None	Block tiling specification (e.g., [FULL_DIM, FULL_DIM])
stream_tiling	TilingSpec | None	Stream tiling specification (e.g., ["SIMD"], [1, 1, 1, "PE"])
datatype	Any | None	Datatype spec (None to use from ONNX, or DatatypeSpec union type to derive/optimize)
required_layout	str | None	Expected input layout (e.g., "NHWC", "NCHW"), None if no requirement

tiling_attrs property

tiling_attrs: list[str]

Extract unique template parameter names from tiling specs.

__post_init__

__post_init__()

Validate interface requirements.

Source code in brainsmith/dataflow/schemas.py

def __post_init__(self):
    """Validate interface requirements."""
    if self.required_layout and self.required_layout not in {"NCHW", "NHWC"}:
        raise ValueError(
            f"Invalid required_layout '{self.required_layout}' for input '{self.name}'. "
            f"Must be 'NCHW' or 'NHWC'."
        )

---

OutputSchema dataclass

OutputSchema(name: str, block_tiling: TilingSpec | None = None, stream_tiling: TilingSpec | None = None, datatype: Any | None = None, required_layout: str | None = None, preserves_input_layout: bool = True)

Output interface specification.

Defines output structure (tiling), datatype derivation, and layout requirements.

Attributes:

Name	Type	Description
name	str	Interface name (e.g., "output", "output0")
block_tiling	TilingSpec | None	Block tiling specification
stream_tiling	TilingSpec | None	Stream tiling specification
datatype	Any | None	Datatype spec (None to use from ONNX, or DatatypeSpec union type to derive)
required_layout	str | None	Expected output layout (e.g., "NHWC"), None if no requirement
preserves_input_layout	bool	Whether output preserves first input's layout (default True)

tiling_attrs property

tiling_attrs: list[str]

Extract unique template parameter names from tiling specs.

__post_init__

__post_init__()

Validate interface requirements.

Source code in brainsmith/dataflow/schemas.py

def __post_init__(self):
    """Validate interface requirements."""
    if self.required_layout and self.required_layout not in {"NCHW", "NHWC"}:
        raise ValueError(
            f"Invalid required_layout '{self.required_layout}' for output '{self.name}'. "
            f"Must be 'NCHW' or 'NHWC'."
        )

---

ParameterSpec dataclass

ParameterSpec(name: str, values: set[int | str] | Callable[[BuildContext], set[int | str]], type: Literal['int', 'string'] | None = None, default: int | str | None = None)

Explorable parameter in design space.

Represents resource allocation or implementation choices that can be explored during DSE (ram_style, res_type, mem_mode, etc.).

Does NOT include tiling dimensions (PE, SIMD) - those are auto-extracted from stream_tiling templates with valid values computed from factoring.

Container Type Convention (Ordered vs Discrete):

The container type determines how the dimension is treated during DSE:

• list/tuple → OrderedParameter (ordered sequences with navigation)

  • Supports min/max access, step_up/step_down, percentage-based indexing
  • Values are sorted automatically
  • Examples: depth=[128, 256, 512], num_layers=[1, 2, 4, 8]

• set/frozenset → Discrete (unordered categories)

  • Membership testing only, no navigation
  • Order doesn't matter
  • Examples: ram_style={"distributed", "block"}, res_type={"lut", "dsp"}

Type Declaration (Hybrid Approach):

• Literal values: Type inferred from first value (optional to specify)
• Callable values: Type MUST be explicitly specified

Attributes:

Name	Type	Description
name	str	Dimension name (e.g., "ram_style", "depth")
values	set[int | str] | Callable[[BuildContext], set[int | str]]	Valid values for this dimension - list/tuple: Ordered sequence (enables navigation methods) - set/frozenset: Discrete categories (membership only) - Callable: Computed from BuildContext (for context-dependent values)
type	Literal['int', 'string'] | None	Value type ("int" or "string") - Required for callable values - Optional for literal values (inferred from first value) - Validated against values if both provided
default	int | str | None	Default value (None = auto-select: min for ordered, first for discrete)

Examples:

>>> # Ordered parameter - type inferred
>>> ParameterSpec("depth", [128, 256, 512, 1024], default=256)

>>> # Discrete parameter - type inferred
>>> ParameterSpec("ram_style", {"distributed", "block"}, default="distributed")

>>> # Callable parameter - type required
>>> ParameterSpec("depth", lambda ctx: compute_depths(ctx), type="int", default=256)

>>> # Explicit type for documentation (optional)
>>> ParameterSpec("mode", {"fast", "accurate"}, type="string")

Validation

• Callable values without type → ValueError
• Type mismatch with literal values → ValueError
• Invalid type (not "int" or "string") → ValueError

Note

Tiling dimensions (PE, SIMD) are ALWAYS ordered (auto-wrapped in OrderedParameter) since they're computed as divisors (naturally ordered sequences).

__post_init__

__post_init__()

Validate type specification against values.

Source code in brainsmith/dataflow/schemas.py

def __post_init__(self):
    """Validate type specification against values."""
    # Callable values MUST specify type
    if callable(self.values):
        if self.type is None:
            raise ValueError(
                f"ParameterSpec '{self.name}': Callable values require explicit type declaration. "
                f"Specify type='int' or type='string'."
            )
        if self.type not in ("int", "string"):
            raise ValueError(
                f"ParameterSpec '{self.name}': Invalid type '{self.type}'. "
                f"Must be 'int' or 'string'."
            )
        return  # Cannot validate callable values without context

    # Validate type matches literal values (if type specified)
    if self.type is not None:
        if self.type not in ("int", "string"):
            raise ValueError(
                f"ParameterSpec '{self.name}': Invalid type '{self.type}'. "
                f"Must be 'int' or 'string'."
            )

        first_val = next(iter(self.values))
        expected_type = int if self.type == "int" else str

        if not isinstance(first_val, expected_type):
            actual_type = type(first_val).__name__
            raise ValueError(
                f"ParameterSpec '{self.name}': Type mismatch. "
                f"Declared type='{self.type}' but values contain {actual_type}. "
                f"First value: {first_val!r}"
            )

Example:

import brainsmith.dataflow as df

# Ordered parameter (list/tuple enables navigation)
depth_param = df.ParameterSpec("depth", [128, 256, 512], default=256)

# Discrete parameter (set for unordered categories)
ram_param = df.ParameterSpec("ram_style", {"distributed", "block"}, default="distributed")

# Callable parameter (explicit type required)
dynamic_param = df.ParameterSpec("depth", lambda ctx: [128, 256], type="int", default=128)

# Use in kernel schema
schema = df.KernelSchema(
    name="MyKernel",
    inputs=[...],
    outputs=[...],
    dse_parameters={
        "ram_style": ram_param,
        "depth": depth_param,
    }
)

---

KernelDesignSpace dataclass

KernelDesignSpace(name: str, inputs: dict[str, InterfaceDesignSpace], outputs: dict[str, InterfaceDesignSpace], internal_datatypes: dict[str, BaseDataType], optimization_constraints: list[Constraint], parameters: dict[str, Union[OrderedParameter, frozenset]])

Kernel design space built once, configured many times.

Built by DesignSpaceBuilder from ONNX context, acts as factory for KernelDesignPoint via configure(). Contains structure constant during DSE plus valid ranges for all explorable dimensions.

Attributes:

Name	Type	Description
name	str	Kernel name
inputs	dict[str, InterfaceDesignSpace]	Input interface design spaces (by name)
outputs	dict[str, InterfaceDesignSpace]	Output interface design spaces (by name)
internal_datatypes	dict[str, BaseDataType]	Internal datatypes (e.g., accumulator)
optimization_constraints	list[Constraint]	Parametric constraints validated at configure()
parameters	dict[str, Union[OrderedParameter, frozenset]]	Explorable parameters - OrderedParameter (with navigation) or frozenset (discrete categories like ram_style)

input_list property

input_list: list[InterfaceDesignSpace]

Inputs in declaration order (for ONNX positional mapping).

Returns inputs as list preserving dict insertion order (Python 3.7+). Useful when mapping to ONNX node.input[i] positions.

output_list property

output_list: list[InterfaceDesignSpace]

Outputs in declaration order (for ONNX positional mapping).

Returns outputs as list preserving dict insertion order (Python 3.7+). Useful when mapping to ONNX node.output[i] positions.

configure

configure(config: dict[str, int | str]) -> KernelDesignPoint

Instantiate kernel at specified point in design space.

Creates a KernelDesignPoint with resolved stream shapes and validates all parametric constraints.

Parameters:

Name	Type	Description	Default
config	dict[str, int | str]	Dimension values (tiling + resource) specifying the instance point	required

Returns:

Type	Description
KernelDesignPoint	KernelDesignPoint with fully resolved configuration

Raises:

Type	Description
ValueError	If config invalid or missing dimensions
ValidationError	If parametric constraints fail

Source code in brainsmith/dataflow/dse_models.py

def configure(self, config: dict[str, int | str]) -> "KernelDesignPoint":
    """Instantiate kernel at specified point in design space.

    Creates a KernelDesignPoint with resolved stream shapes and validates
    all parametric constraints.

    Args:
        config: Dimension values (tiling + resource) specifying the instance point

    Returns:
        KernelDesignPoint with fully resolved configuration

    Raises:
        ValueError: If config invalid or missing dimensions
        ValidationError: If parametric constraints fail
    """
    self._validate_params(config)

    interface_lookup = {}
    inputs = self._instantiate_interfaces(self.inputs, config, interface_lookup)
    outputs = self._instantiate_interfaces(self.outputs, config, interface_lookup)

    instance = KernelDesignPoint(
        design_space=self,
        inputs=inputs,
        outputs=outputs,
        config=config,
    )

    self._validate_instance(instance, config)
    return instance

get_ordered_parameter

get_ordered_parameter(name: str) -> OrderedParameter

Get ordered parameter by name.

Parameters:

Name	Type	Description	Default
name	str	Parameter name	required

Returns:

Type	Description
OrderedParameter	OrderedParameter instance

Raises:

Type	Description
KeyError	If parameter not found
TypeError	If parameter is discrete (not ordered)

Source code in brainsmith/dataflow/dse_models.py

def get_ordered_parameter(self, name: str) -> "OrderedParameter":
    """Get ordered parameter by name.

    Args:
        name: Parameter name

    Returns:
        OrderedParameter instance

    Raises:
        KeyError: If parameter not found
        TypeError: If parameter is discrete (not ordered)
    """
    param = self.parameters[name]
    if not isinstance(param, OrderedParameter):
        raise TypeError(
            f"Parameter '{name}' is discrete (frozenset), not ordered. "
            f"Use get_parameter() for type-agnostic access."
        )
    return param

get_parameter

get_parameter(name: str) -> Union[OrderedParameter, frozenset]

Get parameter by name.

Parameters:

Name	Type	Description	Default
name	str	Parameter name	required

Returns:

Type	Description
Union[OrderedParameter, frozenset]	OrderedParameter for ordered parameters, frozenset for discrete

Raises:

Type	Description
KeyError	If parameter not found

Source code in brainsmith/dataflow/dse_models.py

def get_parameter(self, name: str) -> Union["OrderedParameter", frozenset]:
    """Get parameter by name.

    Args:
        name: Parameter name

    Returns:
        OrderedParameter for ordered parameters, frozenset for discrete

    Raises:
        KeyError: If parameter not found
    """
    return self.parameters[name]

is_discrete_parameter

is_discrete_parameter(name: str) -> bool

Check if parameter is discrete.

Parameters:

Name	Type	Description	Default
name	str	Parameter name	required

Returns:

Type	Description
bool	True if parameter is discrete (frozenset), False if ordered

Raises:

Type	Description
KeyError	If parameter not found

Source code in brainsmith/dataflow/dse_models.py

def is_discrete_parameter(self, name: str) -> bool:
    """Check if parameter is discrete.

    Args:
        name: Parameter name

    Returns:
        True if parameter is discrete (frozenset), False if ordered

    Raises:
        KeyError: If parameter not found
    """
    return isinstance(self.parameters[name], frozenset)

is_ordered_parameter

is_ordered_parameter(name: str) -> bool

Check if parameter is ordered.

Parameters:

Name	Type	Description	Default
name	str	Parameter name	required

Returns:

Type	Description
bool	True if parameter is OrderedParameter, False if discrete (frozenset)

Raises:

Type	Description
KeyError	If parameter not found

Source code in brainsmith/dataflow/dse_models.py

def is_ordered_parameter(self, name: str) -> bool:
    """Check if parameter is ordered.

    Args:
        name: Parameter name

    Returns:
        True if parameter is OrderedParameter, False if discrete (frozenset)

    Raises:
        KeyError: If parameter not found
    """
    return isinstance(self.parameters[name], OrderedParameter)

---

InterfaceDesignSpace dataclass

InterfaceDesignSpace(name: str, tensor_shape: Shape, block_shape: Shape, stream_tiling: TilingSpec, datatype: BaseDataType, is_weight: bool = False, tensor_name: str | None = None, parallelism_dimension: OrderedParameter | None = None, parallelism_param: str | None = None)

Interface design space built once, configured many times.

Defines interface structure constant during DSE. Stream tiling preserved as template for resolution with specific parallelization parameters.

Attributes:

Name	Type	Description
name	str	Interface name
tensor_shape	Shape	Full tensor dimensions
block_shape	Shape	Block dimensions (per-operation tile size)
stream_tiling	TilingSpec	Stream tiling template (e.g., ["SIMD"] or [1, 1, 1, "PE"])
datatype	BaseDataType	Interface datatype
is_weight	bool	Whether this is a weight tensor (constant)
tensor_name	str | None	ONNX tensor name for initializer lookups
parallelism_dimension	OrderedParameter | None	OrderedParameter for stream parameter (None if no parallelism)
parallelism_param	str | None	Parameter name for stream dimension (e.g., "SIMD", "PE")

---

KernelDesignPoint dataclass

KernelDesignPoint(design_space: KernelDesignSpace, inputs: dict[str, InterfaceDesignPoint], outputs: dict[str, InterfaceDesignPoint], config: dict[str, int | str])

Immutable kernel instance at specific design point.

Created by KernelDesignSpace.configure() with specific dimension values. Flyweight pattern minimizes memory - references parent design space, stores only configuration-specific data.

Navigation methods return new instances - the design point itself is immutable. Use with_dimension(), with_step_up(), sweep_dimension() to explore the space.

Attributes:

Name	Type	Description
design_space	KernelDesignSpace	Parent KernelDesignSpace
inputs	dict[str, InterfaceDesignPoint]	Configured input interfaces (by name)
outputs	dict[str, InterfaceDesignPoint]	Configured output interfaces (by name)
config	dict[str, int | str]	Dimension values defining this point (e.g., {"SIMD": 16, "PE": 4})

initiation_interval property

initiation_interval: int

Kernel initiation interval in cycles.

input_list property

input_list: list[InterfaceDesignPoint]

Inputs in declaration order (for ONNX positional mapping).

max_block_folding_factor property

max_block_folding_factor: int

Maximum block folding factor across all inputs.

max_tensor_folding_factor property

max_tensor_folding_factor: int

Maximum tensor folding factor across all inputs.

output_list property

output_list: list[InterfaceDesignPoint]

Outputs in declaration order (for ONNX positional mapping).

total_output_values property

total_output_values: int

Total output values across all outputs.

get_input_stream_dimension

get_input_stream_dimension(index: int) -> Optional[OrderedParameter]

Get parallelism dimension for input interface.

Parameters:

Name	Type	Description	Default
index	int	Input interface index (0-based)	required

Returns:

Type	Description
Optional[OrderedParameter]	OrderedParameter or None if no parallelism

Raises:

Type	Description
IndexError	If index out of range

Source code in brainsmith/dataflow/dse_models.py

def get_input_stream_dimension(self, index: int) -> Optional["OrderedParameter"]:
    """Get parallelism dimension for input interface.

    Args:
        index: Input interface index (0-based)

    Returns:
        OrderedParameter or None if no parallelism

    Raises:
        IndexError: If index out of range
    """
    if index < 0 or index >= len(self.input_list):
        raise IndexError(f"Input index {index} out of range [0, {len(self.input_list)})")

    return self.input_list[index].design_space.parallelism_dimension

get_input_stream_param

get_input_stream_param(index: int) -> str | None

Get parallelism parameter name for input interface.

Parameters:

Name	Type	Description	Default
index	int	Input interface index (0-based)	required

Returns:

Type	Description
str | None	Parameter name (e.g., "SIMD", "PE") or None if no parallelism

Raises:

Type	Description
IndexError	If index out of range

Source code in brainsmith/dataflow/dse_models.py

def get_input_stream_param(self, index: int) -> str | None:
    """Get parallelism parameter name for input interface.

    Args:
        index: Input interface index (0-based)

    Returns:
        Parameter name (e.g., "SIMD", "PE") or None if no parallelism

    Raises:
        IndexError: If index out of range
    """
    if index < 0 or index >= len(self.input_list):
        raise IndexError(f"Input index {index} out of range [0, {len(self.input_list)})")

    return self.input_list[index].design_space.parallelism_param

get_input_stream_value

get_input_stream_value(index: int) -> int | None

Get current parallelism value for input interface.

Parameters:

Name	Type	Description	Default
index	int	Input interface index (0-based)	required

Returns:

Type	Description
int | None	Current parallelism value or None if no parallelism

Raises:

Type	Description
IndexError	If index out of range

Source code in brainsmith/dataflow/dse_models.py

def get_input_stream_value(self, index: int) -> int | None:
    """Get current parallelism value for input interface.

    Args:
        index: Input interface index (0-based)

    Returns:
        Current parallelism value or None if no parallelism

    Raises:
        IndexError: If index out of range
    """
    param = self.get_input_stream_param(index)
    return self.config.get(param) if param else None

get_output_stream_dimension

get_output_stream_dimension(index: int) -> Optional[OrderedParameter]

Get parallelism dimension for output interface.

Parameters:

Name	Type	Description	Default
index	int	Output interface index (0-based)	required

Returns:

Type	Description
Optional[OrderedParameter]	OrderedParameter or None if no parallelism

Raises:

Type	Description
IndexError	If index out of range

Source code in brainsmith/dataflow/dse_models.py

def get_output_stream_dimension(self, index: int) -> Optional["OrderedParameter"]:
    """Get parallelism dimension for output interface.

    Args:
        index: Output interface index (0-based)

    Returns:
        OrderedParameter or None if no parallelism

    Raises:
        IndexError: If index out of range
    """
    if index < 0 or index >= len(self.output_list):
        raise IndexError(f"Output index {index} out of range [0, {len(self.output_list)})")

    return self.output_list[index].design_space.parallelism_dimension

get_output_stream_param

get_output_stream_param(index: int) -> str | None

Get parallelism parameter name for output interface.

Parameters:

Name	Type	Description	Default
index	int	Output interface index (0-based)	required

Returns:

Type	Description
str | None	Parameter name or None if no parallelism

Raises:

Type	Description
IndexError	If index out of range

Source code in brainsmith/dataflow/dse_models.py

def get_output_stream_param(self, index: int) -> str | None:
    """Get parallelism parameter name for output interface.

    Args:
        index: Output interface index (0-based)

    Returns:
        Parameter name or None if no parallelism

    Raises:
        IndexError: If index out of range
    """
    if index < 0 or index >= len(self.output_list):
        raise IndexError(f"Output index {index} out of range [0, {len(self.output_list)})")

    return self.output_list[index].design_space.parallelism_param

get_output_stream_value

get_output_stream_value(index: int) -> int | None

Get current parallelism value for output interface.

Parameters:

Name	Type	Description	Default
index	int	Output interface index (0-based)	required

Returns:

Type	Description
int | None	Current parallelism value or None if no parallelism

Raises:

Type	Description
IndexError	If index out of range

Source code in brainsmith/dataflow/dse_models.py

def get_output_stream_value(self, index: int) -> int | None:
    """Get current parallelism value for output interface.

    Args:
        index: Output interface index (0-based)

    Returns:
        Current parallelism value or None if no parallelism

    Raises:
        IndexError: If index out of range
    """
    param = self.get_output_stream_param(index)
    return self.config.get(param) if param else None

output_stream_shape

output_stream_shape(output_idx: int = 0) -> Shape

Stream shape for output.

Returns the output's stream_shape attribute (resolved during configure).

Source code in brainsmith/dataflow/dse_models.py

def output_stream_shape(self, output_idx: int = 0) -> Shape:
    """Stream shape for output.

    Returns the output's stream_shape attribute (resolved during configure).
    """
    return self.output_list[output_idx].stream_shape

output_stream_width_bits

output_stream_width_bits(output_idx: int = 0) -> int

Stream width in bits for output.

Returns the actual stream width based on the output's stream_shape.

Source code in brainsmith/dataflow/dse_models.py

def output_stream_width_bits(self, output_idx: int = 0) -> int:
    """Stream width in bits for output.

    Returns the actual stream width based on the output's stream_shape.
    """
    output = self.output_list[output_idx]
    return output.stream_width_bits

sweep_dimension

sweep_dimension(name: str, start: int | str | None = None, stop: int | str | None = None) -> Iterator[KernelDesignPoint]

Sweep through all valid values for a dimension.

For ordered dimensions, iterates in order from start to stop. For discrete dimensions, iterates in sorted order (ignores start/stop).

Parameters:

Name	Type	Description	Default
name	str	Dimension to sweep	required
start	int | str | None	Start value (None = use min/first), ordered dims only	None
stop	int | str | None	Stop value (None = use max/last), ordered dims only	None

Yields:

Type	Description
KernelDesignPoint	KernelDesignPoint for each value in range

Raises:

Type	Description
KeyError	If dimension not found

Examples:

>>> # Full sweep (ordered)
>>> for point in base.sweep_dimension("PE"):
...     evaluate(point)

>>> # Partial sweep (ordered)
>>> for point in base.sweep_dimension("SIMD", start=8, stop=64):
...     evaluate(point)

>>> # Discrete sweep (ignores start/stop)
>>> for point in base.sweep_dimension("ram_style"):
...     evaluate(point)

Source code in brainsmith/dataflow/dse_models.py

def sweep_dimension(
    self, name: str, start: int | str | None = None, stop: int | str | None = None
) -> Iterator["KernelDesignPoint"]:
    """Sweep through all valid values for a dimension.

    For ordered dimensions, iterates in order from start to stop.
    For discrete dimensions, iterates in sorted order (ignores start/stop).

    Args:
        name: Dimension to sweep
        start: Start value (None = use min/first), ordered dims only
        stop: Stop value (None = use max/last), ordered dims only

    Yields:
        KernelDesignPoint for each value in range

    Raises:
        KeyError: If dimension not found

    Examples:
        >>> # Full sweep (ordered)
        >>> for point in base.sweep_dimension("PE"):
        ...     evaluate(point)

        >>> # Partial sweep (ordered)
        >>> for point in base.sweep_dimension("SIMD", start=8, stop=64):
        ...     evaluate(point)

        >>> # Discrete sweep (ignores start/stop)
        >>> for point in base.sweep_dimension("ram_style"):
        ...     evaluate(point)
    """
    dim = self.design_space.get_parameter(name)

    if isinstance(dim, OrderedParameter):
        # Ordered: sweep in order from start to stop
        start_idx = 0 if start is None else dim.index_of(start)
        stop_idx = len(dim) - 1 if stop is None else dim.index_of(stop)

        for idx in range(start_idx, stop_idx + 1):
            value = dim.at_index(idx)
            yield self.with_dimension(name, value)
    else:  # frozenset
        # Discrete: iterate in sorted order
        for value in sorted(dim):
            yield self.with_dimension(name, value)

sweep_percentage

sweep_percentage(name: str, percentages: list[float], rounding: Literal['natural', 'down', 'up'] = 'natural') -> Iterator[KernelDesignPoint]

Sweep through ordered dimension at specified percentage points.

Only valid for ordered dimensions.

Parameters:

Name	Type	Description	Default
name	str	Ordered dimension to sweep	required
percentages	list[float]	List of percentage points (0.0-1.0)	required
rounding	Literal['natural', 'down', 'up']	Rounding mode for fractional indices	'natural'

Yields:

Type	Description
KernelDesignPoint	KernelDesignPoint for each percentage

Raises:

Type	Description
KeyError	If dimension not found
TypeError	If dimension is discrete (not ordered)

Examples:

>>> # Quartile sweep
>>> for point in base.sweep_percentage("PE", [0.0, 0.25, 0.5, 0.75, 1.0]):
...     evaluate(point)

>>> # Decile sweep
>>> deciles = [i/10 for i in range(11)]
>>> for point in base.sweep_percentage("SIMD", deciles):
...     evaluate(point)

Source code in brainsmith/dataflow/dse_models.py

def sweep_percentage(
    self,
    name: str,
    percentages: list[float],
    rounding: Literal["natural", "down", "up"] = "natural",
) -> Iterator["KernelDesignPoint"]:
    """Sweep through ordered dimension at specified percentage points.

    Only valid for ordered dimensions.

    Args:
        name: Ordered dimension to sweep
        percentages: List of percentage points (0.0-1.0)
        rounding: Rounding mode for fractional indices

    Yields:
        KernelDesignPoint for each percentage

    Raises:
        KeyError: If dimension not found
        TypeError: If dimension is discrete (not ordered)

    Examples:
        >>> # Quartile sweep
        >>> for point in base.sweep_percentage("PE", [0.0, 0.25, 0.5, 0.75, 1.0]):
        ...     evaluate(point)

        >>> # Decile sweep
        >>> deciles = [i/10 for i in range(11)]
        >>> for point in base.sweep_percentage("SIMD", deciles):
        ...     evaluate(point)
    """
    for pct in percentages:
        yield self.with_percentage(name, pct, rounding)

with_dimension

with_dimension(name: str, value: int | str) -> KernelDesignPoint

Create new design point with specified dimension value.

Works for both ordered and discrete dimensions.

Parameters:

Name	Type	Description	Default
name	str	Dimension name	required
value	int | str	New value for dimension	required

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with updated dimension

Raises:

Type	Description
KeyError	If dimension not found
ValueError	If value not valid for dimension

Examples:

>>> point = design_space.configure({"SIMD": 4, "PE": 1})
>>> point2 = point.with_dimension("SIMD", 8)
>>> point2.config["SIMD"]
8

Source code in brainsmith/dataflow/dse_models.py

def with_dimension(self, name: str, value: int | str) -> "KernelDesignPoint":
    """Create new design point with specified dimension value.

    Works for both ordered and discrete dimensions.

    Args:
        name: Dimension name
        value: New value for dimension

    Returns:
        New KernelDesignPoint with updated dimension

    Raises:
        KeyError: If dimension not found
        ValueError: If value not valid for dimension

    Examples:
        >>> point = design_space.configure({"SIMD": 4, "PE": 1})
        >>> point2 = point.with_dimension("SIMD", 8)
        >>> point2.config["SIMD"]
        8
    """
    new_config = {**self.config, name: value}
    return self.design_space.configure(new_config)

with_input_stream

with_input_stream(index: int, value: int) -> KernelDesignPoint

Set input interface stream parallelism by index.

Convenience method for interface-agnostic parallelism navigation. Automatically resolves the parallelism parameter name from the interface.

Parameters:

Name	Type	Description	Default
index	int	Input interface index (0-based)	required
value	int	Parallelism value	required

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with updated parallelism

Raises:

Type	Description
IndexError	If index out of range
ValueError	If interface has no parallelism parameter or value invalid

Example

Set first input to PE=16

point2 = point.with_input_stream(0, 16)

Source code in brainsmith/dataflow/dse_models.py

def with_input_stream(self, index: int, value: int) -> "KernelDesignPoint":
    """Set input interface stream parallelism by index.

    Convenience method for interface-agnostic parallelism navigation.
    Automatically resolves the parallelism parameter name from the interface.

    Args:
        index: Input interface index (0-based)
        value: Parallelism value

    Returns:
        New KernelDesignPoint with updated parallelism

    Raises:
        IndexError: If index out of range
        ValueError: If interface has no parallelism parameter or value invalid

    Example:
        >>> # Set first input to PE=16
        >>> point2 = point.with_input_stream(0, 16)
    """
    return self._with_stream_helper(self.input_list, "Input", index, value)

with_input_stream_percentage

with_input_stream_percentage(index: int, percentage: float, rounding: Literal['natural', 'down', 'up'] = 'natural') -> KernelDesignPoint

Set input stream parallelism to percentage of range.

Parameters:

Name	Type	Description	Default
index	int	Input interface index (0-based)	required
percentage	float	Value from 0.0 to 1.0 (0.0=min, 1.0=max)	required
rounding	Literal['natural', 'down', 'up']	How to round fractional indices	'natural'

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with parallelism at percentage

Raises:

Type	Description
IndexError	If index out of range
ValueError	If interface has no parallelism parameter or percentage invalid

Source code in brainsmith/dataflow/dse_models.py

def with_input_stream_percentage(
    self, index: int, percentage: float, rounding: Literal["natural", "down", "up"] = "natural"
) -> "KernelDesignPoint":
    """Set input stream parallelism to percentage of range.

    Args:
        index: Input interface index (0-based)
        percentage: Value from 0.0 to 1.0 (0.0=min, 1.0=max)
        rounding: How to round fractional indices

    Returns:
        New KernelDesignPoint with parallelism at percentage

    Raises:
        IndexError: If index out of range
        ValueError: If interface has no parallelism parameter or percentage invalid
    """
    return self._with_stream_percentage_helper(
        self.input_list, "Input", index, percentage, rounding
    )

with_max

with_max(name: str) -> KernelDesignPoint

Create new design point with ordered dimension at maximum.

Parameters:

Name	Type	Description	Default
name	str	Dimension name (must be ordered)	required

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with dimension at maximum

Raises:

Type	Description
KeyError	If dimension not found
TypeError	If dimension is discrete (not ordered)

Examples:

>>> point = design_space.configure({"SIMD": 8, "PE": 4})
>>> point2 = point.with_max("SIMD")
>>> point2.config["SIMD"]
64

Source code in brainsmith/dataflow/dse_models.py

def with_max(self, name: str) -> "KernelDesignPoint":
    """Create new design point with ordered dimension at maximum.

    Args:
        name: Dimension name (must be ordered)

    Returns:
        New KernelDesignPoint with dimension at maximum

    Raises:
        KeyError: If dimension not found
        TypeError: If dimension is discrete (not ordered)

    Examples:
        >>> point = design_space.configure({"SIMD": 8, "PE": 4})
        >>> point2 = point.with_max("SIMD")
        >>> point2.config["SIMD"]
        64
    """
    max_val = self.design_space.get_ordered_parameter(name).max()
    return self.with_dimension(name, max_val)

with_min

with_min(name: str) -> KernelDesignPoint

Create new design point with ordered dimension at minimum.

Parameters:

Name	Type	Description	Default
name	str	Dimension name (must be ordered)	required

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with dimension at minimum

Raises:

Type	Description
KeyError	If dimension not found
TypeError	If dimension is discrete (not ordered)

Examples:

>>> point = design_space.configure({"SIMD": 8, "PE": 4})
>>> point2 = point.with_min("SIMD")
>>> point2.config["SIMD"]
1

Source code in brainsmith/dataflow/dse_models.py

def with_min(self, name: str) -> "KernelDesignPoint":
    """Create new design point with ordered dimension at minimum.

    Args:
        name: Dimension name (must be ordered)

    Returns:
        New KernelDesignPoint with dimension at minimum

    Raises:
        KeyError: If dimension not found
        TypeError: If dimension is discrete (not ordered)

    Examples:
        >>> point = design_space.configure({"SIMD": 8, "PE": 4})
        >>> point2 = point.with_min("SIMD")
        >>> point2.config["SIMD"]
        1
    """
    min_val = self.design_space.get_ordered_parameter(name).min()
    return self.with_dimension(name, min_val)

with_output_stream

with_output_stream(index: int, value: int) -> KernelDesignPoint

Set output interface stream parallelism by index.

Parameters:

Name	Type	Description	Default
index	int	Output interface index (0-based)	required
value	int	Parallelism value	required

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with updated parallelism

Raises:

Type	Description
IndexError	If index out of range
ValueError	If interface has no parallelism parameter or value invalid

Source code in brainsmith/dataflow/dse_models.py

def with_output_stream(self, index: int, value: int) -> "KernelDesignPoint":
    """Set output interface stream parallelism by index.

    Args:
        index: Output interface index (0-based)
        value: Parallelism value

    Returns:
        New KernelDesignPoint with updated parallelism

    Raises:
        IndexError: If index out of range
        ValueError: If interface has no parallelism parameter or value invalid
    """
    return self._with_stream_helper(self.output_list, "Output", index, value)

with_output_stream_percentage

with_output_stream_percentage(index: int, percentage: float, rounding: Literal['natural', 'down', 'up'] = 'natural') -> KernelDesignPoint

Set output stream parallelism to percentage of range.

Parameters:

Name	Type	Description	Default
index	int	Output interface index (0-based)	required
percentage	float	Value from 0.0 to 1.0 (0.0=min, 1.0=max)	required
rounding	Literal['natural', 'down', 'up']	How to round fractional indices	'natural'

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with parallelism at percentage

Raises:

Type	Description
IndexError	If index out of range
ValueError	If interface has no parallelism parameter or percentage invalid

Source code in brainsmith/dataflow/dse_models.py

def with_output_stream_percentage(
    self, index: int, percentage: float, rounding: Literal["natural", "down", "up"] = "natural"
) -> "KernelDesignPoint":
    """Set output stream parallelism to percentage of range.

    Args:
        index: Output interface index (0-based)
        percentage: Value from 0.0 to 1.0 (0.0=min, 1.0=max)
        rounding: How to round fractional indices

    Returns:
        New KernelDesignPoint with parallelism at percentage

    Raises:
        IndexError: If index out of range
        ValueError: If interface has no parallelism parameter or percentage invalid
    """
    return self._with_stream_percentage_helper(
        self.output_list, "Output", index, percentage, rounding
    )

with_percentage

with_percentage(name: str, percentage: float, rounding: str = 'natural') -> KernelDesignPoint

Create new design point with ordered dimension at percentage.

Parameters:

Name	Type	Description	Default
name	str	Dimension name (must be ordered)	required
percentage	float	Position in range [0.0, 1.0]	required
rounding	str	'natural', 'down', or 'up'	'natural'

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with dimension at percentage

Raises:

Type	Description
KeyError	If dimension not found
TypeError	If dimension is discrete (not ordered)
ValueError	If percentage out of range

Examples:

>>> point = design_space.configure({"SIMD": 4, "PE": 1})
>>> point2 = point.with_percentage("SIMD", 0.5)
>>> point2.config["SIMD"]
8

Source code in brainsmith/dataflow/dse_models.py

def with_percentage(
    self, name: str, percentage: float, rounding: str = "natural"
) -> "KernelDesignPoint":
    """Create new design point with ordered dimension at percentage.

    Args:
        name: Dimension name (must be ordered)
        percentage: Position in range [0.0, 1.0]
        rounding: 'natural', 'down', or 'up'

    Returns:
        New KernelDesignPoint with dimension at percentage

    Raises:
        KeyError: If dimension not found
        TypeError: If dimension is discrete (not ordered)
        ValueError: If percentage out of range

    Examples:
        >>> point = design_space.configure({"SIMD": 4, "PE": 1})
        >>> point2 = point.with_percentage("SIMD", 0.5)
        >>> point2.config["SIMD"]
        8
    """
    value = self.design_space.get_ordered_parameter(name).at_percentage(percentage, rounding)
    return self.with_dimension(name, value)

with_step_down

with_step_down(name: str, n: int = 1) -> KernelDesignPoint

Create new design point with ordered dimension stepped down.

Clamps at minimum if n steps would go below bounds.

Parameters:

Name	Type	Description	Default
name	str	Dimension name (must be ordered)	required
n	int	Number of steps to move down (default 1)	1

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with dimension stepped down

Raises:

Type	Description
KeyError	If dimension not found
TypeError	If dimension is discrete (not ordered)
ValueError	If current value not in dimension or n < 0

Examples:

>>> point = design_space.configure({"SIMD": 16, "PE": 4})
>>> point2 = point.with_step_down("SIMD", 1)
>>> point2.config["SIMD"]
8

Source code in brainsmith/dataflow/dse_models.py

def with_step_down(self, name: str, n: int = 1) -> "KernelDesignPoint":
    """Create new design point with ordered dimension stepped down.

    Clamps at minimum if n steps would go below bounds.

    Args:
        name: Dimension name (must be ordered)
        n: Number of steps to move down (default 1)

    Returns:
        New KernelDesignPoint with dimension stepped down

    Raises:
        KeyError: If dimension not found
        TypeError: If dimension is discrete (not ordered)
        ValueError: If current value not in dimension or n < 0

    Examples:
        >>> point = design_space.configure({"SIMD": 16, "PE": 4})
        >>> point2 = point.with_step_down("SIMD", 1)
        >>> point2.config["SIMD"]
        8
    """
    dim = self.design_space.get_ordered_parameter(name)
    current = self.config[name]
    new_val = dim.step_down(current, n)
    return self.with_dimension(name, new_val)

with_step_up

with_step_up(name: str, n: int = 1) -> KernelDesignPoint

Create new design point with ordered dimension stepped up.

Clamps at maximum if n steps would exceed bounds.

Parameters:

Name	Type	Description	Default
name	str	Dimension name (must be ordered)	required
n	int	Number of steps to move up (default 1)	1

Returns:

Type	Description
KernelDesignPoint	New KernelDesignPoint with dimension stepped up

Raises:

Type	Description
KeyError	If dimension not found
TypeError	If dimension is discrete (not ordered)
ValueError	If current value not in dimension or n < 0

Examples:

>>> point = design_space.configure({"SIMD": 4, "PE": 1})
>>> point2 = point.with_step_up("SIMD", 2)
>>> point2.config["SIMD"]
16

Source code in brainsmith/dataflow/dse_models.py

def with_step_up(self, name: str, n: int = 1) -> "KernelDesignPoint":
    """Create new design point with ordered dimension stepped up.

    Clamps at maximum if n steps would exceed bounds.

    Args:
        name: Dimension name (must be ordered)
        n: Number of steps to move up (default 1)

    Returns:
        New KernelDesignPoint with dimension stepped up

    Raises:
        KeyError: If dimension not found
        TypeError: If dimension is discrete (not ordered)
        ValueError: If current value not in dimension or n < 0

    Examples:
        >>> point = design_space.configure({"SIMD": 4, "PE": 1})
        >>> point2 = point.with_step_up("SIMD", 2)
        >>> point2.config["SIMD"]
        16
    """
    dim = self.design_space.get_ordered_parameter(name)
    current = self.config[name]
    new_val = dim.step_up(current, n)
    return self.with_dimension(name, new_val)

Example:

# Get design point from kernel operator
op._ensure_ready(model)
point = op.design_point

# Configure using interface-based API (for stream parameters)
point = point.with_input_stream(0, 32)   # Set input PE=32
point = point.with_output_stream(0, 16)  # Set output PE=16

# Configure using dimension-based API (for generic DSE)
point = point.with_dimension("ram_style", "distributed")
point = point.with_dimension("depth", 256)

# Apply configuration
op.apply_design_point(point)

---

InterfaceDesignPoint dataclass

InterfaceDesignPoint(design_space: InterfaceDesignSpace, stream_shape: Shape)

Interface instance with resolved parallelization.

Flyweight pattern: references parent design space, stores only configuration- specific stream_shape. Delegates tensor_shape, block_shape, and datatype to design space for minimal memory overhead.

Attributes:

Name	Type	Description
design_space	InterfaceDesignSpace	Parent InterfaceDesignSpace
stream_shape	Shape	Resolved stream dimensions for this configuration

block_folding_factor property

block_folding_factor: int

Cycles to stream one block.

Product of stream_cycles_shape. Uses ceiling division: a block of size 32 with stream width 10 requires ceil(32/10) = 4 cycles (3 full + 1 partial).

stream_cycles_shape property

stream_cycles_shape: Shape

Per-dimension cycles needed to stream one block.

Returns shape where each dimension is ceil(block_dim / stream_dim). Describes temporal execution: how we stream each tile.

block_shape=(32, 16), stream_shape=(8, 4)

→ stream_cycles_shape=(4, 4) # 4x4 cycles per block

stream_width_bits property

stream_width_bits: int

Stream width in bits.

streaming_bandwidth property

streaming_bandwidth: int

Elements streamed per cycle.

tensor_blocks_shape property

tensor_blocks_shape: Shape

Per-dimension blocks needed to tile tensor.

Returns shape where each dimension is ceil(tensor_dim / block_dim). Describes spatial decomposition: how we tile the problem.

tensor_shape=(100, 64), block_shape=(32, 16)

→ tensor_blocks_shape=(4, 4) # 4x4 grid of blocks

tensor_folding_factor property

tensor_folding_factor: int

Number of blocks needed to cover full tensor.

Product of tensor_blocks_shape. Uses ceiling division: a tensor of size 100 with block size 32 requires ceil(100/32) = 4 blocks (3 full + 1 partial).

get_shape

get_shape(hierarchy: ShapeHierarchy) -> Shape

Get shape at specified hierarchy level.

Parameters:

Name	Type	Description	Default
hierarchy	ShapeHierarchy	Which level of the shape hierarchy to retrieve	required

Returns:

Type	Description
Shape	Shape at the specified level

Raises:

Type	Description
ValueError	If hierarchy is invalid

Source code in brainsmith/dataflow/dse_models.py

def get_shape(self, hierarchy: ShapeHierarchy) -> Shape:
    """Get shape at specified hierarchy level.

    Args:
        hierarchy: Which level of the shape hierarchy to retrieve

    Returns:
        Shape at the specified level

    Raises:
        ValueError: If hierarchy is invalid
    """
    if hierarchy == ShapeHierarchy.STREAM:
        return self.stream_shape
    elif hierarchy == ShapeHierarchy.BLOCK:
        return self.design_space.block_shape
    elif hierarchy == ShapeHierarchy.TENSOR:
        return self.design_space.tensor_shape
    else:
        raise ValueError(f"Invalid hierarchy: {hierarchy}")

---

OrderedParameter dataclass

OrderedParameter(name: str, values: tuple[int, ...], default: int | None = None)

Ordered parameter for DSE navigation.

Stores discrete values in sorted order, enabling navigation operations like stepping, percentage-based indexing, and min/max access.

Used for parallelization parameters (PE, SIMD, MW, MH) and other explorable parameters with natural ordering (depth, num_layers, etc.).

Attributes:

Name	Type	Description
name	str	Parameter name (e.g., "SIMD", "PE", "depth")
values	tuple[int, ...]	Sorted tuple of valid values
default	int | None	Default value (None = minimum)

Examples:

>>> simd = OrderedParameter("SIMD", (1, 2, 4, 8, 16, 32, 64))
>>> simd.min()
1
>>> simd.at_percentage(0.5)
8
>>> simd.step_up(8, n=2)
32

__contains__

__contains__(value: int) -> bool

Check if value in parameter (for 'value in param' syntax).

Source code in brainsmith/dataflow/ordered_parameter.py

def __contains__(self, value: int) -> bool:
    """Check if value in parameter (for 'value in param' syntax)."""
    return value in self.values

__iter__

__iter__() -> Iterator[int]

Iterate over values in order.

Source code in brainsmith/dataflow/ordered_parameter.py

def __iter__(self) -> Iterator[int]:
    """Iterate over values in order."""
    return iter(self.values)

__len__

__len__() -> int

Number of valid values.

Source code in brainsmith/dataflow/ordered_parameter.py

def __len__(self) -> int:
    """Number of valid values."""
    return len(self.values)

__post_init__

__post_init__()

Validate invariants: sorted, unique, non-empty.

Source code in brainsmith/dataflow/ordered_parameter.py

def __post_init__(self):
    """Validate invariants: sorted, unique, non-empty."""
    if not self.values:
        raise ValueError(f"OrderedParameter '{self.name}' has empty values")

    # Ensure tuple (not list)
    if not isinstance(self.values, tuple):
        object.__setattr__(self, "values", tuple(self.values))

    # Validate sorted
    if self.values != tuple(sorted(self.values)):
        raise ValueError(
            f"OrderedParameter '{self.name}' values must be sorted ascending. "
            f"Got: {self.values}"
        )

    # Validate unique
    if len(self.values) != len(set(self.values)):
        # Find duplicates efficiently (O(n) instead of O(n²))
        seen = set()
        duplicates = {v for v in self.values if v in seen or seen.add(v)}
        raise ValueError(f"OrderedParameter '{self.name}' has duplicate values: {duplicates}")

    # Validate default (if specified)
    if self.default is not None and self.default not in self.values:
        raise ValueError(
            f"Default value {self.default} not in parameter '{self.name}'. "
            f"Valid values: {self.values}"
        )

__repr__

__repr__() -> str

String representation.

Source code in brainsmith/dataflow/ordered_parameter.py

def __repr__(self) -> str:
    """String representation."""
    if len(self.values) <= 6:
        vals = str(self.values)
    else:
        vals = f"({self.values[0]}, {self.values[1]}, ..., {self.values[-1]})"

    default_str = f", default={self.default}" if self.default else ""
    return f"OrderedParameter('{self.name}', {vals}{default_str})"

at_index

at_index(idx: int) -> int

Get value at index (supports negative indexing).

Parameters:

Name	Type	Description	Default
idx	int	Index position (0-based, supports negative like Python lists)	required

Returns:

Type	Description
int	Value at index

Raises:

Type	Description
IndexError	If index out of range

Examples:

>>> param = OrderedParameter("PE", (1, 2, 4, 8, 16))
>>> param.at_index(0)
1
>>> param.at_index(-1)
16
>>> param.at_index(2)
4

Source code in brainsmith/dataflow/ordered_parameter.py

def at_index(self, idx: int) -> int:
    """Get value at index (supports negative indexing).

    Args:
        idx: Index position (0-based, supports negative like Python lists)

    Returns:
        Value at index

    Raises:
        IndexError: If index out of range

    Examples:
        >>> param = OrderedParameter("PE", (1, 2, 4, 8, 16))
        >>> param.at_index(0)
        1
        >>> param.at_index(-1)
        16
        >>> param.at_index(2)
        4
    """
    if not -len(self.values) <= idx < len(self.values):
        raise IndexError(
            f"Index {idx} out of range for parameter '{self.name}' "
            f"(length {len(self.values)})"
        )
    return self.values[idx]

at_percentage

at_percentage(percentage: float, rounding: Literal['natural', 'down', 'up'] = 'natural') -> int

Get value at percentage position in ordered sequence (0.0-1.0).

Maps percentage to continuous index space, then rounds to discrete index. Useful for sweeping through parameter at regular intervals regardless of actual vector length.

Parameters:

Name	Type	Description	Default
percentage	float	Position in range [0.0, 1.0] - 0.0 → first value (min) - 1.0 → last value (max) - 0.5 → middle value	required
rounding	Literal['natural', 'down', 'up']	How to round fractional indices - 'natural': Round to nearest (default, balanced) - 'down': Floor (conservative, prefer smaller values) - 'up': Ceiling (aggressive, prefer larger values)	'natural'

Returns:

Type	Description
int	Value at percentage position

Raises:

Type	Description
ValueError	If percentage not in [0.0, 1.0] or invalid rounding mode

Examples:

>>> param = OrderedParameter("PE", (1, 2, 4, 8, 16))  # 5 values
>>> param.at_percentage(0.0)
1
>>> param.at_percentage(1.0)
16
>>> param.at_percentage(0.5, rounding='natural')
4  # Middle value (index 2 of 0-4)
>>> param.at_percentage(0.75, rounding='down')
8  # 0.75 * 4 = 3.0 → floor(3.0) = 3 → values[3] = 8

>>> # With 4 values, percentages map cleanly to indices
>>> param4 = OrderedParameter("X", (10, 20, 30, 40))
>>> param4.at_percentage(0.0)
10  # 0.0 * 3 = 0
>>> param4.at_percentage(0.333, rounding='natural')
20  # 0.333 * 3 ≈ 1.0 → round(1.0) = 1
>>> param4.at_percentage(1.0)
40  # 1.0 * 3 = 3

Source code in brainsmith/dataflow/ordered_parameter.py

def at_percentage(
    self, percentage: float, rounding: Literal["natural", "down", "up"] = "natural"
) -> int:
    """Get value at percentage position in ordered sequence (0.0-1.0).

    Maps percentage to continuous index space, then rounds to discrete index.
    Useful for sweeping through parameter at regular intervals regardless
    of actual vector length.

    Args:
        percentage: Position in range [0.0, 1.0]
            - 0.0 → first value (min)
            - 1.0 → last value (max)
            - 0.5 → middle value
        rounding: How to round fractional indices
            - 'natural': Round to nearest (default, balanced)
            - 'down': Floor (conservative, prefer smaller values)
            - 'up': Ceiling (aggressive, prefer larger values)

    Returns:
        Value at percentage position

    Raises:
        ValueError: If percentage not in [0.0, 1.0] or invalid rounding mode

    Examples:
        >>> param = OrderedParameter("PE", (1, 2, 4, 8, 16))  # 5 values
        >>> param.at_percentage(0.0)
        1
        >>> param.at_percentage(1.0)
        16
        >>> param.at_percentage(0.5, rounding='natural')
        4  # Middle value (index 2 of 0-4)
        >>> param.at_percentage(0.75, rounding='down')
        8  # 0.75 * 4 = 3.0 → floor(3.0) = 3 → values[3] = 8

        >>> # With 4 values, percentages map cleanly to indices
        >>> param4 = OrderedParameter("X", (10, 20, 30, 40))
        >>> param4.at_percentage(0.0)
        10  # 0.0 * 3 = 0
        >>> param4.at_percentage(0.333, rounding='natural')
        20  # 0.333 * 3 ≈ 1.0 → round(1.0) = 1
        >>> param4.at_percentage(1.0)
        40  # 1.0 * 3 = 3
    """
    if not 0.0 <= percentage <= 1.0:
        raise ValueError(f"Percentage must be in [0.0, 1.0], got {percentage}")

    # Map percentage to continuous index space
    max_idx = len(self.values) - 1
    float_idx = percentage * max_idx

    # Apply rounding strategy
    if rounding == "down":
        idx = int(math.floor(float_idx))
    elif rounding == "up":
        idx = int(math.ceil(float_idx))
    elif rounding == "natural":
        idx = round(float_idx)
    else:
        raise ValueError(
            f"Invalid rounding mode '{rounding}'. " f"Must be 'natural', 'down', or 'up'."
        )

    # Clamp to valid range (defensive, should be unnecessary)
    idx = max(0, min(idx, max_idx))

    return self.values[idx]

get_default

get_default() -> int

Get default value (explicit or minimum).

Source code in brainsmith/dataflow/ordered_parameter.py

def get_default(self) -> int:
    """Get default value (explicit or minimum)."""
    return self.default if self.default is not None else self.values[0]

index_of

index_of(value: int) -> int

Get index of value in ordered sequence.

Parameters:

Name	Type	Description	Default
value	int	Value to find	required

Returns:

Type	Description
int	Zero-based index of value

Raises:

Type	Description
ValueError	If value not in parameter

Examples:

>>> param = OrderedParameter("SIMD", (1, 2, 4, 8, 16))
>>> param.index_of(4)
2
>>> param.index_of(16)
4

Source code in brainsmith/dataflow/ordered_parameter.py

def index_of(self, value: int) -> int:
    """Get index of value in ordered sequence.

    Args:
        value: Value to find

    Returns:
        Zero-based index of value

    Raises:
        ValueError: If value not in parameter

    Examples:
        >>> param = OrderedParameter("SIMD", (1, 2, 4, 8, 16))
        >>> param.index_of(4)
        2
        >>> param.index_of(16)
        4
    """
    try:
        return self.values.index(value)
    except ValueError:
        raise ValueError(
            f"Value {value} not in parameter '{self.name}'. "
            f"Valid range: [{self.min()}, {self.max()}], "
            f"values: {self.values}"
        )

max

max() -> int

Get maximum value (last in ordered sequence).

Source code in brainsmith/dataflow/ordered_parameter.py

def max(self) -> int:
    """Get maximum value (last in ordered sequence)."""
    return self.values[-1]

min

min() -> int

Get minimum value (first in ordered sequence).

Source code in brainsmith/dataflow/ordered_parameter.py

def min(self) -> int:
    """Get minimum value (first in ordered sequence)."""
    return self.values[0]

step_down

step_down(current: int, n: int = 1) -> int

Step down n positions from current value.

Clamps at minimum if n steps would go below bounds.

Parameters:

Name	Type	Description	Default
current	int	Current value (must be in parameter)	required
n	int	Number of steps to move down (positive integer)	1

Returns:

Type	Description
int	New value n steps down (clamped at min)

Raises:

Type	Description
ValueError	If current value not in parameter or n < 0

Examples:

>>> param = OrderedParameter("SIMD", (1, 2, 4, 8, 16, 32, 64))
>>> param.step_down(16, 1)
8
>>> param.step_down(16, 2)
4
>>> param.step_down(4, 10)
1  # Clamped at min

Source code in brainsmith/dataflow/ordered_parameter.py

def step_down(self, current: int, n: int = 1) -> int:
    """Step down n positions from current value.

    Clamps at minimum if n steps would go below bounds.

    Args:
        current: Current value (must be in parameter)
        n: Number of steps to move down (positive integer)

    Returns:
        New value n steps down (clamped at min)

    Raises:
        ValueError: If current value not in parameter or n < 0

    Examples:
        >>> param = OrderedParameter("SIMD", (1, 2, 4, 8, 16, 32, 64))
        >>> param.step_down(16, 1)
        8
        >>> param.step_down(16, 2)
        4
        >>> param.step_down(4, 10)
        1  # Clamped at min
    """
    if n < 0:
        raise ValueError(f"step_down requires n >= 0, got {n}")

    idx = self.index_of(current)
    new_idx = max(idx - n, 0)
    return self.values[new_idx]

step_up

step_up(current: int, n: int = 1) -> int

Step up n positions from current value.

Clamps at maximum if n steps would exceed bounds.

Parameters:

Name	Type	Description	Default
current	int	Current value (must be in parameter)	required
n	int	Number of steps to move up (positive integer)	1

Returns:

Type	Description
int	New value n steps up (clamped at max)

Raises:

Type	Description
ValueError	If current value not in parameter or n < 0

Examples:

>>> param = OrderedParameter("PE", (1, 2, 4, 8, 16, 32, 64))
>>> param.step_up(4, 1)
8
>>> param.step_up(4, 2)
16
>>> param.step_up(32, 10)
64  # Clamped at max

Source code in brainsmith/dataflow/ordered_parameter.py

def step_up(self, current: int, n: int = 1) -> int:
    """Step up n positions from current value.

    Clamps at maximum if n steps would exceed bounds.

    Args:
        current: Current value (must be in parameter)
        n: Number of steps to move up (positive integer)

    Returns:
        New value n steps up (clamped at max)

    Raises:
        ValueError: If current value not in parameter or n < 0

    Examples:
        >>> param = OrderedParameter("PE", (1, 2, 4, 8, 16, 32, 64))
        >>> param.step_up(4, 1)
        8
        >>> param.step_up(4, 2)
        16
        >>> param.step_up(32, 10)
        64  # Clamped at max
    """
    if n < 0:
        raise ValueError(f"step_up requires n >= 0, got {n}")

    idx = self.index_of(current)
    new_idx = min(idx + n, len(self.values) - 1)
    return self.values[new_idx]

validate

validate(value: int) -> bool

Check if value is valid for this parameter.

Source code in brainsmith/dataflow/ordered_parameter.py

def validate(self, value: int) -> bool:
    """Check if value is valid for this parameter."""
    return value in self.values

---

DesignSpaceBuilder

Builds kernel design space from schema and ONNX context.

Two-phase construction: 1. build() creates KernelDesignSpace once (tensor/block shapes, datatypes, valid ranges) 2. design_space.configure() creates KernelDesignPoint many times (stream shapes for specific params)

Example

builder = DesignSpaceBuilder() context = BuildContext( ... schema=kernel_schema, ... model_w=model_wrapper, ... node_inputs=list(node.input), ... node_outputs=list(node.output), ... param_getter=self.get_nodeattr, ... param_setter=self.set_nodeattr, ... node_name=node.name ... ) design_space = builder.build(context) point = design_space.configure({"SIMD": 64, "PE": 1})

build

build(ctx: BuildContext) -> KernelDesignSpace

Build kernel design space from ONNX context.

Resolves all properties constant across parallelization configs: - Tensor shapes (from ONNX graph) - Block shapes (from block_tiling templates) - Datatypes (from ONNX graph + union type derivation) - Internal datatypes (from union type derivation) - Structural constraints (validated once) - Valid parallelization parameter ranges (divisor sets)

Stream shapes are left as templates for later resolution via configure().

Parameters:

Name	Type	Description	Default
ctx	BuildContext	Build context with ONNX node and ModelWrapper	required

Returns:

Type	Description
KernelDesignSpace	KernelDesignSpace ready for configuration exploration

Raises:

Type	Description
ValueError	If structural constraints fail

Source code in brainsmith/dataflow/builder.py

def build(self, ctx: BuildContext) -> KernelDesignSpace:
    """Build kernel design space from ONNX context.

    Resolves all properties constant across parallelization configs:
    - Tensor shapes (from ONNX graph)
    - Block shapes (from block_tiling templates)
    - Datatypes (from ONNX graph + union type derivation)
    - Internal datatypes (from union type derivation)
    - Structural constraints (validated once)
    - Valid parallelization parameter ranges (divisor sets)

    Stream shapes are left as templates for later resolution via configure().

    Args:
        ctx: Build context with ONNX node and ModelWrapper

    Returns:
        KernelDesignSpace ready for configuration exploration

    Raises:
        ValueError: If structural constraints fail
    """
    from .dse_models import KernelDesignSpace
    from .validation import DesignSpaceValidationContext

    self._ctx = ctx
    self._interfaces: dict[str, Any] = {}

    logger.debug(f"Building KernelDesignSpace for {ctx.node_name}")

    # Build input interfaces from ONNX graph
    inputs: dict[str, InterfaceDesignSpace] = {}

    for i, inp_name in enumerate(ctx.node_inputs):
        if not inp_name:
            continue

        if i >= len(ctx.schema.inputs):
            logger.warning(
                f"Node has input {i} but schema only defines {len(ctx.schema.inputs)} inputs"
            )
            continue

        schema = ctx.schema.inputs[i]

        try:
            interface = self._build_interface(
                direction="input", index=i, tensor_name=inp_name, schema=schema
            )
            inputs[schema.name] = interface
            self._interfaces[schema.name] = interface
        except ValueError as e:
            raise ValueError(f"Failed to build input '{schema.name}': {e}") from e

    # Derive internal datatypes from inputs and parameters
    internal_datatypes = {}

    if ctx.schema.internal_datatypes:
        for internal_name, datatype_spec in ctx.schema.internal_datatypes.items():
            try:
                # Use unified DatatypeSpec resolver (supports union types)
                datatype = self._resolve_datatype_spec(
                    spec=datatype_spec,
                    tensor_name=None,  # Internals have no ONNX tensor
                    fallback_datatype=None,  # Internal datatypes must be explicit
                )
                ctx.param_setter(f"{internal_name}Datatype", datatype.name)

                # Store datatype directly (no shapes for internal datatypes)
                self._interfaces[internal_name] = datatype
                internal_datatypes[internal_name] = datatype

                logger.debug(f"  Internal '{internal_name}': dtype={datatype.name}")
            except ValueError as e:
                raise ValueError(
                    f"Failed to resolve internal datatype '{internal_name}': {e}"
                ) from e

    # Build output interfaces (may derive datatypes from inputs)
    outputs: dict[str, InterfaceDesignSpace] = {}

    for i, out_name in enumerate(ctx.node_outputs):
        if i >= len(ctx.schema.outputs):
            logger.warning(
                f"Node has output {i} but schema only defines {len(ctx.schema.outputs)} outputs"
            )
            continue

        schema = ctx.schema.outputs[i]

        try:
            interface = self._build_interface(
                direction="output", index=i, tensor_name=out_name, schema=schema
            )
            outputs[schema.name] = interface
            self._interfaces[schema.name] = interface
        except ValueError as e:
            raise ValueError(f"Failed to build output '{schema.name}': {e}") from e

    # Separate constraints by evaluation phase (structural vs optimization)
    structural_constraints = [
        c for c in ctx.schema.constraints if c.evaluation_phase == "structural"
    ]
    optimization_constraints = [
        c for c in ctx.schema.constraints if c.evaluation_phase != "structural"
    ]

    logger.debug(
        f"  Split {len(ctx.schema.constraints)} constraints: "
        f"{len(structural_constraints)} structural, {len(optimization_constraints)} optimization"
    )

    # Validate structural constraints against design space
    if structural_constraints:
        validation_ctx = DesignSpaceValidationContext(
            inputs=inputs,
            outputs=outputs,
            internal_datatypes=internal_datatypes,
            param_getter=ctx.param_getter,
        )

        failed = [
            f"{c.describe()}: {e}"
            for c in structural_constraints
            if (e := c.check(validation_ctx))
        ]
        if failed:
            raise ValueError(f"{ctx.node_name} validation failed:\n" + "\n".join(failed))

        logger.debug(f"  All {len(structural_constraints)} structural constraints passed")

    # Compute valid dimension values (tiling from divisors + DSE from schema)
    all_dimensions = self._compute_dimension_ranges(inputs, outputs, ctx.schema)

    # Link parallelism metadata to interfaces (shared dimension references)
    # Must happen AFTER dimension computation so dimensions dict is available
    inputs = self._link_parallelism_metadata(inputs, all_dimensions)
    outputs = self._link_parallelism_metadata(outputs, all_dimensions)

    # Assemble immutable design space model
    # Note: structural_constraints validated above but not stored (never re-validated)
    design_space = KernelDesignSpace(
        name=ctx.schema.name,
        inputs=inputs,
        outputs=outputs,
        internal_datatypes=internal_datatypes,
        optimization_constraints=optimization_constraints,
        parameters=all_dimensions,
    )

    logger.debug(f"KernelDesignSpace built successfully for {ctx.node_name}")
    return design_space

---

BuildContext dataclass

BuildContext(schema: KernelSchema, model_w: ModelWrapper, node_inputs: list[str], node_outputs: list[str], param_getter: Callable[[str], Any], param_setter: Callable[[str, Any], None], node_name: str = '<unknown>')

Build context for kernel design space construction.

Encapsulates all data needed to build a KernelDesignSpace from a schema.

Attributes:

Name	Type	Description
schema	KernelSchema	KernelSchema defining structure
model_w	ModelWrapper	ModelWrapper for ONNX graph access
node_inputs	list[str]	ONNX node input tensor names
node_outputs	list[str]	ONNX node output tensor names
param_getter	Callable[[str], Any]	Function to retrieve nodeattr values
param_setter	Callable[[str, Any], None]	Function to store nodeattr values
node_name	str	Node name for error messages

---

Constraint

Bases: Protocol

Validation rule for kernel constraints.

Pure predicate that validates kernel properties during construction. Uses duck typing to work with any validation context providing required methods.

Required methods: - check(ctx) → Optional[str] - describe() → str

evaluation_phase property

evaluation_phase: str

When to evaluate this constraint during kernel construction.

Returns:

Type	Description
str	'structural' - Evaluated once during design space construction (Phase 1) Constraints that determine backend compatibility (tensor shapes, block shapes, datatypes, etc.)
str	'optimization' - Evaluated per-configuration during configure() (Phase 2) Constraints that bound optimization space (stream shapes, parallelization parameters, etc.)

Default implementation uses heuristic: - Constraints with hierarchy == STREAM are optimization constraints - All other constraints are structural

Subclasses can override this property for explicit classification.

Examples:

DatatypeInteger: 'structural' (no hierarchy, datatype determines compatibility) ShapesEqual(hierarchy=TENSOR): 'structural' (tensor shape determines compatibility) ShapesEqual(hierarchy=BLOCK): 'structural' (block shape determines compatibility) ShapesEqual(hierarchy=STREAM): 'optimization' (stream shape bounds optimization) DimensionDivisible(hierarchy=STREAM): 'optimization' (stream dim bounds optimization)

check

check(ctx) -> str | None

Check constraint in given context.

Parameters:

Name	Type	Description	Default
ctx		Validation context (DesignSpaceValidationContext or ConfigurationValidationContext)	required

Returns:

Type	Description
str | None	None if satisfied, error message string if violated

Source code in brainsmith/dataflow/constraints.py

def check(self, ctx) -> str | None:
    """Check constraint in given context.

    Args:
        ctx: Validation context (DesignSpaceValidationContext or ConfigurationValidationContext)

    Returns:
        None if satisfied, error message string if violated
    """
    ...

describe

describe() -> str

Human-readable description of constraint.

Source code in brainsmith/dataflow/constraints.py

def describe(self) -> str:
    """Human-readable description of constraint."""
    ...

---

ValidationError

ValidationError(message: str, location: str = '', suggestions: list = None)

Bases: ValueError

Validation error with context and suggestions.

Parameters:

Name	Type	Description	Default
message	str	Error message	required
location	str	Optional context (e.g., "input.stream[1]")	''
suggestions	list	Optional list of suggestions	None

Examples:

>>> raise ValidationError("Invalid parameter")
>>> raise ValidationError("PE must divide 768", location="output.stream[1]")

Source code in brainsmith/dataflow/validation.py

def __init__(self, message: str, location: str = "", suggestions: list = None):
    self.message = message
    self.location = location
    self.suggestions = suggestions or []
    super().__init__(self._format_message())

__str__

__str__() -> str

Return formatted message (delegates to parent Exception).

Source code in brainsmith/dataflow/validation.py

def __str__(self) -> str:
    """Return formatted message (delegates to parent Exception)."""
    return super().__str__()

---

DesignSpaceValidationContext dataclass

DesignSpaceValidationContext(inputs: dict[str, Any], outputs: dict[str, Any], internal_datatypes: dict[str, DataType], param_getter: Callable[[str], Any] | None = None)

Validation context for structural constraints during design space build.

Used during KernelDesignSpace construction to validate tensor shapes, block shapes, and datatypes. Stream shapes not available until configure().

Attributes:

Name	Type	Description
inputs	dict[str, Any]	Input interfaces (InterfaceDesignSpace)
outputs	dict[str, Any]	Output interfaces (InterfaceDesignSpace)
internal_datatypes	dict[str, DataType]	Internal datatypes
param_getter	Callable[[str], Any] | None	Optional nodeattr getter

Example

ctx = DesignSpaceValidationContext( inputs=interfaces_input, outputs=interfaces_output, internal_datatypes=internal_datatypes, param_getter=get_nodeattr ) for constraint in structural_constraints: if error := constraint.check(ctx): raise ValueError(error)

get_datatype

get_datatype(name: str) -> DataType

Get datatype from interface or internal datatypes.

Parameters:

Name	Type	Description	Default
name	str	Interface or internal datatype name	required

Returns:

Type	Description
DataType	DataType

Raises:

Type	Description
KeyError	If interface/datatype not found

Source code in brainsmith/dataflow/validation.py

def get_datatype(self, name: str) -> DataType:
    """Get datatype from interface or internal datatypes.

    Args:
        name: Interface or internal datatype name

    Returns:
        DataType

    Raises:
        KeyError: If interface/datatype not found
    """
    # Check internal datatypes first
    if name in self.internal_datatypes:
        return self.internal_datatypes[name]

    # Check interfaces
    return self._find_interface(name).datatype

get_param

get_param(name: str) -> Any

Get kernel parameter value.

Note: Primarily for rare block_tiling parameters. Most parameters are stream_tiling and only available during configure().

Parameters:

Name	Type	Description	Default
name	str	Parameter name	required

Returns:

Type	Description
Any	Parameter value

Raises:

Type	Description
RuntimeError	If no param_getter provided
KeyError	If parameter not found

Source code in brainsmith/dataflow/validation.py

def get_param(self, name: str) -> Any:
    """Get kernel parameter value.

    Note: Primarily for rare block_tiling parameters. Most parameters
    are stream_tiling and only available during configure().

    Args:
        name: Parameter name

    Returns:
        Parameter value

    Raises:
        RuntimeError: If no param_getter provided
        KeyError: If parameter not found
    """
    if self.param_getter is None:
        raise RuntimeError(f"No param_getter available. Cannot get parameter '{name}'.")

    try:
        return self.param_getter(name)
    except (AttributeError, KeyError) as e:
        raise KeyError(f"Parameter '{name}' not found in nodeattrs") from e

get_shape

get_shape(name: str, hierarchy: ShapeHierarchy = ShapeHierarchy.TENSOR) -> tuple[int, ...]

Get shape at hierarchy level.

Parameters:

Name	Type	Description	Default
name	str	Interface name	required
hierarchy	ShapeHierarchy	Which level of hierarchy (TENSOR or BLOCK only)	TENSOR

Returns:

Type	Description
tuple[int, ...]	Shape tuple

Raises:

Type	Description
KeyError	If interface not found
RuntimeError	If STREAM hierarchy requested (not available in design space)

Source code in brainsmith/dataflow/validation.py

def get_shape(
    self, name: str, hierarchy: ShapeHierarchy = ShapeHierarchy.TENSOR
) -> tuple[int, ...]:
    """Get shape at hierarchy level.

    Args:
        name: Interface name
        hierarchy: Which level of hierarchy (TENSOR or BLOCK only)

    Returns:
        Shape tuple

    Raises:
        KeyError: If interface not found
        RuntimeError: If STREAM hierarchy requested (not available in design space)
    """
    if hierarchy == ShapeHierarchy.STREAM:
        raise RuntimeError(
            "Stream shapes not available in design space validation context. "
            "Stream-level constraints are optimization constraints and should be "
            "validated during configure(), not during build()."
        )

    interface = self._find_interface(name)

    if hierarchy == ShapeHierarchy.BLOCK:
        return interface.block_shape
    elif hierarchy == ShapeHierarchy.TENSOR:
        return interface.tensor_shape
    else:
        raise ValueError(f"Unknown hierarchy: {hierarchy}")

is_dynamic

is_dynamic(name: str) -> bool

Check if interface is dynamic (no initializer).

Parameters:

Name	Type	Description	Default
name	str	Interface name	required

Returns:

Type	Description
bool	True if dynamic (activations), False if static (weights)

Source code in brainsmith/dataflow/validation.py

def is_dynamic(self, name: str) -> bool:
    """Check if interface is dynamic (no initializer).

    Args:
        name: Interface name

    Returns:
        True if dynamic (activations), False if static (weights)
    """
    if name in self.inputs:
        return not self.inputs[name].is_weight

    if name in self.outputs:
        return True  # Outputs always dynamic

    # Not found
    raise KeyError(f"Interface '{name}' not found")

---

ConfigurationValidationContext dataclass

ConfigurationValidationContext(configured_model: Any, params: dict[str, int])

Validation context for optimization constraints during configure().

Used during KernelDesignSpace.configure() to validate constraints on stream shapes and parallelization parameters.

Attributes:

Name	Type	Description
configured_model	Any	KernelDesignPoint with configured interfaces
params	dict[str, int]	Parallelization parameters

Example

ctx = ConfigurationValidationContext( configured_model=instance, params=params ) for constraint in parametric_constraints: if error := constraint.check(ctx): raise ValueError(error)

get_datatype

get_datatype(name: str) -> DataType

Get datatype from interface or internal datatypes.

Parameters:

Name	Type	Description	Default
name	str	Interface or internal datatype name	required

Returns:

Type	Description
DataType	DataType

Raises:

Type	Description
KeyError	If interface/datatype not found

Source code in brainsmith/dataflow/validation.py

def get_datatype(self, name: str) -> DataType:
    """Get datatype from interface or internal datatypes.

    Args:
        name: Interface or internal datatype name

    Returns:
        DataType

    Raises:
        KeyError: If interface/datatype not found
    """
    # Check internal datatypes
    if name in self.configured_model.internal_datatypes:
        return self.configured_model.internal_datatypes[name]

    # Check interfaces
    return self._find_interface(name).datatype

get_param

get_param(name: str) -> Any

Get kernel parameter value.

Parameters:

Name	Type	Description	Default
name	str	Parameter name (e.g., "SIMD", "PE", "epsilon")	required

Returns:

Type	Description
Any	Parameter value

Raises:

Type	Description
KeyError	If parameter not found

Source code in brainsmith/dataflow/validation.py

def get_param(self, name: str) -> Any:
    """Get kernel parameter value.

    Args:
        name: Parameter name (e.g., "SIMD", "PE", "epsilon")

    Returns:
        Parameter value

    Raises:
        KeyError: If parameter not found
    """
    if name not in self.params:
        raise KeyError(
            f"Parameter '{name}' not found in configuration. "
            f"Available: {list(self.params.keys())}"
        )
    return self.params[name]

get_shape

get_shape(name: str, hierarchy: ShapeHierarchy = ShapeHierarchy.TENSOR) -> tuple[int, ...]

Get shape at hierarchy level.

All hierarchies available (TENSOR, BLOCK, STREAM).

Parameters:

Name	Type	Description	Default
name	str	Interface name	required
hierarchy	ShapeHierarchy	Which level of hierarchy	TENSOR

Returns:

Type	Description
tuple[int, ...]	Shape tuple

Raises:

Type	Description
KeyError	If interface not found

Source code in brainsmith/dataflow/validation.py

def get_shape(
    self, name: str, hierarchy: ShapeHierarchy = ShapeHierarchy.TENSOR
) -> tuple[int, ...]:
    """Get shape at hierarchy level.

    All hierarchies available (TENSOR, BLOCK, STREAM).

    Args:
        name: Interface name
        hierarchy: Which level of hierarchy

    Returns:
        Shape tuple

    Raises:
        KeyError: If interface not found
    """
    interface = self._find_interface(name)
    return interface.get_shape(hierarchy)

is_dynamic

is_dynamic(name: str) -> bool

Check if interface is dynamic (no initializer).

Parameters:

Name	Type	Description	Default
name	str	Interface name	required

Returns:

Type	Description
bool	True if dynamic (activations), False if static (weights)

Source code in brainsmith/dataflow/validation.py

def is_dynamic(self, name: str) -> bool:
    """Check if interface is dynamic (no initializer).

    Args:
        name: Interface name

    Returns:
        True if dynamic (activations), False if static (weights)
    """
    if name in self.configured_model.inputs:
        return not self.configured_model.inputs[name].is_weight

    if name in self.configured_model.outputs:
        return True  # Outputs always dynamic

    # Not found
    raise KeyError(f"Interface '{name}' not found")

---

TransformationResult dataclass

TransformationResult(nodes_to_insert: list[NodeProto], nodes_to_remove: list[NodeProto], metadata: dict[str, Any] = dict())

Result of ONNX node to hardware kernel transformation.

Attributes:

Name	Type	Description
nodes_to_insert	list[NodeProto]	HW nodes to insert into graph
nodes_to_remove	list[NodeProto]	ONNX nodes to remove from graph
metadata	dict[str, Any]	Optional transformation metadata

Example:

import brainsmith.dataflow as df
from onnx import helper

# Create transformation result when converting ONNX to HW node
hw_node = helper.make_node(
    "LayerNorm",
    inputs=list(node.input),
    outputs=list(node.output),
    domain="brainsmith.kernels",
    name=f"LayerNorm_{node.name}",
)

result = df.TransformationResult(
    nodes_to_insert=[hw_node],
    nodes_to_remove=[node]
)

---

Shape module-attribute

Shape = tuple[int, ...]

Immutable tensor shape (e.g., (1, 784))

---

ShapeHierarchy

Bases: Enum

Shape hierarchy level for constraints and relationships.

Attributes:

Name	Type	Description
STREAM		Stream shape (parallelism, elements per cycle)
BLOCK		Block shape (tiling dimensions)
TENSOR		Tensor shape (full logical dimensions)

---

FULL_DIM module-attribute

FULL_DIM = _FullDimType()

---

FULL_SHAPE module-attribute

FULL_SHAPE = _FullShapeType()

See Also

• Component Registry - Register custom kernels, backends, and steps
• Getting Started - Installation and quickstart
• GitHub - Issues and questions