Python wrapper for OpenCL enabling GPU and parallel computing with comprehensive array operations and mathematical functions
86
Core OpenCL functionality for platform discovery, device selection, context creation, program compilation, and kernel execution. These components provide the foundation for all OpenCL operations and complete control over parallel execution.
Discover available OpenCL platforms and devices, with support for interactive selection and automatic device filtering.
def get_platforms():
"""
Get list of available OpenCL platforms.
Returns:
list[Platform]: Available platforms
Raises:
RuntimeError: If no platforms found or ICD loader issues
"""
def choose_devices(interactive=None, answers=None):
"""
Choose OpenCL devices interactively or programmatically.
Parameters:
- interactive (bool, optional): Enable interactive selection
- answers (list[str], optional): Pre-provided answers for selection
Returns:
list[Device]: Selected devices
"""
class Platform:
"""
Represents an OpenCL platform (driver implementation).
Attributes:
- name (str): Platform name
- vendor (str): Platform vendor
- version (str): OpenCL version supported
"""
def get_devices(self, device_type=None):
"""Get devices available on this platform."""
class Device:
"""
Represents an OpenCL compute device.
Attributes:
- name (str): Device name
- type (device_type): Device type (GPU, CPU, etc.)
- vendor (str): Device vendor
- max_compute_units (int): Number of compute units
- max_work_group_size (int): Maximum work group size
- local_mem_size (int): Local memory size in bytes
- global_mem_size (int): Global memory size in bytes
"""
@property
def platform(self):
"""Platform this device belongs to."""Create execution contexts and command queues for managing parallel operations and device communication.
def create_some_context(interactive=None, answers=None):
"""
Create an OpenCL context interactively or programmatically.
Parameters:
- interactive (bool, optional): Enable interactive device selection
- answers (list[str], optional): Pre-provided device selection answers
Returns:
Context: Created OpenCL context
"""
class Context:
"""
OpenCL execution context managing devices and memory objects.
Attributes:
- devices (list[Device]): Devices in this context
"""
def __init__(self, devices=None, properties=None, dev_type=None):
"""
Create context from devices or device type.
Parameters:
- devices (list[Device], optional): Specific devices to use
- properties (list, optional): Context properties
- dev_type (device_type, optional): Device type filter
"""
class CommandQueue:
"""
Command queue for scheduling operations on a device.
Attributes:
- context (Context): Associated context
- device (Device): Target device
"""
def __init__(self, context, device=None, properties=None):
"""
Create command queue for device operations.
Parameters:
- context (Context): OpenCL context
- device (Device, optional): Target device (uses first device if None)
- properties (command_queue_properties, optional): Queue properties
"""
def finish(self):
"""Wait for all enqueued commands to complete."""
def flush(self):
"""Submit all enqueued commands to device."""Compile OpenCL programs from source code with automatic caching and create executable kernels.
class Program:
"""
Compiled OpenCL program with automatic caching support.
"""
def __init__(self, context, source):
"""
Create program from OpenCL source code.
Parameters:
- context (Context): OpenCL context
- source (str): OpenCL source code
"""
def build(self, options=None, devices=None, cache_dir=None):
"""
Build program for devices.
Parameters:
- options (str | list[str], optional): Build options
- devices (list[Device], optional): Target devices
- cache_dir (str, optional): Cache directory for binaries
Returns:
Program: Self for method chaining
"""
def compile(self, options=None, devices=None, headers=None):
"""Compile program without linking."""
def get_info(self, param):
"""Get program information."""
def get_build_info(self, device, param):
"""Get build information for device."""
def all_kernels(self):
"""Get all kernels in program."""
def __getattr__(self, name):
"""Access kernel by name (e.g., prg.my_kernel)."""
def create_program_with_built_in_kernels(context, devices, kernel_names):
"""
Create program with built-in kernels.
Parameters:
- context (Context): OpenCL context
- devices (list[Device]): Target devices
- kernel_names (str | list[str]): Built-in kernel names
Returns:
Program: Program with built-in kernels
"""
def link_program(context, programs, options=None, devices=None):
"""
Link compiled programs into executable program.
Parameters:
- context (Context): OpenCL context
- programs (list[Program]): Compiled programs to link
- options (str | list[str], optional): Link options
- devices (list[Device], optional): Target devices
Returns:
Program: Linked executable program
"""
class Kernel:
"""
Executable OpenCL kernel function.
Attributes:
- function_name (str): Kernel function name
- num_args (int): Number of kernel arguments
- context (Context): Associated context
"""
def __init__(self, program, name):
"""
Create kernel from program.
Parameters:
- program (Program): Compiled program
- name (str): Kernel function name
"""
def get_info(self, param):
"""Get kernel information."""
def get_work_group_info(self, param, device):
"""Get work group information for device."""
def set_arg(self, index, arg):
"""Set kernel argument by index."""
def set_args(self, *args):
"""Set all kernel arguments."""
def __call__(self, queue, global_size, local_size, *args, **kwargs):
"""Execute kernel with arguments."""Execute kernels on devices with support for ND-range execution and event synchronization.
def enqueue_nd_range_kernel(queue, kernel, global_size, local_size=None,
global_offset=None, wait_for=None, g_times_l=False):
"""
Execute kernel over ND-range of work items.
Parameters:
- queue (CommandQueue): Command queue
- kernel (Kernel): Kernel to execute
- global_size (tuple[int, ...]): Global work size
- local_size (tuple[int, ...], optional): Local work group size
- global_offset (tuple[int, ...], optional): Global work offset
- wait_for (list[Event], optional): Events to wait for
- g_times_l (bool): Whether global_size is multiple of local_size
Returns:
Event: Kernel execution event
"""Manage asynchronous operations with events for synchronization and profiling.
class Event:
"""
Represents an OpenCL event for asynchronous operations.
Attributes:
- command_execution_status (int): Execution status
- command_type (int): Command type that created event
"""
def wait(self):
"""Wait for event completion."""
def get_info(self, param):
"""Get event information."""
def get_profiling_info(self, param):
"""Get profiling information."""
class NannyEvent(Event):
"""Event with automatic host memory management."""
class UserEvent(Event):
"""User-created event for custom synchronization."""
def __init__(self, context):
"""Create user event."""
def set_status(self, status):
"""Set event status."""
def wait_for_events(events):
"""
Wait for multiple events to complete.
Parameters:
- events (list[Event]): Events to wait for
"""Insert synchronization primitives into command queues.
def enqueue_marker(queue, wait_for=None):
"""
Insert marker into command queue.
Parameters:
- queue (CommandQueue): Command queue
- wait_for (list[Event], optional): Events to wait for
Returns:
Event: Marker event
"""
def enqueue_barrier(queue, wait_for=None):
"""
Insert barrier into command queue.
Parameters:
- queue (CommandQueue): Command queue
- wait_for (list[Event], optional): Events to wait for
Returns:
Event: Barrier event
"""Enable debugging and optimization for OpenCL development.
def enable_debugging(platform_or_context):
"""
Enable debugging for OpenCL platform or context.
Parameters:
- platform_or_context (Platform | Context): Target for debugging
"""import pyopencl as cl
# Interactive context creation
ctx = cl.create_some_context()
queue = cl.CommandQueue(ctx)
# Alternative: specific device selection
platforms = cl.get_platforms()
devices = platforms[0].get_devices()
ctx = cl.Context([devices[0]])
queue = cl.CommandQueue(ctx)import pyopencl as cl
import numpy as np
# Create context and queue
ctx = cl.create_some_context()
queue = cl.CommandQueue(ctx)
# OpenCL kernel source
kernel_source = """
__kernel void vector_add(__global const float* a,
__global const float* b,
__global float* result) {
int gid = get_global_id(0);
result[gid] = a[gid] + b[gid];
}
"""
# Compile program
program = cl.Program(ctx, kernel_source).build()
# Create kernel
kernel = program.vector_add
# Create buffers and execute
a = np.random.randn(1000).astype(np.float32)
b = np.random.randn(1000).astype(np.float32)
result = np.empty_like(a)
a_buf = cl.Buffer(ctx, cl.mem_flags.READ_ONLY | cl.mem_flags.COPY_HOST_PTR, hostbuf=a)
b_buf = cl.Buffer(ctx, cl.mem_flags.READ_ONLY | cl.mem_flags.COPY_HOST_PTR, hostbuf=b)
result_buf = cl.Buffer(ctx, cl.mem_flags.WRITE_ONLY, result.nbytes)
# Execute kernel
event = kernel(queue, a.shape, None, a_buf, b_buf, result_buf)
cl.enqueue_copy(queue, result, result_buf, wait_for=[event])
print(f"Result: {result[:5]}")Install with Tessl CLI
npx tessl i tessl/pypi-pyopencldocs
evals
scenario-1
scenario-2
scenario-3
scenario-4
scenario-5
scenario-6
scenario-7
scenario-8
scenario-9
scenario-10