tensor

BufferizedNDArray

Bases: OverrideTensor

Source code in src/finch/tensor/bufferized_ndarray.py

class BufferizedNDArray(OverrideTensor):
    def __init__(
        self,
        val: NumpyBuffer,
        shape: tuple[np.intp, ...],
        strides: tuple[np.intp, ...],
        fill_value: Any = 0,
        device=None,
    ):
        self.val = val
        self._shape = shape
        pos_t = val.ftype.length_type
        self.strides = tuple(pos_t(stride) for stride in strides)
        elem_t = val.ftype.element_type
        self._fill_value = (
            fill_value
            if isinstance(fill_value, AbstractFill)
            else StaticFill(elem_t(fill_value))
        )

        self._device = normalize_device(device)

    def with_fill(self, fill_value: AbstractFill) -> "BufferizedNDArray":
        return BufferizedNDArray(
            self.val,
            self._shape,
            self.strides,
            fill_value,
            device=self._device,
        )

    def to_numpy(self):
        """
        Convert the bufferized NDArray to a NumPy array.
        This is used to get the underlying NumPy array from the bufferized NDArray.
        """
        return self.val.arr.reshape(self._shape, copy=False)

    def to_scipy(self):
        raise NotImplementedError(f"{type(self).__name__} does not support to_scipy.")

    @classmethod
    def from_numpy(
        cls, arr: np.ndarray, fill_value: Any = 0, device=None
    ) -> "BufferizedNDArray":
        if not arr.flags["C_CONTIGUOUS"]:
            arr = np.ascontiguousarray(arr)
        itemsize = arr.dtype.itemsize
        strides = tuple(np.intp(stride // itemsize) for stride in arr.strides)
        shape = tuple(np.intp(s) for s in arr.shape)
        val = NumpyBuffer(arr.reshape(-1, copy=False))
        fill = as_fill(fill_value)
        coerced = np.asarray(fill.value, dtype=arr.dtype).flat[0]
        return BufferizedNDArray(
            val,
            shape,
            strides,
            DynamicFill(coerced) if is_dynamic(fill) else StaticFill(coerced),
            device=device,
        )

    def __array__(self, dtype=None, copy=None):
        out = self.to_numpy()
        if dtype is not None and out.dtype != dtype:
            if copy is not None and not copy:
                raise ValueError(
                    "Unable to avoid copy while creating an array as requested."
                )
            out = out.astype(dtype)
        return out

    @property
    def ftype(self):
        """
        Returns the ftype of the buffer, which is a BufferizedNDArrayFType.
        """
        return BufferizedNDArrayFType(
            buffer_type=ftype(self.val),
            ndim=self.ndim,
            dimension_type=ftype(self._shape),
            fill_type=self._fill_value.ftype,
            device=self._device,
        )

    @property
    def shape(self) -> tuple[int, ...]:
        return tuple(int(s) for s in self._shape)

    @property
    def ndim(self) -> int:
        return len(self._shape)

    @property
    def fill_value(self) -> Any:
        """Default value to fill the tensor."""
        return self._fill_value.value

    @property
    def fill(self) -> AbstractFill:
        return self._fill_value

    @property
    def device(self):
        return self._device

    def to_device(self, device, /, *, stream=None):
        if stream is not None:
            raise ValueError(f"stream argument is not supported; got {stream!r}")
        device = normalize_device(device)
        if device == self.device:
            return self
        return BufferizedNDArray(
            self.val,
            self._shape,
            self.strides,
            self._fill_value,
            device=device,
        )

    @property
    def element_type(self) -> FType:
        """Data type of the tensor elements."""
        return self.ftype.element_type

    @property
    def shape_type(self) -> tuple:
        """Shape type of the tensor."""
        return self.ftype.shape_type

    def declare(self, init, op, shape):
        """
        Declare a bufferized NDArray with the given initialization value,
        operation, and shape.
        """
        if is_dynamic(self._fill_value):
            # Declaring a tensor sets its dynamic fill to the init.
            init = self.val.ftype.element_type(init)
            self._fill_value = DynamicFill(init, self.val.ftype.element_type)
        elif not bool(ffuncs.same(init, self._fill_value.value)):
            raise ValueError(
                f"Declaring with {init!r} would change the static fill "
                f"{self._fill_value.value!r}"
            )
        for i in range(self.val.length()):
            self.val.store(i, init)
        return self

    def freeze(self, op):
        return self

    def thaw(self, op):
        return self

    @property
    def lvl(self):
        return BufferizedNDArrayLevel(self)

    def access(self, indices, op):
        return BufferizedNDArrayAccessor(self).access(indices, op)

    def item(self):
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to Python scalar.")
        val = self.val.load(0)
        return val.item() if hasattr(val, "item") else val

    def __getitem__(self, index):
        """
        Get an item from the bufferized NDArray.
        This allows for indexing into the bufferized array.
        """
        if isinstance(index, slice | np.ndarray) or (
            isinstance(index, tuple) and any(isinstance(i, slice) for i in index)
        ):
            result = self.to_numpy()[index]
            if isinstance(result, np.ndarray):
                return BufferizedNDArray.from_numpy(
                    result, fill_value=self.fill_value, device=self.device
                )
            return Scalar(result, fill_value=self.fill_value, device=self.device)
        if not isinstance(index, tuple) and self.ndim != 1:
            result = self.to_numpy()[index]
            if isinstance(result, np.ndarray):
                return BufferizedNDArray.from_numpy(
                    result, fill_value=self.fill_value, device=self.device
                )
            return Scalar(result, fill_value=self.fill_value, device=self.device)
        if isinstance(index, tuple):
            index = tuple(i for i in index if i is not Ellipsis)
            if index == () and self.ndim == 0:
                return Scalar(
                    self.val.load(0), fill_value=self.fill_value, device=self.device
                )
            if len(index) < self.ndim:
                return BufferizedNDArray.from_numpy(
                    self.to_numpy()[index],
                    fill_value=self.fill_value,
                    device=self.device,
                )
            index = 0 if index == () else np.dot(index, self.strides)
        return Scalar(
            self.val.load(index), fill_value=self.fill_value, device=self.device
        )

    def __setitem__(self, index, value):
        """
        Set an item in the bufferized NDArray.
        This allows for indexing into the bufferized array.
        """
        if isinstance(index, tuple):
            index = tuple(i % s for i, s in zip(index, self._shape, strict=False))
            index = np.ravel_multi_index(index, self._shape)
        self.val.store(index, value)

    def reshape(self, shape, /, *, copy=None):
        return BufferizedNDArray.from_numpy(
            self.to_numpy().reshape(shape),
            fill_value=self.fill_value,
            device=self.device,
        )

    def __str__(self):
        return f"{self.ftype}(shape={tuple(int(d) for d in self.shape)})"

    __repr__ = __str__

element_type property

Data type of the tensor elements.

fill_value property

Default value to fill the tensor.

ftype property

Returns the ftype of the buffer, which is a BufferizedNDArrayFType.

shape_type property

Shape type of the tensor.

__getitem__(index)

Get an item from the bufferized NDArray. This allows for indexing into the bufferized array.

Source code in src/finch/tensor/bufferized_ndarray.py

def __getitem__(self, index):
    """
    Get an item from the bufferized NDArray.
    This allows for indexing into the bufferized array.
    """
    if isinstance(index, slice | np.ndarray) or (
        isinstance(index, tuple) and any(isinstance(i, slice) for i in index)
    ):
        result = self.to_numpy()[index]
        if isinstance(result, np.ndarray):
            return BufferizedNDArray.from_numpy(
                result, fill_value=self.fill_value, device=self.device
            )
        return Scalar(result, fill_value=self.fill_value, device=self.device)
    if not isinstance(index, tuple) and self.ndim != 1:
        result = self.to_numpy()[index]
        if isinstance(result, np.ndarray):
            return BufferizedNDArray.from_numpy(
                result, fill_value=self.fill_value, device=self.device
            )
        return Scalar(result, fill_value=self.fill_value, device=self.device)
    if isinstance(index, tuple):
        index = tuple(i for i in index if i is not Ellipsis)
        if index == () and self.ndim == 0:
            return Scalar(
                self.val.load(0), fill_value=self.fill_value, device=self.device
            )
        if len(index) < self.ndim:
            return BufferizedNDArray.from_numpy(
                self.to_numpy()[index],
                fill_value=self.fill_value,
                device=self.device,
            )
        index = 0 if index == () else np.dot(index, self.strides)
    return Scalar(
        self.val.load(index), fill_value=self.fill_value, device=self.device
    )

__setitem__(index, value)

Set an item in the bufferized NDArray. This allows for indexing into the bufferized array.

Source code in src/finch/tensor/bufferized_ndarray.py

def __setitem__(self, index, value):
    """
    Set an item in the bufferized NDArray.
    This allows for indexing into the bufferized array.
    """
    if isinstance(index, tuple):
        index = tuple(i % s for i, s in zip(index, self._shape, strict=False))
        index = np.ravel_multi_index(index, self._shape)
    self.val.store(index, value)

declare(init, op, shape)

Declare a bufferized NDArray with the given initialization value, operation, and shape.

Source code in src/finch/tensor/bufferized_ndarray.py

def declare(self, init, op, shape):
    """
    Declare a bufferized NDArray with the given initialization value,
    operation, and shape.
    """
    if is_dynamic(self._fill_value):
        # Declaring a tensor sets its dynamic fill to the init.
        init = self.val.ftype.element_type(init)
        self._fill_value = DynamicFill(init, self.val.ftype.element_type)
    elif not bool(ffuncs.same(init, self._fill_value.value)):
        raise ValueError(
            f"Declaring with {init!r} would change the static fill "
            f"{self._fill_value.value!r}"
        )
    for i in range(self.val.length()):
        self.val.store(i, init)
    return self

to_numpy()

Convert the bufferized NDArray to a NumPy array. This is used to get the underlying NumPy array from the bufferized NDArray.

Source code in src/finch/tensor/bufferized_ndarray.py

def to_numpy(self):
    """
    Convert the bufferized NDArray to a NumPy array.
    This is used to get the underlying NumPy array from the bufferized NDArray.
    """
    return self.val.arr.reshape(self._shape, copy=False)

BufferizedNDArrayFType

Bases: FinchTensorFType, MutableStructFType

A ftype for bufferized NumPy arrays that provides metadata about the array. This includes the fill value, element type, and shape type.

Source code in src/finch/tensor/bufferized_ndarray.py

class BufferizedNDArrayFType(FinchTensorFType, MutableStructFType):
    """
    A ftype for bufferized NumPy arrays that provides metadata about the array.
    This includes the fill value, element type, and shape type.
    """

    @property
    def struct_name(self):
        return "BufferizedNDArray"

    @property
    def struct_fields(self):
        fields = [
            ("val", self.buf_t),
            ("shape", self.shape_t),
            ("strides", self.strides_t),
        ]
        if is_dynamic(self._fill_type):
            # The fill value is bound at call time through a struct field.
            fields.append(("fill", self.element_type))
        return fields

    def struct_getattr(self, obj, attr):
        # A dynamic fill is bound at call time through the "fill" field.
        if attr == "fill":
            return obj.fill_value
        return getattr(obj, attr)

    def struct_setattr(self, obj, attr, value):
        match attr:
            case "shape":
                obj._shape = value
            case "fill":
                obj._fill_value = DynamicFill(value, self.element_type)
            case _:
                setattr(obj, attr, value)

    @property
    def lvl_t(self):
        return BufferizedNDArrayLevelFType(self)

    def get_child_type(self, attr):
        if attr != "lvl":
            raise TypeError(f"{self} does not support child {attr!r}")
        return self.lvl_t

    def get_child(self, obj, attr):
        self.get_child_type(attr)
        return obj

    def from_fields(self, buf, shape, strides, fill=None):
        if is_dynamic(self.fill_type):
            # The fill field holds the value of the dynamic fill.
            fill = DynamicFill(fill, self.element_type)
        return BufferizedNDArray(
            buf, shape, strides, resolve_fill(self.fill_type, fill), device=self.device
        )

    def from_numpy(self, arr: np.ndarray, fill_value=None):
        val = NumpyBuffer(arr.reshape(-1, copy=False))
        strides = _get_default_strides(arr.shape)
        return BufferizedNDArray(
            val=val,
            shape=tuple(
                t(s)
                for s, t in zip(arr.shape, self.shape_t.struct_fieldtypes, strict=True)
            ),
            strides=tuple(
                t(s)
                for (s, t) in zip(
                    strides, self.strides_t.struct_fieldtypes, strict=True
                )
            ),
            fill_value=resolve_fill(
                self.fill_type, None if fill_value is None else as_fill(fill_value)
            ),
            device=self.device,
        )

    def __init__(
        self,
        *,
        buffer_type: NumpyBufferFType,
        ndim: int,
        dimension_type: TupleFType | tuple[FType, ...],
        fill_type: AbstractFillFType | None = None,
        device=None,
    ):
        if not isinstance(dimension_type, TupleFType):
            dimension_type = TupleFType.from_tuple(dimension_type)
        assert isinstance(dimension_type, TupleFType)
        # Normalize dimension field types to Finch ftypes so generated
        # result_type values are consistent with strict asm type checks.
        dimension_type = TupleFType.from_tuple(
            tuple(ftype(t) for t in dimension_type.struct_fieldtypes)
        )
        self.buf_t = buffer_type
        self._ndim = ndim
        self.shape_t = dimension_type
        self.strides_t = TupleFType.from_tuple((self.buf_t.length_type,) * ndim)
        elem_t = self.buf_t.element_type
        # The fill defaults to a static zero, and takes the buffer's element type.
        if fill_type is None:
            fill_type = StaticFillFType(elem_t(0))
        self._fill_type = fill_type.astype(elem_t)
        self._device = normalize_device(device)

    def construct(
        self,
        shape: tuple[int, ...],
        fill_value: Any = None,
    ) -> BufferizedNDArray:
        # The fill value may be raw, which is a static fill.
        fill = resolve_fill(
            self.fill_type, None if fill_value is None else as_fill(fill_value)
        )
        arr = np.empty(shape, dtype=np_dtype(self.element_type))
        arr[...] = fill.value
        return self.from_numpy(arr, fill_value=fill)

    def __call__(
        self,
        val: Any,
    ) -> BufferizedNDArray:
        """
        Convert a tensor to this bufferized ndarray type.

        Args:
            val: A tensor to convert to this type.
        Returns:
            A BufferizedNDArray instance of this type.
        """
        raise NotImplementedError(
            f"Tensor conversion not yet implemented for {type(self).__name__}"
        )

    def __eq__(self, other):
        if not isinstance(other, BufferizedNDArrayFType):
            return False
        return (
            self.buf_t == other.buf_t
            and self.ndim == other.ndim
            and self.shape_t == other.shape_t
            and self.fill_type == other.fill_type
            and self.device == other.device
        )

    def with_fill(self, fill_type: AbstractFillFType) -> "BufferizedNDArrayFType":
        return BufferizedNDArrayFType(
            buffer_type=self.buf_t,
            ndim=self.ndim,
            dimension_type=self.shape_t,
            fill_type=fill_type,
            device=self.device,
        )

    def fisinstance(self, other):
        other_t = ftype(other)
        if is_dynamic(self._fill_type) and isinstance(other_t, BufferizedNDArrayFType):
            # A dynamic-fill ftype accepts any fill value of matching dtype.
            other_t = other_t.with_fill(self._fill_type)
        return other_t == self

    def __hash__(self):
        return hash(
            (
                self.buf_t,
                self.ndim,
                self.shape_t,
                self.fill_type,
                self.device,
            )
        )

    def __str__(self):
        return str(self.struct_name)

    def __repr__(self):
        return (
            f"BufferizedNDArrayFType(buffer_type={repr(self.buf_t)},"
            f" ndim = {self.ndim}, dimension_type ={repr(self.shape_t)},"
            f" fill_type={self.fill_type!r})"
        )

    @property
    def ndim(self) -> int:
        return int(self._ndim)

    @property
    def fill_type(self) -> AbstractFillFType:
        return self._fill_type

    @property
    def device(self):
        return self._device

    @property
    def element_type(self):
        return self.buf_t.element_type

    @property
    def shape_type(self) -> tuple:
        return tuple(self.shape_t.struct_fieldtypes)

    @property
    def level_format_properties(self) -> list[FormatProperty]:
        return [Dense(tuple(range(n + 1))) for n in range(self.ndim)]

    def lower_dim(self, ctx, obj, r):
        return self.lvl_t.level_lower_dim(ctx, ctx(obj.lvl), r)

    def lower_fill(self, ctx, obj):
        return self.lvl_t.lower_fill(ctx(obj.lvl))

    def lower_declare(self, ctx, tns: ntn.Fiber, init, op, shape):
        root = ctx(tns.lvl.root)
        if is_dynamic(self.fill_type):
            # The init becomes the fill of the tensor.
            if init.result_type != self.element_type:
                init = asm.Call(asm.Literal(ffuncs.astype(self.element_type)), (init,))
            ctx.exec(asm.SetAttr(root, asm.Literal("fill"), init))
            init = asm.GetAttr(root, asm.Literal("fill"))
        i_var = asm.Variable("i", self.buf_t.length_type)
        buf = asm.GetAttr(root, asm.Literal("val"))
        body = asm.Store(
            buf,
            i_var,
            init,
        )
        ctx.exec(asm.ForLoop(i_var, asm.Literal(np.intp(0)), asm.Length(buf), body))
        return

    def lower_freeze(self, ctx, tns, op):
        return tns

    def lower_thaw(self, ctx, tns, op):
        return tns

    def unfurl(self, ctx, tns, ext, mode, proto):
        tns = ctx.resolve(tns)
        return self.lvl_t.level_unfurl(ctx, tns, ext, mode, proto, ctx(tns.pos))

    def reshape(self, arr: BufferizedNDArray, new_shape: tuple[np.intp, ...]):
        old_size = functools.reduce(operator.mul, arr.shape, np.intp(1))
        new_size = functools.reduce(operator.mul, new_shape, np.intp(1))
        if old_size != new_size:
            raise ValueError(
                f"Cannot reshape array of size {old_size} into shape {new_shape}"
            )
        new_strides = _get_default_strides(new_shape)
        default_strides = _get_default_strides(arr.shape)
        if arr.strides == default_strides:
            return BufferizedNDArray(
                arr.val,
                new_shape,
                new_strides,
                fill_value=arr.fill_value,
                device=arr.device,
            )
        return BufferizedNDArray.from_numpy(
            arr.to_numpy().reshape(new_shape),
            fill_value=arr.fill_value,
            device=arr.device,
        )

    def lower_unwrap(self, ctx, obj):
        return self.lvl_t.level_lower_unwrap(ctx, obj, ctx(obj.pos))

    def lower_increment(self, ctx, obj, op, val):
        return self.lvl_t.level_lower_increment(ctx, obj, op, val, ctx(obj.pos))

__call__(val)

Convert a tensor to this bufferized ndarray type.

Parameters:

Name	Type	Description	Default
val	Any	A tensor to convert to this type.	required

Returns: A BufferizedNDArray instance of this type.

Source code in src/finch/tensor/bufferized_ndarray.py

def __call__(
    self,
    val: Any,
) -> BufferizedNDArray:
    """
    Convert a tensor to this bufferized ndarray type.

    Args:
        val: A tensor to convert to this type.
    Returns:
        A BufferizedNDArray instance of this type.
    """
    raise NotImplementedError(
        f"Tensor conversion not yet implemented for {type(self).__name__}"
    )

ChunkMaskTensor

Bases: PatternTensor

Map n indices to chunks of size b.

shape is (ceil(n / b), n), and row j holds the indices of chunk j, so that the efficient row-major traversal visits each chunk and then its indices. The last chunk may be shorter.

Source code in src/finch/tensor/patterns.py

class ChunkMaskTensor(PatternTensor):
    """Map ``n`` indices to chunks of size ``b``.

    ``shape`` is ``(ceil(n / b), n)``, and row ``j`` holds the indices of chunk
    ``j``, so that the efficient row-major traversal visits each chunk and then
    its indices. The last chunk may be shorter.
    """

    def __init__(self, shape, *, b: int, dtype=None):
        self._b = operator.index(b)
        if self._b <= 0:
            raise ValueError("b must be positive")
        super().__init__(
            shape,
            dtype=dtype,
            default_dtype=np.bool_,
            fill_value=False,
            pattern_value=True,
            b=self._b,
        )
        chunks, n = map(operator.index, self.shape)
        if n < 0:
            raise ValueError("n must be nonnegative")
        if chunks != (n + self._b - 1) // self._b:
            raise ValueError("shape[0] must equal ceil(shape[1] / b)")

    def contains(self, j, i) -> bool:
        return j == i // self._b

ConstantScalar

Bases: Scalar

A scalar whose value is treated as a compile-time constant. Kernels may specialize on the value (and recompile per distinct value), unlike plain Scalars, whose values are bound at call time.

Source code in src/finch/tensor/scalar.py

class ConstantScalar(Scalar):
    """
    A scalar whose value is treated as a compile-time constant. Kernels may
    specialize on the value (and recompile per distinct value), unlike plain
    `Scalar`s, whose values are bound at call time.
    """

    def __init__(self, val: Any, device=None):
        super().__init__(val, device=device)

    @property
    def ftype(self):
        return ScalarFType(
            ftype(self.val), StaticFillFType(self._fill_value.value), self._device
        )

DenseLevel dataclass

Bases: SingleDimensionLevel

A class representing dense level.

Source code in src/finch/tensor/level/dense_level.py

@dataclass
class DenseLevel(SingleDimensionLevel):
    """
    A class representing dense level.
    """

    lvl: Level
    dimension: np.integer
    stride: np.integer = field(init=False)

    def __post_init__(self):
        # The stride is stored so that kernels can write it back; it must be
        # recomputed whenever the dimensions of the dense levels below change.
        dim_t = ftype(self.dimension)
        match self.lvl:
            case DenseLevel():
                self.stride = dim_t(self.lvl.dimension * self.lvl.stride)
            case _:
                self.stride = dim_t(1)

    @property
    def shape(self) -> tuple:
        return (self.dimension, *self.lvl.shape)

    @property
    def ftype(self) -> DenseLevelFType:
        return DenseLevelFType(self.lvl.ftype, ftype(self.dimension))

    @property
    def val(self) -> Any:
        return self.lvl.val

    def __str__(self):
        return f"DenseLevel(lvl={self.lvl}, dim={self.dimension})"

DenseLevelFType dataclass

Bases: SingleDimensionLevelFType, MutableStructFType

Source code in src/finch/tensor/level/dense_level.py

@dataclass(unsafe_hash=True)
class DenseLevelFType(SingleDimensionLevelFType, MutableStructFType):
    _lvl_t: LevelFType
    dimension_type: ftypes.FDTypeInteger = ftypes.intp

    @property
    def struct_name(self):
        return "DenseLevelFType"

    @property
    def struct_fields(self):
        return [
            ("lvl", self.lvl_t),
            ("dimension", self.dimension_type),
            ("stride", self.dimension_type),
        ]

    def struct_setattr(self, obj, attr, value):
        setattr(obj, attr, value)

    def level_cost(self, fields, stats, stats_factory, num_pos, lvl) -> float:
        n = stats.get_dim_size(fields[lvl])
        return self.lvl_t.level_cost(fields, stats, stats_factory, num_pos * n, lvl + 1)

    def __post_init__(self):
        self.dimension_type = ftype(self.dimension_type)

    def construct(self, shape: tuple[Any, ...], *, pos: int) -> DenseLevel:
        """
        Creates an instance of DenseLevel with the given ftype.

        Args:
            shape: The shape to be used for the level. (mandatory)
        Returns:
            An instance of DenseLevel.
        """
        dimension = self.dimension_type(shape[0])
        lvl = self.lvl_t.construct(
            shape=shape[1:],
            pos=int(pos) * int(dimension),
        )
        return DenseLevel(lvl, dimension)

    def __call__(self, val: Any) -> DenseLevel:
        """
        Convert a level to this dense level type.

        Args:
            val: A value to convert to this type.
        Returns:
            A DenseLevel instance of this type.
        """
        raise NotImplementedError(
            f"Level conversion not yet implemented for {type(self).__name__}"
        )

    def __str__(self):
        return f"DenseLevelFType({self.lvl_t})"

    @property
    def ndim(self):
        return 1 + self.lvl_t.ndim

    @property
    def fill_type(self):
        return self.lvl_t.fill_type

    @property
    def element_type(self):
        """
        Returns the type of elements stored in the fibers.
        """
        return self.lvl_t.element_type

    @property
    def shape_type(self):
        """
        Returns the type of the shape of the fibers.
        """
        return (self.dimension_type, *self.lvl_t.shape_type)

    @property
    def position_type(self):
        """
        Returns the type of positions within the levels.
        """
        return self.lvl_t.position_type

    @property
    def buffer_type(self):
        return self.lvl_t.buffer_type

    @property
    def buffer_factory(self):
        """
        Returns the ftype of the buffer used for the fibers.
        """
        return self.lvl_t.buffer_factory

    @property
    def lvl_t(self):
        return self._lvl_t

    def level_format_properties(self, n):
        return [Dense(tuple(range(n + 1)))] + self.lvl_t.level_format_properties(n + 1)

    def level_lower_dim(self, ctx, obj, r):
        if r == 0:
            return asm.GetAttr(obj, asm.Literal("dimension"))
        return self.lvl_t.level_lower_dim(
            ctx, asm.GetAttr(obj, asm.Literal("lvl")), r - 1
        )

    def level_lower_declare(self, ctx, tns, init, op, shape, pos):
        pos = asm.Call(
            asm.Literal(ffuncs.mul),
            (pos, asm.GetAttr(tns, asm.Literal("dimension"))),
        )
        return self.lvl_t.level_lower_declare(
            ctx, asm.GetAttr(tns, asm.Literal("lvl")), init, op, shape, pos
        )

    def level_lower_freeze(self, ctx, lvl, op, pos):
        pos = asm.Call(
            asm.Literal(ffuncs.mul),
            (pos, asm.GetAttr(lvl, asm.Literal("dimension"))),
        )
        return self.lvl_t.level_lower_freeze(
            ctx, asm.GetAttr(lvl, asm.Literal("lvl")), op, pos
        )

    def level_lower_thaw(self, ctx, lvl, op, pos):
        pos = asm.Call(
            asm.Literal(ffuncs.mul),
            (pos, asm.GetAttr(lvl, asm.Literal("dimension"))),
        )
        return self.lvl_t.level_lower_thaw(
            ctx, asm.GetAttr(lvl, asm.Literal("lvl")), op, pos
        )

    def level_lower_assemble(self, ctx, lvl, start, stop):
        dimension = asm.GetAttr(lvl, asm.Literal("dimension"))
        self.lvl_t.level_lower_assemble(
            ctx,
            asm.GetAttr(lvl, asm.Literal("lvl")),
            asm.Call(asm.Literal(ffuncs.mul), (start, dimension)),
            asm.Call(asm.Literal(ffuncs.mul), (stop, dimension)),
        )

    def level_lower_increment(self, ctx, obj, op, val, pos):
        raise NotImplementedError(
            "DenseLevelFType does not support level_lower_increment."
        )

    def level_lower_unwrap(self, ctx, obj, pos):
        raise NotImplementedError(
            "DenseLevelFType does not support level_lower_unwrap."
        )

    def level_unfurl(
        self,
        ctx: AssemblyContext,
        fiber: ntn.Fiber,
        ext: SymbolicExtent,
        mode,
        proto,
        pos: asm.AssemblyExpression,
    ):
        tns = fiber
        level = tns.lvl
        lvl = ctx(level)

        def child_accessor(ctx: LoopletContext, idx: ntn.Variable):
            if idx.type_ is None:
                raise TypeError(f"Expected loop variable type for {idx.name}")
            pos_2 = asm.Variable(
                ctx.freshen(idx, f"_pos_{self.ndim - 1}"), self.position_type
            )
            child_pos = asm.Call(
                asm.Literal(ffuncs.add),
                (
                    asm.Call(
                        asm.Literal(ffuncs.mul),
                        (pos, asm.GetAttr(lvl, asm.Literal("dimension"))),
                    ),
                    ctx.ctx(idx),
                ),
            )
            if child_pos.result_type != self.position_type:
                child_pos = asm.Call(
                    asm.Literal(ffuncs.astype(self.position_type)), (child_pos,)
                )
            ctx.exec(asm.Assign(pos_2, child_pos))
            return lplt.Run(
                replace(
                    tns,
                    lvl=ntn.Child(level),
                    pos=ntn.Value(pos_2, self.position_type),
                    idxs=(*tns.idxs, idx),
                )
            )

        return lplt.Lookup(child_accessor)

    def from_fields(self, lvl, dimension, stride) -> DenseLevel:
        return DenseLevel(lvl=lvl, dimension=dimension)

buffer_factory property

Returns the ftype of the buffer used for the fibers.

element_type property

Returns the type of elements stored in the fibers.

position_type property

Returns the type of positions within the levels.

shape_type property

Returns the type of the shape of the fibers.

__call__(val)

Convert a level to this dense level type.

Parameters:

Name	Type	Description	Default
val	Any	A value to convert to this type.	required

Returns: A DenseLevel instance of this type.

Source code in src/finch/tensor/level/dense_level.py

def __call__(self, val: Any) -> DenseLevel:
    """
    Convert a level to this dense level type.

    Args:
        val: A value to convert to this type.
    Returns:
        A DenseLevel instance of this type.
    """
    raise NotImplementedError(
        f"Level conversion not yet implemented for {type(self).__name__}"
    )

construct(shape, *, pos)

Creates an instance of DenseLevel with the given ftype.

Parameters:

Name	Type	Description	Default
shape	tuple[Any, ...]	The shape to be used for the level. (mandatory)	required

Returns: An instance of DenseLevel.

Source code in src/finch/tensor/level/dense_level.py

def construct(self, shape: tuple[Any, ...], *, pos: int) -> DenseLevel:
    """
    Creates an instance of DenseLevel with the given ftype.

    Args:
        shape: The shape to be used for the level. (mandatory)
    Returns:
        An instance of DenseLevel.
    """
    dimension = self.dimension_type(shape[0])
    lvl = self.lvl_t.construct(
        shape=shape[1:],
        pos=int(pos) * int(dimension),
    )
    return DenseLevel(lvl, dimension)

ElementLevel dataclass

Bases: Level

A class representing the leaf level of Finch tensors.

Source code in src/finch/tensor/level/element_level.py

@dataclass
class ElementLevel(Level):
    """
    A class representing the leaf level of Finch tensors.
    """

    _format: ElementLevelFType = field(repr=False)
    _val: Any | None = None
    # The fill belongs to the level, so that a dynamic fill may change without
    # changing the format. Only a static fill can default to that of the format.
    _fill: Any = None

    def __post_init__(self):
        if self._val is None:
            self._val = self._format.buffer_type(len=0)
        if self._fill is None:
            fill_type = self._format.fill_type
            assert fill_type is not None
            self._fill = resolve_fill(fill_type).value

    @property
    def shape(self) -> tuple:
        return ()

    @property
    def ftype(self) -> ElementLevelFType:
        return self._format

    def with_fill(self, fill_value: AbstractFill) -> "ElementLevel":
        return replace(
            self,
            _format=self._format.with_fill(fill_value.ftype),
            _fill=fill_value.value,
        )

    @property
    def val(self) -> Any:
        return self._val

    @property
    def fill_value(self) -> Any:
        return self._fill

    def __str__(self):
        return f"ElementLevel(val={self._val})"

ElementLevelFType dataclass

Bases: LevelFType, MutableStructFType

Source code in src/finch/tensor/level/element_level.py

@dataclass(unsafe_hash=True)
class ElementLevelFType(LevelFType, MutableStructFType):
    fill_type: AbstractFillFType | None = None
    element_type: FType | None = None
    position_type: FType | None = None
    buffer_factory: Any = NumpyBufferFType
    buffer_type: Any = None

    @property
    def struct_name(self):
        return "ElementLevelFType"

    @property
    def struct_fields(self):
        fields = [
            ("val", self.buffer_type),
        ]
        if is_dynamic(self.fill_type):
            # The fill value is bound at call time through a struct field.
            fields.append(("fill", self.element_type))
        return fields

    def struct_getattr(self, obj, attr):
        # A dynamic fill is bound at call time through the "fill" field.
        if attr == "fill":
            return obj.fill_value
        return getattr(obj, attr)

    def struct_setattr(self, obj, attr, value):
        match attr:
            case "val":
                obj._val = value
            case "fill":
                obj._fill = value

    def level_cost(self, fields, stats, stats_factory, num_pos, lvl) -> float:
        # no inner level
        # cost = num_pos * bytes per value
        assert self.element_type is not None
        val_size = np_dtype(self.element_type).itemsize
        return num_pos * val_size

    def __post_init__(self):
        # Ensure element_type is an FType
        if self.element_type is None:
            assert self.fill_type is not None, (
                "Must provide either element_type or fill_type."
            )
            self.element_type = self.fill_type.element_type
        assert isinstance(self.element_type, FType), (
            "element_type must be an instance of FType"
        )
        if self.buffer_type is None:
            self.buffer_type = self.buffer_factory(self.element_type)
        if self.position_type is None:
            self.position_type = ftype(np.intp)
        self.position_type = ftype(self.position_type)
        self.element_type = self.buffer_type.element_type
        # The fill defaults to a static zero, and takes the buffer's element type.
        fill_type = self.fill_type
        if fill_type is None:
            fill_type = StaticFillFType(self.element_type(0))
        self.fill_type = fill_type.astype(self.element_type)

    def construct(self, shape: tuple[Any, ...], *, pos: int) -> "ElementLevel":
        """
        Creates an instance of ElementLevel with the given ftype.

        Args:
            shape: Should be always `()`, used for validation.
        Returns:
            An instance of ElementLevel.
        """
        if len(shape) != 0:
            raise ValueError("ElementLevelFType must be called with an empty shape.")
        val = self.buffer_type(len=pos)
        return ElementLevel(self, val)

    def __call__(self, val: Any) -> "ElementLevel":
        """
        Convert a level to this element level type.

        Args:
            val: A value to convert to this type.
        Returns:
            An ElementLevel instance of this type.
        """
        raise NotImplementedError(
            f"Level conversion not yet implemented for {type(self).__name__}"
        )

    def __str__(self):
        match self.fill_type:
            case StaticFillFType() as fill_type:
                return f"ElementLevelFType(fv={fill_type.value})"
            case fill_type:
                return f"ElementLevelFType(fv={fill_type})"

    @property
    def ndim(self):
        return 0

    @property
    def shape_type(self):
        return ()

    @property
    def lvl_t(self):
        raise Exception("ElementLevelFType is the leaf level.")

    def level_format_properties(self, n):
        return []

    def with_fill(self, fill_type: Any) -> "ElementLevelFType":
        return ElementLevelFType(
            fill_type=fill_type,
            element_type=self.element_type,
            position_type=self.position_type,
            buffer_factory=self.buffer_factory,
            buffer_type=self.buffer_type,
        )

    def lower_fill(self, lvl_expr):
        return asm.GetAttr(lvl_expr, asm.Literal("fill"))

    def from_fields(self, val=None, fill=None) -> "ElementLevel":
        # Wrap numpy arrays in NumpyBuffer and flatten, similar to BufferizedNDArray
        if val is not None and isinstance(val, np.ndarray):
            from finch.codegen import NumpyBuffer

            val = NumpyBuffer(np.asarray(val).reshape(-1, copy=False))
        return ElementLevel(_format=self, _val=val, _fill=fill)

    def level_lower_declare(self, ctx, tns, init, op, shape, pos):
        if is_dynamic(self.fill_type):
            # The init becomes the fill of the tensor.
            elem_t = self.fill_type.element_type
            if init.result_type != elem_t:
                init = asm.Call(asm.Literal(ffuncs.astype(elem_t)), (init,))
            ctx.exec(asm.SetAttr(tns, asm.Literal("fill"), init))
            init = asm.GetAttr(tns, asm.Literal("fill"))
        buf = asm.GetAttr(tns, asm.Literal("val"))
        i_var = asm.Variable("i", self.buffer_type.length_type)
        body = asm.Store(buf, i_var, init)
        ctx.exec(asm.ForLoop(i_var, asm.Literal(np.intp(0)), asm.Length(buf), body))

    def level_lower_unwrap(self, ctx, obj, pos):
        buf = asm.GetAttr(ctx(obj.lvl), asm.Literal("val"))
        if pos.result_type != self.buffer_type.length_type:
            pos = asm.Call(
                asm.Literal(ffuncs.astype(self.buffer_type.length_type)), (pos,)
            )
        return asm.Load(buf, pos)

    def level_lower_increment(
        self,
        ctx: AssemblyContext,
        obj,
        op: ntn.Literal,
        val: ntn.NotationExpression,
        pos: asm.AssemblyExpression,
    ):
        buf = asm.GetAttr(ctx(obj.lvl), asm.Literal("val"))
        if pos.result_type != self.buffer_type.length_type:
            pos = asm.Call(
                asm.Literal(ffuncs.astype(self.buffer_type.length_type)), (pos,)
            )
        pos_e, op_e, val_e = pos, ctx(op), ctx(val)
        ctx.exec(
            asm.Store(
                buf,
                pos_e,
                asm.Call(op_e, (asm.Load(buf, pos_e), val_e)),
            )
        )
        match obj:
            case ntn.HollowFiber(dirty=dirty):
                fill = (
                    asm.Literal(self.fill_type.value)
                    if is_static(self.fill_type)
                    else self.lower_fill(ctx(obj.lvl))
                )
                ctx.exec(
                    asm.If(
                        asm.Call(
                            asm.Literal(ffuncs.not_),
                            (
                                asm.Call(
                                    asm.Literal(ffuncs.same),
                                    (asm.Load(buf, pos_e), fill),
                                ),
                            ),
                        ),
                        asm.Assign(ctx(dirty), asm.Literal(True)),
                    )
                )

    def level_lower_assemble(self, ctx, lvl, start, stop):
        buf = asm.GetAttr(lvl, asm.Literal("val"))
        p_t = self.buffer_type.length_type
        to_size = asm.Literal(ffuncs.astype(p_t))
        p = asm.Variable(ctx.freshen("p"), p_t)
        p_start = asm.Variable(ctx.freshen("p_start"), p_t)
        p_stop = asm.Variable(ctx.freshen("p_stop"), p_t)
        length = asm.Length(buf)
        fill = (
            asm.Literal(self.fill_type.value)
            if is_static(self.fill_type)
            else self.lower_fill(lvl)
        )
        expr = """finch
        p_start = to_size(start)
        p_stop = to_size(stop)
        if (length < p_stop)
            resize(buf, p_stop)
        end
        for (p in p_start:p_stop)
            buf[p] = fill
        end
        """
        ctx.exec(parse_assembly(expr, locals(), position_type=p_t))

    def level_lower_freeze(self, ctx, lvl, op, pos):
        buf = asm.GetAttr(lvl, asm.Literal("val"))
        size = asm.Call(
            asm.Literal(ffuncs.astype(self.buffer_type.length_type)), (pos,)
        )
        length = asm.Length(buf)
        expr = """finch
        if (length != size)
            resize(buf, size)
        end
        """
        ctx.exec(parse_assembly(expr, locals()))

    def level_lower_thaw(self, ctx, lvl, op, pos):
        pass

    def level_lower_dim(self, ctx, obj, r):
        raise NotImplementedError("ElementLevelFType does not support level_lower_dim.")

    def level_unfurl(self, ctx, lvl, ext, mode, proto, pos):
        raise NotImplementedError("ElementLevelFType does not support level_unfurl.")

__call__(val)

Convert a level to this element level type.

Parameters:

Name	Type	Description	Default
val	Any	A value to convert to this type.	required

Returns: An ElementLevel instance of this type.

Source code in src/finch/tensor/level/element_level.py

def __call__(self, val: Any) -> "ElementLevel":
    """
    Convert a level to this element level type.

    Args:
        val: A value to convert to this type.
    Returns:
        An ElementLevel instance of this type.
    """
    raise NotImplementedError(
        f"Level conversion not yet implemented for {type(self).__name__}"
    )

construct(shape, *, pos)

Creates an instance of ElementLevel with the given ftype.

Parameters:

Name	Type	Description	Default
shape	tuple[Any, ...]	Should be always (), used for validation.	required

Returns: An instance of ElementLevel.

Source code in src/finch/tensor/level/element_level.py

def construct(self, shape: tuple[Any, ...], *, pos: int) -> "ElementLevel":
    """
    Creates an instance of ElementLevel with the given ftype.

    Args:
        shape: Should be always `()`, used for validation.
    Returns:
        An instance of ElementLevel.
    """
    if len(shape) != 0:
        raise ValueError("ElementLevelFType must be called with an empty shape.")
    val = self.buffer_type(len=pos)
    return ElementLevel(self, val)

FiberTensor dataclass

Bases: OverrideTensor

A class representing a tensor with fiber structure.

Attributes:

Name	Type	Description
lvl	Level	a fiber allocator that manages the fibers in the tensor.

Source code in src/finch/tensor/fiber_tensor.py

@dataclass
class FiberTensor(OverrideTensor):
    """
    A class representing a tensor with fiber structure.

    Attributes:
        lvl: a fiber allocator that manages the fibers in the tensor.
    """

    lvl: Level
    pos: np.integer = np.intp(0)
    _device: Any = None

    def __post_init__(self):
        self._device = normalize_device(self._device)

    def __repr__(self):
        return f"FiberTensor(lvl={self.lvl})"

    @property
    def ftype(self):
        """
        Returns the ftype of the fiber tensor, which is a FiberTensorFType.
        """
        return FiberTensorFType(self.lvl.ftype, self._device)

    def with_fill(self, fill_value: AbstractFill) -> FiberTensor:
        return replace(self, lvl=self.lvl.with_fill(fill_value))

    @property
    def shape(self):
        return self.lvl.shape

    @property
    def val(self):
        return self.lvl.val

    @property
    def ndim(self):
        return self.lvl.ndim

    @property
    def shape_type(self):
        return self.lvl.shape_type

    @property
    def element_type(self):
        return self.lvl.element_type

    @property
    def fill_value(self):
        return self.lvl.fill_value

    @property
    def device(self):
        return self._device

    def to_device(self, device, /, *, stream=None):
        if stream is not None:
            raise ValueError(f"stream argument is not supported; got {stream!r}")
        device = normalize_device(device)
        if device == self.device:
            return self
        return FiberTensor(self.lvl, self.pos, device)

    @property
    def position_type(self):
        return self.lvl.position_type

    @property
    def buffer_factory(self):
        return self.lvl.buffer_factory

    def to_numpy(self) -> np.ndarray:
        # TODO: temporary for dense only. TBD in sparse_level PR
        return np.reshape(self.lvl.val.arr, self.shape, copy=False)

    def to_scipy(self):
        from .level import (
            DenseLevel,
            ElementLevel,
            SparseCOOLevel,
            SparseListLevel,
        )

        if self.ndim != 2:
            raise ValueError("SciPy sparse arrays must be two-dimensional.")

        if self.fill_value != 0:
            raise ValueError("SciPy CSR conversion requires a zero fill value.")

        assert self.pos == 0

        match self.lvl:
            case SparseCOOLevel(
                lvl=ElementLevel() as element,
                tbl=(row, col),
            ):
                return sps.coo_array(
                    (element.val.arr, (row.arr, col.arr)),
                    shape=self.shape,
                    copy=False,
                )

            case DenseLevel(
                lvl=SparseListLevel(
                    lvl=ElementLevel() as element,
                    ptr=ptr,
                    idx=idx,
                )
            ):
                return sps.csr_array(
                    (element.val.arr, idx.arr, ptr.arr),
                    shape=self.shape,
                    copy=False,
                )

            case _:
                raise NotImplementedError(
                    f"Finch format {self.ftype} is not supported by SciPy conversion."
                )

    def item(self):
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to Python scalar.")
        return self.to_numpy().item()

    @classmethod
    def from_scipy_csr(
        cls,
        obj: Any,
        /,
        *,
        dtype=None,
        device=None,
        copy=None,
    ):
        from .level import (
            DenseLevel,
            ElementLevel,
            SparseListLevel,
            dense,
            element,
            sparse_list,
        )

        if dtype is not None:
            if copy is False:
                obj = obj.astype(dtype, copy=False)
            else:
                obj = obj.astype(dtype, copy=True)
        elif copy is True:
            obj = obj.copy()

        if not obj.has_canonical_format:
            if copy is False:
                raise ValueError(
                    "Unable to avoid copy while creating an array as requested."
                )
            obj = obj.copy()
            obj.sum_duplicates()

        data = obj.data
        indices = obj.indices
        indptr = obj.indptr

        element_lvl = element(
            fill_type=np.zeros((), dtype=data.dtype)[()],
            element_type=ftype(data.dtype),
            position_type=ftype(indices.dtype),
        )
        sparse_lvl = sparse_list(element_lvl, dimension_type=ftype(indices.dtype))
        dense_lvl = dense(sparse_lvl, dimension_type=ftype(indices.dtype))

        return cls(
            DenseLevel(
                SparseListLevel(
                    ElementLevel(
                        _format=element_lvl,
                        _val=NumpyBuffer(data),
                    ),
                    dimension=sparse_lvl.dimension_type(obj.shape[1]),
                    ptr=NumpyBuffer(indptr),
                    idx=NumpyBuffer(indices),
                ),
                dimension=dense_lvl.dimension_type(obj.shape[0]),
            ),
            _device=device,
        )

    @classmethod
    def from_scipy_coo(
        cls,
        obj: Any,
        /,
        *,
        dtype=None,
        device=None,
        copy=None,
    ):
        from .level import (
            ElementLevel,
            SparseCOOLevel,
            element,
        )

        if dtype is not None:
            if copy is False:
                obj = obj.astype(dtype, copy=False)
            else:
                obj = obj.astype(dtype, copy=True)
        elif copy is True:
            obj = obj.copy()

        if not obj.has_canonical_format:
            if copy is False:
                raise ValueError(
                    "Unable to avoid copy while creating an array as requested."
                )
            obj = obj.copy()
            obj.sum_duplicates()

        data = obj.data
        row = obj.row
        col = obj.col
        index_type = ftype(row.dtype)

        element_lvl = element(
            fill_type=np.zeros((), dtype=data.dtype)[()],
            element_type=ftype(data.dtype),
            position_type=index_type,
        )

        return cls(
            SparseCOOLevel(
                ElementLevel(
                    _format=element_lvl,
                    _val=NumpyBuffer(data),
                ),
                shape=tuple(index_type(d) for d in obj.shape),
                tbl=(NumpyBuffer(row), NumpyBuffer(col)),
            ),
            _device=device,
        )

ftype property

Returns the ftype of the fiber tensor, which is a FiberTensorFType.

FiberTensorFType dataclass

Bases: FinchTensorFType, ImmutableStructFType

An abstract base class representing the ftype of a fiber tensor.

Attributes:

Name	Type	Description
lvl_t	LevelFType	The level ftype to be used for the tensor.

Source code in src/finch/tensor/fiber_tensor.py

@dataclass(unsafe_hash=True)
class FiberTensorFType(FinchTensorFType, ImmutableStructFType):
    """
    An abstract base class representing the ftype of a fiber tensor.

    Attributes:
        lvl_t: The level ftype to be used for the tensor.
    """

    lvl_t: LevelFType
    _device: Any = None

    def __post_init__(self):
        self._device = normalize_device(self._device)

    @property
    def struct_name(self):
        # TODO: include dt = np.dtype(self.buf_t.element_type)
        return "FiberTensorFType"

    @property
    def struct_fields(self):
        return [
            ("lvl", self.lvl_t),
            ("shape", TupleFType.from_tuple(self.shape_type)),
            ("pos", self.position_type),
        ]

    def get_child_type(self, attr):
        if attr != "lvl":
            raise TypeError(f"{self} does not support child {attr!r}")
        return self.lvl_t

    def get_child(self, obj, attr):
        self.get_child_type(attr)
        return asm.GetAttr(obj, asm.Literal(attr))

    def with_fill(self, fill_type: AbstractFillFType) -> FiberTensorFType:
        """Rebuild this ftype with the leaf fill replaced."""
        return FiberTensorFType(self.lvl_t.with_fill(fill_type), self._device)

    def fisinstance(self, other):
        other_t = ftype(other)
        if is_dynamic(self.fill_type) and isinstance(other_t, FiberTensorFType):
            # A dynamic-fill ftype accepts any fill value of matching dtype.
            other_t = other_t.with_fill(self.fill_type)
        return other_t == self

    def construct(self, shape: tuple[int, ...], fill_value: Any = None) -> FiberTensor:
        """
        Creates an instance of a FiberTensor with the given arguments.
        """
        # The fill value may be raw, which is a static fill.
        fill = resolve_fill(
            self.fill_type, None if fill_value is None else as_fill(fill_value)
        )
        # The levels are built with a static fill, which the tensor then
        # replaces with its own.
        fmt = self.with_fill(StaticFillFType(fill.value))
        tns = FiberTensor(
            fmt.lvl_t.construct(shape=shape, pos=1),
            fmt.position_type(0),
            _device=fmt.device,
        )
        return tns.with_fill(fill)

    def __call__(self, val: Any) -> FiberTensor:
        """
        Convert a tensor to this fiber tensor type.

        Args:
            val: A tensor to convert to this type.
        Returns:
            A FiberTensor instance of this type.
        """
        raise NotImplementedError(
            f"Tensor conversion not yet implemented for {type(self).__name__}"
        )

    def __str__(self):
        return f"FiberTensorFType({self.lvl_t})"

    @property
    def ndim(self):
        return self.lvl_t.ndim

    @property
    def shape_type(self):
        return self.lvl_t.shape_type

    @property
    def element_type(self):
        return self.lvl_t.element_type

    @property
    def fill_type(self):
        return self.lvl_t.fill_type

    @property
    def device(self):
        return self._device

    @property
    def position_type(self):
        return self.lvl_t.position_type

    @property
    def buffer_factory(self):
        return self.lvl_t.buffer_factory

    @property
    def buffer_type(self):
        return self.lvl_t.buffer_type

    @property
    def level_format_properties(self) -> list[FormatProperty]:
        return self.lvl_t.level_format_properties(0)

    def unfurl(self, ctx, tns, ext, mode, proto):
        tns = ctx.resolve(tns)
        return self.lvl_t.level_unfurl(ctx, tns, ext, mode, proto, ctx(tns.pos))

    def lower_freeze(self, ctx, tns, op):
        return self.lvl_t.level_lower_freeze(
            ctx, ctx(tns.lvl), op, asm.Literal(self.position_type(1))
        )

    def lower_thaw(self, ctx, tns, op):
        return self.lvl_t.level_lower_thaw(
            ctx, ctx(tns.lvl), op, asm.Literal(self.position_type(1))
        )

    def lower_unwrap(self, ctx, obj):
        return self.lvl_t.level_lower_unwrap(ctx, obj, ctx(obj.pos))

    def lower_increment(self, ctx, obj, op, val):
        return self.lvl_t.level_lower_increment(ctx, obj, op, val, ctx(obj.pos))

    def lower_declare(self, ctx, tns, init, op, shape):
        return self.lvl_t.level_lower_declare(
            ctx, ctx(tns.lvl), init, op, shape, asm.Literal(self.position_type(1))
        )

    def lower_dim(self, ctx, obj, r):
        return self.lvl_t.level_lower_dim(ctx, ctx(obj.lvl), r)

    def lower_fill(self, ctx, obj):
        if is_static(self.fill_type):
            return asm.Literal(self.fill_type.value)
        return self.lvl_t.lower_fill(ctx(obj.lvl))

    def from_fields(self, *args) -> FiberTensor:
        lvl, shape, pos = args
        return FiberTensor(lvl, pos, self.device)

    def from_numpy(self, arr: np.ndarray, fill_value: Any = None) -> FiberTensor:
        # The fill value may be raw, which is a static fill.
        fill = resolve_fill(
            self.fill_type, None if fill_value is None else as_fill(fill_value)
        )

        def build_level(lvl_t, shape):
            match lvl_t:
                case levels.DenseLevelFType():
                    if not shape:
                        raise ValueError("Array rank does not match the fiber format")
                    return levels.DenseLevel(
                        build_level(lvl_t.lvl_t, shape[1:]),
                        lvl_t.dimension_type(shape[0]),
                    )
                case levels.ElementLevelFType():
                    if shape:
                        raise ValueError("Array rank does not match the fiber format")
                    return lvl_t.with_fill(fill.ftype).from_fields(arr, fill.value)
                case _:
                    raise NotImplementedError(
                        f"NumPy conversion does not support {type(lvl_t).__name__}"
                    )

        return FiberTensor(
            build_level(self.lvl_t, arr.shape),
            pos=self.position_type(0),
            _device=self.device,
        )

__call__(val)

Convert a tensor to this fiber tensor type.

Parameters:

Name	Type	Description	Default
val	Any	A tensor to convert to this type.	required

Returns: A FiberTensor instance of this type.

Source code in src/finch/tensor/fiber_tensor.py

def __call__(self, val: Any) -> FiberTensor:
    """
    Convert a tensor to this fiber tensor type.

    Args:
        val: A tensor to convert to this type.
    Returns:
        A FiberTensor instance of this type.
    """
    raise NotImplementedError(
        f"Tensor conversion not yet implemented for {type(self).__name__}"
    )

construct(shape, fill_value=None)

Creates an instance of a FiberTensor with the given arguments.

Source code in src/finch/tensor/fiber_tensor.py

def construct(self, shape: tuple[int, ...], fill_value: Any = None) -> FiberTensor:
    """
    Creates an instance of a FiberTensor with the given arguments.
    """
    # The fill value may be raw, which is a static fill.
    fill = resolve_fill(
        self.fill_type, None if fill_value is None else as_fill(fill_value)
    )
    # The levels are built with a static fill, which the tensor then
    # replaces with its own.
    fmt = self.with_fill(StaticFillFType(fill.value))
    tns = FiberTensor(
        fmt.lvl_t.construct(shape=shape, pos=1),
        fmt.position_type(0),
        _device=fmt.device,
    )
    return tns.with_fill(fill)

with_fill(fill_type)

Rebuild this ftype with the leaf fill replaced.

Source code in src/finch/tensor/fiber_tensor.py

def with_fill(self, fill_type: AbstractFillFType) -> FiberTensorFType:
    """Rebuild this ftype with the leaf fill replaced."""
    return FiberTensorFType(self.lvl_t.with_fill(fill_type), self._device)

FillTensor

Bases: OverrideTensor

A tensor that has a specific shape but contains no actual data. Used primarily for broadcasting operations where a tensor of a specific shape is needed but the values are irrelevant.

Source code in src/finch/tensor/patterns.py

class FillTensor(OverrideTensor):
    """
    A tensor that has a specific shape but contains no actual data.
    Used primarily for broadcasting operations where a tensor of a specific
    shape is needed but the values are irrelevant.
    """

    def with_fill(self, fill_value: AbstractFill) -> FillTensor:
        return FillTensor(self._shape, fill_value)

    def __init__(self, shape, fill_value):
        self._shape = shape
        self._fill = as_fill(fill_value)

    def __getitem__(self, idxs):
        return Scalar(self._fill.value, fill_value=self._fill)

    def item(self):
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to Python scalar.")
        return self._fill.value

    def to_numpy(self):
        raise NotImplementedError(f"{type(self).__name__} does not support to_numpy.")

    def to_scipy(self):
        raise NotImplementedError(f"{type(self).__name__} does not support to_scipy.")

    @property
    def shape(self):
        return self._shape

    @property
    def fill_value(self) -> Any:
        """Default fill value."""
        return self._fill.value

    @property
    def element_type(self) -> FType:
        """Data type of the tensor's elements."""
        return self.ftype.element_type

    @property
    def shape_type(self) -> tuple[FType, ...]:
        """Shape type of the tensor."""
        return self.ftype.shape_type

    @property
    def ftype(self):
        return FillTensorFType(
            self._fill.ftype,
            self._fill.element_type,
            tuple(ftype(dim) for dim in self.shape),
        )

element_type property

Data type of the tensor's elements.

fill_value property

Default fill value.

shape_type property

Shape type of the tensor.

IndexTensor

Bases: OverrideTensor

A tensor that has a specific shape and returns the linear (flattened) index used to access it.

Source code in src/finch/tensor/patterns.py

class IndexTensor(OverrideTensor):
    """
    A tensor that has a specific shape and returns the linear (flattened) index used
    to access it.
    """

    def __init__(
        self,
        shape,
        element_type: FType | type = np.intp,
        fill_value: AbstractFill | None = None,
    ):
        self._shape = tuple(shape)
        self._element_type = ftype(element_type)
        self._fill = fill_value

    def with_fill(self, fill_value: AbstractFill) -> IndexTensor:
        return IndexTensor(self._shape, self._element_type, fill_value)

    def __getitem__(self, idxs):
        if self.ndim == 0 and idxs in ((), Ellipsis, (...,)):
            return Scalar(self.fill_value, fill_value=self.fill_value)
        if not isinstance(idxs, tuple):
            idxs = (idxs,)
        idxs = tuple(idx for idx in idxs if idx is not Ellipsis)
        if len(idxs) != self.ndim:
            raise IndexError("Incorrect number of indices for IndexTensor.")
        flat_index = 0
        for idx, dim in zip(idxs, self.shape, strict=True):
            flat_index = flat_index * dim + idx
        flat_index = self._element_type(flat_index)
        return Scalar(flat_index, fill_value=self.fill_value)

    def item(self):
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to Python scalar.")
        return self.fill_value

    def to_numpy(self):
        raise NotImplementedError(f"{type(self).__name__} does not support to_numpy.")

    def to_scipy(self):
        raise NotImplementedError(f"{type(self).__name__} does not support to_scipy.")

    @property
    def shape(self):
        return self._shape

    @property
    def fill_value(self) -> Any:
        """Default fill value."""
        return resolve_fill(self.ftype.fill_type, self._fill).value

    @property
    def element_type(self) -> FType:
        """Data type of the tensor's elements."""
        return self.ftype.element_type

    @property
    def shape_type(self) -> tuple[FType, ...]:
        """Shape type of the tensor."""
        return self.ftype.shape_type

    @property
    def ftype(self):
        return IndexTensorFType(
            self._element_type,
            tuple(ftype(dim) for dim in self.shape),
            None if self._fill is None else self._fill.ftype,
        )

element_type property

Data type of the tensor's elements.

fill_value property

Default fill value.

shape_type property

Shape type of the tensor.

Level dataclass

Bases: FTyped[LT], ABC

An abstract base class representing a fiber allocator that manages fibers in a tensor.

Source code in src/finch/tensor/level/level.py

@dataclass
class Level(FTyped[LT], ABC):
    """
    An abstract base class representing a fiber allocator that manages fibers in
    a tensor.
    """

    @property
    @abstractmethod
    def shape(self) -> tuple:
        """
        Shape of the fibers in the structure.
        """
        ...

    @property
    @abstractmethod
    def val(self) -> Any: ...

    @property
    def ndim(self):
        return self.ftype.ndim

    @property
    def fill_value(self):
        """The fill value itself; see `self.ftype.fill_type` for
        dynamic vs static. A level with a dynamic fill holds its own value."""
        fill_type = self.ftype.fill_type
        if is_static(fill_type):
            return fill_type.value
        raise NotImplementedError(
            f"{type(self).__name__} must hold its dynamic fill value"
        )

    @property
    def element_type(self):
        return self.ftype.element_type

    @property
    def shape_type(self):
        return self.ftype.shape_type

    @property
    def position_type(self):
        return self.ftype.position_type

    @property
    def buffer_factory(self):
        return self.ftype.buffer_factory

    @property
    def buffer_type(self):
        return self.ftype.buffer_type

    @abstractmethod
    def with_fill(self, fill_value: AbstractFill) -> Self: ...

fill_value property

The fill value itself; see self.ftype.fill_type for dynamic vs static. A level with a dynamic fill holds its own value.

shape abstractmethod property

Shape of the fibers in the structure.

LevelFType dataclass

Bases: FType, ABC

An abstract base class representing the ftype of levels.

Source code in src/finch/tensor/level/level.py

@dataclass
class LevelFType(FType, ABC):
    """
    An abstract base class representing the ftype of levels.
    """

    def level_get_child_type(self, attr: str) -> LevelFType:
        match self:
            case StructFType() if self.struct_hasattr(attr):
                child = self.struct_attrtype(attr)
                match child:
                    case LevelFType():
                        return child
        raise TypeError(f"{self} does not support child {attr!r}")

    def level_get_child(
        self, obj: asm.AssemblyExpression, attr: str
    ) -> asm.AssemblyExpression:
        self.level_get_child_type(attr)
        return asm.GetAttr(obj, asm.Literal(attr))

    def with_fill(self, fill_type: AbstractFillFType) -> LevelFType:
        """Rebuild this level ftype with the leaf fill replaced."""
        # Level ftypes are dataclasses with a `_lvl_t` child field; non-
        # dataclass levels must override.
        return replace(self, _lvl_t=self.lvl_t.with_fill(fill_type))  # type: ignore[type-var]

    def lower_fill(self, lvl_expr):
        """Assembly expression reading the runtime fill from the leaf level's
        fill field, descending from this level's struct expression."""
        return self.lvl_t.lower_fill(asm.GetAttr(lvl_expr, asm.Literal("lvl")))

    @property
    @abstractmethod
    def ndim(self):
        """
        Number of dimensions of the fibers in the structure.
        """
        ...

    @property
    @abstractmethod
    def fill_type(self) -> AbstractFillFType:
        """
        Fill of the fibers, and whether kernels may specialize on its value.
        """
        ...

    @property
    @abstractmethod
    def element_type(self) -> FType:
        """
        Type of elements stored in the fibers.
        """
        ...

    @property
    @abstractmethod
    def shape_type(self):
        """
        Tuple of types of the dimensions in the shape.
        """
        ...

    @property
    @abstractmethod
    def position_type(self):
        """
        Type of positions within the levels.
        """
        ...

    @property
    @abstractmethod
    def buffer_factory(self):
        """
        Function to create default buffers for the fibers.
        """
        ...

    @property
    @abstractmethod
    def buffer_type(self): ...

    @property
    @abstractmethod
    def lvl_t(self):
        """
        Get the nested level.
        """
        ...

    @abstractmethod
    def level_cost(self, fields, stats, stats_factory, num_pos, lvl):
        """
        Defines cost for a level
        """
        ...

    @abstractmethod
    def level_unfurl(self, ctx, lvl, ext, mode, proto, pos):
        """
        Emit code to unfurl the fiber at position `pos` in the level.
        """
        ...

    @abstractmethod
    def level_lower_freeze(self, ctx, lvl, op, pos):
        """
        Emit code to freeze `pos` previously assembled positions in the level.
        """
        ...

    def level_lower_assemble(self, ctx, lvl, start, stop):
        """Assemble parent positions in the half-open range [start, stop)."""
        raise NotImplementedError(
            f"Assembly is not implemented for {type(self).__name__}"
        )

    @abstractmethod
    def level_lower_thaw(self, ctx, lvl, op, pos):
        """
        Emit code to thaw `pos` previously assembled positions in the level.
        """
        ...

    @abstractmethod
    def level_lower_unwrap(self, ctx, obj, pos):
        """
        Emit code to return the unwrapped scalar at position `pos` in the level.
        """
        ...

    @abstractmethod
    def level_lower_increment(self, ctx, obj, op, val, pos):
        """
        Emit code to increment position `pos` in the level.
        """
        ...

    @abstractmethod
    def level_lower_declare(self, ctx, tns, init, op, shape, pos):
        """
        Emit code to lower a declare of `pos` previously assembled positions in
        the level.
        """
        ...

    @abstractmethod
    def level_lower_dim(self, ctx, obj, r):
        """
        Emit code to return the size of dimension `r` of the subtensors in the level.
        """
        ...

    @abstractmethod
    def construct(self, shape: tuple[Any, ...], *, pos: int) -> Level:
        """
        Construct a level instance with the given shape.
        """
        ...

    @abstractmethod
    def level_format_properties(self, n: int) -> list[FormatProperty]:
        """
        Return the format properties contributed by this level type and children.

        ``n`` is the outer dimension index represented by this level. Nested
        levels use increasing indices, so the returned properties can describe how
        this dimension constrains dimensions further inside the fiber tree.
        """
        ...

buffer_factory abstractmethod property

Function to create default buffers for the fibers.

element_type abstractmethod property

Type of elements stored in the fibers.

fill_type abstractmethod property

Fill of the fibers, and whether kernels may specialize on its value.

lvl_t abstractmethod property

Get the nested level.

ndim abstractmethod property

Number of dimensions of the fibers in the structure.

position_type abstractmethod property

Type of positions within the levels.

shape_type abstractmethod property

Tuple of types of the dimensions in the shape.

construct(shape, *, pos) abstractmethod

Construct a level instance with the given shape.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def construct(self, shape: tuple[Any, ...], *, pos: int) -> Level:
    """
    Construct a level instance with the given shape.
    """
    ...

level_cost(fields, stats, stats_factory, num_pos, lvl) abstractmethod

Defines cost for a level

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_cost(self, fields, stats, stats_factory, num_pos, lvl):
    """
    Defines cost for a level
    """
    ...

level_format_properties(n) abstractmethod

Return the format properties contributed by this level type and children.

n is the outer dimension index represented by this level. Nested levels use increasing indices, so the returned properties can describe how this dimension constrains dimensions further inside the fiber tree.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_format_properties(self, n: int) -> list[FormatProperty]:
    """
    Return the format properties contributed by this level type and children.

    ``n`` is the outer dimension index represented by this level. Nested
    levels use increasing indices, so the returned properties can describe how
    this dimension constrains dimensions further inside the fiber tree.
    """
    ...

level_lower_assemble(ctx, lvl, start, stop)

Assemble parent positions in the half-open range [start, stop).

Source code in src/finch/tensor/level/level.py

def level_lower_assemble(self, ctx, lvl, start, stop):
    """Assemble parent positions in the half-open range [start, stop)."""
    raise NotImplementedError(
        f"Assembly is not implemented for {type(self).__name__}"
    )

level_lower_declare(ctx, tns, init, op, shape, pos) abstractmethod

Emit code to lower a declare of pos previously assembled positions in the level.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_lower_declare(self, ctx, tns, init, op, shape, pos):
    """
    Emit code to lower a declare of `pos` previously assembled positions in
    the level.
    """
    ...

level_lower_dim(ctx, obj, r) abstractmethod

Emit code to return the size of dimension r of the subtensors in the level.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_lower_dim(self, ctx, obj, r):
    """
    Emit code to return the size of dimension `r` of the subtensors in the level.
    """
    ...

level_lower_freeze(ctx, lvl, op, pos) abstractmethod

Emit code to freeze pos previously assembled positions in the level.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_lower_freeze(self, ctx, lvl, op, pos):
    """
    Emit code to freeze `pos` previously assembled positions in the level.
    """
    ...

level_lower_increment(ctx, obj, op, val, pos) abstractmethod

Emit code to increment position pos in the level.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_lower_increment(self, ctx, obj, op, val, pos):
    """
    Emit code to increment position `pos` in the level.
    """
    ...

level_lower_thaw(ctx, lvl, op, pos) abstractmethod

Emit code to thaw pos previously assembled positions in the level.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_lower_thaw(self, ctx, lvl, op, pos):
    """
    Emit code to thaw `pos` previously assembled positions in the level.
    """
    ...

level_lower_unwrap(ctx, obj, pos) abstractmethod

Emit code to return the unwrapped scalar at position pos in the level.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_lower_unwrap(self, ctx, obj, pos):
    """
    Emit code to return the unwrapped scalar at position `pos` in the level.
    """
    ...

level_unfurl(ctx, lvl, ext, mode, proto, pos) abstractmethod

Emit code to unfurl the fiber at position pos in the level.

Source code in src/finch/tensor/level/level.py

@abstractmethod
def level_unfurl(self, ctx, lvl, ext, mode, proto, pos):
    """
    Emit code to unfurl the fiber at position `pos` in the level.
    """
    ...

lower_fill(lvl_expr)

Assembly expression reading the runtime fill from the leaf level's fill field, descending from this level's struct expression.

Source code in src/finch/tensor/level/level.py

def lower_fill(self, lvl_expr):
    """Assembly expression reading the runtime fill from the leaf level's
    fill field, descending from this level's struct expression."""
    return self.lvl_t.lower_fill(asm.GetAttr(lvl_expr, asm.Literal("lvl")))

with_fill(fill_type)

Rebuild this level ftype with the leaf fill replaced.

Source code in src/finch/tensor/level/level.py

def with_fill(self, fill_type: AbstractFillFType) -> LevelFType:
    """Rebuild this level ftype with the leaf fill replaced."""
    # Level ftypes are dataclasses with a `_lvl_t` child field; non-
    # dataclass levels must override.
    return replace(self, _lvl_t=self.lvl_t.with_fill(fill_type))  # type: ignore[type-var]

OverrideTensor

Bases: Tensor

Source code in src/finch/tensor/override_tensor.py

class OverrideTensor(Tensor):
    @property
    def mT(self):
        return self.mod.matrix_transpose(self)

    @property
    def T(self):
        if self.ndim != 2:
            raise ValueError("T is only defined for 2D arrays")
        return self.mod.matrix_transpose(self)

    @property
    def mod(self):
        return self.override_module()

    def override_module(self):
        return sys.modules["finch.interface"]

    def __array_function__(self, func, types, args, kwargs):
        """Override NumPy functions using the __array_function__ protocol."""
        # https://numpy.org/neps/nep-0018-array-function-protocol.html
        override_func = getattr(self.mod, func.__name__, None)
        if override_func is None:
            return NotImplemented
        return override_func(*args, **kwargs)

    def __array_ufunc__(self, ufunc: np.ufunc, method, *inputs, **kwargs):
        """Override NumPy ufuncs using the __array_ufunc__ protocol."""
        # https://numpy.org/devdocs/user/basics.ufuncs.html#ufuncs-basics
        # https://numpy.org/devdocs/reference/ufuncs.html#ufuncs-methods
        if kwargs.get("out") is not None:
            raise NotImplementedError("out parameter is not supported")
        if kwargs.get("where") is not None:
            raise NotImplementedError("where parameter is not supported")
        if kwargs.get("casting") is not None:
            raise NotImplementedError("casting parameter is not supported")
        if kwargs.get("order") is not None:
            raise NotImplementedError("order parameter is not supported")
        if kwargs.get("axes") is not None:
            kwargs["axis"] = kwargs.pop("axes")
        if ufunc in element_wise_ufunc_map:
            if method == "__call__":
                return self.mod.elementwise(
                    element_wise_ufunc_map[ufunc], *inputs, **kwargs
                )
            if method == "reduce":
                return self.mod.reduce(ufunc, *inputs, **kwargs)
        if ufunc in ufunc_map:
            func_name = ufunc_map[ufunc]
            if method == "__call__":
                return getattr(self.mod, func_name)(*inputs, **kwargs)
        return NotImplemented

    def __array_namespace__(self, *, api_version=None):
        # https://data-apis.org/array-api/latest/API_specification/generated/array_api.array.__array_namespace__.html#array_api.array.__array_namespace__
        if api_version is None:
            api_version = "2024.12"

        if api_version not in {"2024.12"}:
            raise ValueError(f'"{api_version}" Array API version not supported.')

        return sys.modules["finch"]

    def __add__(self, other):
        return self.mod.add(self, other)

    def __radd__(self, other):
        return self.mod.add(other, self)

    def __sub__(self, other):
        return self.mod.subtract(self, other)

    def __rsub__(self, other):
        return self.mod.subtract(other, self)

    def __mul__(self, other):
        return self.mod.multiply(self, other)

    def __rmul__(self, other):
        return self.mod.multiply(other, self)

    def __abs__(self):
        return self.mod.abs(self)

    def __pos__(self):
        return self.mod.positive(self)

    def __neg__(self):
        return self.mod.negative(self)

    def __invert__(self):
        return self.mod.bitwise_invert(self)

    def __and__(self, other):
        return self.mod.bitwise_and(self, other)

    def __rand__(self, other):
        return self.mod.bitwise_and(other, self)

    def __lshift__(self, other):
        return self.mod.bitwise_left_shift(self, other)

    def __rlshift__(self, other):
        return self.mod.bitwise_left_shift(other, self)

    def __or__(self, other):
        return self.mod.bitwise_or(self, other)

    def __ror__(self, other):
        return self.mod.bitwise_or(other, self)

    def __rshift__(self, other):
        return self.mod.bitwise_right_shift(self, other)

    def __rrshift__(self, other):
        return self.mod.bitwise_right_shift(other, self)

    def __xor__(self, other):
        return self.mod.bitwise_xor(self, other)

    def __rxor__(self, other):
        return self.mod.bitwise_xor(other, self)

    def __truediv__(self, other):
        return self.mod.truediv(self, other)

    def __rtruediv__(self, other):
        return self.mod.truediv(other, self)

    def __floordiv__(self, other):
        return self.mod.floor_divide(self, other)

    def __rfloordiv__(self, other):
        return self.mod.floor_divide(other, self)

    def __mod__(self, other):
        return self.mod.mod(self, other)

    def __rmod__(self, other):
        return self.mod.mod(other, self)

    def __pow__(self, other):
        return self.mod.power(self, other)

    def __rpow__(self, other):
        return self.mod.power(other, self)

    def __matmul__(self, other):
        return self.mod.matmul(self, other)

    def __rmatmul__(self, other):
        return self.mod.matmul(other, self)

    def __sin__(self):
        return self.mod.sin(self)

    def __sinh__(self):
        return self.mod.sinh(self)

    def __cos__(self):
        return self.mod.cos(self)

    def __cosh__(self):
        return self.mod.cosh(self)

    def __tan__(self):
        return self.mod.tan(self)

    def __tanh__(self):
        return self.mod.tanh(self)

    def __asin__(self):
        return self.mod.asin(self)

    def __asinh__(self):
        return self.mod.asinh(self)

    def __acos__(self):
        return self.mod.acos(self)

    def __acosh__(self):
        return self.mod.acosh(self)

    def __atan__(self):
        return self.mod.atan(self)

    def __atanh__(self):
        return self.mod.atanh(self)

    def __atan2__(self, other):
        return self.mod.atan2(self, other)

    def __complex__(self):
        """
        Converts a zero-dimensional array to a Python `complex` object.
        """
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to complex.")
        # dispatch to the scalar value's `__complex__` method
        return complex(self.item())

    def __float__(self):
        """
        Converts a zero-dimensional array to a Python `float` object.
        """
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to float.")
        # dispatch to the scalar value's `__float__` method
        return float(self.item())

    def __int__(self):
        """
        Converts a zero-dimensional array to a Python `int` object.
        """
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to int.")
        # dispatch to the scalar value's `__int__` method
        return int(self.item())

    def __bool__(self):
        """
        Converts a zero-dimensional array to a Python `bool` object.
        """
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to bool.")
        # dispatch to the scalar value's `__bool__` method
        return bool(self.item())

    def __index__(self) -> int:
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to index.")
        return operator.index(self.item())

    def __log__(self):
        return self.mod.log(self)

    def __log1p__(self):
        return self.mod.log1p(self)

    def __log2__(self):
        return self.mod.log2(self)

    def __log10__(self):
        return self.mod.log10(self)

    def __logaddexp__(self, other):
        return self.mod.logaddexp(self, other)

    def __logical_and__(self, other):
        return self.mod.logical_and(self, other)

    def __logical_or__(self, other):
        return self.mod.logical_or(self, other)

    def __logical_xor__(self, other):
        return self.mod.logical_xor(self, other)

    def __logical_not__(self):
        return self.mod.logical_not(self)

    def __lt__(self, other):
        return self.mod.less(self, other)

    def __le__(self, other):
        return self.mod.less_equal(self, other)

    def __gt__(self, other):
        return self.mod.greater(self, other)

    def __ge__(self, other):
        return self.mod.greater_equal(self, other)

    def __eq__(self, other):
        return self.mod.equal(self, other)

    def __ne__(self, other):
        return self.mod.not_equal(self, other)

__array_function__(func, types, args, kwargs)

Override NumPy functions using the array_function protocol.

Source code in src/finch/tensor/override_tensor.py

def __array_function__(self, func, types, args, kwargs):
    """Override NumPy functions using the __array_function__ protocol."""
    # https://numpy.org/neps/nep-0018-array-function-protocol.html
    override_func = getattr(self.mod, func.__name__, None)
    if override_func is None:
        return NotImplemented
    return override_func(*args, **kwargs)

__array_ufunc__(ufunc, method, *inputs, **kwargs)

Override NumPy ufuncs using the array_ufunc protocol.

Source code in src/finch/tensor/override_tensor.py

def __array_ufunc__(self, ufunc: np.ufunc, method, *inputs, **kwargs):
    """Override NumPy ufuncs using the __array_ufunc__ protocol."""
    # https://numpy.org/devdocs/user/basics.ufuncs.html#ufuncs-basics
    # https://numpy.org/devdocs/reference/ufuncs.html#ufuncs-methods
    if kwargs.get("out") is not None:
        raise NotImplementedError("out parameter is not supported")
    if kwargs.get("where") is not None:
        raise NotImplementedError("where parameter is not supported")
    if kwargs.get("casting") is not None:
        raise NotImplementedError("casting parameter is not supported")
    if kwargs.get("order") is not None:
        raise NotImplementedError("order parameter is not supported")
    if kwargs.get("axes") is not None:
        kwargs["axis"] = kwargs.pop("axes")
    if ufunc in element_wise_ufunc_map:
        if method == "__call__":
            return self.mod.elementwise(
                element_wise_ufunc_map[ufunc], *inputs, **kwargs
            )
        if method == "reduce":
            return self.mod.reduce(ufunc, *inputs, **kwargs)
    if ufunc in ufunc_map:
        func_name = ufunc_map[ufunc]
        if method == "__call__":
            return getattr(self.mod, func_name)(*inputs, **kwargs)
    return NotImplemented

__bool__()

Converts a zero-dimensional array to a Python bool object.

Source code in src/finch/tensor/override_tensor.py

def __bool__(self):
    """
    Converts a zero-dimensional array to a Python `bool` object.
    """
    if self.ndim != 0:
        raise ValueError("Cannot convert non-scalar tensor to bool.")
    # dispatch to the scalar value's `__bool__` method
    return bool(self.item())

__complex__()

Converts a zero-dimensional array to a Python complex object.

Source code in src/finch/tensor/override_tensor.py

def __complex__(self):
    """
    Converts a zero-dimensional array to a Python `complex` object.
    """
    if self.ndim != 0:
        raise ValueError("Cannot convert non-scalar tensor to complex.")
    # dispatch to the scalar value's `__complex__` method
    return complex(self.item())

__float__()

Converts a zero-dimensional array to a Python float object.

Source code in src/finch/tensor/override_tensor.py

def __float__(self):
    """
    Converts a zero-dimensional array to a Python `float` object.
    """
    if self.ndim != 0:
        raise ValueError("Cannot convert non-scalar tensor to float.")
    # dispatch to the scalar value's `__float__` method
    return float(self.item())

__int__()

Converts a zero-dimensional array to a Python int object.

Source code in src/finch/tensor/override_tensor.py

def __int__(self):
    """
    Converts a zero-dimensional array to a Python `int` object.
    """
    if self.ndim != 0:
        raise ValueError("Cannot convert non-scalar tensor to int.")
    # dispatch to the scalar value's `__int__` method
    return int(self.item())

RandomMaskTensor

Bases: PatternTensor

A reproducible random mask with true probability p and no stored entries.

shape may be an integer for a vector, or a tuple (including ()). An omitted seed draws one random 64-bit seed from rng (a new NumPy generator by default). Reads consume no randomness; a fixed seed and coordinates give the same value even if the shape changes.

Like Finch.jl, indexed masks mix the seed, then each one-based coordinate from the first axis to the last, which is Finch.jl's order once the axes are reversed like other tensors. Scalar masks use the seed directly. The high 53 bits give a value in [0, 1) that is compared with p.

Source code in src/finch/tensor/patterns.py

class RandomMaskTensor(PatternTensor):
    """A reproducible random mask with true probability ``p`` and no stored entries.

    ``shape`` may be an integer for a vector, or a tuple (including ``()``).
    An omitted ``seed`` draws one random 64-bit seed from ``rng`` (a new NumPy
    generator by default). Reads consume no randomness; a fixed seed and
    coordinates give the same value even if the shape changes.

    Like Finch.jl, indexed masks mix the seed, then each one-based coordinate
    from the first axis to the last, which is Finch.jl's order once the axes
    are reversed like other tensors. Scalar masks use the seed directly.
    The high 53 bits give a value in ``[0, 1)`` that is compared with ``p``.
    """

    def __init__(
        self,
        shape,
        p: float,
        *,
        seed: int | None = None,
        rng: np.random.Generator | None = None,
        dtype=None,
    ):
        if not 0 <= p <= 1:
            raise ValueError("p must lie in [0, 1]")
        if isinstance(shape, int | np.integer):
            shape = (shape,)
        shape = tuple(operator.index(dim) for dim in shape)
        if any(dim < 0 for dim in shape):
            raise ValueError("shape dimensions must be nonnegative")
        if any(dim > np.iinfo(np.intp).max for dim in shape):
            raise OverflowError("shape dimensions must fit in intp")
        self._p = float(p)
        if seed is None:
            if rng is None:
                rng = np.random.default_rng()
            self._seed = int(rng.integers(0, 1 << 64, dtype=np.uint64))
        else:
            self._seed = operator.index(seed)
        if not 0 <= self._seed < 1 << 64:
            raise ValueError("seed must lie in [0, 2**64 - 1]")
        super().__init__(
            shape,
            ndim=len(shape),
            dtype=dtype,
            default_dtype=np.bool_,
            fill_value=False,
            pattern_value=True,
            p=self._p,
            seed=self._seed,
        )

    def contains(self, *idxs) -> bool:
        if not idxs:
            return _randommask_uniform(self._seed) < self._p
        state = _randommask_mix(self._seed)
        for idx in idxs:
            state = _randommask_mix(state ^ (operator.index(idx) + 1))
        return _randommask_uniform(state) < self._p

    def item(self):
        if self.ndim != 0:
            raise ValueError("Cannot convert non-scalar tensor to Python scalar.")
        return self[()].item()

RepeatTensor

Bases: PatternTensor

Map each of n indices to k repeated indices.

shape is (n, n * k), and row i holds the repeats j of i, so that the efficient row-major traversal visits each index and then its repeats. For k <= 0, all entries are false.

Source code in src/finch/tensor/patterns.py

class RepeatTensor(PatternTensor):
    """Map each of ``n`` indices to ``k`` repeated indices.

    ``shape`` is ``(n, n * k)``, and row ``i`` holds the repeats ``j`` of ``i``,
    so that the efficient row-major traversal visits each index and then its
    repeats. For ``k <= 0``, all entries are false.
    """

    def __init__(self, shape, *, k: int = 0, dtype=None):
        self._k = k
        super().__init__(shape, dtype=dtype, k=self._k)

    @classmethod
    def _level_format_properties(cls, ndim: int) -> list[FormatProperty]:
        return [Blocked((0,), (1,)), Repeated((0,), (1,))]

    def contains(self, i, j) -> bool:
        return self._k > 0 and i == j // self._k

Scalar

Bases: OverrideTensor

Source code in src/finch/tensor/scalar.py

class Scalar(OverrideTensor):
    def __init__(self, val: Any, fill_value: Any = None, device=None):
        if fill_value is None:
            fill_value = val
        self.val = val
        self._fill_value = as_fill(fill_value)
        self._device = normalize_device(device)

    @property
    def ftype(self):
        elem_t = ftype(self.val)
        return ScalarFType(elem_t, DynamicFillFType(elem_t), self._device)

    @property
    def shape(self):
        return ()

    @property
    def fill_value(self) -> Any:
        """Default value to fill the scalar."""
        return self._fill_value.value

    @property
    def device(self):
        return self._device

    def with_fill(self, fill_value: AbstractFill) -> Scalar:
        if is_dynamic(fill_value):
            return Scalar(self.val, fill_value=fill_value, device=self._device)
        if not bool(ffuncs.same(fill_value.value, self.val)):
            raise NotImplementedError(
                "a ConstantScalar's fill is its value, so a scalar of "
                f"{self.val!r} cannot take the static fill {fill_value.value!r}"
            )
        return ConstantScalar(self.val, device=self._device)

    def to_device(self, device, /, *, stream=None):
        if stream is not None:
            raise ValueError(f"stream argument is not supported; got {stream!r}")
        device = normalize_device(device)
        if device == self.device:
            return self
        return Scalar(self.val, fill_value=self._fill_value, device=device)

    @property
    def element_type(self) -> FType:
        """Data type of the scalar."""
        return self.ftype.element_type

    @property
    def shape_type(self) -> tuple:
        """Shape type of the scalar."""
        return self.ftype.shape_type

    def item(self):
        return self.val.item() if hasattr(self.val, "item") else self.val

    def __array__(self, dtype=None, copy=None):
        if copy is None:
            return np.asarray(self.val, dtype=dtype)
        return np.array(self.val, dtype=dtype, copy=copy)

    def __getitem__(self, idx):
        if idx == () or idx is Ellipsis or idx == (...,):
            return self
        raise IndexError("Too many indices for scalar tensor.")

    def __str__(self):
        return str(self.val)

    def to_numpy(self):
        return self.val

    def to_scipy(self):
        raise NotImplementedError(f"{type(self).__name__} does not support to_scipy.")

element_type property

Data type of the scalar.

fill_value property

Default value to fill the scalar.

shape_type property

Shape type of the scalar.

ScalarFType

Bases: TensorFType, ImmutableStructFType

Source code in src/finch/tensor/scalar.py

class ScalarFType(TensorFType, ImmutableStructFType):
    def __init__(
        self, _element_type: FType, _fill_type: AbstractFillFType, _device=None
    ):
        self._element_type = _element_type
        self._fill_type = _fill_type
        self._device = normalize_device(_device)

    def __eq__(self, other):
        if isinstance(other, ScalarFType):
            return (
                self._element_type == other._element_type
                and self._fill_type == other._fill_type
                and self.device == other.device
            )
        return False

    def __hash__(self):
        return hash((self._element_type, self._fill_type, self.device))

    def construct(self, shape: tuple) -> Scalar:
        if shape != ():
            raise ValueError("ScalarFType can only be called with empty shape ()")
        return self._element_type(resolve_fill(self._fill_type).value)

    def __call__(self, val: Any) -> Scalar:
        """
        Convert a tensor to this scalar tensor type.

        Args:
            val: A value to convert to this type.
        Returns:
            A Scalar instance of this type.
        """
        raise NotImplementedError(
            f"Tensor conversion not yet implemented for {type(self).__name__}"
        )

    def from_numpy(self, arr):
        return self(arr)

    @property
    def fill_type(self) -> AbstractFillFType:
        return self._fill_type

    @property
    def device(self):
        return self._device

    @property
    def element_type(self) -> FType:
        return self._element_type

    @property
    def shape_type(self):
        return ()

    @property
    def struct_name(self) -> str:
        return "Scalar"

    @property
    def struct_fields(self) -> list[tuple[str, FType]]:
        return [("val", self._element_type)]

    def from_fields(self, val):
        if is_dynamic(self._fill_type):
            # A scalar's fill defaults to its value.
            return Scalar(val, device=self._device)
        return Scalar(
            val, fill_value=resolve_fill(self._fill_type), device=self._device
        )

    def with_fill(self, fill_type: AbstractFillFType) -> ScalarFType:
        return ScalarFType(self._element_type, fill_type, self._device)

    def fisinstance(self, other):
        other_t = ftype(other)
        if is_dynamic(self._fill_type) and isinstance(other_t, ScalarFType):
            # A dynamic-fill ftype accepts any fill value of matching dtype.
            other_t = other_t.with_fill(self._fill_type)
        return other_t == self

    def lower_fill(self, ctx, obj):
        if is_static(self._fill_type):
            return asm.Literal(self._fill_type.value)
        # The dynamic fill of a scalar is its value.
        return self.lower_unwrap(ctx, obj)

    def lower_unwrap(self, ctx, obj):
        match obj:
            case ntn.Fiber():
                # A slot-bound scalar argument: read the value struct field.
                return asm.GetAttr(ctx(obj.lvl.root), asm.Literal("val"))
            case _:
                # An inline scalar value (e.g. a sparse gap read).
                return ctx(obj)

__call__(val)

Convert a tensor to this scalar tensor type.

Parameters:

Name	Type	Description	Default
val	Any	A value to convert to this type.	required

Returns: A Scalar instance of this type.

Source code in src/finch/tensor/scalar.py

def __call__(self, val: Any) -> Scalar:
    """
    Convert a tensor to this scalar tensor type.

    Args:
        val: A value to convert to this type.
    Returns:
        A Scalar instance of this type.
    """
    raise NotImplementedError(
        f"Tensor conversion not yet implemented for {type(self).__name__}"
    )

SparseListLevel dataclass

Bases: SingleDimensionLevel

A class representing sparse list level.

Source code in src/finch/tensor/level/sparse_list_level.py

@dataclass
class SparseListLevel(SingleDimensionLevel):
    """
    A class representing sparse list level.
    """

    lvl: Level
    dimension: np.integer
    ptr: Any = None
    idx: Any = None
    # These cursors must survive serialization while ptr holds per-parent counts.
    qos_fill: Any = None
    qos_stop: Any = None
    # One past the last occupied parent position; zero means none.
    prev_pos: Any = 0
    frozen: bool = True

    @property
    def shape(self) -> tuple:
        return (self.dimension, *self.lvl.shape)

    def __post_init__(self):
        if self.ptr is None:
            self.ptr = self.lvl.buffer_factory(self.lvl.position_type)(1)
        if self.idx is None:
            self.idx = self.lvl.buffer_factory(ftype(self.dimension))(0)
        p_t = self.lvl.position_type
        self.qos_fill = p_t(
            self.idx.length() if self.qos_fill is None else self.qos_fill
        )
        self.qos_stop = p_t(
            self.idx.length() if self.qos_stop is None else self.qos_stop
        )
        self.prev_pos = p_t(self.prev_pos)

    @property
    def ftype(self) -> SparseListLevelFType:
        return SparseListLevelFType(self.lvl.ftype, ftype(self.dimension))

    @property
    def val(self) -> Any:
        return self.lvl.val

    def __str__(self):
        return f"SparseListLevel(lvl={self.lvl}, dim={self.dimension})"

SparseListLevelFType dataclass

Bases: SingleDimensionLevelFType, MutableStructFType

Source code in src/finch/tensor/level/sparse_list_level.py

@dataclass(unsafe_hash=True)
class SparseListLevelFType(SingleDimensionLevelFType, MutableStructFType):
    _lvl_t: LevelFType
    dimension_type: ftypes.FDTypeInteger = ftypes.intp

    def __post_init__(self):
        self.dimension_type = ftype(self.dimension_type)

    @property
    def struct_name(self):
        return "SparseListLevelFType"

    @property
    def p_t(self):
        return self.position_type

    @property
    def struct_fields(self):
        return [
            ("lvl", self.lvl_t),
            ("dimension", self.dimension_type),
            ("ptr", self.ptr_type),
            ("idx", self.idx_type),
            ("qos_fill", self.position_type),
            ("qos_stop", self.position_type),
            ("prev_pos", self.position_type),
            ("frozen", ftypes.bool_),
        ]

    def __str__(self):
        return f"SparseListLevelFType({self.lvl_t})"

    @property
    def ndim(self):
        return 1 + self.lvl_t.ndim

    @property
    def fill_type(self):
        return self.lvl_t.fill_type

    @property
    def element_type(self):
        """
        Returns the type of elements stored in the fibers.
        """
        return self.lvl_t.element_type

    @property
    def shape_type(self):
        """
        Returns the type of the shape of the fibers.
        """
        return (self.dimension_type, *self.lvl_t.shape_type)

    @property
    def position_type(self):
        """
        Returns the type of positions within the levels.
        """
        return self.lvl_t.position_type

    @property
    def buffer_type(self):
        return self.lvl_t.buffer_type

    @property
    def buffer_factory(self):
        """
        Returns the ftype of the buffer used for the fibers.
        """
        return self.lvl_t.buffer_factory

    @property
    def ptr_type(self):
        return self.buffer_factory(self.position_type)

    @property
    def idx_type(self):
        return self.buffer_factory(self.dimension_type)

    def level_cost(self, fields, stats, stats_factory, num_pos, lvl) -> float:
        pos_size = np_dtype(self.position_type).itemsize
        size_ptr = (num_pos + 1) * pos_size
        reduce_fields = fields[lvl + 1 :]
        if reduce_fields:
            reduced_stats = stats_factory.aggregate(
                ffuncs.or_, False, reduce_fields, stats_factory.support(stats)
            )
        else:
            reduced_stats = stats
        nnz_prefix = reduced_stats.estimate_non_fill_values()
        size_idx = nnz_prefix * pos_size

        return (
            size_ptr
            + size_idx
            + self.lvl_t.level_cost(fields, stats, stats_factory, nnz_prefix, lvl + 1)
        )

    def construct(self, shape: tuple[Any, ...], *, pos: int) -> "SparseListLevel":
        """
        Creates an instance of SparseListLevel.

        Args:
            shape: The shape to be used for the level. (mandatory)
        Returns:
            An instance of DenseLevel.
        """
        lvl = self.lvl_t.construct(shape=shape[1:], pos=0)
        return SparseListLevel(
            lvl,
            self.dimension_type(shape[0]),
            self.ptr_type(int(pos) + 1),
            self.idx_type(0),
        )

    def __call__(self, val: Any) -> "SparseListLevel":
        """
        Convert a level to this sparse list level type.

        Args:
            val: A value to convert to this type.
        Returns:
            A SparseListLevel instance of this type.
        """
        raise NotImplementedError(
            f"Level conversion not yet implemented for {type(self).__name__}"
        )

    @property
    def lvl_t(self):
        return self._lvl_t

    def level_format_properties(self, n):
        return self.lvl_t.level_format_properties(n + 1)

    def level_lower_dim(self, ctx, obj, r):
        if r == 0:
            return asm.GetAttr(obj, asm.Literal("dimension"))
        return self.lvl_t.level_lower_dim(
            ctx, asm.GetAttr(obj, asm.Literal("lvl")), r - 1
        )

    def level_lower_declare(self, ctx, tns, init, op, shape, pos):
        p_t = self.ptr_type.length_type
        p = asm.Variable(ctx.freshen("p"), p_t)
        stop = asm.Variable(ctx.freshen("pos_stop"), p_t)
        to_size = asm.Literal(ffuncs.astype(p_t))
        zero = asm.Literal(self.position_type(0))
        expr = """finch
        stop = to_size(pos) + 1
        resize(tns.ptr, stop)
        for (p in 0:stop)
            tns.ptr[p] = zero
        end
        tns.qos_fill = zero
        tns.qos_stop = zero
        tns.prev_pos = zero
        tns.frozen = false
        """
        ctx.exec(parse_assembly(expr, locals(), position_type=p_t))
        return self.lvl_t.level_lower_declare(
            ctx, asm.GetAttr(tns, asm.Literal("lvl")), init, op, shape, zero
        )

    def level_lower_thaw(self, ctx, lvl, op, pos):
        p_t = self.ptr_type.length_type
        pos_stop = asm.Variable(ctx.freshen("pos_stop"), p_t)
        p = asm.Variable(ctx.freshen("p"), p_t)
        to_size = asm.Literal(ffuncs.astype(p_t))
        to_pos = asm.Literal(ffuncs.astype(self.position_type))
        zero = asm.Literal(self.position_type(0))
        expr = """finch
        pos_stop = to_size(pos)
        if (lvl.frozen)
            lvl.qos_fill = lvl.ptr[pos_stop]
            lvl.qos_stop = lvl.qos_fill
            lvl.prev_pos = zero
            p = pos_stop
            // Difference backwards while preceding entries are still prefix sums.
            while (p > 0)
                lvl.ptr[p] -= lvl.ptr[p - 1]
                if (lvl.ptr[p] != zero)
                    if (lvl.prev_pos == zero)
                        lvl.prev_pos = to_pos(p)
                    end
                end
                p -= 1
            end
            lvl.frozen = false
        end
        """
        ctx.exec(parse_assembly(expr, locals(), position_type=p_t))
        return self.lvl_t.level_lower_thaw(
            ctx,
            asm.GetAttr(lvl, asm.Literal("lvl")),
            op,
            asm.GetAttr(lvl, asm.Literal("qos_fill")),
        )

    def level_lower_freeze(self, ctx, lvl, op, pos):
        p_t = self.ptr_type.length_type
        pos_stop = asm.Variable(ctx.freshen("pos_stop"), p_t)
        qos_stop = asm.Variable(ctx.freshen("qos_stop"), self.position_type)
        p = asm.Variable(ctx.freshen("p"), p_t)
        to_size = asm.Literal(ffuncs.astype(p_t))
        to_idx_size = asm.Literal(ffuncs.astype(self.idx_type.length_type))
        expr = """finch
        pos_stop = to_size(pos)
        resize(lvl.ptr, pos_stop + 1)
        if (lvl.frozen == false)
            for (p in 0:pos_stop)
                lvl.ptr[p + 1] += lvl.ptr[p]
            end
        end
        qos_stop = lvl.ptr[pos_stop]
        resize(lvl.idx, to_idx_size(qos_stop))
        lvl.qos_fill = qos_stop
        lvl.qos_stop = qos_stop
        lvl.frozen = true
        """
        ctx.exec(parse_assembly(expr, locals(), position_type=p_t))
        return self.lvl_t.level_lower_freeze(
            ctx,
            asm.GetAttr(lvl, asm.Literal("lvl")),
            op,
            asm.GetAttr(lvl, asm.Literal("qos_fill")),
        )

    def level_lower_increment(self, ctx, obj, op, val, pos):
        raise NotImplementedError(
            "SparseListLevelFType does not support level_lower_increment."
        )

    def level_lower_assemble(self, ctx, lvl, start, stop):
        p_t = self.ptr_type.length_type
        p = asm.Variable(ctx.freshen("p"), p_t)
        p_start = asm.Variable(ctx.freshen("p_start"), p_t)
        p_stop = asm.Variable(ctx.freshen("p_stop"), p_t)
        to_size = asm.Literal(ffuncs.astype(p_t))
        length = asm.Length(asm.GetAttr(lvl, asm.Literal("ptr")))
        zero = asm.Literal(self.position_type(0))
        expr = """finch
        p_start = to_size(start) + 1
        p_stop = to_size(stop) + 1
        if (length < p_stop)
            resize(lvl.ptr, p_stop)
        end
        for (p in p_start:p_stop)
            lvl.ptr[p] = zero
        end
        """
        ctx.exec(parse_assembly(expr, locals(), position_type=p_t))

    def level_lower_unwrap(self, ctx, obj, pos):
        raise NotImplementedError(
            "SparseListLevelFType does not support level_lower_unwrap."
        )

    def level_unfurl(
        self, ctx, fiber: ntn.Fiber, ext, mode: ntn.AccessMode, proto, pos
    ):
        match mode:
            case ntn.Update():
                return self.level_unfurl_update(ctx, fiber, ext, mode, proto, pos)
        tns = fiber
        level = tns.lvl
        lvl_asm = ctx(level)
        ptr_s = asm.GetAttr(lvl_asm, asm.Literal("ptr"))
        idx_s = asm.GetAttr(lvl_asm, asm.Literal("idx"))

        q = asm.Variable(ctx.freshen("q"), self.position_type)
        q_stop = asm.Variable(ctx.freshen("q_stop"), self.position_type)
        i_stop = asm.Variable(ctx.freshen("i_stop"), self.position_type)
        i_last = asm.Variable(ctx.freshen("i_last"), self.position_type)
        fill = (
            ntn.Value(self.lower_fill(lvl_asm), self.element_type)
            if is_dynamic(self.fill_type)
            else ntn.Literal(self.fill_type.value)
        )
        full = ntn.Full(
            fill,
            tuple(
                ntn.Value(self.level_lower_dim(ctx, lvl_asm, r), self.shape_type[r])
                for r in range(1, self.ndim)
            ),
        )
        tmp_locals = locals()

        def thunk_preamble(ctx, idx):
            expr = """finch
            q = ptr_s[pos]
            q_stop = ptr_s[pos + 1]
            i_stop = 1
            i_last = 0
            if (q < q_stop)
                i_stop = idx_s[q]
                i_last = idx_s[q_stop - 1]
            end
            """
            return parse_assembly(expr, tmp_locals, position_type=self.position_type)

        def seek_fn(ctx, ext):
            start = ctx.ctx(ext.get_start())

            code = f"""finch
            if (idx_s[q] < {start})
              q = scansearch(idx_s, {start}, q, q_stop - 1)
            end
            """

            return parse_assembly(
                code,
                tmp_locals | asm.get_vars_in_expr(start),
                position_type=self.position_type,
            )

        def chunk_tail_fn(ctx, idx):
            pos_2 = asm.Variable(
                ctx.freshen(idx, f"_pos_{self.ndim - 1}"), self.position_type
            )
            ctx.exec(asm.Assign(pos_2, q))
            return lplt.Run(
                ntn.Fiber(
                    ntn.Child(level),
                    ntn.Value(pos_2, self.position_type),
                    (*tns.idxs, idx),
                )
            )

        return lplt.Thunk(
            preamble=thunk_preamble,
            body=lambda ctx, ext: lplt.Sequence(
                head=lambda ctx, idx: lplt.Stepper(
                    preamble=lambda ctx: asm.IfElse(
                        asm.Call(asm.L(ffuncs.lt), (q, q_stop)),
                        asm.Block((asm.Assign(i_stop, asm.Load(idx_s, q)),)),
                        asm.Block(
                            (
                                asm.Assign(
                                    i_stop,
                                    asm.GetAttr(lvl_asm, asm.Literal("dimension")),
                                ),
                            )
                        ),
                    ),
                    stop=lambda ctx: ntn.Variable(i_stop.name, self.position_type),
                    chunk=lplt.Sequence(
                        head=lambda ctx, idx: lplt.Run(full),
                        split=lambda ctx, ext: ntn.Variable(
                            i_stop.name, self.position_type
                        ),
                        tail=chunk_tail_fn,
                    ),
                    next=lambda ctx: asm.Block(
                        (
                            asm.Assign(
                                q,
                                asm.Call(asm.L(ffuncs.add), (q, asm.L(self.p_t(1)))),
                            ),
                        )
                    ),
                    seek=seek_fn,
                ),
                split=lambda ctx, idx: ntn.Call(
                    ntn.L(ffuncs.add),
                    (ntn.Variable(i_last.name, self.position_type), ext.get_unit()),
                ),
                tail=lambda ctx, idx: lplt.Run(full),
            ),
        )

    def level_unfurl_update(self, ctx, fiber, ext, mode, proto, pos):
        lvl = ctx(fiber.lvl)
        child = asm.GetAttr(lvl, asm.Literal("lvl"))
        p_t = self.position_type
        qos = asm.Variable(ctx.freshen("qos"), p_t)
        dirty = ntn.Variable(ctx.freshen("dirty"), ftypes.bool_)
        dirty_asm = ctx(dirty)
        to_ptr_size = asm.Literal(ffuncs.astype(self.ptr_type.length_type))
        to_idx_size = asm.Literal(ffuncs.astype(self.idx_type.length_type))
        to_index = asm.Literal(ffuncs.astype(self.dimension_type))
        to_pos = asm.Literal(ffuncs.astype(p_t))
        bindings = locals()

        def preamble(ctx, idx):
            body = parse_assembly(
                """finch
            qos = lvl.qos_fill
            """,
                bindings,
                position_type=p_t,
            )
            if ctx.mode.safe or ctx.mode.debug:
                return asm.Block(
                    (
                        *body.bodies,
                        asm.Assert(
                            asm.Call(
                                asm.Literal(ffuncs.not_),
                                (asm.GetAttr(lvl, asm.Literal("frozen")),),
                            )
                        ),
                        asm.Assert(
                            asm.Call(
                                asm.Literal(ffuncs.le),
                                (asm.GetAttr(lvl, asm.Literal("prev_pos")), pos),
                            )
                        ),
                    )
                )
            return body

        def lookup(ctx, idx):
            idx_asm = ctx.ctx(idx)

            def prepare(ctx, _):
                grow = ctx.ctx.block()
                grow.exec(
                    parse_assembly(
                        """finch
                if (lvl.qos_stop == 0)
                    lvl.qos_stop = 1
                else
                    lvl.qos_stop += lvl.qos_stop
                end
                resize(lvl.idx, to_idx_size(lvl.qos_stop))
                """,
                        bindings,
                        position_type=p_t,
                    )
                )
                self.lvl_t.level_lower_assemble(
                    grow, child, qos, asm.GetAttr(lvl, asm.Literal("qos_stop"))
                )
                return asm.Block(
                    (
                        asm.If(
                            asm.Call(
                                asm.Literal(ffuncs.ge),
                                (qos, asm.GetAttr(lvl, asm.Literal("qos_stop"))),
                            ),
                            asm.Block(tuple(grow.emit())),
                        ),
                        asm.Assign(dirty_asm, asm.Literal(False)),
                    )
                )

            def finish(ctx, _):
                written = parse_assembly(
                    """finch
                lvl.idx[to_idx_size(qos)] = to_index(idx_asm)
                qos += 1
                lvl.prev_pos = to_pos(pos + 1)
                """,
                    bindings | {"idx_asm": idx_asm},
                    position_type=p_t,
                )
                match fiber:
                    case ntn.HollowFiber(dirty=parent_dirty):
                        written = asm.Block(
                            (
                                *written.bodies,
                                asm.Assign(ctx.ctx(parent_dirty), asm.Literal(True)),
                            )
                        )
                return asm.If(dirty_asm, written)

            return lplt.Thunk(
                preamble=prepare,
                body=lambda ctx, ext: lplt.Run(
                    ntn.HollowFiber(
                        ntn.Child(fiber.lvl),
                        ntn.Value(qos, p_t),
                        (*fiber.idxs, idx),
                        dirty=dirty,
                    )
                ),
                epilogue=finish,
            )

        def epilogue(ctx, idx):
            return parse_assembly(
                """finch
            lvl.ptr[to_ptr_size(pos + 1)] += qos - lvl.qos_fill
            lvl.qos_fill = qos
            """,
                bindings,
                position_type=p_t,
            )

        return lplt.Thunk(
            preamble=preamble,
            body=lambda ctx, ext: lplt.Lookup(lookup),
            epilogue=epilogue,
        )

    def from_fields(
        self,
        lvl,
        dimension,
        ptr,
        idx,
        qos_fill=None,
        qos_stop=None,
        prev_pos=0,
        frozen=True,
    ) -> "SparseListLevel":
        return SparseListLevel(
            lvl, dimension, ptr, idx, qos_fill, qos_stop, prev_pos, frozen
        )

buffer_factory property

Returns the ftype of the buffer used for the fibers.

element_type property

Returns the type of elements stored in the fibers.

position_type property

Returns the type of positions within the levels.

shape_type property

Returns the type of the shape of the fibers.

__call__(val)

Convert a level to this sparse list level type.

Parameters:

Name	Type	Description	Default
val	Any	A value to convert to this type.	required

Returns: A SparseListLevel instance of this type.

Source code in src/finch/tensor/level/sparse_list_level.py

def __call__(self, val: Any) -> "SparseListLevel":
    """
    Convert a level to this sparse list level type.

    Args:
        val: A value to convert to this type.
    Returns:
        A SparseListLevel instance of this type.
    """
    raise NotImplementedError(
        f"Level conversion not yet implemented for {type(self).__name__}"
    )

construct(shape, *, pos)

Creates an instance of SparseListLevel.

Parameters:

Name	Type	Description	Default
shape	tuple[Any, ...]	The shape to be used for the level. (mandatory)	required

Returns: An instance of DenseLevel.

Source code in src/finch/tensor/level/sparse_list_level.py

def construct(self, shape: tuple[Any, ...], *, pos: int) -> "SparseListLevel":
    """
    Creates an instance of SparseListLevel.

    Args:
        shape: The shape to be used for the level. (mandatory)
    Returns:
        An instance of DenseLevel.
    """
    lvl = self.lvl_t.construct(shape=shape[1:], pos=0)
    return SparseListLevel(
        lvl,
        self.dimension_type(shape[0]),
        self.ptr_type(int(pos) + 1),
        self.idx_type(0),
    )

SplitMaskTensor

Bases: PatternTensor

Partition n indices into p contiguous regions of nearly equal size.

shape is (p, n), and row j covers n*j//p <= i < n*(j+1)//p, so that the efficient row-major traversal visits each region and then its indices.

Source code in src/finch/tensor/patterns.py

class SplitMaskTensor(PatternTensor):
    """Partition ``n`` indices into ``p`` contiguous regions of nearly equal size.

    ``shape`` is ``(p, n)``, and row ``j`` covers ``n*j//p <= i < n*(j+1)//p``,
    so that the efficient row-major traversal visits each region and then its
    indices.
    """

    def __init__(self, shape, *, dtype=None):
        super().__init__(
            shape,
            dtype=dtype,
            default_dtype=np.bool_,
            fill_value=False,
            pattern_value=True,
        )
        p, n = map(operator.index, self.shape)
        if n < 0:
            raise ValueError("n must be nonnegative")
        if p <= 0:
            raise ValueError("shape[0] must be positive")

    def contains(self, j, i) -> bool:
        p, n = self.shape
        return n * j // p <= i < n * (j + 1) // p

element(fill_type=None, element_type=None, position_type=None, buffer_factory=NumpyBufferFType, buffer_type=None)

Creates an ElementLevelFType with the given parameters.

Parameters:

Name	Type	Description	Default
fill_type	Any	The fill of the level's format, or the value of a static fill.	None
element_type	FType | None	The type of elements stored in the level.	None
position_type	FType | None	The type of positions within the level.	None
buffer_factory	Any	The factory used to create buffers for the level.	NumpyBufferFType
buffer_type	Any	Format of the value stored in the level.	None

Returns:

Type	Description
ElementLevelFType	An instance of ElementLevelFType.

Source code in src/finch/tensor/level/element_level.py

def element(
    fill_type: Any = None,
    element_type: FType | None = None,
    position_type: FType | None = None,
    buffer_factory: Any = NumpyBufferFType,
    buffer_type: Any = None,
) -> ElementLevelFType:
    """
    Creates an ElementLevelFType with the given parameters.

    Args:
        fill_type: The fill of the level's format, or the value of a static
            fill.
        element_type: The type of elements stored in the level.
        position_type: The type of positions within the level.
        buffer_factory: The factory used to create buffers for the level.
        buffer_type: Format of the value stored in the level.

    Returns:
        An instance of ElementLevelFType.
    """
    match fill_type:
        case AbstractFillFType() | None:
            pass
        case AbstractFill():
            raise TypeError(f"element takes the fill of a format, not {fill_type}")
        case _:
            fill_type = StaticFillFType(fill_type)
    return ElementLevelFType(
        fill_type=fill_type,
        element_type=element_type,
        position_type=position_type,
        buffer_factory=buffer_factory,
        buffer_type=buffer_type,
    )