Host Runtime (`aie.utils.hostruntime`)¶

Host-side helpers for selecting a device, allocating NPU-accessible tensors, running compiled designs, and sharing common CLI / argument plumbing across programming examples.

Import the public surface from aie.utils.hostruntime (or via the aie.utils / iron re-exports where noted). The symbols below are rendered from the Python package under python/utils/hostruntime/ using the same mkdocstrings path layout as the other Python API pages (utils.*, not the installed aie.utils.* import name).

Runtime abstractions¶

Abstract runtime types that concrete backends (for example XRT) implement.

HostRuntimeError ¶

Bases: Exception

Error raised when a NPU kernel encounters an error during runtime operations.

KernelHandle ¶

Bases: ABC

Abstract representation that represents a kernel already registered/loaded with a runtime.

KernelResult ¶

KernelResult(
    npu_time: int, trace_config: TraceConfig | None = None
)

Bases: ABC

A wrapper around data produced as the result of running a kernel.

Initialize the KernelResult.

Parameters:

Name	Type	Description	Default
`npu_time`	`int`	The execution time on the NPU in nanoseconds.	required
`trace_config`	`TraceConfig \| None`	Configuration for tracing. Defaults to None.	`None`

Source code in python/utils/hostruntime/hostruntime.py

def __init__(
    self,
    npu_time: int,
    trace_config: TraceConfig | None = None,
):
    """Initialize the KernelResult.

    Args:
        npu_time (int): The execution time on the NPU in nanoseconds.
        trace_config (TraceConfig | None, optional): Configuration for tracing. Defaults to None.
    """
    self._npu_time = npu_time
    self._trace_config = trace_config

npu_time `property` ¶

npu_time: int

Get the NPU execution time.

Returns:

Name	Type	Description
`int`	`int`	The execution time in nanoseconds.

trace_config `property` ¶

trace_config: TraceConfig | None

Get the trace configuration.

Returns:

Type	Description
`TraceConfig \| None`	TraceConfig \| None: The trace configuration if available, else None.

has_trace ¶

has_trace() -> bool

Check if trace data is available.

Returns:

Name	Type	Description
`bool`	`bool`	True if trace configuration is present, False otherwise.

Source code in python/utils/hostruntime/hostruntime.py

def has_trace(self) -> bool:
    """Check if trace data is available.

    Returns:
        bool: True if trace configuration is present, False otherwise.
    """
    return self._trace_config is not None

is_success `abstractmethod` ¶

is_success() -> bool

Check if the kernel execution was successful.

Returns:

Name	Type	Description
`bool`	`bool`	True if successful, False otherwise.

Source code in python/utils/hostruntime/hostruntime.py

@abstractmethod
def is_success(self) -> bool:
    """Check if the kernel execution was successful.

    Returns:
        bool: True if successful, False otherwise.
    """
    pass

HostRuntime ¶

Bases: ABC

An abstract class for a generic host runtime.

check_device_consistency ¶

check_device_consistency()

Check if the overridden device is loadable on the runtime device.

A 1- or N-column variant of a generation (e.g. NPU1Col1) is loadable on a wider device of the same generation (e.g. a 4-column NPU1), so we accept any override whose arch matches and whose column count is <= the runtime device's column count.

Source code in python/utils/hostruntime/hostruntime.py

def check_device_consistency(self):
    """Check if the overridden device is loadable on the runtime device.

    A 1- or N-column variant of a generation (e.g. NPU1Col1) is loadable
    on a wider device of the same generation (e.g. a 4-column NPU1), so we
    accept any override whose arch matches and whose column count is <=
    the runtime device's column count.
    """
    assert __package__ is not None
    mod = sys.modules[__package__]
    override = getattr(mod, "_CURRENT_DEVICE", None)
    if override is None:
        return
    runtime_device = self.device()
    try:
        same_arch = override.arch == runtime_device.arch
        fits = override.cols <= runtime_device.cols
    except AttributeError:
        same_arch = fits = False
    if not (same_arch and fits):
        raise RuntimeError(
            f"Overridden device {override} is not loadable on runtime "
            f"device {runtime_device}"
        )

cleanup ¶

cleanup() -> None

Release any cached device/runtime resources held by this runtime.

Base implementation is a no-op: a plain runtime holds nothing to release. Caching runtimes override this to free hardware contexts, loaded executables, instruction buffers, etc. Safe to call even if the runtime never ran anything.

Source code in python/utils/hostruntime/hostruntime.py

def cleanup(self) -> None:
    """Release any cached device/runtime resources held by this runtime.

    Base implementation is a no-op: a plain runtime holds nothing to
    release. Caching runtimes override this to free hardware contexts,
    loaded executables, instruction buffers, etc. Safe to call even if the
    runtime never ran anything.
    """
    return

evict_context ¶

evict_context(xclbin_path: Path) -> None

Drop any cached device context associated with xclbin_path.

Recovery hook invoked after the driver rejects a submit against a stale context (e.g. an XRT IOCTL EINVAL) so the next load rebuilds a fresh context. Base implementation is a no-op for runtimes that keep no evictable context cache.

Source code in python/utils/hostruntime/hostruntime.py

def evict_context(self, xclbin_path: Path) -> None:
    """Drop any cached device context associated with ``xclbin_path``.

    Recovery hook invoked after the driver rejects a submit against a stale
    context (e.g. an XRT IOCTL EINVAL) so the next ``load`` rebuilds a fresh
    context. Base implementation is a no-op for runtimes that keep no
    evictable context cache.
    """
    return

load `abstractmethod` ¶

load(npu_kernel: NPUKernel, **kwargs) -> KernelHandle

Load an NPU kernel into the runtime.

Parameters:

Name	Type	Description	Default
`npu_kernel`	`NPUKernel`	The NPU kernel to load.	required
`**kwargs`		Additional arguments for loading.	`{}`

Returns:

Name	Type	Description
`KernelHandle`	`KernelHandle`	A handle to the loaded kernel.

Source code in python/utils/hostruntime/hostruntime.py

@abstractmethod
def load(self, npu_kernel: NPUKernel, **kwargs) -> KernelHandle:
    """Load an NPU kernel into the runtime.

    Args:
        npu_kernel (NPUKernel): The NPU kernel to load.
        **kwargs: Additional arguments for loading.

    Returns:
        KernelHandle: A handle to the loaded kernel.
    """
    pass

run `abstractmethod` ¶

run(
    kernel_handle: KernelHandle,
    args,
    trace_config: TraceConfig | None = None,
    fail_on_error: bool = True,
    only_if_loaded=False,
    **kwargs
) -> KernelResult

Run a loaded kernel.

Parameters:

Name	Type	Description	Default
`kernel_handle`	`KernelHandle`	The handle to the loaded kernel.	required
`args`		Arguments to pass to the kernel.	required
`trace_config`	`TraceConfig \| None`	Configuration for tracing. Defaults to None.	`None`
`fail_on_error`	`bool`	Whether to raise an exception on kernel failure. Defaults to True.	`True`
`only_if_loaded`	`bool`	If True, only run if already loaded. Defaults to False.	`False`
`**kwargs`		Additional arguments.	`{}`

Returns:

Name	Type	Description
`KernelResult`	`KernelResult`	The result of the kernel execution.

Source code in python/utils/hostruntime/hostruntime.py

@abstractmethod
def run(
    self,
    kernel_handle: KernelHandle,
    args,
    trace_config: TraceConfig | None = None,
    fail_on_error: bool = True,
    only_if_loaded=False,
    **kwargs,
) -> KernelResult:
    """Run a loaded kernel.

    Args:
        kernel_handle (KernelHandle): The handle to the loaded kernel.
        args: Arguments to pass to the kernel.
        trace_config (TraceConfig | None, optional): Configuration for tracing. Defaults to None.
        fail_on_error (bool, optional): Whether to raise an exception on kernel failure. Defaults to True.
        only_if_loaded (bool, optional): If True, only run if already loaded. Defaults to False.
        **kwargs: Additional arguments.

    Returns:
        KernelResult: The result of the kernel execution.
    """
    pass

load_and_run ¶

load_and_run(
    npu_kernel: NPUKernel, run_args: list, **kwargs
) -> tuple[KernelHandle, KernelResult]

Load and run an NPU kernel.

Parameters:

Name	Type	Description	Default
`npu_kernel`	`NPUKernel`	The NPU kernel to load and run.	required
`run_args`	`list`	Arguments to pass to the kernel.	required
`**kwargs`		Additional arguments passed to load.	`{}`

Returns:

