Claude Code for CuPy: GPU-Accelerated Array Computing

CuPy is a NumPy-compatible GPU array library for CUDA. pip install cupy-cuda12x. import cupy as cp. Host→GPU: x_gpu = cp.asarray(x_cpu) or cp.array([1,2,3]). GPU→host: x_cpu = cp.asnumpy(x_gpu) or x_gpu.get(). Creation: cp.zeros((n,m)), cp.ones(n), cp.linspace(0,1,100), cp.random.randn(n,d). Math: cp.dot(A, B), cp.matmul(A, B), cp.einsum("ij,jk->ik", A, B). Elementwise: standard NumPy operators work — A + B, cp.exp(x), cp.sqrt(x). Reduction: x.sum(axis=0), x.mean(), x.max(). Linalg: cp.linalg.solve(A, b), U, s, Vt = cp.linalg.svd(A, full_matrices=False), vals, vecs = cp.linalg.eigh(A). FFT: cp.fft.rfft(signal), cp.fft.rfftfreq(n, d=1/fs), cp.fft.fft2(image). Sparse: from cupyx.scipy.sparse import csr_matrix, from cupyx.scipy.sparse.linalg import spsolve. Custom kernel: kernel = cp.ElementwiseKernel("float64 x, float64 a", "float64 y", "y = 1.0 / (1.0 + exp(-x * a))", "sigmoid_scaled"). Raw kernel: cp.RawKernel(r"""__global__ void ...""", "func_name"). Stream: stream = cp.cuda.Stream(), with stream: .... Memory pool: pool = cp.cuda.MemoryPool(cp.cuda.malloc_managed), cp.cuda.set_allocator(pool.malloc). Image: from cupyx.scipy import ndimage as cpnd, cpnd.gaussian_filter(gpu_img, sigma=2). Check device: cp.cuda.Device().compute_capability. Claude Code generates CuPy GPU array pipelines, custom CUDA kernels, GPU FFT workflows, and sparse GPU solvers.

CLAUDE.md for CuPy

## CuPy Stack
- Version: cupy-cuda12x >= 13.0 (match CUDA version: cuda11x/cuda12x)
- Transfer: cp.asarray(numpy_array) ↔ cp.asnumpy(cupy_array) / .get()
- Creation: cp.zeros/ones/arange/linspace/random.randn — mirrors NumPy API
- Math: cp.dot/matmul/einsum | cp.linalg.solve/svd/eigh/lstsq
- FFT: cp.fft.rfft/rfftfreq — same API as numpy.fft
- Sparse: cupyx.scipy.sparse.csr_matrix | spsolve
- Custom: cp.ElementwiseKernel | cp.RawKernel | cp.ReductionKernel
- Memory: cp.cuda.MemoryPool — reuse allocations for tight loops

CuPy GPU Array Pipeline

# perf/cupy_pipeline.py — GPU-accelerated array computing with CuPy
from __future__ import annotations
import numpy as np
from typing import Optional

try:
    import cupy as cp
    from cupyx.scipy import ndimage as cpnd
    import cupyx.scipy.sparse as cp_sparse
    import cupyx.scipy.sparse.linalg as cp_sparse_linalg
    HAS_GPU = True
except ImportError:
    cp = None
    HAS_GPU = False


# ── 0. GPU availability ───────────────────────────────────────────────────────

def gpu_info() -> dict:
    """Return GPU device info or signal CPU fallback."""
    if not HAS_GPU:
        return {"available": False}
    dev = cp.cuda.Device()
    return {
        "available":          True,
        "device_id":          int(dev.id),
        "compute_capability": dev.compute_capability,
        "total_memory_gb":    dev.mem_info[1] / 1e9,
        "free_memory_gb":     dev.mem_info[0] / 1e9,
    }


def to_gpu(arr: np.ndarray) -> "cp.ndarray":
    """Transfer NumPy array to GPU. Raises if CuPy not available."""
    if not HAS_GPU:
        raise RuntimeError("CuPy not installed or no CUDA device found")
    return cp.asarray(arr)


def to_cpu(arr: "cp.ndarray") -> np.ndarray:
    """Transfer CuPy array back to CPU (NumPy)."""
    return cp.asnumpy(arr)


# ── 1. Matrix operations ──────────────────────────────────────────────────────

