Source code for holovec.backends.base

"""Abstract backend interface for holovec operations.

This module defines the abstract interface that all computational backends
must implement. Backends provide the underlying array operations for
different computation frameworks (NumPy, PyTorch, JAX, etc.).
"""

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any, List, Optional, Sequence, Tuple, Union

# Type alias for arrays (backend-specific)
Array = Any


[docs] class Backend(ABC): """Abstract base class for computational backends. All backends must implement these operations to support VSA computations across different frameworks (NumPy, PyTorch, JAX). """ @property @abstractmethod def name(self) -> str: """Return the backend name (e.g., 'numpy', 'torch', 'jax').""" pass
[docs] @abstractmethod def is_available(self) -> bool: """Check if the backend is available in the current environment.""" pass
# ===== Capability Probes =====
[docs] def supports_complex(self) -> bool: """Check if backend supports complex number operations. Complex operations are required for FHRR (Fourier HRR) and other frequency-domain VSA models. Returns: True if backend can handle complex dtypes (complex64, complex128) """ return True # Default: assume support (override in backend if needed)
[docs] def supports_sparse(self) -> bool: """Check if backend supports sparse array operations. Sparse operations are beneficial for BSC (Binary Spatter Codes) and BSDC (Binary Sparse Distributed Codes) which have high sparsity. Returns: True if backend has native sparse array support """ return False # Default: no sparse support (override if available)
[docs] def supports_gpu(self) -> bool: """Check if backend has GPU acceleration support. GPU support enables significant speedups for large-scale operations and is critical for production deployments. Returns: True if backend can utilize GPU hardware """ return False # Default: CPU only (override for PyTorch/JAX)
[docs] def supports_jit(self) -> bool: """Check if backend supports Just-In-Time (JIT) compilation. JIT compilation can provide 10-100x speedups for certain operations by compiling Python code to optimized machine code. Returns: True if backend has JIT compilation (e.g., JAX, Numba) """ return False # Default: no JIT (override for JAX)
[docs] def supports_device(self, device: str) -> bool: """Check if backend supports a specific device. Args: device: Device identifier (e.g., 'cpu', 'cuda', 'cuda:0', 'mps') Returns: True if the specified device is available Examples: >>> backend.supports_device('cpu') # Always True >>> backend.supports_device('cuda') # True if CUDA GPU available >>> backend.supports_device('mps') # True if Apple Metal available """ # Default: only CPU supported return device.lower() in ('cpu', 'cpu:0')
# ===== Array Creation =====
[docs] @abstractmethod def zeros(self, shape: Union[int, Tuple[int, ...]], dtype: str = 'float32') -> Array: """Create an array of zeros with the given shape and dtype.""" pass
[docs] @abstractmethod def ones(self, shape: Union[int, Tuple[int, ...]], dtype: str = 'float32') -> Array: """Create an array of ones with the given shape and dtype.""" pass
[docs] @abstractmethod def random_normal( self, shape: Union[int, Tuple[int, ...]], mean: float = 0.0, std: float = 1.0, dtype: str = 'float32', seed: Optional[int] = None ) -> Array: """Create an array of random values from a normal distribution.""" pass
[docs] @abstractmethod def random_uniform( self, shape: Union[int, Tuple[int, ...]], low: float = 0.0, high: float = 1.0, dtype: str = 'float32', seed: Optional[int] = None ) -> Array: """Create an array of random values from a uniform distribution.""" pass
[docs] @abstractmethod def random_binary( self, shape: Union[int, Tuple[int, ...]], p: float = 0.5, dtype: str = 'int32', seed: Optional[int] = None ) -> Array: """Create a binary array with probability p of being 1.""" pass
[docs] @abstractmethod def random_bipolar( self, shape: Union[int, Tuple[int, ...]], p: float = 0.5, dtype: str = 'float32', seed: Optional[int] = None ) -> Array: """Create a bipolar array {-1, +1} with probability p of being +1.""" pass
[docs] @abstractmethod def random_phasor( self, shape: Union[int, Tuple[int, ...]], dtype: str = 'complex64', seed: Optional[int] = None ) -> Array: """Create an array of random unit phasors (complex numbers with magnitude 1).""" pass
[docs] @abstractmethod def array(self, data: Any, dtype: Optional[str] = None) -> Array: """Create an array from Python data (list, tuple, etc.).""" pass
# ===== Element-wise Operations =====
[docs] @abstractmethod def multiply(self, a: Array, b: Array) -> Array: """Element-wise multiplication.""" pass
[docs] @abstractmethod def add(self, a: Array, b: Array) -> Array: """Element-wise addition.""" pass
[docs] @abstractmethod def subtract(self, a: Array, b: Array) -> Array: """Element-wise subtraction.""" pass
[docs] @abstractmethod def divide(self, a: Array, b: Array) -> Array: """Element-wise division.""" pass
[docs] @abstractmethod def xor(self, a: Array, b: Array) -> Array: """Element-wise XOR (for binary/bipolar).""" pass
[docs] @abstractmethod def conjugate(self, a: Array) -> Array: """Complex conjugate (for complex arrays).""" pass
[docs] @abstractmethod def exp(self, a: Array) -> Array: """Element-wise exponential: e^a. Args: a: Input array Returns: Array with exp applied element-wise """ pass
[docs] @abstractmethod def log(self, a: Array) -> Array: """Element-wise natural logarithm: ln(a). Args: a: Input array (must be positive) Returns: Array with log applied element-wise """ pass
# ===== Additional Element-wise Utilities =====
[docs] @abstractmethod def power(self, a: Array, exponent: float) -> Array: """Element-wise power: a**exponent.""" pass
[docs] @abstractmethod def angle(self, a: Array) -> Array: """Element-wise phase/angle for complex arrays (radians).""" pass
[docs] @abstractmethod def real(self, a: Array) -> Array: """Element-wise real part of (possibly complex) array.""" pass
[docs] @abstractmethod def imag(self, a: Array) -> Array: """Element-wise imaginary part of (possibly complex) array.""" pass
[docs] @abstractmethod def multiply_scalar(self, a: Array, scalar: float) -> Array: """Multiply array by a Python scalar.""" pass
[docs] @abstractmethod def linspace(self, start: float, stop: float, num: int) -> Array: """Create linearly spaced array of length num in [start, stop].""" pass
# ===== Reductions =====
[docs] @abstractmethod def sum(self, a: Array, axis: Optional[int] = None, keepdims: bool = False) -> Array: """Sum along an axis.""" pass
[docs] @abstractmethod def mean(self, a: Array, axis: Optional[int] = None, keepdims: bool = False) -> Array: """Mean along an axis.""" pass
[docs] @abstractmethod def norm(self, a: Array, ord: Union[int, str] = 2, axis: Optional[int] = None) -> Array: """Compute the norm of an array.""" pass
[docs] @abstractmethod def dot(self, a: Array, b: Array) -> Array: """Dot product of two vectors.""" pass
[docs] @abstractmethod def max(self, a: Array, axis: Optional[int] = None, keepdims: bool = False) -> Array: """Maximum value along an axis. Args: a: Input array axis: Axis along which to compute max (None for global max) keepdims: Whether to keep dimensions Returns: Maximum value(s) """ pass
[docs] @abstractmethod def min(self, a: Array, axis: Optional[int] = None, keepdims: bool = False) -> Array: """Minimum value along an axis. Args: a: Input array axis: Axis along which to compute min (None for global min) keepdims: Whether to keep dimensions Returns: Minimum value(s) """ pass
[docs] @abstractmethod def argmax(self, a: Array, axis: Optional[int] = None) -> Array: """Index of maximum value along an axis. Args: a: Input array axis: Axis along which to find argmax (None for global argmax) Returns: Index/indices of maximum value(s) """ pass
[docs] @abstractmethod def argmin(self, a: Array, axis: Optional[int] = None) -> Array: """Index of minimum value along an axis. Args: a: Input array axis: Axis along which to find argmin (None for global argmin) Returns: Index/indices of minimum value(s) """ pass
# ===== Normalization =====
[docs] @abstractmethod def normalize(self, a: Array, ord: Union[int, str] = 2, axis: Optional[int] = None, eps: float = 1e-12) -> Array: """Normalize an array to unit norm.""" pass
[docs] @abstractmethod def softmax(self, a: Array, axis: int = -1) -> Array: """Softmax function with numerical stability. Computes: softmax(x_i) = exp(x_i - max(x)) / Σ exp(x_j - max(x)) The max subtraction provides numerical stability by preventing overflow in the exponential function. Args: a: Input array axis: Axis along which to compute softmax Returns: Array with softmax applied along specified axis References: - Bricken & Pehlevan (2022): "Attention Approximates Sparse Distributed Memory" - Furlong & Eliasmith (2023): "Fractional binding in VSAs" """ pass
# ===== FFT Operations =====
[docs] @abstractmethod def fft(self, a: Array) -> Array: """1D Fast Fourier Transform.""" pass
[docs] @abstractmethod def ifft(self, a: Array) -> Array: """1D Inverse Fast Fourier Transform.""" pass
# ===== Circular Operations =====
[docs] @abstractmethod def circular_convolve(self, a: Array, b: Array) -> Array: """Circular convolution of two vectors.""" pass
[docs] @abstractmethod def circular_correlate(self, a: Array, b: Array) -> Array: """Circular correlation of two vectors.""" pass
# ===== Permutations =====
[docs] @abstractmethod def permute(self, a: Array, indices: Array) -> Array: """Permute array elements according to indices.""" pass
[docs] @abstractmethod def roll(self, a: Array, shift: int, axis: Optional[int] = None) -> Array: """Roll array elements along an axis.""" pass
# ===== Similarity Measures =====
[docs] @abstractmethod def cosine_similarity(self, a: Array, b: Array) -> float: """Compute cosine similarity between two vectors.""" pass
[docs] @abstractmethod def hamming_distance(self, a: Array, b: Array) -> float: """Compute Hamming distance between two binary/bipolar vectors.""" pass
[docs] @abstractmethod def euclidean_distance(self, a: Array, b: Array) -> float: """Compute Euclidean distance between two vectors.""" pass
# ===== Utilities =====
[docs] @abstractmethod def shape(self, a: Array) -> Tuple[int, ...]: """Return the shape of an array.""" pass
[docs] @abstractmethod def dtype(self, a: Array) -> str: """Return the dtype of an array as a string.""" pass
[docs] @abstractmethod def to_numpy(self, a: Array) -> Any: """Convert array to NumPy array (for compatibility).""" pass
[docs] @abstractmethod def from_numpy(self, a: Any) -> Array: """Create backend array from NumPy array.""" pass
[docs] @abstractmethod def clip(self, a: Array, min_val: float, max_val: float) -> Array: """Clip array values to [min_val, max_val].""" pass
[docs] @abstractmethod def abs(self, a: Array) -> Array: """Element-wise absolute value.""" pass
[docs] @abstractmethod def sign(self, a: Array) -> Array: """Element-wise sign.""" pass
[docs] @abstractmethod def threshold(self, a: Array, threshold: float, above: float = 1.0, below: float = 0.0) -> Array: """Threshold array values.""" pass
[docs] @abstractmethod def where(self, condition: Array, x: Array, y: Array) -> Array: """Select elements from x or y depending on boolean condition.""" pass
[docs] @abstractmethod def stack(self, arrays: Sequence[Array], axis: int = 0) -> Array: """Stack arrays along a new axis.""" pass
[docs] @abstractmethod def concatenate(self, arrays: Sequence[Array], axis: int = 0) -> Array: """Concatenate arrays along an existing axis.""" pass
# ===== Matrix Operations (for GHRR, VTB) =====
[docs] @abstractmethod def matmul(self, a: Array, b: Array) -> Array: """Matrix multiplication (or batched matrix multiplication). Args: a: Matrix or batch of matrices b: Matrix or batch of matrices Returns: Matrix product """ pass
[docs] @abstractmethod def matrix_transpose(self, a: Array) -> Array: """Transpose last two dimensions of array. For 2D: standard transpose For 3D+: transpose last two dimensions (batch transpose) Args: a: Array with at least 2 dimensions Returns: Transposed array """ pass
[docs] @abstractmethod def matrix_trace(self, a: Array) -> Array: """Compute trace of matrix or batch of matrices. For 2D array: returns scalar For 3D+ array: returns trace of each matrix in batch Args: a: Matrix or batch of matrices (last 2 dims are matrix) Returns: Scalar or array of traces """ pass
[docs] @abstractmethod def svd(self, a: Array, full_matrices: bool = True) -> Tuple[Array, Array, Array]: """Compute Singular Value Decomposition (SVD). Decomposes matrix A as A = U @ diag(S) @ Vh, where: - U: left singular vectors (unitary) - S: singular values (non-negative, sorted descending) - Vh: conjugate transpose of right singular vectors (unitary) For batched matrices (3D+), computes SVD for each matrix in batch. Args: a: Matrix or batch of matrices (shape [..., m, n]) full_matrices: If True, U and Vh have shapes [..., m, m] and [..., n, n]. If False, shapes are [..., m, k] and [..., k, n] where k=min(m,n). Returns: Tuple of (U, S, Vh) arrays Examples: >>> A = backend.random_normal((3, 3)) >>> U, S, Vh = backend.svd(A) >>> # Verify: A ≈ U @ diag(S) @ Vh """ pass
[docs] @abstractmethod def reshape(self, a: Array, shape: Tuple[int, ...]) -> Array: """Reshape array to new shape. Args: a: Array to reshape shape: Target shape Returns: Reshaped array """ pass
class BackendError(Exception): """Exception raised when a backend operation fails.""" pass class BackendNotAvailableError(BackendError): """Exception raised when a requested backend is not available.""" pass