Type	Description
`tuple[KernelHandle, KernelResult]`	tuple[KernelHandle, KernelResult]: A tuple containing the kernel handle and the execution result.

Source code in python/utils/hostruntime/hostruntime.py

def load_and_run(
    self,
    npu_kernel: NPUKernel,
    run_args: list,
    **kwargs,
) -> tuple[KernelHandle, KernelResult]:
    """Load and run an NPU kernel.

    Args:
        npu_kernel (NPUKernel): The NPU kernel to load and run.
        run_args (list): Arguments to pass to the kernel.
        **kwargs: Additional arguments passed to load.

    Returns:
        tuple[KernelHandle, KernelResult]: A tuple containing the kernel handle and the execution result.
    """
    trace_config = npu_kernel.trace_config
    handle = self.load(npu_kernel, **kwargs)
    if trace_config:
        if trace_config.reuse_output_buffer and len(run_args) > 0:
            trace_config.last_tensor_shape = run_args[-1].shape
            trace_config.last_tensor_dtype = np.dtype(run_args[-1].dtype)
        self.prepare_args_for_trace(run_args, trace_config)

        # Passing a trace_config to a design that never called enable_trace
        # means the lowering appended no trace operand, yet the host just
        # appended a trace buffer above. The extra buffer has no matching
        # runtime_sequence operand and would run with an empty trace (or,
        # before the firmware-ABI floor over-declared kernels.json, segfault
        # in XRT argument setup). Compare against the design's true operand
        # count -- floor-independent, unlike the kernels.json boN slot count.
        num_host_bos = npu_kernel.num_host_bos
        if num_host_bos is not None and len(run_args) > num_host_bos:
            raise HostRuntimeError(
                f"A trace_config was supplied but the compiled design has "
                f"{num_host_bos} host buffer argument(s), while running with "
                f"a trace buffer requires {len(run_args)}. The design must "
                f"call enable_trace(...) so trace lowering appends a trace "
                f"buffer operand; otherwise the trace buffer has nowhere to "
                f"land."
            )

    ret = self.run(handle, list(run_args), trace_config=trace_config)

    if trace_config:
        trace_buffer, ctrl_buffer = self.extract_trace_from_args(
            run_args, trace_config
        )
        self.process_trace(trace_buffer, ctrl_buffer, trace_config)

    return handle, ret

device `abstractmethod` ¶

device() -> Device

Get the device associated with this runtime.

Returns:

Name	Type	Description
`Device`	`Device`	The device object.

Source code in python/utils/hostruntime/hostruntime.py

@abstractmethod
def device(self) -> "Device":
    """Get the device associated with this runtime.

    Returns:
        Device: The device object.
    """
    pass

read_insts_binary `classmethod` ¶

read_insts_binary(insts_path: Path)

Read instructions from a binary file.

Parameters:

Name	Type	Description	Default
`insts_path`	`Path`	Path to the binary instruction file.	required

Returns:

Type	Description
	np.ndarray: Array of uint32 instructions.

Source code in python/utils/hostruntime/hostruntime.py

@classmethod
def read_insts_binary(cls, insts_path: Path):
    """Read instructions from a binary file.

    Args:
        insts_path (Path): Path to the binary instruction file.

    Returns:
        np.ndarray: Array of uint32 instructions.
    """
    with open(insts_path, "rb") as f:
        data = f.read()
    # Interpret the binary data as an array of uint32 values.
    return np.frombuffer(data, dtype=np.uint32)

read_insts `classmethod` ¶

read_insts(insts_path: Path)

Read instructions from the given file.

If the file extension is .bin, uses binary read. If the file extension is .txt, uses sequence (text) read.

Parameters:

Name	Type	Description	Default
`insts_path`	`Path`	Path to the instruction file.	required

Returns:

Type	Description
	np.ndarray: Array of instructions.

Raises:

Type	Description
`HostRuntimeError`	If the file extension is not supported.

Source code in python/utils/hostruntime/hostruntime.py

@classmethod
def read_insts(cls, insts_path: Path):
    """Read instructions from the given file.

    If the file extension is .bin, uses binary read.
    If the file extension is .txt, uses sequence (text) read.

    Args:
        insts_path (Path): Path to the instruction file.

    Returns:
        np.ndarray: Array of instructions.

    Raises:
        HostRuntimeError: If the file extension is not supported.
    """
    ext = insts_path.suffix.lower()
    if ext == ".bin":
        return cls.read_insts_binary(insts_path)
    else:
        raise HostRuntimeError(
            "Unsupported file extension for instruction file: expected .bin"
        )

prepare_args_for_trace `classmethod` ¶

prepare_args_for_trace(
    args: list[NpuTensor], trace_config: TraceConfig
) -> list[NpuTensor]

Prepare arguments for tracing by appending necessary buffers.

Parameters:

Name	Type	Description	Default
`args`	`list[NpuTensor]`	List of input/output tensors.	required
`trace_config`	`TraceConfig`	Trace configuration.	required

Returns:

Type	Description
`list[NpuTensor]`	list[NpuTensor]: The updated list of tensors with trace buffers appended.

Source code in python/utils/hostruntime/hostruntime.py

@classmethod
def prepare_args_for_trace(
    cls, args: list[NpuTensor], trace_config: TraceConfig
) -> list[NpuTensor]:
    """Prepare arguments for tracing by appending necessary buffers.

    Args:
        args (list[NpuTensor]): List of input/output tensors.
        trace_config (TraceConfig): Trace configuration.

    Returns:
        list[NpuTensor]: The updated list of tensors with trace buffers appended.
    """
    if trace_config.reuse_output_buffer:
        # Trace data is written into the tail of the last output buffer.
        # Extend that buffer by the trace size; no new host buffer is added.
        out_size = trace_config.trace_size
        if len(args) > 0:
            out_size += args[-1].nbytes
            # TODO: should really copy previous contents of output into this buffer...? What if it's in/out?
            args[-1] = tensor((out_size,), dtype=np.uint8)
        else:
            out = tensor((out_size,), dtype=np.uint8)
            args.append(out)
    else:
        # Dedicated trace buffer: trace lowering appended one trailing
        # argument to the runtime_sequence, so the host appends exactly one
        # trailing buffer here. The trace buffer lands at index len(args),
        # which matches the appended argument's index by construction -- no
        # positional padding is needed.
        if trace_config.enable_ctrl_pkts:
            # write ctrl packets
            ctrl_pkts = [
                create_ctrl_pkt(1, 0, 0x32004),  # core status
                create_ctrl_pkt(1, 0, 0x340D8),  # trace status
            ]
            # Pad to 8 words
            ctrl_pkts += [0] * (8 - len(ctrl_pkts))

            header = tensor(np.array(ctrl_pkts, dtype=np.uint32))
            args.append(header)

        # Allocate extra space for control packets if enabled
        alloc_size = trace_config.trace_size
        if trace_config.enable_ctrl_pkts:
            alloc_size = trace_config.trace_size * 4

        trace_buff = tensor((alloc_size,), dtype=np.uint8)
        args.append(trace_buff)
    return args

extract_trace_from_args `classmethod` ¶

extract_trace_from_args(
    args: list[NpuTensor], trace_config: TraceConfig
) -> tuple[ndarray, ndarray | None]

Extract trace and control buffers from the arguments.

Parameters:

Name	Type	Description	Default
`args`	`list[NpuTensor]`	List of tensors used in execution.	required
`trace_config`	`TraceConfig`	Trace configuration.	required

Returns:

Type	Description
`tuple[ndarray, ndarray \| None]`	tuple[np.ndarray, np.ndarray \| None]: A tuple containing the trace buffer and optionally the control buffer.

Source code in python/utils/hostruntime/hostruntime.py

@classmethod
def extract_trace_from_args(
    cls, args: list[NpuTensor], trace_config: TraceConfig
) -> tuple[np.ndarray, np.ndarray | None]:
    """Extract trace and control buffers from the arguments.

    Args:
        args (list[NpuTensor]): List of tensors used in execution.
        trace_config (TraceConfig): Trace configuration.

    Returns:
        tuple[np.ndarray, np.ndarray | None]: A tuple containing the trace buffer and optionally the control buffer.
    """
    trace_buff = None
    ctrl_buff = None

    if trace_config.reuse_output_buffer:
        prefix, trace_buff = cls._extract_prefix(
            args[-1], trace_config.last_tensor_shape, trace_config.last_tensor_dtype
        )
        args[-1] = prefix  # pyright: ignore[reportCallIssue, reportArgumentType]
    else:
        # The trace position is always last.
        trace_buff = args[-1].numpy()

    if trace_config.enable_ctrl_pkts:
        trace_buff, ctrl_buff = cls._extract_prefix(
            trace_buff, trace_config.trace_size, np.dtype(np.uint8)
        )
    trace_buff = trace_buff.view(np.uint32).reshape(
        trace_config.trace_size // np.dtype(np.uint32).itemsize
    )
    return trace_buff, ctrl_buff

