FFT Computation Module

FFT Computation Module

Core FFT computation functionality moved from old_fft_module.py and main.py. Provides low-level FFT calculations without user interface elements.

class mmpp.fft.compute_fft.FFTComputeConfig(window_function='hann', filter_type='remove_mean', fft_engine='auto', zero_padding=True, nfft=None)[source]

Bases: object

Configuration for FFT computations.

Parameters:

window_function (Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian'])
filter_type (Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static'])
fft_engine (Literal['numpy', 'pyfftw', 'scipy', 'auto'])
zero_padding (bool)
nfft (Optional[int])

window_function: Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian'] = 'hann'

filter_type: Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static'] = 'remove_mean'

fft_engine: Literal['numpy', 'pyfftw', 'scipy', 'auto'] = 'auto'

zero_padding: bool = True

nfft: Optional[int] = None

__post_init__()[source]

Validate configuration.

Return type:: None

__init__(window_function='hann', filter_type='remove_mean', fft_engine='auto', zero_padding=True, nfft=None)

Parameters:

window_function (Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian'])
filter_type (Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static'])
fft_engine (Literal['numpy', 'pyfftw', 'scipy', 'auto'])
zero_padding (bool)
nfft (Optional[int])

class mmpp.fft.compute_fft.FFTComputeResult(frequencies, spectrum, metadata, config)[source]

Bases: object

Result of FFT computation.

Parameters:

frequencies (ndarray)
spectrum (ndarray)
metadata (dict[str, Any])
config (FFTComputeConfig)

frequencies: ndarray

spectrum: ndarray

metadata: dict[str, Any]

config: FFTComputeConfig

property peak_frequency: float: Get frequency with maximum power.

save_to_zarr(zarr_path, dataset_name='fft', force=False)[source]

Save FFT result to zarr file.

Return type:: None

Parameters:

zarr_pathstr: Path to zarr file
dataset_namestr, optional: Base dataset name (default: “fft”)
forcebool, optional: Overwrite existing data (default: False)

param zarr_path:
type zarr_path:: str
param dataset_name:
type dataset_name:: str
param force:
type force:: bool

__init__(frequencies, spectrum, metadata, config)

Parameters:

frequencies (ndarray)
spectrum (ndarray)
metadata (dict[str, Any])
config (FFTComputeConfig)

class mmpp.fft.compute_fft.FFTCompute(debug=False)[source]

Bases: object

Core FFT computation engine.

Handles low-level FFT calculations without user interface elements.

Parameters:: debug (bool)

__init__(debug=False)[source]

Initialize FFT compute engine.

Parameters:

debugbool, optional: Enable debug logging (default: False)

param debug:
type debug:: bool

determine_engine(data_size)[source]

Determine best FFT engine based on data size.

Return type:: str

Parameters:

data_sizeint: Total number of elements to transform

Returns:

: str

Selected engine name

param data_size:
type data_size:: int

apply_window(data, window_type)[source]

Apply window function to data.

Return type:: ndarray

Parameters:

datanp.ndarray: Input data (time axis first)
window_typestr: Window function type

Returns:

: np.ndarray

Windowed data

param data:
type data:: ndarray
param window_type:
type window_type:: Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian']

apply_filter(data, filter_type)[source]

Apply filtering to data.

Return type:: ndarray

Parameters:

datanp.ndarray: Input data (time axis first)
filter_typestr: Filter type

Returns:

: np.ndarray

Filtered data

param data:
type data:: ndarray
param filter_type:
type filter_type:: Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static']

compute_fft(data, dt, engine)[source]

Compute FFT using specified engine.

Return type:: tuple[ndarray, ndarray]

Parameters:

datanp.ndarray: Input data (time axis first)
dtfloat: Time step
enginestr: FFT engine to use

Returns:

: tuple

(frequencies, fft_data)

param data:
type data:: ndarray
param dt:
type dt:: float
param engine:
type engine:: str

calculate_fft_method1(data, dt, window='hann', filter_type='remove_mean', engine=None)[source]

FFT Method 1: Apply filtering and windowing, then FFT, then average spatially.

Return type:: FFTComputeResult

Parameters:

datanp.ndarray: Input data (time, …, components)
dtfloat: Time step
windowstr: Window type
filter_typestr: Filter type
enginestr, optional: FFT engine

Returns:

: FFTComputeResult

FFT computation result

param data:
type data:: ndarray
param dt:
type dt:: float
param window:
type window:: Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian']
param filter_type:
type filter_type:: Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static']
param engine:
type engine:: Optional[str]

calculate_fft_method2(data, dt, window='hann', filter_type='remove_mean', engine=None)[source]

FFT Method 2: Apply filtering, average spatially, then windowing and FFT.

Return type:: FFTComputeResult

Parameters:

datanp.ndarray: Input data (time, …, components)
dtfloat: Time step
windowstr: Window type
filter_typestr: Filter type
enginestr, optional: FFT engine

Returns:

: FFTComputeResult

FFT computation result

param data:
type data:: ndarray
param dt:
type dt:: float
param window:
type window:: Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian']
param filter_type:
type filter_type:: Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static']
param engine:
type engine:: Optional[str]

load_data_from_zarr(zarr_path, dataset, z_layer=-1)[source]

Load data from zarr file.

Return type:: tuple[ndarray, float]

Parameters:

zarr_pathstr: Path to zarr file
datasetstr: Dataset name
z_layerint: Z-layer index (-1 for last layer)

Returns:

: tuple

(data, dt) where data is the loaded array and dt is time step

param zarr_path:
type zarr_path:: str
param dataset:
type dataset:: str
param z_layer:
type z_layer:: int

get_available_options()[source]

Get available configuration options.

Return type:: dict[str, Any]

load_existing_fft_data(zarr_path, dataset_name='fft')[source]

Load existing FFT data from zarr file.

Return type:: Optional[FFTComputeResult]

Parameters:

zarr_pathstr: Path to zarr file
dataset_namestr, optional: Dataset name (default: “fft”)

Returns:

: Optional[FFTComputeResult]

Loaded FFT result or None if not found

param zarr_path:
type zarr_path:: str
param dataset_name:
type dataset_name:: str

calculate_fft_data(zarr_path, dataset, z_layer=-1, method=1, save=False, force=False, save_dataset_name=None, **kwargs)[source]

Calculate FFT for data from zarr file.

Return type:: FFTComputeResult

Parameters:

zarr_pathstr: Path to zarr file
datasetstr: Dataset name
z_layerint: Z-layer index (-1 for last layer)
methodint: FFT method (1 or 2)
savebool, optional: Save result to zarr file (default: False)
forcebool, optional: Force recalculation and overwrite existing (default: False)
save_dataset_namestr, optional: Custom name for saved dataset (default: auto-generated)
**kwargsAny: Additional FFT configuration options

Returns:

: FFTComputeResult

FFT computation result

param zarr_path:
type zarr_path:: str
param dataset:
type dataset:: str
param z_layer:
type z_layer:: int
param method:
type method:: int
param save:
type save:: bool
param force:
type force:: bool
param save_dataset_name:
type save_dataset_name:: Optional[str]

Main Classes

FFTCompute

class mmpp.fft.compute_fft.FFTCompute(debug=False)[source]

Bases: object

Core FFT computation engine.

Handles low-level FFT calculations without user interface elements.

Parameters:: debug (bool)

__init__(debug=False)[source]

Initialize FFT compute engine.

Parameters:

debugbool, optional: Enable debug logging (default: False)

param debug:
type debug:: bool

determine_engine(data_size)[source]

Determine best FFT engine based on data size.

Return type:: str

Parameters:

data_sizeint: Total number of elements to transform

Returns:

: str

Selected engine name

param data_size:
type data_size:: int

apply_window(data, window_type)[source]

Apply window function to data.

Return type:: ndarray

Parameters:

datanp.ndarray: Input data (time axis first)
window_typestr: Window function type

Returns:

: np.ndarray

Windowed data

param data:
type data:: ndarray
param window_type:
type window_type:: Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian']

apply_filter(data, filter_type)[source]

Apply filtering to data.

Return type:: ndarray

Parameters:

datanp.ndarray: Input data (time axis first)
filter_typestr: Filter type

Returns:

: np.ndarray

Filtered data

param data:
type data:: ndarray
param filter_type:
type filter_type:: Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static']

compute_fft(data, dt, engine)[source]

Compute FFT using specified engine.

Return type:: tuple[ndarray, ndarray]

Parameters:

datanp.ndarray: Input data (time axis first)
dtfloat: Time step
enginestr: FFT engine to use

Returns:

: tuple

(frequencies, fft_data)

param data:
type data:: ndarray
param dt:
type dt:: float
param engine:
type engine:: str

calculate_fft_method1(data, dt, window='hann', filter_type='remove_mean', engine=None)[source]

FFT Method 1: Apply filtering and windowing, then FFT, then average spatially.

Return type:: FFTComputeResult

Parameters:

datanp.ndarray: Input data (time, …, components)
dtfloat: Time step
windowstr: Window type
filter_typestr: Filter type
enginestr, optional: FFT engine

Returns:

: FFTComputeResult

FFT computation result

param data:
type data:: ndarray
param dt:
type dt:: float
param window:
type window:: Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian']
param filter_type:
type filter_type:: Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static']
param engine:
type engine:: Optional[str]

calculate_fft_method2(data, dt, window='hann', filter_type='remove_mean', engine=None)[source]

FFT Method 2: Apply filtering, average spatially, then windowing and FFT.

Return type:: FFTComputeResult

Parameters:

datanp.ndarray: Input data (time, …, components)
dtfloat: Time step
windowstr: Window type
filter_typestr: Filter type
enginestr, optional: FFT engine

Returns:

: FFTComputeResult

FFT computation result

param data:
type data:: ndarray
param dt:
type dt:: float
param window:
type window:: Literal['none', 'hann', 'hamming', 'blackman', 'bartlett', 'kaiser', 'tukey', 'gaussian']
param filter_type:
type filter_type:: Literal['none', 'remove_mean', 'remove_static', 'detrend_linear', 'remove_mean_and_static']
param engine:
type engine:: Optional[str]

load_data_from_zarr(zarr_path, dataset, z_layer=-1)[source]

Load data from zarr file.

Return type:: tuple[ndarray, float]

Parameters:

zarr_pathstr: Path to zarr file
datasetstr: Dataset name
z_layerint: Z-layer index (-1 for last layer)

Returns:

: tuple

(data, dt) where data is the loaded array and dt is time step

param zarr_path:
type zarr_path:: str
param dataset:
type dataset:: str
param z_layer:
type z_layer:: int

get_available_options()[source]

Get available configuration options.

Return type:: dict[str, Any]

load_existing_fft_data(zarr_path, dataset_name='fft')[source]

Load existing FFT data from zarr file.

Return type:: Optional[FFTComputeResult]

Parameters:

zarr_pathstr: Path to zarr file
dataset_namestr, optional: Dataset name (default: “fft”)

Returns:

: Optional[FFTComputeResult]

Loaded FFT result or None if not found

param zarr_path:
type zarr_path:: str
param dataset_name:
type dataset_name:: str

calculate_fft_data(zarr_path, dataset, z_layer=-1, method=1, save=False, force=False, save_dataset_name=None, **kwargs)[source]

Calculate FFT for data from zarr file.

Return type:: FFTComputeResult

Parameters:

zarr_pathstr: Path to zarr file
datasetstr: Dataset name
z_layerint: Z-layer index (-1 for last layer)
methodint: FFT method (1 or 2)
savebool, optional: Save result to zarr file (default: False)
forcebool, optional: Force recalculation and overwrite existing (default: False)
save_dataset_namestr, optional: Custom name for saved dataset (default: auto-generated)
**kwargsAny: Additional FFT configuration options

Returns:

: FFTComputeResult

FFT computation result

param zarr_path:
type zarr_path:: str
param dataset:
type dataset:: str
param z_layer:
type z_layer:: int
param method:
type method:: int
param save:
type save:: bool
param force:
type force:: bool
param save_dataset_name:
type save_dataset_name:: Optional[str]

FFTComputeResult

class mmpp.fft.compute_fft.FFTComputeResult(frequencies, spectrum, metadata, config)[source]

Bases: object

Result of FFT computation.

Parameters:

frequencies (ndarray)
spectrum (ndarray)
metadata (dict[str, Any])
config (FFTComputeConfig)

frequencies: ndarray

spectrum: ndarray

metadata: dict[str, Any]

config: FFTComputeConfig

property peak_frequency: float: Get frequency with maximum power.

save_to_zarr(zarr_path, dataset_name='fft', force=False)[source]

Save FFT result to zarr file.

Return type:: None

Parameters:

zarr_pathstr: Path to zarr file
dataset_namestr, optional: Base dataset name (default: “fft”)
forcebool, optional: Overwrite existing data (default: False)

param zarr_path:
type zarr_path:: str
param dataset_name:
type dataset_name:: str
param force:
type force:: bool

__init__(frequencies, spectrum, metadata, config)

Parameters:

frequencies (ndarray)
spectrum (ndarray)
metadata (dict[str, Any])
config (FFTComputeConfig)

FFTConfig

Usage Examples

Basic FFT Computation

import mmpp

# Setup FFT computation
db = mmpp.open('/path/to/data')
result = db.find(solver=3)[0]

# Compute FFT with custom parameters
fft_result = result.fft.spectrum(
    dset='m_z11',
    z_layer=-1,
    method=1,
    save=True,
    force=False
)

# Access computed data
frequencies = result.fft.frequencies()
power_spectrum = result.fft.power()
magnitude = result.fft.magnitude()
phase = result.fft.phase()

Advanced Configuration

# Custom FFT configuration
from mmpp.fft.compute_fft import FFTConfig

config = FFTConfig(
    window_function='hann',
    fft_engine='scipy',
    zero_padding=True,
    nfft=1024
)

# Use configuration
fft_result = result.fft._compute_fft(
    dataset_name='m_z11',
    z_layer=-1,
    method=1,
    config=config
)