use pyo3::prelude::*;
use super::dtype::{
Dtype, array2_to_py, complex_2d_owned, parse_dtype, real_1d_vec, real_2d_owned, vec1_to_py,
};
use super::spectrogram::PyScalar;
use crate::fft2d as rust_fft2d;
use crate::fft2d::Fft2dPlanner as RustFft2dPlanner;
use crate::image_ops;
#[pyfunction]
#[inline]
#[pyo3(signature = (data: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(data: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn fft2d(py: Python, data: &Bound<'_, PyAny>, dtype: Option<&str>) -> PyResult<Py<PyAny>> {
fn run<T>(py: Python<'_>, data: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>>
where
T: PyScalar,
num_complex::Complex<T>: numpy::Element,
{
let owned = real_2d_owned::<T>(py, data)?;
let result = py.detach(|| rust_fft2d::fft2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, data),
Dtype::F64 => run::<f64>(py, data),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (spectrum: "numpy.typing.NDArray[numpy.complex64]", output_ncols: "int", dtype: "str" = None), text_signature = "(spectrum: numpy.typing.NDArray[numpy.complex64], output_ncols: int, dtype: str = \"float64\")")]
pub fn ifft2d(
py: Python,
spectrum: &Bound<'_, PyAny>,
output_ncols: usize,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T>(py: Python<'_>, spectrum: &Bound<'_, PyAny>, output_ncols: usize) -> PyResult<Py<PyAny>>
where
T: PyScalar,
num_complex::Complex<T>: numpy::Element,
{
let owned = complex_2d_owned::<T>(py, spectrum)?;
let result = py.detach(|| rust_fft2d::ifft2d(&owned, output_ncols))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, spectrum, output_ncols),
Dtype::F64 => run::<f64>(py, spectrum, output_ncols),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (data: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(data: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn power_spectrum_2d(
py: Python,
data: &Bound<'_, PyAny>,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, data: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
let owned = real_2d_owned::<T>(py, data)?;
let result = py.detach(|| rust_fft2d::power_spectrum_2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, data),
Dtype::F64 => run::<f64>(py, data),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (data: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(data: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn magnitude_spectrum_2d(
py: Python,
data: &Bound<'_, PyAny>,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, data: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
let owned = real_2d_owned::<T>(py, data)?;
let result = py.detach(|| rust_fft2d::magnitude_spectrum_2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, data),
Dtype::F64 => run::<f64>(py, data),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (arr: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(arr: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn fftshift(py: Python, arr: &Bound<'_, PyAny>, dtype: Option<&str>) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, arr: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
let owned = real_2d_owned::<T>(py, arr)?;
Ok(array2_to_py(py, rust_fft2d::fftshift(owned)))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, arr),
Dtype::F64 => run::<f64>(py, arr),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (arr: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(arr: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn ifftshift(py: Python, arr: &Bound<'_, PyAny>, dtype: Option<&str>) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, arr: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
let owned = real_2d_owned::<T>(py, arr)?;
Ok(array2_to_py(py, rust_fft2d::ifftshift(owned)))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, arr),
Dtype::F64 => run::<f64>(py, arr),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (arr: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(arr: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn fftshift_1d(py: Python, arr: &Bound<'_, PyAny>, dtype: Option<&str>) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, arr: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
let v = real_1d_vec::<T>(py, arr)?;
Ok(vec1_to_py(py, rust_fft2d::fftshift_1d(v)))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, arr),
Dtype::F64 => run::<f64>(py, arr),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (arr: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(arr: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn ifftshift_1d(py: Python, arr: &Bound<'_, PyAny>, dtype: Option<&str>) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, arr: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
let v = real_1d_vec::<T>(py, arr)?;
Ok(vec1_to_py(py, rust_fft2d::ifftshift_1d(v)))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, arr),
Dtype::F64 => run::<f64>(py, arr),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (n, d = 1.0, dtype = None), text_signature = "(n: int, d: float = 1.0, dtype: str = \"float64\")")]
pub fn fftfreq(py: Python, n: usize, d: f64, dtype: Option<&str>) -> PyResult<Py<PyAny>> {
match parse_dtype(dtype)? {
Dtype::F32 => Ok(vec1_to_py(py, rust_fft2d::fftfreq::<f32>(n, d))),
Dtype::F64 => Ok(vec1_to_py(py, rust_fft2d::fftfreq::<f64>(n, d))),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (n, d = 1.0, dtype = None), text_signature = "(n: int, d: float = 1.0, dtype: str = \"float64\")")]
pub fn rfftfreq(py: Python, n: usize, d: f64, dtype: Option<&str>) -> PyResult<Py<PyAny>> {
match parse_dtype(dtype)? {
Dtype::F32 => Ok(vec1_to_py(py, rust_fft2d::rfftfreq::<f32>(n, d))),
Dtype::F64 => Ok(vec1_to_py(py, rust_fft2d::rfftfreq::<f64>(n, d))),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (size: "int", sigma: "float", dtype: "str" = None), text_signature = "(size: int, sigma: float, dtype: str = \"float64\")")]
pub fn gaussian_kernel_2d(
py: Python,
size: usize,
sigma: f64,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
let size = std::num::NonZeroUsize::new(size).ok_or_else(|| {
pyo3::exceptions::PyValueError::new_err("size must be a non-zero odd integer")
})?;
match parse_dtype(dtype)? {
Dtype::F32 => {
let result = py.detach(|| image_ops::gaussian_kernel_2d::<f32>(size, sigma))?;
Ok(array2_to_py(py, result))
}
Dtype::F64 => {
let result = py.detach(|| image_ops::gaussian_kernel_2d::<f64>(size, sigma))?;
Ok(array2_to_py(py, result))
}
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (image: "numpy.typing.NDArray[numpy.float64]", kernel: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(image: numpy.typing.NDArray[numpy.float64], kernel: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn convolve_fft(
py: Python,
image: &Bound<'_, PyAny>,
kernel: &Bound<'_, PyAny>,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(
py: Python<'_>,
image: &Bound<'_, PyAny>,
kernel: &Bound<'_, PyAny>,
) -> PyResult<Py<PyAny>> {
let image = real_2d_owned::<T>(py, image)?;
let kernel = real_2d_owned::<T>(py, kernel)?;
let result = py.detach(|| image_ops::convolve_fft(&image.view(), &kernel.view()))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, image, kernel),
Dtype::F64 => run::<f64>(py, image, kernel),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (image: "numpy.typing.NDArray[numpy.float64]", cutoff_fraction: "float", dtype: "str" = None), text_signature = "(image: numpy.typing.NDArray[numpy.float64], cutoff_fraction: float, dtype: str = \"float64\")")]
pub fn lowpass_filter(
py: Python,
image: &Bound<'_, PyAny>,
cutoff_fraction: f64,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(
py: Python<'_>,
image: &Bound<'_, PyAny>,
cutoff_fraction: f64,
) -> PyResult<Py<PyAny>> {
let image = real_2d_owned::<T>(py, image)?;
let result = py.detach(|| image_ops::lowpass_filter(&image.view(), cutoff_fraction))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, image, cutoff_fraction),
Dtype::F64 => run::<f64>(py, image, cutoff_fraction),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (image: "numpy.typing.NDArray[numpy.float64]", cutoff_fraction: "float", dtype: "str" = None), text_signature = "(image: numpy.typing.NDArray[numpy.float64], cutoff_fraction: float, dtype: str = \"float64\")")]
pub fn highpass_filter(
py: Python,
image: &Bound<'_, PyAny>,
cutoff_fraction: f64,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(
py: Python<'_>,
image: &Bound<'_, PyAny>,
cutoff_fraction: f64,
) -> PyResult<Py<PyAny>> {
let image = real_2d_owned::<T>(py, image)?;
let result = py.detach(|| image_ops::highpass_filter(&image.view(), cutoff_fraction))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, image, cutoff_fraction),
Dtype::F64 => run::<f64>(py, image, cutoff_fraction),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (image: "numpy.typing.NDArray[numpy.float64]", low_cutoff: "float", high_cutoff: "float", dtype: "str" = None), text_signature = "(image: numpy.typing.NDArray[numpy.float64], low_cutoff: float, high_cutoff: float, dtype: str = \"float64\")")]
pub fn bandpass_filter(
py: Python,
image: &Bound<'_, PyAny>,
low_cutoff: f64,
high_cutoff: f64,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(
py: Python<'_>,
image: &Bound<'_, PyAny>,
low_cutoff: f64,
high_cutoff: f64,
) -> PyResult<Py<PyAny>> {
let image = real_2d_owned::<T>(py, image)?;
let result =
py.detach(|| image_ops::bandpass_filter(&image.view(), low_cutoff, high_cutoff))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, image, low_cutoff, high_cutoff),
Dtype::F64 => run::<f64>(py, image, low_cutoff, high_cutoff),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (image: "numpy.typing.NDArray[numpy.float64]", dtype: "str" = None), text_signature = "(image: numpy.typing.NDArray[numpy.float64], dtype: str = \"float64\")")]
pub fn detect_edges_fft(
py: Python,
image: &Bound<'_, PyAny>,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, image: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
let image = real_2d_owned::<T>(py, image)?;
let result = py.detach(|| image_ops::detect_edges_fft(&image.view()))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, image),
Dtype::F64 => run::<f64>(py, image),
}
}
#[pyfunction]
#[inline]
#[pyo3(signature = (image: "numpy.typing.NDArray[numpy.float64]", amount: "float", dtype: "str" = None), text_signature = "(image: numpy.typing.NDArray[numpy.float64], amount: float, dtype: str = \"float64\")")]
pub fn sharpen_fft(
py: Python,
image: &Bound<'_, PyAny>,
amount: f64,
dtype: Option<&str>,
) -> PyResult<Py<PyAny>> {
fn run<T: PyScalar>(py: Python<'_>, image: &Bound<'_, PyAny>, amount: f64) -> PyResult<Py<PyAny>> {
let image = real_2d_owned::<T>(py, image)?;
let result = py.detach(|| image_ops::sharpen_fft(&image.view(), amount))?;
Ok(array2_to_py(py, result))
}
match parse_dtype(dtype)? {
Dtype::F32 => run::<f32>(py, image, amount),
Dtype::F64 => run::<f64>(py, image, amount),
}
}
enum PlannerInner {
F32(RustFft2dPlanner<f32>),
F64(RustFft2dPlanner<f64>),
}
#[pyclass(name = "Fft2dPlanner", skip_from_py_object)]
pub struct PyFft2dPlanner {
inner: PlannerInner,
dtype: Dtype,
}
#[pymethods]
impl PyFft2dPlanner {
#[new]
#[pyo3(signature = (dtype = None), text_signature = "(dtype: str = \"float64\")")]
fn new(dtype: Option<&str>) -> PyResult<Self> {
let d = parse_dtype(dtype)?;
let inner = match d {
Dtype::F32 => PlannerInner::F32(RustFft2dPlanner::<f32>::new()),
Dtype::F64 => PlannerInner::F64(RustFft2dPlanner::<f64>::new()),
};
Ok(Self { inner, dtype: d })
}
#[getter]
const fn dtype(&self) -> &'static str {
match self.dtype {
Dtype::F32 => "float32",
Dtype::F64 => "float64",
}
}
#[pyo3(signature = (data: "numpy.typing.NDArray[numpy.float64]"), text_signature = "(data: numpy.typing.NDArray[numpy.float64])")]
fn fft2d(&mut self, py: Python, data: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
match &mut self.inner {
PlannerInner::F32(p) => {
let owned = real_2d_owned::<f32>(py, data)?;
let result = py.detach(|| p.fft2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
PlannerInner::F64(p) => {
let owned = real_2d_owned::<f64>(py, data)?;
let result = py.detach(|| p.fft2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
}
}
#[pyo3(signature = (spectrum: "numpy.typing.NDArray[numpy.complex128]", output_ncols: "int"), text_signature = "(spectrum: numpy.typing.NDArray[numpy.complex128], output_ncols: int)")]
fn ifft2d(
&mut self,
py: Python,
spectrum: &Bound<'_, PyAny>,
output_ncols: usize,
) -> PyResult<Py<PyAny>> {
match &mut self.inner {
PlannerInner::F32(p) => {
let owned = complex_2d_owned::<f32>(py, spectrum)?;
let result = py.detach(|| p.ifft2d(&owned.view(), output_ncols))?;
Ok(array2_to_py(py, result))
}
PlannerInner::F64(p) => {
let owned = complex_2d_owned::<f64>(py, spectrum)?;
let result = py.detach(|| p.ifft2d(&owned.view(), output_ncols))?;
Ok(array2_to_py(py, result))
}
}
}
#[pyo3(signature = (data: "numpy.typing.NDArray[numpy.float64]"), text_signature = "(data: numpy.typing.NDArray[numpy.float64])")]
fn power_spectrum_2d(&mut self, py: Python, data: &Bound<'_, PyAny>) -> PyResult<Py<PyAny>> {
match &mut self.inner {
PlannerInner::F32(p) => {
let owned = real_2d_owned::<f32>(py, data)?;
let result = py.detach(|| p.power_spectrum_2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
PlannerInner::F64(p) => {
let owned = real_2d_owned::<f64>(py, data)?;
let result = py.detach(|| p.power_spectrum_2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
}
}
#[pyo3(signature = (data: "numpy.typing.NDArray[numpy.float64]"), text_signature = "(data: numpy.typing.NDArray[numpy.float64])")]
fn magnitude_spectrum_2d(
&mut self,
py: Python,
data: &Bound<'_, PyAny>,
) -> PyResult<Py<PyAny>> {
match &mut self.inner {
PlannerInner::F32(p) => {
let owned = real_2d_owned::<f32>(py, data)?;
let result = py.detach(|| p.magnitude_spectrum_2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
PlannerInner::F64(p) => {
let owned = real_2d_owned::<f64>(py, data)?;
let result = py.detach(|| p.magnitude_spectrum_2d(&owned.view()))?;
Ok(array2_to_py(py, result))
}
}
}
}
pub fn register(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_function(wrap_pyfunction!(fft2d, m)?)?;
m.add_function(wrap_pyfunction!(ifft2d, m)?)?;
m.add_function(wrap_pyfunction!(power_spectrum_2d, m)?)?;
m.add_function(wrap_pyfunction!(magnitude_spectrum_2d, m)?)?;
m.add_function(wrap_pyfunction!(fftshift, m)?)?;
m.add_function(wrap_pyfunction!(ifftshift, m)?)?;
m.add_function(wrap_pyfunction!(fftshift_1d, m)?)?;
m.add_function(wrap_pyfunction!(ifftshift_1d, m)?)?;
m.add_function(wrap_pyfunction!(fftfreq, m)?)?;
m.add_function(wrap_pyfunction!(rfftfreq, m)?)?;
m.add_function(wrap_pyfunction!(gaussian_kernel_2d, m)?)?;
m.add_function(wrap_pyfunction!(convolve_fft, m)?)?;
m.add_function(wrap_pyfunction!(lowpass_filter, m)?)?;
m.add_function(wrap_pyfunction!(highpass_filter, m)?)?;
m.add_function(wrap_pyfunction!(bandpass_filter, m)?)?;
m.add_function(wrap_pyfunction!(detect_edges_fft, m)?)?;
m.add_function(wrap_pyfunction!(sharpen_fft, m)?)?;
m.add_class::<PyFft2dPlanner>()?;
Ok(())
}