process_trace `classmethod` ¶

process_trace(
    trace_buffer, ctrl_buffer, trace_config, verbosity=0
)

Process the trace buffer and control buffer.

Parameters:

Name	Type	Description	Default
`trace_buffer`	`ndarray`	The trace data buffer.	required
`ctrl_buffer`	`ndarray`	The control packet buffer.	required
`trace_config`	`TraceConfig`	Trace configuration.	required
`verbosity`	`int`	Verbosity level. Defaults to 0.	`0`

Source code in python/utils/hostruntime/hostruntime.py

@classmethod
def process_trace(cls, trace_buffer, ctrl_buffer, trace_config, verbosity=0):
    """Process the trace buffer and control buffer.

    Args:
        trace_buffer (np.ndarray): The trace data buffer.
        ctrl_buffer (np.ndarray): The control packet buffer.
        trace_config (TraceConfig): Trace configuration.
        verbosity (int, optional): Verbosity level. Defaults to 0.
    """
    logger.debug("trace_buffer shape: %s", trace_buffer.shape)
    logger.debug("trace_buffer dtype: %s", trace_buffer.dtype)
    trace_config.write_trace(trace_buffer)

    if trace_config.enable_ctrl_pkts:
        logger.debug("ctrl_buffer shape: %s", ctrl_buffer.shape)
        logger.debug("ctrl_buffer dtype: %s", ctrl_buffer.dtype)
        logger.debug("ctrl buffer: %s", [hex(d) for d in ctrl_buffer])
        for i in range(ctrl_buffer.size // 2):
            col, row, pkt_type, pkt_id = extract_tile(ctrl_buffer[i * 2])
            overflow = True if (ctrl_buffer[i * 2 + 1] >> 8) == 3 else False
            if overflow:
                logger.warning(
                    "Trace overflow detected in tile(%d,%d). Trace results may be invalid.",
                    row,
                    col,
                )

verify_results `classmethod` ¶

verify_results(io_args, refs=None, verbosity=0)

Verify the results of the kernel execution against reference data.

Parameters:

Name	Type	Description	Default
`io_args`	`list[NpuTensor]`	List of input/output tensors.	required
`refs`	`dict \| None`	Dictionary mapping index to reference numpy array. Defaults to None (empty dict).	`None`
`verbosity`	`int`	Verbosity level. Defaults to 0.	`0`

Returns:

Name	Type	Description
`int`		Number of errors found.

Raises:

Type	Description
`HostRuntimeError`	If a reference index is out of bounds.

Source code in python/utils/hostruntime/hostruntime.py

@classmethod
def verify_results(cls, io_args, refs=None, verbosity=0):
    """Verify the results of the kernel execution against reference data.

    Args:
        io_args (list[NpuTensor]): List of input/output tensors.
        refs (dict | None, optional): Dictionary mapping index to reference numpy array. Defaults to None (empty dict).
        verbosity (int, optional): Verbosity level. Defaults to 0.

    Returns:
        int: Number of errors found.

    Raises:
        HostRuntimeError: If a reference index is out of bounds.
    """
    if refs is None:
        refs = {}
    errors = 0
    if verbosity >= 1:
        logger.info("Verifying results ...")

    for idx, ref in refs.items():
        if idx >= len(io_args):
            raise HostRuntimeError(
                f"Error: Reference index {idx} out of bounds for {len(io_args)} IO buffers"
            )
        io_args[idx].to("cpu")
        o = io_args[idx].numpy()
        e = bfloat16_safe_allclose(ref.dtype, ref, o)
        errors += np.size(e) - np.count_nonzero(e)
    return errors

run_test ¶

run_test(
    npu_kernel,
    io_args,
    ref,
    verify: bool = True,
    verbosity: int = 0,
) -> int

Run a test for the given NPU kernel.

Parameters:

Name	Type	Description	Default
`npu_kernel`	`NPUKernel`	The NPU kernel to test.	required
`io_args`	`list[NpuTensor]`	List of input/output tensors.	required
`ref`	`dict`	Reference data for verification.	required
`verify`	`bool`	Whether to verify results. Defaults to True.	`True`
`verbosity`	`int`	Verbosity level. Defaults to 0.	`0`

Returns:

Name	Type	Description
`int`	`int`	0 if successful, 1 otherwise.

Source code in python/utils/hostruntime/hostruntime.py

def run_test(
    self,
    npu_kernel,
    io_args,
    ref,
    verify: bool = True,
    verbosity: int = 0,
) -> int:
    """Run a test for the given NPU kernel.

    Args:
        npu_kernel (NPUKernel): The NPU kernel to test.
        io_args (list[NpuTensor]): List of input/output tensors.
        ref (dict): Reference data for verification.
        verify (bool, optional): Whether to verify results. Defaults to True.
        verbosity (int, optional): Verbosity level. Defaults to 0.

    Returns:
        int: 0 if successful, 1 otherwise.
    """
    kernel_handle = self.load(npu_kernel)
    trace_config = npu_kernel.trace_config

    # Ensure io_args is a list
    if not isinstance(io_args, list):
        io_args = [io_args] if io_args else []

    buffers = io_args
    last_out = buffers[-1] if buffers else None

    if trace_config:
        trace_config.last_tensor_shape = last_out.shape if last_out else None
        trace_config.last_tensor_dtype = last_out.dtype if last_out else None
        self.prepare_args_for_trace(buffers, trace_config)

    ret = self.run(kernel_handle, buffers)

    if verbosity >= 1:
        logger.info("npu_time: %s us", ret.npu_time / 1000.0)

    if trace_config:
        trace_buffer, ctrl_buffer = self.extract_trace_from_args(
            buffers, trace_config
        )
        self.process_trace(trace_buffer, ctrl_buffer, trace_config, verbosity)

    errors = 0
    if verify:
        errors = self.verify_results(io_args, ref, verbosity)

    if not errors:
        return 0
    else:
        logger.error("Error count: %d", errors)
        logger.error("Failed.")
        return 1

Tensor and device utilities¶

Device selection, the NpuTensor tensor type, and numerical helpers used when comparing host results. NpuTensor was previously called Tensor, which remains available as an alias.

A tensor is a shape and a dtype over bytes it does not own. The bytes, and the record of which agent currently holds each range of them, belong to the Storage below.

Host runtime utilities: device selection, tensor allocation, and numerical helpers.

NpuTensor ¶

NpuTensor(
    shape_or_data, dtype: DTypeLike = uint32, device="npu"
)

Bases: ABC

A host-mapped, device-resident buffer of fixed shape and dtype.

This is a buffer with a residency state machine, not a general array. Its invariant is host/device coherence: the host and the device each hold a view of the same storage, and the two are reconciled only at the points this class defines. Everything else it offers (indexing, filling, the numpy and torch bridges, :meth:subview) exists to keep that reconciliation correct while still letting callers treat the buffer as data.

The invariant in full:

Writes through the declared paths (:meth:__setitem__, :meth:fill_, and the factories) are reconciled: the factories transfer as they construct, and the in-place writes record the region so the next :meth:to sends it. A write through the raw data array does neither, and is the one way to leave host and device disagreeing.
A method that spans regions in different states reconciles per region. Asking a whole tensor where it lives collapses a mixed extent to one answer, which is the right answer for :meth:to and the wrong one to transfer by.
:meth:to moves residency and is a no-op when the buffer is already on the target device, so a caller that has written through a declared path never pays for a redundant transfer, and a caller that has bypassed one gets no transfer at all.
Reconciliation is not byte-granular. It acts on whole cache lines, which is why :meth:subview requires its regions to be granule-aligned.

Subclasses supply the storage and the two transfer primitives; the invariant itself lives here so every backend states it the same way.

Named for the role rather than the mechanism: XRTTensor and HRXTensor name what implements the buffer, NpuTensor names what it is. Tensor remains as an alias for existing callers.

Initialize the tensor.

Parameters:

Name	Type	Description	Default
`shape_or_data`	`tuple or array - like`	If a tuple, creates a new tensor with the given shape and dtype. If array-like, wraps the data into a tensor with optional dtype casting.	required
`dtype`	`dtype`	Data type of the tensor. Defaults to np.uint32.	`uint32`
`device`	`str`	Device string identifier (e.g., 'npu', 'cpu'). Defaults to 'npu'.	`'npu'`

Source code in python/utils/hostruntime/tensor_class.py

def __init__(self, shape_or_data, dtype: npt.DTypeLike = np.uint32, device="npu"):
    """Initialize the tensor.

    Args:
        shape_or_data (tuple or array-like):
            - If a tuple, creates a new tensor with the given shape and dtype.
            - If array-like, wraps the data into a tensor with optional dtype casting.
        dtype (np.dtype, optional): Data type of the tensor. Defaults to np.uint32.
        device (str, optional): Device string identifier (e.g., 'npu', 'cpu'). Defaults to 'npu'.
    """
    if device not in self.__class__.DEVICES:
        raise ValueError(f"Unsupported device: {device}")
    self._initial_device = device
    self.dtype = dtype

storage `property` ¶

storage: Storage

The allocation this tensor is a view of.

A backend that still owns its bytes directly is wrapped on first use in a storage whose transport reconciles whole-extent, so every tensor has one whether or not its backend has adopted the split.

device `property` `writable` ¶

device

Which agent holds the authoritative copy of this tensor's bytes.

Read from the allocation, so two tensors over the same bytes can never disagree. A tensor spanning regions in different states reports cpu while any part of it still needs flushing, since that is the answer that keeps a caller's reconcile from being skipped.

base `property` ¶

base

The buffer that owns this one's storage, or None if it owns it itself.

Mirrors :attr:numpy.ndarray.base, including collapsing a chain of views: the base of a view of a view is the buffer that actually owns the storage, not the intermediate view. The intermediate is still referenced internally, so the whole chain stays alive for as long as any view of it does.

is_view `property` ¶

is_view

Whether this buffer shares another buffer's storage.

storage_offset `property` ¶

storage_offset

Where this buffer starts within :attr:base's storage, in bytes.

Zero for a buffer that owns its storage. Accumulated through nesting, so it is always measured from the owner rather than from the view this one was carved out of. Compare :meth:torch.Tensor.storage_offset, which is in elements; this is in bytes because a view may reinterpret the dtype.

data `abstractmethod` `property` ¶

data: ndarray

Subclasses must implement a data property.

Returns:

Type	Description
`ndarray`	np.ndarray: The underlying data of the tensor.

shape `abstractmethod` `property` ¶

shape: tuple[int, ...]

Subclasses must implement a shape property.

Returns:

Name	Type	Description
`tuple`	`tuple[int, ...]`	The shape of the tensor.

nbytes `cached` `property` ¶

nbytes: int

Number of bytes consumed by elements in the tensor.

element_size `cached` `property` ¶

element_size: int

Number of bytes per element.

to ¶

to(target_device: str)

Move the tensor to a specified target device.

Parameters:

Name	Type	Description	Default
`target_device`	`str`	The target device.	required

Returns:

Type	Description
	The tensor object on the target device.

Source code in python/utils/hostruntime/tensor_class.py

def to(self, target_device: str):
    """Move the tensor to a specified target device.

    Args:
        target_device (str): The target device.

    Returns:
       The tensor object on the target device.
    """
    coherence = self._coherence_ref or self._coherence()
    if coherence.uniform == target_device:
        return self
    start, end = self._extent
    if coherence.get(start, end) == target_device:
        # Already wholly where it is wanted: nothing to transfer and nothing
        # to record. A dispatch takes this path for every argument it does
        # not have to move, so it stays off the map entirely.
        return self
    if target_device == "npu":
        # Send the dirty ranges, not the whole extent. Ranges are coalesced,
        # so a buffer written end to end costs one transfer and a buffer
        # with a few dirty windows costs those windows.
        for lo, hi in coherence.ranges(start, end, _CoherenceMap.HOST):
            self.storage.sync_to_device(lo, hi - lo)
        coherence.set(start, end, "npu")
    elif target_device == "cpu":
        for lo, hi in coherence.ranges(start, end, _CoherenceMap.DEVICE):
            self.storage.sync_from_device(lo, hi - lo)
        coherence.set(start, end, "cpu")
    else:
        raise ValueError(f"Unknown device '{target_device}'")
    return self

mutate ¶

mutate()

Borrow this buffer's bytes for a host write.

Reconciles on entry so partial updates see current contents, records the write on exit, and retires the borrowed array so a reference kept past the block fails loudly instead of writing bytes nobody will flush.

The flush itself is deferred: the region is marked host-written and the next transfer to the device sends every dirty range in one pass, so a caller filling many windows of one buffer pays for one flush, not one per window.

with tensor.mutate() as buf:
    buf[:] = values

Source code in python/utils/hostruntime/tensor_class.py

def mutate(self):
    """Borrow this buffer's bytes for a host write.

    Reconciles on entry so partial updates see current contents, records the
    write on exit, and retires the borrowed array so a reference kept past
    the block fails loudly instead of writing bytes nobody will flush.

    The flush itself is deferred: the region is marked host-written and the
    next transfer to the device sends every dirty range in one pass, so a
    caller filling many windows of one buffer pays for one flush, not one
    per window.

        with tensor.mutate() as buf:
            buf[:] = values
    """
    return _WriteBorrow(self, reconcile=True)

overwrite ¶

overwrite()

Borrow for a write that replaces every byte of this tensor.

The same as :meth:mutate without the reconcile on entry, which nothing can observe if all of it is about to be replaced. Filling a buffer this way costs one transfer rather than two.

The caller is promising to write the whole region. Bytes left unwritten keep whatever the host last had there and are sent to the device with the rest, so use :meth:mutate for a partial update. This is the one promise here that cannot be checked; it is still narrower than reaching for data, which makes the same promise and does not record the write.

Source code in python/utils/hostruntime/tensor_class.py

def overwrite(self):
    """Borrow for a write that replaces every byte of this tensor.

    The same as :meth:`mutate` without the reconcile on entry, which nothing
    can observe if all of it is about to be replaced. Filling a buffer this
    way costs one transfer rather than two.

    The caller is promising to write the whole region. Bytes left unwritten
    keep whatever the host last had there and are sent to the device with
    the rest, so use :meth:`mutate` for a partial update. This is the one
    promise here that cannot be checked; it is still narrower than reaching
    for ``data``, which makes the same promise and does not record the write.
    """
    return _WriteBorrow(self, reconcile=False)

subview ¶

subview(offset, shape, dtype=None)

Return a tensor viewing a sub-region of this tensor's underlying storage.

The returned tensor shares this tensor's buffer (no new allocation, no copy), holds a reference to this tensor so the storage outlives the view, and synchronizes its own slice. It is a plain tensor of the same backend class, not a distinct type.

The region must be aligned to :data:COHERENCE_GRANULE, because host and device are reconciled a cache line at a time, not a byte at a time. Two views sharing a line are not independent: synchronizing one acts on the other's bytes in that line, so a view whose host copy is stale can be written back over data the device just produced in its neighbor. The check makes that unrepresentable instead of leaving it to callers to remember. It is enforced for every backend, including the CPU-only one that has no coherence concern of its own, so a layout validated against the test backend stays valid on a device.

A view may end anywhere if it ends where this tensor ends: its last line is shared with no sibling, so it is no worse than synchronizing the whole buffer. This also keeps a whole-buffer view (offset=0) legal for a tensor whose own size is not a multiple of the granule.

Note that alignment bounds the damage but does not make a sync exactly slice-scoped: some driver paths (an imported buffer, or one with no kernel mapping) maintain the whole buffer regardless of the requested range. Correctness must not depend on a sync being narrow, only on views not sharing a coherence granule.

Three ways this departs from numpy and torch, none of them accidental:

Out-of-range and negative arguments are rejected, where numpy slicing clamps and wraps them. Clamping a region silently hands back a smaller one than asked for, and a DMA region that is quietly the wrong size is worse than an error. The method-with-arguments spelling is far enough from [start:stop] that it should not invite the slicing intuition.

The alignment requirement has no analogue in either library, because neither reconciles anything between two agents.

offset is in bytes, where numpy counts offsets in elements of the array's own dtype and torch reports storage_offset in elements. Bytes because that is the unit the thing being carved is measured in: a buffer has no dtype, the alignment rule below is in bytes, and :attr:storage_offset reports bytes, so the argument going in and the value reported back are the same number. shape stays in elements of the view's dtype, since it describes the tensor rather than the region.

Parameters:

Name	Type	Description	Default
`offset`		Start of the region, in bytes from the start of this tensor's own region.	required
`shape`		Logical shape of the view.	required
`dtype`	`dtype`	dtype to interpret the region as. Defaults to this tensor's dtype.	`None`

Returns:

Name	Type	Description
`NpuTensor`		A view sharing this tensor's storage.

Raises:

Type	Description
`ValueError`	If the region falls outside this tensor's buffer, or is not aligned to :data:`COHERENCE_GRANULE`.

Source code in python/utils/hostruntime/tensor_class.py

def subview(self, offset, shape, dtype=None):
    """Return a tensor viewing a sub-region of this tensor's underlying storage.

    The returned tensor shares this tensor's buffer (no new allocation, no
    copy), holds a reference to this tensor so the storage outlives the view,
    and synchronizes its own slice. It is a plain tensor of the same backend
    class, not a distinct type.

    The region must be aligned to :data:`COHERENCE_GRANULE`, because host and
    device are reconciled a cache line at a time, not a byte at a time. Two
    views sharing a line are not independent: synchronizing one acts on the
    other's bytes in that line, so a view whose host copy is stale can be
    written back over data the device just produced in its neighbor. The
    check makes that unrepresentable instead of leaving it to callers to
    remember. It is enforced for every backend, including the CPU-only one
    that has no coherence concern of its own, so a layout validated against
    the test backend stays valid on a device.

    A view may end anywhere if it ends where this tensor ends: its last line
    is shared with no sibling, so it is no worse than synchronizing the whole
    buffer. This also keeps a whole-buffer view (``offset=0``) legal for a
    tensor whose own size is not a multiple of the granule.

    Note that alignment bounds the damage but does not make a sync exactly
    slice-scoped: some driver paths (an imported buffer, or one with no
    kernel mapping) maintain the whole buffer regardless of the requested
    range. Correctness must not depend on a sync being narrow, only on views
    not sharing a coherence granule.

    Three ways this departs from numpy and torch, none of them accidental:

    Out-of-range and negative arguments are rejected, where numpy slicing
    clamps and wraps them. Clamping a region silently hands back a smaller
    one than asked for, and a DMA region that is quietly the wrong size is
    worse than an error. The method-with-arguments spelling is far enough
    from ``[start:stop]`` that it should not invite the slicing intuition.

    The alignment requirement has no analogue in either library, because
    neither reconciles anything between two agents.

    ``offset`` is in bytes, where numpy counts offsets in elements of the
    array's own dtype and torch reports ``storage_offset`` in elements.
    Bytes because that is the unit the thing being carved is measured in: a
    buffer has no dtype, the alignment rule below is in bytes, and
    :attr:`storage_offset` reports bytes, so the argument going in and the
    value reported back are the same number. ``shape`` stays in elements of
    the view's dtype, since it describes the tensor rather than the region.

    Args:
        offset: Start of the region, in bytes from the start of this
            tensor's own region.
        shape: Logical shape of the view.
        dtype (np.dtype, optional): dtype to interpret the region as.
            Defaults to this tensor's dtype.

    Returns:
        NpuTensor: A view sharing this tensor's storage.

    Raises:
        ValueError: If the region falls outside this tensor's buffer, or is
            not aligned to :data:`COHERENCE_GRANULE`.
    """
    if type(self)._subview is NpuTensor._subview:
        # Answer the capability question before complaining about a region
        # this backend could not have produced whatever its shape.
        raise NotImplementedError(
            f"{type(self).__name__} does not support subview()"
        )
    view_dtype = np.dtype(dtype) if dtype is not None else np.dtype(self.dtype)
    shape = _as_shape(shape)
    try:
        offset = operator.index(offset)
    except TypeError:
        raise TypeError(
            f"subview() offset must be an integer number of bytes, got "
            f"{offset!r}."
        ) from None
    offset_bytes = offset
    # math.prod over validated ints: np.prod would accumulate in int64 and
    # wrap silently, which defeats the bounds check below rather than
    # tripping it.
    nbytes = math.prod(shape) * view_dtype.itemsize
    if offset_bytes < 0 or offset_bytes + nbytes > self.nbytes:
        raise ValueError(
            f"subview(offset={offset}, shape={shape}, dtype={view_dtype}) "
            f"is out of bounds for a buffer of {self.nbytes} bytes"
        )
    granule = self._resolve_coherence_granule()
    ends_at_parent_end = offset_bytes + nbytes == self.nbytes
    if offset_bytes % granule or (nbytes % granule and not ends_at_parent_end):
        raise ValueError(
            f"subview(offset={offset}, shape={tuple(shape)}, dtype={view_dtype}) "
            f"spans bytes [{offset_bytes}, {offset_bytes + nbytes}) of this "
            f"buffer, which is not aligned to the {granule}-byte coherence "
            f"granule. Host and device are reconciled a cache line at a time, "
            f"so a view sharing a line with a neighbor cannot be synchronized "
            f"independently of it. Pad the region layout so each view starts "
            f"at a multiple of {granule} bytes and (unless it ends where this "
            f"buffer ends) is a multiple of {granule} bytes long."
        )
    return self._subview(offset_bytes, tuple(shape), view_dtype)

numpy ¶

numpy()

Return a NumPy view of the tensor data on host memory.

This method ensures that data is first synchronized from the device (e.g., NPU) to the host before returning the array.

Returns:

Type	Description
	np.ndarray: The tensor's data as a NumPy array.

Note: For NPU tensors, this method causes implicit data synchronization from device to host to ensure the returned array reflects the current device state.

Source code in python/utils/hostruntime/tensor_class.py

def numpy(self):
    """Return a NumPy view of the tensor data on host memory.

    This method ensures that data is first synchronized from the device
    (e.g., NPU) to the host before returning the array.

    Returns:
        np.ndarray: The tensor's data as a NumPy array.

    Note: For NPU tensors, this method causes implicit data synchronization from device to host
    to ensure the returned array reflects the current device state.
    """
    self._reconcile_for_read()
    return self.data

to_torch ¶

to_torch()

Return a torch tensor sharing the data in this tensor if possible.

Syncs from device first if the tensor is on the NPU.

Returns:

Type	Description
	torch.Tensor: A torch tensor containing the data.

Raises:

Type	Description
`ImportError`	If torch is not installed.

Source code in python/utils/hostruntime/tensor_class.py

def to_torch(self):
    """Return a torch tensor sharing the data in this tensor if possible.

    Syncs from device first if the tensor is on the NPU.

    Returns:
        torch.Tensor: A torch tensor containing the data.

    Raises:
        ImportError: If torch is not installed.
    """
    return _array_to_torch(self.numpy())

torch_view ¶

torch_view()

Return a torch tensor sharing this buffer's host memory without syncing from device.

Unlike to_torch(), this does NOT sync from the NPU first. Marks the buffer as CPU-resident so that a subsequent .to("npu") call (or the NPU operator's implicit sync) will push the written data to device. Use this on write paths where the caller is about to overwrite the buffer contents.

Returns:

Type	Description
	torch.Tensor: A zero-copy torch tensor view of the host-side buffer.

Raises:

Type	Description
`ImportError`	If torch is not installed.

Source code in python/utils/hostruntime/tensor_class.py

def torch_view(self):
    """Return a torch tensor sharing this buffer's host memory without syncing from device.

    Unlike to_torch(), this does NOT sync from the NPU first. Marks the buffer as
    CPU-resident so that a subsequent .to("npu") call (or the NPU operator's implicit
    sync) will push the written data to device. Use this on write paths where the
    caller is about to overwrite the buffer contents.

    Returns:
        torch.Tensor: A zero-copy torch tensor view of the host-side buffer.

    Raises:
        ImportError: If torch is not installed.
    """
    self.device = "cpu"  # mark dirty so next to("npu") will actually sync
    return _array_to_torch(self.data)

from_torch `classmethod` ¶

from_torch(torch_tensor, device=None, **kwargs)

Return a tensor with a copy of the data in the torch_tensor.

Parameters:

Name	Type	Description	Default
`torch_tensor`	`Tensor`	The source torch tensor.	required
`device`	`str`	The target device. Defaults to None.	`None`
`**kwargs`		Additional arguments for tensor creation.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A new tensor containing the data from the torch tensor.

Raises:

Type	Description
`ImportError`	If torch is not installed.

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def from_torch(cls, torch_tensor, device=None, **kwargs):
    """Return a tensor with a copy of the data in the torch_tensor.

    Args:
        torch_tensor (torch.Tensor): The source torch tensor.
        device (str, optional): The target device. Defaults to None.
        **kwargs: Additional arguments for tensor creation.

    Returns:
        NpuTensor: A new tensor containing the data from the torch tensor.

    Raises:
        ImportError: If torch is not installed.
    """
    import torch  # pyright: ignore[reportMissingImports]
    from ml_dtypes import bfloat16

    # Detach (to drop grad) and ensure on CPU
    t = torch_tensor.detach()
    if t.device.type != "cpu":
        t = t.cpu()
    # Ensure contiguous for safe view operations
    if not t.is_contiguous():
        t = t.contiguous()

    if t.dtype == torch.bfloat16:
        # View the same memory as int16, then as NumPy bfloat16
        # This avoids numeric conversion and extra passes over memory.
        u16_np = t.view(torch.uint16).numpy()  # shares memory
        np_array = u16_np.view(bfloat16)  # reinterpret
    else:
        np_array = t.numpy()

    return cls(
        np_array,
        dtype=np_array.dtype,
        device=device or cls.DEFAULT_DEVICE,
        **kwargs,
    )

fill_ ¶

fill_(value)

Fill the tensor with a scalar value (in-place operation).

Parameters:

Name	Type	Description	Default
`value`		The scalar value to fill the tensor with.	required

Note: this replaces every byte, so it skips the reconcile a partial write would need, and records the write rather than flushing it. The next transfer to the device sends it.

Source code in python/utils/hostruntime/tensor_class.py

def fill_(self, value):
    """Fill the tensor with a scalar value (in-place operation).

    Args:
        value: The scalar value to fill the tensor with.

    Note: this replaces every byte, so it skips the reconcile a partial
    write would need, and records the write rather than flushing it. The
    next transfer to the device sends it.
    """
    with self.overwrite() as array:
        array.fill(value)

numel ¶

numel()

Calculate the number of elements in the tensor.

Returns:

Name	Type	Description
`int`		The total number of elements in the tensor.

Source code in python/utils/hostruntime/tensor_class.py

def numel(self):
    """Calculate the number of elements in the tensor.

    Returns:
        int: The total number of elements in the tensor.
    """
    return int(np.prod(self.shape))

ones `classmethod` ¶

ones(*size, out=None, dtype=None, device=None, **kwargs)

Return a tensor filled with ones, with shape defined by size.

Parameters:

Name	Type	Description	Default
`*size`	`int...`	Shape of the tensor, passed as separate ints or a single tuple/list.	`()`
`out`	`NpuTensor`	Optional output tensor to write into.	`None`
`dtype`	`dtype`	Desired dtype. Defaults to np.float32.	`None`
`device`	`str`	Target device. Defaults to 'npu'.	`None`
`**kwargs`		Additional keyword args.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A one-filled tensor.

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def ones(cls, *size, out=None, dtype=None, device=None, **kwargs):
    """Return a tensor filled with ones, with shape defined by size.

    Args:
        *size (int...): Shape of the tensor, passed as separate ints or a single tuple/list.
        out (NpuTensor, optional): Optional output tensor to write into.
        dtype (np.dtype, optional): Desired dtype. Defaults to np.float32.
        device (str, optional): Target device. Defaults to 'npu'.
        **kwargs: Additional keyword args.

    Returns:
        NpuTensor: A one-filled tensor.
    """
    t = cls.__check_or_create(*size, out=out, dtype=dtype, device=device, **kwargs)
    t.fill_(1)
    return t

zeros `classmethod` ¶

zeros(*size, out=None, dtype=None, device=None, **kwargs)

Return a tensor filled with zeros, with shape defined by size.

Parameters:

Name	Type	Description	Default
`*size`	`int...`	Shape of the tensor, passed as separate ints or a single tuple/list.	`()`
`out`	`NpuTensor`	Optional output tensor to write into.	`None`
`dtype`	`dtype`	Desired dtype. Defaults to np.float32.	`None`
`device`	`str`	Target device. Defaults to 'npu'.	`None`
`**kwargs`		Additional keyword args.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A zero-filled tensor.

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def zeros(cls, *size, out=None, dtype=None, device=None, **kwargs):
    """Return a tensor filled with zeros, with shape defined by size.

    Args:
        *size (int...): Shape of the tensor, passed as separate ints or a single tuple/list.
        out (NpuTensor, optional): Optional output tensor to write into.
        dtype (np.dtype, optional): Desired dtype. Defaults to np.float32.
        device (str, optional): Target device. Defaults to 'npu'.
        **kwargs: Additional keyword args.

    Returns:
        NpuTensor: A zero-filled tensor.
    """
    t = cls.__check_or_create(*size, out=out, dtype=dtype, device=device, **kwargs)
    t.fill_(0)
    return t

full `classmethod` ¶

full(
    size,
    fill_value,
    *,
    out=None,
    dtype=None,
    device=None,
    **kwargs
)

Return a tensor of shape size filled with fill_value.

Parameters:

Name	Type	Description	Default
`size`	`int or tuple/list of int`	Shape of the returned tensor.	required
`fill_value`	`scalar`	Value to fill the tensor with.	required
`out`	`NpuTensor`	Optional output tensor to write into.	`None`
`dtype`	`dtype`	Desired dtype. Defaults to np.float32.	`None`
`device`	`str`	Target device. Defaults to 'npu'.	`None`
`**kwargs`		Additional keyword args.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A tensor filled with `fill_value`.

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def full(cls, size, fill_value, *, out=None, dtype=None, device=None, **kwargs):
    """Return a tensor of shape `size` filled with `fill_value`.

    Args:
        size (int or tuple/list of int): Shape of the returned tensor.
        fill_value (scalar): Value to fill the tensor with.
        out (NpuTensor, optional): Optional output tensor to write into.
        dtype (np.dtype, optional): Desired dtype. Defaults to np.float32.
        device (str, optional): Target device. Defaults to 'npu'.
        **kwargs: Additional keyword args.

    Returns:
        NpuTensor: A tensor filled with `fill_value`.
    """
    t = cls.__check_or_create(size, out=out, dtype=dtype, device=device, **kwargs)
    t.fill_(fill_value)
    return t

randint `classmethod` ¶

randint(
    low,
    high,
    size,
    *,
    out=None,
    dtype=None,
    device=None,
    generator=None,
    **kwargs
)

Return a tensor filled with random integers uniformly sampled from [low, high).

Parameters:

Name	Type	Description	Default
`low`	`int`	Lowest integer to be drawn (inclusive).	required
`high`	`int`	One above the highest integer to be drawn (exclusive).	required
`size`	`tuple`	Shape of the returned tensor.	required
`out`	`NpuTensor`	Optional tensor to write the result into.	`None`
`dtype`	`dtype`	Data type. Defaults to np.int64.	`None`
`device`	`str`	Target device. Defaults to 'npu'.	`None`
`generator`	`Generator`	Source RNG for reproducibility. If None, uses np.random module-level state.	`None`
`**kwargs`		Additional arguments passed to the constructor.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A tensor with random integers.

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def randint(
    cls,
    low,
    high,
    size,
    *,
    out=None,
    dtype=None,
    device=None,
    generator=None,
    **kwargs,
):
    """Return a tensor filled with random integers uniformly sampled from [low, high).

    Args:
        low (int): Lowest integer to be drawn (inclusive).
        high (int): One above the highest integer to be drawn (exclusive).
        size (tuple): Shape of the returned tensor.
        out (NpuTensor, optional): Optional tensor to write the result into.
        dtype (np.dtype, optional): Data type. Defaults to np.int64.
        device (str, optional): Target device. Defaults to 'npu'.
        generator (np.random.Generator, optional): Source RNG for reproducibility.
            If None, uses np.random module-level state.
        **kwargs: Additional arguments passed to the constructor.

    Returns:
        NpuTensor: A tensor with random integers.
    """
    dtype = dtype or np.int64
    device = device or cls.DEFAULT_DEVICE

    t = cls.__check_or_create(size, out=out, dtype=dtype, device=device, **kwargs)
    if generator is not None:
        random_val = generator.integers(low, high, size=size, dtype=dtype)
    else:
        random_val = np.random.randint(low, high, size=size, dtype=dtype)
    if size == ():
        t.data.fill(random_val)
    else:
        t.data[:] = random_val
    if device == "npu":
        t._sync_to_device()
    return t

rand `classmethod` ¶

rand(
    *size,
    out=None,
    dtype=None,
    device=None,
    generator=None,
    **kwargs
)

Return a tensor filled with random numbers from a uniform distribution on [0, 1).

Parameters:

Name	Type	Description	Default
`*size`	`int...`	Variable number of integers or a single tuple defining the shape.	`()`
`out`	`NpuTensor`	Output tensor to write into.	`None`
`dtype`	`dtype`	Desired data type. Defaults to np.float32.	`None`
`device`	`str`	Target device. Defaults to 'npu'.	`None`
`generator`	`Generator`	Source RNG for reproducibility. If None, uses np.random module-level state.	`None`
`**kwargs`		Additional arguments passed to constructor.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A tensor with random values in [0, 1).

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def rand(cls, *size, out=None, dtype=None, device=None, generator=None, **kwargs):
    """Return a tensor filled with random numbers from a uniform distribution on [0, 1).

    Args:
        *size (int...): Variable number of integers or a single tuple defining the shape.
        out (NpuTensor, optional): Output tensor to write into.
        dtype (np.dtype, optional): Desired data type. Defaults to np.float32.
        device (str, optional): Target device. Defaults to 'npu'.
        generator (np.random.Generator, optional): Source RNG for reproducibility.
            If None, uses np.random module-level state.
        **kwargs: Additional arguments passed to constructor.

    Returns:
        NpuTensor: A tensor with random values in [0, 1).
    """
    if not size:
        raise ValueError("rand() received no arguments")
    dtype = dtype or np.float32
    device = device or cls.DEFAULT_DEVICE

    t = cls.__check_or_create(*size, out=out, dtype=dtype, device=device, **kwargs)
    if generator is not None:
        random_val = generator.uniform(0.0, 1.0, size=t.shape).astype(dtype)
    else:
        random_val = np.random.uniform(0.0, 1.0, size=t.shape).astype(dtype)
    # Ensure values are < 1.0 for low-precision types
    is_bfloat16 = False
    try:
        from ml_dtypes import bfloat16

        if dtype == bfloat16:
            is_bfloat16 = True
    except ImportError:
        pass

    if np.issubdtype(dtype, np.floating) or is_bfloat16:
        max_val = np.nextafter(dtype(1.0), dtype(0.0))
        random_val = np.clip(random_val, 0.0, max_val)

    if t.shape == ():
        t.data.fill(random_val)
    else:
        t.data[:] = random_val
    if device == "npu":
        t._sync_to_device()
    return t

arange `classmethod` ¶

arange(
    start=0,
    end=None,
    step=1,
    *,
    shape=None,
    out=None,
    dtype=None,
    device=None,
    **kwargs
)

Return a tensor with values from the interval [start, end) with spacing step.

Parameters:

Name	Type	Description	Default
`start`	`number`	Start of interval. Defaults to 0.	`0`
`end`	`number`	End of interval (exclusive). Required if only one argument is given.	`None`
`step`	`number`	Gap between elements. Defaults to 1.	`1`
`shape`	`tuple`	If given, reshape the 1-D sequence to this shape. `prod(shape)` must equal the length of the generated range.	`None`
`dtype`	`dtype`	Desired output data type. Inferred if not provided.	`None`
`out`	`NpuTensor`	Optional tensor to write output to (must match shape and dtype).	`None`
`device`	`str`	Target device. Defaults to 'npu'.	`None`
`**kwargs`		Additional keyword arguments forwarded to the underlying tensor constructor.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A tensor containing the sequence (1-D by default, or `shape` if given).

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def arange(
    cls,
    start=0,
    end=None,
    step=1,
    *,
    shape=None,
    out=None,
    dtype=None,
    device=None,
    **kwargs,
):
    """Return a tensor with values from the interval [start, end) with spacing `step`.

    Args:
        start (number): Start of interval. Defaults to 0.
        end (number): End of interval (exclusive). Required if only one argument is given.
        step (number): Gap between elements. Defaults to 1.
        shape (tuple, optional): If given, reshape the 1-D sequence to this shape.
            `prod(shape)` must equal the length of the generated range.
        dtype (np.dtype, optional): Desired output data type. Inferred if not provided.
        out (NpuTensor, optional): Optional tensor to write output to (must match shape and dtype).
        device (str, optional): Target device. Defaults to 'npu'.
        **kwargs: Additional keyword arguments forwarded to the underlying tensor constructor.

    Returns:
        NpuTensor: A tensor containing the sequence (1-D by default, or `shape` if given).
    """
    if end is None:
        start, end = 0, start

    if dtype is None:
        if any(isinstance(x, float) for x in (start, end, step)):
            dtype = np.float32
        else:
            dtype = np.int64

    device = device or cls.DEFAULT_DEVICE

    data = np.arange(start, end, step, dtype=dtype)

    if shape is not None:
        shape = tuple(shape)
        if int(np.prod(shape)) != data.size:
            raise ValueError(
                f"iron.arange: shape={shape} (prod={int(np.prod(shape))}) does "
                f"not match generated range size {data.size}"
            )
        data = data.reshape(shape)
    else:
        shape = (data.size,)

    if out is not None:
        if out.shape != shape or out.dtype != dtype or out.device != device:
            raise ValueError(
                "Provided `out` tensor must match shape, dtype, and device"
            )
        out.data[...] = data
        if device == "npu":
            out._sync_to_device()
        return out

    t = cls(shape, dtype=dtype, device=device, **kwargs)
    t.data[...] = data
    if device == "npu":
        t._sync_to_device()
    return t

zeros_like `classmethod` ¶

zeros_like(other, dtype=None, device=None, **kwargs)

Create a new tensor with the same shape as other, filled with zeros.

Parameters:

Name	Type	Description	Default
`other`	`NpuTensor`	The reference tensor to copy shape from.	required
`dtype`	`dtype`	Data type of the new tensor. Defaults to other's dtype.	`None`
`device`	`str`	Target device. Defaults to other's device.	`None`
`**kwargs`		Additional keyword arguments forwarded to the constructor.	`{}`

Returns:

Name	Type	Description
`NpuTensor`		A new zero-filled tensor with the same shape.

Source code in python/utils/hostruntime/tensor_class.py

@classmethod
def zeros_like(cls, other, dtype=None, device=None, **kwargs):
    """Create a new tensor with the same shape as `other`, filled with zeros.

    Args:
        other (NpuTensor): The reference tensor to copy shape from.
        dtype (np.dtype, optional): Data type of the new tensor. Defaults to other's dtype.
        device (str, optional): Target device. Defaults to other's device.
        **kwargs: Additional keyword arguments forwarded to the constructor.

    Returns:
        NpuTensor: A new zero-filled tensor with the same shape.
    """
    dtype = dtype or other.dtype
    device = device or other.device
    t = cls(other.shape, dtype=dtype, device=device, **kwargs)
    t.data.fill(0)

    if device == "npu":
        t._sync_to_device()

    return t

set_current_device ¶

set_current_device(device: Device | None)

Set (or clear) the current device.

Parameters:

Name	Type	Description	Default
`device`	`Device \| None`	The device to set as current. Passing `None` clears the current selection (used by test teardown and to reset between designs), so a `Device \| None` from a resolver can be forwarded here directly.	required

Source code in python/utils/hostruntime/__init__.py

def set_current_device(device: "Device | None"):
    """Set (or clear) the current device.

    Args:
        device (Device | None): The device to set as current. Passing ``None``
            clears the current selection (used by test teardown and to reset
            between designs), so a ``Device | None`` from a resolver can be
            forwarded here directly.
    """
    global _CURRENT_DEVICE
    _CURRENT_DEVICE = device

bfloat16_safe_allclose ¶

bfloat16_safe_allclose(dtype, arr1, arr2)

Check if two arrays are element-wise equal within a tolerance, handling bfloat16 safely.

Parameters:

Name	Description	Default
`dtype`	The data type of the arrays.	required
`arr1`	First input array.	required
`arr2`	Second input array.	required

Returns:

Name	Type	Description
`bool`		True if the arrays are equal within tolerance, False otherwise.

Source code in python/utils/hostruntime/__init__.py

def bfloat16_safe_allclose(dtype, arr1, arr2):
    """Check if two arrays are element-wise equal within a tolerance, handling bfloat16 safely.

    Args:
        dtype: The data type of the arrays.
        arr1: First input array.
        arr2: Second input array.

    Returns:
        bool: True if the arrays are equal within tolerance, False otherwise.
    """
    if dtype == bfloat16:
        if isinstance(arr1, NpuTensor):
            arr1 = np.array(arr1, dtype=np.float16)
        else:
            arr1 = arr1.astype(np.float16)
        if isinstance(arr2, NpuTensor):
            arr2 = np.array(arr2, dtype=np.float16)
        else:
            arr2 = arr2.astype(np.float16)
    return np.allclose(arr1, arr2)

The host tensor: a shaped, typed view over a coherence-managed allocation.

The allocation itself, its residency bookkeeping, and the torch bridge live in :mod:buffer, :mod:coherence and :mod:torch_interop. Each is re-exported here, so from .tensor_class import Storage and friends keep resolving.

NpuTensor ¶

NpuTensor(
    shape_or_data, dtype: DTypeLike = uint32, device="npu"
)

Bases: ABC

A host-mapped, device-resident buffer of fixed shape and dtype.

This is a buffer with a residency state machine, not a general array. Its invariant is host/device coherence: the host and the device each hold a view of the same storage, and the two are reconciled only at the points this class defines. Everything else it offers (indexing, filling, the numpy and torch bridges, :meth:subview) exists to keep that reconciliation correct while still letting callers treat the buffer as data.

The invariant in full:

Writes through the declared paths (:meth:__setitem__, :meth:fill_, and the factories) are reconciled: the factories transfer as they construct, and the in-place writes record the region so the next :meth:to sends it. A write through the raw data array does neither, and is the one way to leave host and device disagreeing.
A method that spans regions in different states reconciles per region. Asking a whole tensor where it lives collapses a mixed extent to one answer, which is the right answer for :meth:to and the wrong one to transfer by.
:meth:to moves residency and is a no-op when the buffer is already on the target device, so a caller that has written through a declared path never pays for a redundant transfer, and a caller that has bypassed one gets no transfer at all.
Reconciliation is not byte-granular. It acts on whole cache lines, which is why :meth:subview requires its regions to be granule-aligned.

Subclasses supply the storage and the two transfer primitives; the invariant itself lives here so every backend states it the same way.

Named for the role rather than the mechanism: XRTTensor and HRXTensor name what implements the buffer, NpuTensor names what it is. Tensor remains as an alias for existing callers.

Initialize the tensor.

Parameters:

Name	Type	Description	Default
`shape_or_data`	`tuple or array - like`	If a tuple, creates a new tensor with the given shape and dtype. If array-like, wraps the data into a tensor with optional dtype casting.	required
`dtype`	`dtype`	Data type of the tensor. Defaults to np.uint32.	`uint32`
`device`	`str`	Device string identifier (e.g., 'npu', 'cpu'). Defaults to 'npu'.	`'npu'`

Source code in python/utils/hostruntime/tensor_class.py

def __init__(self, shape_or_data, dtype: npt.DTypeLike = np.uint32, device="npu"):
    """Initialize the tensor.

    Args:
        shape_or_data (tuple or array-like):
            - If a tuple, creates a new tensor with the given shape and dtype.
            - If array-like, wraps the data into a tensor with optional dtype casting.
        dtype (np.dtype, optional): Data type of the tensor. Defaults to np.uint32.
        device (str, optional): Device string identifier (e.g., 'npu', 'cpu'). Defaults to 'npu'.
    """
    if device not in self.__class__.DEVICES:
        raise ValueError(f"Unsupported device: {device}")
    self._initial_device = device
    self.dtype = dtype

storage `property` ¶

storage: Storage

The allocation this tensor is a view of.

A backend that still owns its bytes directly is wrapped on first use in a storage whose transport reconciles whole-extent, so every tensor has one whether or not its backend has adopted the split.

device `property` `writable` ¶

device

Which agent holds the authoritative copy of this tensor's bytes.

Read from the allocation, so two tensors over the same bytes can never disagree. A tensor spanning regions in different states reports cpu while any part of it still needs flushing, since that is the answer that keeps a caller's reconcile from being skipped.

base `property` ¶

base

The buffer that owns this one's storage, or None if it owns it itself.

Mirrors :attr:numpy.ndarray.base, including collapsing a chain of views: the base of a view of a view is the buffer that actually owns the storage, not the intermediate view. The intermediate is still referenced internally, so the whole chain stays alive for as long as any view of it does.

is_view `property` ¶

is_view

Whether this buffer shares another buffer's storage.

storage_offset `property` ¶

storage_offset

Where this buffer starts within :attr:base's storage, in bytes.

Zero for a buffer that owns its storage. Accumulated through nesting, so it is always measured from the owner rather than from the view this one was carved out of. Compare :meth:torch.Tensor.storage_offset, which is in elements; this is in bytes because a view may reinterpret the dtype.

data `abstractmethod` `property` ¶

data: ndarray

Subclasses must implement a data property.

Returns:

Type	Description
`ndarray`	np.ndarray: The underlying data of the tensor.

shape `abstractmethod` `property` ¶

shape: tuple[int, ...]

Subclasses must implement a shape property.

Returns:

Name	Type	Description
`tuple`	`tuple[int, ...]`	The shape of the tensor.

nbytes `cached` `property` ¶

nbytes: int

Number of bytes consumed by elements in the tensor.

element_size `cached` `property` ¶

element_size: int

Number of bytes per element.

to ¶

to(target_device: str)

Move the tensor to a specified target device.

Parameters:

Name	Type	Description	Default
`target_device`	`str`	The target device.	required

Returns:

Type	Description
	The tensor object on the target device.

Source code in python/utils/hostruntime/tensor_class.py

def to(self, target_device: str):
    """Move the tensor to a specified target device.

    Args:
        target_device (str): The target device.

    Returns:
       The tensor object on the target device.
    """
    coherence = self._coherence_ref or self._coherence()
    if coherence.uniform == target_device:
        return self
    start, end = self._extent
    if coherence.get(start, end) == target_device:
        # Already wholly where it is wanted: nothing to transfer and nothing
        # to record. A dispatch takes this path for every argument it does
        # not have to move, so it stays off the map entirely.
        return self
    if target_device == "npu":
        # Send the dirty ranges, not the whole extent. Ranges are coalesced,
        # so a buffer written end to end costs one transfer and a buffer
        # with a few dirty windows costs those windows.
        for lo, hi in coherence.ranges(start, end, _CoherenceMap.HOST):
            self.storage.sync_to_device(lo, hi - lo)
        coherence.set(start, end, "npu")
    elif target_device == "cpu":
        for lo, hi in coherence.ranges(start, end, _CoherenceMap.DEVICE):
            self.storage.sync_from_device(lo, hi - lo)
        coherence.set(start, end, "cpu")
    else:
        raise ValueError(f"Unknown device '{target_device}'")
    return self

mutate ¶

mutate()

Borrow this buffer's bytes for a host write.

Reconciles on entry so partial updates see current contents, records the write on exit, and retires the borrowed array so a reference kept past the block fails loudly instead of writing bytes nobody will flush.

The flush itself is deferred: the region is marked host-written and the next transfer to the device sends every dirty range in one pass, so a caller filling many windows of one buffer pays for one flush, not one per window.

with tensor.mutate() as buf:
    buf[:] = values

Source code in python/utils/hostruntime/tensor_class.py

def mutate(self):
    """Borrow this buffer's bytes for a host write.

    Reconciles on entry so partial updates see current contents, records the
    write on exit, and retires the borrowed array so a reference kept past
    the block fails loudly instead of writing bytes nobody will flush.

    The flush itself is deferred: the region is marked host-written and the
    next transfer to the device sends every dirty range in one pass, so a
    caller filling many windows of one buffer pays for one flush, not one
    per window.

        with tensor.mutate() as buf:
            buf[:] = values
    """
    return _WriteBorrow(self, reconcile=True)

overwrite ¶

overwrite()

Borrow for a write that replaces every byte of this tensor.

The same as :meth:mutate without the reconcile on entry, which nothing can observe if all of it is about to be replaced. Filling a buffer this way costs one transfer rather than two.

The caller is promising to write the whole region. Bytes left unwritten keep whatever the host last had there and are sent to the device with the rest, so use :meth:mutate for a partial update. This is the one promise here that cannot be checked; it is still narrower than reaching for data, which makes the same promise and does not record the write.

Source code in python/utils/hostruntime/tensor_class.py

def overwrite(self):
    """Borrow for a write that replaces every byte of this tensor.

    The same as :meth:`mutate` without the reconcile on entry, which nothing
    can observe if all of it is about to be replaced. Filling a buffer this
    way costs one transfer rather than two.

    The caller is promising to write the whole region. Bytes left unwritten
    keep whatever the host last had there and are sent to the device with
    the rest, so use :meth:`mutate` for a partial update. This is the one
    promise here that cannot be checked; it is still narrower than reaching
    for ``data``, which makes the same promise and does not record the write.
    """
    return _WriteBorrow(self, reconcile=False)

subview ¶

subview(offset, shape, dtype=None)