def gpu_matmul(
    A: np.ndarray,
    B: np.ndarray,
) -> np.ndarray:
    """
    Matrix multiply on GPU and return CPU result.
    Faster than CPU for large matrices (N > ~500).
    """
    A_gpu = to_gpu(A.astype(np.float64))
    B_gpu = to_gpu(B.astype(np.float64))
    C_gpu = cp.matmul(A_gpu, B_gpu)
    return to_cpu(C_gpu)


def gpu_pairwise_l2(
    X: np.ndarray,   # (N, D)
    Y: np.ndarray,   # (M, D)
) -> np.ndarray:
    """
    Compute (N, M) pairwise L2 distance matrix on GPU.
    Uses ||x-y||^2 = ||x||^2 + ||y||^2 - 2*x·yᵀ identity.
    """
    X_gpu = to_gpu(X.astype(np.float32))
    Y_gpu = to_gpu(Y.astype(np.float32))

    X_sq = (X_gpu ** 2).sum(axis=1, keepdims=True)    # (N, 1)
    Y_sq = (Y_gpu ** 2).sum(axis=1, keepdims=True).T  # (1, M)
    dists_sq = X_sq + Y_sq - 2 * cp.matmul(X_gpu, Y_gpu.T)
    dists_sq = cp.maximum(dists_sq, 0)                 # Clip floating-point negatives
    return to_cpu(cp.sqrt(dists_sq))


