A pythonic wrapper around FFTW, the FFT library, presenting a unified interface for all the supported transforms.
—
Global configuration settings, utility functions for finding optimal transform sizes, and system information access. These utilities help optimize performance and provide information about the pyFFTW installation.
Functions to find optimal sizes and parameters for FFT operations.
def next_fast_len(target):
"""
Find the next efficient length for FFT transforms.
FFTW is most efficient for sizes that have small prime factors.
This function finds the smallest length >= target that gives
good performance.
Parameters:
- target: int - Target transform length
Returns:
- int: Optimal length >= target for efficient transforms
"""Global configuration values that affect pyFFTW behavior.
# Optimal SIMD alignment for the current CPU (in bytes)
simd_alignment: int
# Default number of threads for multi-threaded operations
NUM_THREADS: int
# Default planner effort level
PLANNER_EFFORT: str # Usually 'FFTW_ESTIMATE'Constants providing information about supported types and FFTW version.
# PyFFTW package version string (e.g., '0.15.0')
__version__: str
# FFTW library version string (e.g., '3.3.10')
fftw_version: str
# FFTW version as tuple (e.g., (3, 3, 10))
fftw_version_tuple: tuple
# Compiler flags used to build FFTW
fftw_cc: str
# Threading type ('OMP', 'pthreads', or None)
_threading_type: str
# Supported precision types
_supported_types: list # ['32', '64', 'ld']
# Supported numpy complex types
_supported_nptypes_complex: list
# Supported numpy real types
_supported_nptypes_real: list
# Human-readable type names
_all_types_human_readable: list
# Numpy type mappings
_all_types_np: list
# FFTW version information dictionary
_fftw_version_dict: dict
# FFTW compiler information dictionary
_fftw_cc_dict: dictThe pyfftw.config module provides runtime configuration management.
# Configuration variables (from pyfftw.config)
NUM_THREADS: int # Number of threads (from env PYFFTW_NUM_THREADS or OMP_NUM_THREADS)
PLANNER_EFFORT: str # Planner effort (from env PYFFTW_PLANNER_EFFORT)import pyfftw
import numpy as np
# Find optimal sizes for various targets
targets = [100, 500, 1000, 1500, 2000]
print("Target -> Optimal (overhead)")
for target in targets:
optimal = pyfftw.next_fast_len(target)
overhead = (optimal - target) / target * 100
print(f"{target:4d} -> {optimal:4d} ({overhead:5.1f}%)")
# Use optimal size for better performance
target_size = 1000
optimal_size = pyfftw.next_fast_len(target_size)
# Create data with optimal size
data = np.random.randn(optimal_size) + 1j * np.random.randn(optimal_size)
result = pyfftw.empty_aligned(optimal_size, dtype='complex128')
# This will be more efficient than using target_size directly
fft_obj = pyfftw.FFTW(data, result)import pyfftw
# Display system and version information
print("=== pyFFTW System Information ===")
print(f"FFTW Version: {pyfftw.fftw_version}")
print(f"FFTW Version Tuple: {pyfftw.fftw_version_tuple}")
print(f"FFTW Compiler: {pyfftw.fftw_cc}")
print(f"Threading Type: {pyfftw._threading_type}")
print(f"SIMD Alignment: {pyfftw.simd_alignment} bytes")
print(f"\nSupported Types: {pyfftw._supported_types}")
print(f"Complex Types: {pyfftw._supported_nptypes_complex}")
print(f"Real Types: {pyfftw._supported_nptypes_real}")
# Check configuration
import pyfftw.config
print(f"\n=== Configuration ===")
print(f"Default Threads: {pyfftw.config.NUM_THREADS}")
print(f"Default Planner Effort: {pyfftw.config.PLANNER_EFFORT}")import os
import pyfftw.config
# Configuration can be controlled via environment variables
print("Current configuration:")
print(f"NUM_THREADS: {pyfftw.config.NUM_THREADS}")
print(f"PLANNER_EFFORT: {pyfftw.config.PLANNER_EFFORT}")
# Example of setting environment variables
# (these would typically be set before importing pyfftw)
os.environ['PYFFTW_NUM_THREADS'] = '8'
os.environ['PYFFTW_PLANNER_EFFORT'] = 'FFTW_PATIENT'
# Reload configuration (normally not needed in real applications)
pyfftw.config._reload_config()
print("\nAfter setting environment variables:")
print(f"NUM_THREADS: {pyfftw.config.NUM_THREADS}")
print(f"PLANNER_EFFORT: {pyfftw.config.PLANNER_EFFORT}")import pyfftw
import numpy as np
import time
def benchmark_sizes(base_size, num_variations=10):
"""Benchmark different sizes around a base size."""
# Test the base size and optimal size
optimal_size = pyfftw.next_fast_len(base_size)
sizes_to_test = [base_size, optimal_size]
# Add some variations
for i in range(1, num_variations):
test_size = base_size + i * 10
sizes_to_test.append(test_size)
sizes_to_test.append(pyfftw.next_fast_len(test_size))
# Remove duplicates and sort
sizes_to_test = sorted(list(set(sizes_to_test)))
results = {}
for size in sizes_to_test:
# Create aligned arrays
data = pyfftw.empty_aligned(size, dtype='complex128')
result = pyfftw.empty_aligned(size, dtype='complex128')
# Fill with random data
data[:] = np.random.randn(size) + 1j * np.random.randn(size)
# Create FFTW object
fft_obj = pyfftw.FFTW(data, result, flags=('FFTW_MEASURE',))
# Benchmark execution time
num_runs = max(10, 1000000 // size) # Adjust runs based on size
start = time.time()
for _ in range(num_runs):
fft_obj()
elapsed = time.time() - start
avg_time = elapsed / num_runs * 1e6 # microseconds
results[size] = avg_time
# Display results
print(f"Size benchmarks (base size: {base_size}, optimal: {optimal_size})")
print("Size Time (μs) Relative")
print("-" * 30)
base_time = results[base_size]
for size in sorted(results.keys()):
time_us = results[size]
relative = time_us / base_time
marker = " (optimal)" if size == optimal_size else ""
print(f"{size:5d} {time_us:8.2f} {relative:6.2f}{marker}")
# Run benchmark
benchmark_sizes(1000)import pyfftw
import numpy as np
def check_type_support():
"""Check which data types are supported."""
test_size = 64
# Test different data types
types_to_test = [
('float32', np.float32),
('float64', np.float64),
('longdouble', np.longdouble),
('complex64', np.complex64),
('complex128', np.complex128),
('clongdouble', np.clongdouble if hasattr(np, 'clongdouble') else None)
]
print("Data Type Support:")
print("Type Supported Size (bytes)")
print("-" * 40)
for type_name, numpy_type in types_to_test:
if numpy_type is None:
print(f"{type_name:12s} No -")
continue
try:
# Try to create arrays and FFTW object
if 'complex' in type_name:
input_array = pyfftw.empty_aligned(test_size, dtype=numpy_type)
output_array = pyfftw.empty_aligned(test_size, dtype=numpy_type)
else:
# For real types, output is complex
input_array = pyfftw.empty_aligned(test_size, dtype=numpy_type)
if numpy_type == np.float32:
output_dtype = np.complex64
elif numpy_type == np.longdouble:
output_dtype = np.clongdouble if hasattr(np, 'clongdouble') else np.complex128
else:
output_dtype = np.complex128
output_array = pyfftw.empty_aligned(test_size//2 + 1, dtype=output_dtype)
# Try to create FFTW object
fft_obj = pyfftw.FFTW(input_array, output_array)
supported = "Yes"
type_size = numpy_type().itemsize
except Exception as e:
supported = "No"
type_size = numpy_type().itemsize if hasattr(numpy_type(), 'itemsize') else 0
print(f"{type_name:12s} {supported:3s} {type_size}")
check_type_support()import pyfftw
import pyfftw.config
import threading
class PyFFTWConfig:
"""Custom configuration manager for pyFFTW applications."""
def __init__(self):
self._lock = threading.Lock()
self.load_defaults()
def load_defaults(self):
"""Load default configuration values."""
self.num_threads = pyfftw.config.NUM_THREADS
self.planner_effort = pyfftw.config.PLANNER_EFFORT
self.simd_alignment = pyfftw.simd_alignment
self.auto_align = True
self.auto_contiguous = True
def get_fft_kwargs(self):
"""Get standard kwargs for FFTW functions."""
with self._lock:
return {
'threads': self.num_threads,
'planner_effort': self.planner_effort,
'auto_align_input': self.auto_align,
'auto_contiguous': self.auto_contiguous
}
def set_performance_mode(self, mode='balanced'):
"""Set predefined performance modes."""
with self._lock:
if mode == 'speed':
self.planner_effort = 'FFTW_ESTIMATE'
self.num_threads = pyfftw.config.NUM_THREADS
elif mode == 'balanced':
self.planner_effort = 'FFTW_MEASURE'
self.num_threads = min(4, pyfftw.config.NUM_THREADS)
elif mode == 'quality':
self.planner_effort = 'FFTW_PATIENT'
self.num_threads = 1 # Single thread for reproducibility
else:
raise ValueError(f"Unknown mode: {mode}")
def create_fft_object(self, input_array, output_array, **kwargs):
"""Create FFTW object with current configuration."""
config_kwargs = self.get_fft_kwargs()
config_kwargs.update(kwargs) # Allow override
return pyfftw.FFTW(input_array, output_array, **config_kwargs)
# Usage example
config = PyFFTWConfig()
# Set different performance modes
config.set_performance_mode('quality')
# Create FFT object with configuration
data = pyfftw.empty_aligned(1024, dtype='complex128')
result = pyfftw.empty_aligned(1024, dtype='complex128')
fft_obj = config.create_fft_object(data, result)Install with Tessl CLI
npx tessl i tessl/pypi-pyfftw