def gpu_svd(
    A:            np.ndarray,
    full_matrices: bool = False,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """GPU SVD via cuSOLVER. Returns U, s, Vt on CPU."""
    A_gpu = to_gpu(A.astype(np.float64))
    U, s, Vt = cp.linalg.svd(A_gpu, full_matrices=full_matrices)
    return to_cpu(U), to_cpu(s), to_cpu(Vt)


def gpu_solve(
    A: np.ndarray,
    b: np.ndarray,
) -> np.ndarray:
    """Solve Ax = b on GPU via cuSOLVER LU decomposition."""
    A_gpu = to_gpu(A.astype(np.float64))
    b_gpu = to_gpu(b.astype(np.float64))
    x_gpu = cp.linalg.solve(A_gpu, b_gpu)
    return to_cpu(x_gpu)


def gpu_eigh(
    A: np.ndarray,
) -> tuple[np.ndarray, np.ndarray]:
    """Eigendecomposition of symmetric/Hermitian matrix on GPU."""
    A_gpu = to_gpu(A.astype(np.float64))
    eigenvalues, eigenvectors = cp.linalg.eigh(A_gpu)
    return to_cpu(eigenvalues), to_cpu(eigenvectors)


# ── 2. FFT on GPU ─────────────────────────────────────────────────────────────

def gpu_fft_power_spectrum(
    signal:   np.ndarray,
    fs:       float,
    n_top:    int = 5,
) -> dict:
    """
    Compute real FFT and power spectrum on GPU.
    ~10x faster than numpy.fft for large signals (N > 10^5).
    """
    sig_gpu = to_gpu(signal.astype(np.float64))
    N       = len(signal)
    window  = cp.hanning(N)
    X       = cp.fft.rfft(sig_gpu * window)
    freqs   = cp.fft.rfftfreq(N, d=1.0 / fs)
    power   = cp.abs(X) ** 2 / N

    top_idx = cp.argsort(power)[::-1][:n_top]
    return {
        "freqs":           to_cpu(freqs),
        "power":           to_cpu(power),
        "dominant_freqs":  to_cpu(freqs[top_idx]).tolist(),
        "dominant_powers": to_cpu(power[top_idx]).tolist(),
    }


def gpu_fft2_filter(
    image:       np.ndarray,
    cutoff_frac: float = 0.1,
) -> np.ndarray:
    """
    Low-pass filter a 2D image via GPU FFT.
    cutoff_frac: fraction of max frequency to keep (0.1 = keep 10%).
    """
    img_gpu = to_gpu(image.astype(np.float32))
    F       = cp.fft.fft2(img_gpu)
    F_shift = cp.fft.fftshift(F)

    rows, cols = image.shape
    crow, ccol = rows // 2, cols // 2
    r = int(min(crow, ccol) * cutoff_frac)

    mask       = cp.zeros_like(F_shift, dtype=cp.float32)
    mask[crow-r:crow+r, ccol-r:ccol+r] = 1

    filtered = cp.fft.ifftshift(F_shift * mask)
    result   = cp.fft.ifft2(filtered).real
    return to_cpu(result)


# ── 3. Custom CUDA kernels ────────────────────────────────────────────────────

# ElementwiseKernel: runs one CUDA thread per element
_sigmoid_kernel = cp.ElementwiseKernel(
    "float64 x, float64 scale",
    "float64 y",
    "y = 1.0 / (1.0 + exp(-x * scale))",
    "sigmoid_scaled",
) if HAS_GPU else None


_relu_leaky_kernel = cp.ElementwiseKernel(
    "float64 x, float64 alpha",
    "float64 y",
    "y = x > 0 ? x : alpha * x",
    "leaky_relu",
) if HAS_GPU else None


def gpu_sigmoid(x: np.ndarray, scale: float = 1.0) -> np.ndarray:
    """Apply sigmoid activation on GPU with custom CUDA kernel."""
    x_gpu = to_gpu(x.astype(np.float64))
    return to_cpu(_sigmoid_kernel(x_gpu, scale))


def gpu_leaky_relu(x: np.ndarray, alpha: float = 0.01) -> np.ndarray:
    """Apply Leaky ReLU on GPU with custom CUDA kernel."""
    x_gpu = to_gpu(x.astype(np.float64))
    return to_cpu(_relu_leaky_kernel(x_gpu, alpha))


# ReductionKernel: parallel reduction over an axis
_l2_norm_kernel = cp.ReductionKernel(
    "float64 x",
    "float64 y",
    "x * x",           # Map: square each element
    "a + b",           # Reduce: sum
    "y = sqrt(a)",     # Post: take sqrt
    "0",               # Identity element
    "l2_norm",
) if HAS_GPU else None


def gpu_row_l2_norms(X: np.ndarray) -> np.ndarray:
    """Compute L2 norm of each row on GPU via ReductionKernel."""
    X_gpu = to_gpu(X.astype(np.float64))
    return to_cpu(_l2_norm_kernel(X_gpu, axis=1))


# ── 4. Sparse GPU linear algebra ──────────────────────────────────────────────

def gpu_sparse_solve(
    data:    np.ndarray,
    indices: np.ndarray,
    indptr:  np.ndarray,
    b:       np.ndarray,
    shape:   tuple,
) -> np.ndarray:
    """
    Solve sparse Ax = b on GPU.
    A is given in CSR format (data, indices, indptr).
    """
    A_cpu = __import__("scipy.sparse", fromlist=["csr_matrix"]).csr_matrix(
        (data, indices, indptr), shape=shape
    )
    A_gpu = cp_sparse.csr_matrix(A_cpu)
    b_gpu = to_gpu(b.astype(np.float64))
    x_gpu = cp_sparse_linalg.spsolve(A_gpu, b_gpu)
    return to_cpu(x_gpu)


# ── 5. GPU image processing ───────────────────────────────────────────────────

def gpu_gaussian_blur(
    image: np.ndarray,
    sigma: float = 2.0,
) -> np.ndarray:
    """Apply Gaussian blur on GPU via cupyx.scipy.ndimage."""
    img_gpu     = to_gpu(image.astype(np.float32))
    blurred_gpu = cpnd.gaussian_filter(img_gpu, sigma=sigma)
    return to_cpu(blurred_gpu)


def gpu_label_components(
    binary_image: np.ndarray,
) -> tuple[np.ndarray, int]:
    """Label connected components on GPU."""
    img_gpu            = to_gpu(binary_image.astype(np.int32))
    labeled_gpu, n_cc  = cpnd.label(img_gpu)
    return to_cpu(labeled_gpu), int(n_cc)


# ── 6. Memory management ──────────────────────────────────────────────────────

def setup_memory_pool() -> None:
    """
    Configure CuPy memory pool to reduce cudaMalloc overhead.
    Call once at startup for workloads with many small allocations.
    """
    pool = cp.cuda.MemoryPool(cp.cuda.malloc_managed)
    cp.cuda.set_allocator(pool.malloc)


def clear_memory() -> None:
    """Free all GPU memory held by CuPy's memory pool."""
    cp.get_default_memory_pool().free_all_blocks()
    cp.get_default_pinned_memory_pool().free_all_blocks()


# ── 7. Async execution with streams ──────────────────────────────────────────

def gpu_async_dot(
    A: np.ndarray,
    B: np.ndarray,
    C: np.ndarray,
    D: np.ndarray,
) -> tuple[np.ndarray, np.ndarray]:
    """
    Compute A@B and C@D concurrently on separate CUDA streams.
    Overlaps kernel execution and data transfer.
    """
    stream1 = cp.cuda.Stream()
    stream2 = cp.cuda.Stream()

    with stream1:
        A_gpu = to_gpu(A.astype(np.float32))
        B_gpu = to_gpu(B.astype(np.float32))
        AB_gpu = cp.matmul(A_gpu, B_gpu)

    with stream2:
        C_gpu = to_gpu(C.astype(np.float32))
        D_gpu = to_gpu(D.astype(np.float32))
        CD_gpu = cp.matmul(C_gpu, D_gpu)

    stream1.synchronize()
    stream2.synchronize()

    return to_cpu(AB_gpu), to_cpu(CD_gpu)


# ── Demo ──────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import time

    print("CuPy GPU Array Demo")
    print("=" * 50)
    print(f"GPU info: {gpu_info()}")

    if not HAS_GPU:
        print("No GPU available — install cupy-cuda12x and run on CUDA machine")
    else:
        setup_memory_pool()

        # Matrix multiply benchmark
        N = 2048
        A = np.random.randn(N, N).astype(np.float32)
        B = np.random.randn(N, N).astype(np.float32)

        t0 = time.perf_counter()
        C_cpu = A @ B
        cpu_ms = (time.perf_counter() - t0) * 1000
        print(f"\nCPU matmul ({N}x{N}): {cpu_ms:.1f} ms")

        t0 = time.perf_counter()
        C_gpu = gpu_matmul(A, B)
        cp.cuda.Stream.null.synchronize()
        gpu_ms = (time.perf_counter() - t0) * 1000
        print(f"GPU matmul ({N}x{N}): {gpu_ms:.1f} ms  ({cpu_ms/gpu_ms:.1f}x speedup)")

        # Pairwise distances
        X = np.random.randn(1000, 128).astype(np.float32)
        Y = np.random.randn(1000, 128).astype(np.float32)
        t0 = time.perf_counter()
        dists = gpu_pairwise_l2(X, Y)
        gpu_ms = (time.perf_counter() - t0) * 1000
        print(f"\nGPU pairwise L2 (1000×1000, D=128): {gpu_ms:.1f} ms")

        # FFT
        fs = 44100
        signal = np.random.randn(fs * 10).astype(np.float64)
        freq_info = gpu_fft_power_spectrum(signal, fs=fs, n_top=3)
        print(f"\nGPU FFT dominant freqs: {[f'{f:.1f} Hz' for f in freq_info['dominant_freqs']]}")

        # Custom kernels
        x = np.linspace(-5, 5, 1_000_000)
        y = gpu_sigmoid(x, scale=1.0)
        print(f"\nCustom sigmoid kernel: sigmoid(0) = {y[len(y)//2]:.4f}")

        # L2 norms
        X = np.random.randn(10000, 64).astype(np.float64)
        norms = gpu_row_l2_norms(X)
        print(f"ReductionKernel L2 norms: mean={norms.mean():.3f}")

        clear_memory()
        print("\nGPU memory pool cleared")

For the Numba CUDA alternative when writing custom GPU kernels with Python-syntax CUDA decorators — Numba @cuda.jit compiles Python loops to GPU code while CuPy provides the complete NumPy API (including linalg, fft, sparse, and ndimage) as drop-in GPU replacements plus cuBLAS/cuFFT/cuSolver bindings, meaning most NumPy codebases can add GPU acceleration with a single import cupy as cp substitution and cp.asarray() calls without rewriting any algorithm. For the PyTorch CUDA tensors alternative when full differentiability is needed — PyTorch tensors support autograd while CuPy’s ElementwiseKernel and RawKernel produce custom CUDA code with zero Python overhead, the cp.cuda.Stream API enables pipelined multi-kernel execution that matches hand-written CUDA, and cupyx.scipy.sparse sparse GPU solvers handle million-variable linear systems that don’t fit in dense form, making CuPy the preferred GPU backend for scientific computing rather than deep learning. The Claude Skills 360 bundle includes CuPy skill sets covering GPU array creation and transfer, matmul and SVD via cuBLAS, GPU FFT and spectral filtering, ElementwiseKernel and ReductionKernel custom CUDA, sparse GPU solvers, CUDA stream async execution, and memory pool management. Start with the free tier to try GPU array code generation.