use num_complex::Complex;
use rustfft::{Fft, FftPlanner};
use std::str::FromStr;
use std::sync::Arc;
#[derive(Debug, Clone)]
pub enum FftMode {
TwoSided,
Centered,
OneSided,
OneSided2X,
}
impl FromStr for FftMode {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"twosided" => Ok(FftMode::TwoSided),
"centered" => Ok(FftMode::Centered),
"onesided" => Ok(FftMode::OneSided),
"onesided2X" => Ok(FftMode::OneSided2X),
_ => Err(format!("Unknown fft_mode: {}", s)),
}
}
}
impl FftMode {
pub fn is_onesided(&self) -> bool {
matches!(self, FftMode::OneSided | FftMode::OneSided2X)
}
}
pub struct StandaloneSTFT {
win: Vec<Complex<f64>>,
hop: usize,
fs: f64,
fft_mode: FftMode,
mfft: usize,
phase_shift: i32,
dual_win: Option<Vec<Complex<f64>>>,
forward_fft: Option<Arc<dyn Fft<f64>>>,
inverse_fft: Option<Arc<dyn Fft<f64>>>,
}
impl StandaloneSTFT {
pub fn new(
win: Vec<f64>,
hop: usize,
fs: f64,
fft_mode: Option<&str>,
mfft: Option<usize>,
dual_win: Option<Vec<f64>>,
phase_shift: Option<i32>,
) -> Result<Self, String> {
if win.is_empty() {
return Err("Parameter win must be non-empty!".to_string());
}
if !win.iter().all(|&x| x.is_finite()) {
return Err("Parameter win must have finite entries!".to_string());
}
if hop < 1 {
return Err(format!("Parameter hop={} is not >= 1!", hop));
}
if fs <= 0.0 || fs.is_nan() {
return Err(format!("Sampling frequency fs={} must be positive!", fs));
}
let win_complex: Vec<Complex<f64>> =
win.into_iter().map(|x| Complex::new(x, 0.0)).collect();
let fft_mode = fft_mode.unwrap_or("onesided").parse()?;
let mfft = mfft.unwrap_or(win_complex.len());
let phase_shift = phase_shift.unwrap_or(0);
if mfft < win_complex.len() {
return Err(format!(
"mfft={} must be at least the window length {}!",
mfft,
win_complex.len()
));
}
if !(-(mfft as i64) < phase_shift as i64 && (phase_shift as i64) < mfft as i64) {
return Err(format!(
"-mfft < phase_shift < mfft does not hold for mfft={}, phase_shift={}!",
mfft, phase_shift
));
}
let dual_win_complex = if let Some(dw) = dual_win {
if dw.len() != win_complex.len() {
return Err("dual_win.len() must equal win.len()!".to_string());
}
if !dw.iter().all(|&x| x.is_finite()) {
return Err("Parameter dual_win must have finite entries!".to_string());
}
Some(dw.into_iter().map(|x| Complex::new(x, 0.0)).collect())
} else {
None
};
let mut planner = FftPlanner::new();
let forward_fft = Some(planner.plan_fft_forward(mfft));
let inverse_fft = Some(planner.plan_fft_inverse(mfft));
Ok(StandaloneSTFT {
win: win_complex,
hop,
fs,
fft_mode,
mfft,
phase_shift,
dual_win: dual_win_complex,
forward_fft,
inverse_fft,
})
}
pub fn win(&self) -> &[Complex<f64>] {
&self.win
}
pub fn hop(&self) -> usize {
self.hop
}
pub fn fs(&self) -> f64 {
self.fs
}
pub fn fft_mode(&self) -> &FftMode {
&self.fft_mode
}
pub fn mfft(&self) -> usize {
self.mfft
}
pub fn phase_shift(&self) -> i32 {
self.phase_shift
}
pub fn sampling_period(&self) -> f64 {
1.0 / self.fs
}
pub fn delta_t(&self) -> f64 {
self.hop as f64 * self.sampling_period()
}
pub fn delta_f(&self) -> f64 {
1.0 / (self.mfft as f64 * self.sampling_period())
}
pub fn m_num(&self) -> usize {
self.win.len()
}
pub fn m_num_mid(&self) -> usize {
self.m_num() / 2
}
pub fn f_pts(&self) -> usize {
if self.fft_mode.is_onesided() {
self.mfft / 2 + 1
} else {
self.mfft
}
}
pub fn onesided_fft(&self) -> bool {
self.fft_mode.is_onesided()
}
fn calc_dual_canonical_window(&self) -> Result<Vec<Complex<f64>>, String> {
let win = &self.win;
let hop = self.hop;
if hop > win.len() {
return Err(format!(
"hop={} is larger than window length of {} => STFT not invertible!",
hop,
win.len()
));
}
let w2: Vec<f64> = win.iter().map(|w| w.norm_sqr()).collect();
let mut dd = w2.clone();
for k in (hop..win.len()).step_by(hop) {
for i in k..win.len() {
dd[i] += w2[i - k];
}
for i in 0..(win.len() - k) {
dd[i] += w2[i + k];
}
}
let max_dd = dd.iter().fold(0.0f64, |a, &b| a.max(b));
let relative_resolution = f64::EPSILON * max_dd;
if !dd.iter().all(|&x| x >= relative_resolution) {
return Err("Short-time Fourier Transform not invertible!".to_string());
}
Ok(win.iter().zip(dd.iter()).map(|(w, &d)| w / d).collect())
}
pub fn dual_win(&mut self) -> Result<&[Complex<f64>], String> {
if self.dual_win.is_none() {
self.dual_win = Some(self.calc_dual_canonical_window()?);
}
Ok(self.dual_win.as_ref().unwrap())
}
pub fn invertible(&mut self) -> bool {
self.dual_win().is_ok()
}
fn fft_func(&self, mut x: Vec<Complex<f64>>) -> Vec<Complex<f64>> {
if x.len() < self.mfft {
x.resize(self.mfft, Complex::new(0.0, 0.0));
}
let p_s =
(self.phase_shift + self.m_num_mid() as i32).rem_euclid(self.m_num() as i32) as usize;
x.rotate_left(p_s);
if x.len() != self.mfft {
x.resize(self.mfft, Complex::new(0.0, 0.0));
}
match &self.fft_mode {
FftMode::TwoSided => {
if let Some(ref fft) = self.forward_fft {
fft.process(&mut x);
}
x
}
FftMode::Centered => {
if let Some(ref fft) = self.forward_fft {
fft.process(&mut x);
}
self.fftshift(x)
}
FftMode::OneSided => {
if let Some(ref fft) = self.forward_fft {
fft.process(&mut x);
}
let n_out = self.mfft / 2 + 1;
x.truncate(n_out);
if n_out > 0 {
x[0] = Complex::new(x[0].re, 0.0); }
if self.mfft.is_multiple_of(2) && n_out > 1 {
x[n_out - 1] = Complex::new(x[n_out - 1].re, 0.0); }
x
}
FftMode::OneSided2X => {
if let Some(ref fft) = self.forward_fft {
fft.process(&mut x);
}
let n_out = self.mfft / 2 + 1;
x.truncate(n_out);
let fac = 2.0; if self.mfft.is_multiple_of(2) {
let len = x.len();
for x_i in x.iter_mut().skip(1).take(len - 2) {
*x_i *= fac;
}
} else {
for x_i in x.iter_mut().skip(1) {
*x_i *= fac;
}
}
x
}
}
}
fn ifft_func(&self, mut x: Vec<Complex<f64>>) -> Vec<Complex<f64>> {
match &self.fft_mode {
FftMode::TwoSided => {
if let Some(ref ifft) = self.inverse_fft {
ifft.process(&mut x);
}
}
FftMode::Centered => {
x = self.ifftshift(x);
if let Some(ref ifft) = self.inverse_fft {
ifft.process(&mut x);
}
}
FftMode::OneSided => {
let mut full_x = vec![Complex::new(0.0, 0.0); self.mfft];
full_x[..x.len()].copy_from_slice(&x);
let mirror_end = if self.mfft.is_multiple_of(2) {
(self.mfft / 2).min(x.len() - 1)
} else {
(self.mfft / 2).min(x.len() - 1)
};
for i in 1..=mirror_end {
if i < x.len() && (self.mfft - i) < full_x.len() {
full_x[self.mfft - i] = x[i].conj();
}
}
if let Some(ref ifft) = self.inverse_fft {
ifft.process(&mut full_x);
}
x = full_x;
}
FftMode::OneSided2X => {
let mut xc = x.clone();
let fac = 2.0;
if self.mfft.is_multiple_of(2) {
let len = xc.len();
for xc_i in xc.iter_mut().skip(1).take(len - 2) {
*xc_i /= fac;
}
} else {
for xc_i in xc.iter_mut().skip(1) {
*xc_i /= fac;
}
}
let mut full_x = vec![Complex::new(0.0, 0.0); self.mfft];
full_x[..xc.len()].copy_from_slice(&xc);
let mirror_end = (self.mfft / 2).min(xc.len() - 1);
for i in 1..=mirror_end {
if i < xc.len() && (self.mfft - i) < full_x.len() {
full_x[self.mfft - i] = xc[i].conj();
}
}
if let Some(ref ifft) = self.inverse_fft {
ifft.process(&mut full_x);
}
x = full_x;
}
}
let normalization_factor = 1.0 / (self.mfft as f64);
for val in &mut x {
*val *= normalization_factor;
}
let p_s =
(self.phase_shift + self.m_num_mid() as i32).rem_euclid(self.m_num() as i32) as usize;
x.rotate_right(p_s);
x.truncate(self.m_num());
x
}
fn fftshift(&self, mut x: Vec<Complex<f64>>) -> Vec<Complex<f64>> {
let n = x.len();
let mid = n.div_ceil(2);
x.rotate_left(mid);
x
}
fn ifftshift(&self, mut x: Vec<Complex<f64>>) -> Vec<Complex<f64>> {
let n = x.len();
let mid = n / 2;
x.rotate_left(mid);
x
}
fn pre_padding(&self) -> (i32, i32) {
let w2: Vec<f64> = self.win.iter().map(|w| w.norm_sqr()).collect();
let n0 = -(self.m_num_mid() as i32);
let mut p = 0;
let mut n = n0;
loop {
let n_next = n - self.hop as i32;
if n_next + self.m_num() as i32 <= 0 {
return (n, -p);
}
let suffix_start = if n_next < 0 {
self.m_num() as i32 + n_next
} else {
n_next
};
if suffix_start >= 0
&& (suffix_start as usize) < w2.len()
&& w2[(suffix_start as usize)..].iter().all(|&x| x == 0.0)
{
return (n, -p);
}
n = n_next;
p += 1;
if n < n0 - self.m_num() as i32 {
break;
}
}
(
n0 - self.m_num() as i32,
-((self.m_num() / self.hop + 1) as i32),
)
}
fn post_padding(&self, n: usize) -> (i32, i32) {
let m2p = self.m_num() - self.m_num_mid();
if n < m2p {
panic!("Parameter n must be >= ceil(m_num/2) = {}!", m2p);
}
let w2: Vec<f64> = self.win.iter().map(|w| w.norm_sqr()).collect();
let q1 = n / self.hop;
let k1 = q1 as i32 * self.hop as i32 - self.m_num_mid() as i32;
for (q_offset, k) in (k1..(n as i32 + self.m_num() as i32))
.step_by(self.hop)
.enumerate()
{
let q = q1 + q_offset;
let n_next = k + self.hop as i32;
if n_next >= n as i32 || w2[..(n as i32 - n_next) as usize].iter().all(|&x| x == 0.0) {
return (k + self.m_num() as i32, q as i32 + 1);
}
}
(
n as i32 + self.m_num() as i32,
((n + self.m_num_mid()) / self.hop + 1) as i32,
)
}
pub fn p_min(&self) -> i32 {
self.pre_padding().1
}
pub fn p_max(&self, n: usize) -> i32 {
self.post_padding(n).1
}
pub fn p_range(
&self,
n: usize,
p0: Option<i32>,
p1: Option<i32>,
) -> Result<(i32, i32), String> {
let p0 = p0.unwrap_or(self.p_min());
let p1 = p1.unwrap_or(self.p_max(n));
if !(self.p_min() <= p0 && p0 < p1 && p1 <= self.p_max(n)) {
return Err(format!("Invalid slice range: p0={}, p1={}", p0, p1));
}
Ok((p0, p1))
}
fn x_slices(&self, x: &[f64], k_offset: i32, p0: i32, p1: i32) -> Vec<Vec<Complex<f64>>> {
let n = x.len();
let mut slices = Vec::new();
for p in p0..p1 {
let k_center = p * self.hop as i32 + k_offset;
let k_start = k_center - self.m_num_mid() as i32;
let k_end = k_start + self.m_num() as i32;
let mut slice_data = vec![Complex::new(0.0, 0.0); self.m_num()];
if k_start >= 0 && k_end <= n as i32 {
let start_idx = k_start as usize;
let end_idx = k_end as usize;
for (i, &val) in x[start_idx..end_idx].iter().enumerate() {
slice_data[i] = Complex::new(val, 0.0);
}
} else {
let valid_start = 0.max(k_start);
let valid_end = (n as i32).min(k_end);
if valid_start < valid_end {
let slice_start = valid_start - k_start;
let valid_start_idx = valid_start as usize;
let valid_end_idx = valid_end as usize;
let slice_start_idx = slice_start as usize;
for (i, &val) in x[valid_start_idx..valid_end_idx].iter().enumerate() {
slice_data[slice_start_idx + i] = Complex::new(val, 0.0);
}
}
}
slices.push(slice_data);
}
slices
}
pub fn stft(
&self,
x: &[f64],
p0: Option<i32>,
p1: Option<i32>,
k_offset: Option<i32>,
) -> Result<Vec<Vec<Complex<f64>>>, String> {
if self.onesided_fft() && x.iter().any(|&val| val.is_nan()) {
return Err("Complex-valued input not allowed for one-sided FFT modes".to_string());
}
let n = x.len();
let m2p = self.m_num() - self.m_num_mid();
if n < m2p {
return Err(format!(
"Input length {} must be >= ceil(m_num/2) = {}",
n, m2p
));
}
let (p0, p1) = self.p_range(n, p0, p1)?;
let k_offset = k_offset.unwrap_or(0);
let slices = self.x_slices(x, k_offset, p0, p1);
let mut time_slice_results = Vec::new();
for slice in slices.iter() {
let windowed: Vec<Complex<f64>> = slice
.iter()
.zip(self.win.iter())
.map(|(x, w)| {
let result = x * w.conj();
if self.onesided_fft() {
Complex::new(result.re, 0.0)
} else {
result
}
})
.collect();
let fft_result = self.fft_func(windowed);
time_slice_results.push(fft_result);
}
let num_freq_bins = self.f_pts();
let num_time_slices = time_slice_results.len();
let mut stft_result = vec![vec![Complex::new(0.0, 0.0); num_time_slices]; num_freq_bins];
for (time_idx, time_slice) in time_slice_results.iter().enumerate() {
for (freq_idx, &freq_val) in time_slice.iter().enumerate() {
if freq_idx < num_freq_bins {
stft_result[freq_idx][time_idx] = freq_val;
}
}
}
Ok(stft_result)
}
pub fn istft(
&mut self,
stft_data: &[Vec<Complex<f64>>],
k0: Option<i32>,
k1: Option<i32>,
) -> Result<Vec<f64>, String> {
if stft_data.is_empty() {
return Err("STFT data cannot be empty".to_string());
}
let f_pts_expected = self.f_pts();
let time_slices = stft_data[0].len();
if stft_data.len() != f_pts_expected {
return Err(format!(
"STFT frequency dimension {} must equal {}",
stft_data.len(),
f_pts_expected
));
}
if stft_data.iter().any(|row| row.len() != time_slices) {
return Err("STFT data must have equal-length frequency rows".to_string());
}
let n_min = self.m_num() - self.m_num_mid();
let q_num = self.p_max(n_min) - self.p_min();
if (time_slices as i32) < q_num {
return Err(format!(
"STFT time dimension {} needs to have at least {} slices",
time_slices, q_num
));
}
let q_max = time_slices as i32 + self.p_min();
let k_max = (q_max - 1) * self.hop as i32 + self.m_num() as i32 - self.m_num_mid() as i32;
let k0 = k0.unwrap_or(0);
let k1 = k1.unwrap_or(k_max);
let k_min = self.pre_padding().0;
if !(k_min <= k0 && k0 < k1 && k1 <= k_max) {
return Err(format!(
"k_min={} <= k0={} < k1={} <= k_max={} does not hold",
k_min, k0, k1, k_max
));
}
let num_pts = (k1 - k0) as usize;
if num_pts < n_min {
return Err(format!(
"Output length {} has to be at least {}",
num_pts, n_min
));
}
let q0 = if k0 >= 0 {
k0.div_euclid(self.hop as i32) + self.p_min()
} else {
k0.div_euclid(self.hop as i32)
};
let q1 = (self.p_max(k1 as usize)).min(q_max);
let dual_win = self.dual_win()?.to_vec();
let mut x = vec![0.0f64; num_pts];
for q in q0..q1 {
let stft_idx = q - self.p_min();
if stft_idx < 0 || stft_idx >= time_slices as i32 {
continue; }
let mut stft_slice = vec![Complex::new(0.0, 0.0); f_pts_expected];
for f in 0..f_pts_expected {
stft_slice[f] = stft_data[f][stft_idx as usize];
}
let xs_raw = self.ifft_func(stft_slice);
let xs: Vec<Complex<f64>> = xs_raw
.iter()
.zip(dual_win.iter())
.map(|(x, w)| x * w)
.collect();
let i0 = q * self.hop as i32 - self.m_num_mid() as i32;
let i1 = (i0 + self.m_num() as i32).min(num_pts as i32 + k0);
let mut j0 = 0i32;
let mut j1 = i1 - i0;
let mut actual_i0 = i0;
if i0 < k0 {
j0 += k0 - i0;
actual_i0 = k0;
}
if i1 > k0 + num_pts as i32 {
j1 -= i1 - k0 - num_pts as i32;
}
if actual_i0 < i1 && j0 < j1 {
let target_start = (actual_i0 - k0) as usize;
let _target_end = target_start + (j1 - j0) as usize;
let source_start = j0 as usize;
let source_end = j1 as usize;
for (i, &val) in xs[source_start..source_end].iter().enumerate() {
if target_start + i < x.len() {
if self.onesided_fft() {
x[target_start + i] += val.re;
} else {
x[target_start + i] += val.re; }
}
}
}
}
Ok(x)
}
pub fn t(
&self,
n: usize,
p0: Option<i32>,
p1: Option<i32>,
k_offset: Option<i32>,
) -> Result<Vec<f64>, String> {
let (p0, p1) = self.p_range(n, p0, p1)?;
let k_offset = k_offset.unwrap_or(0);
Ok((p0..p1)
.map(|p| (p * self.hop as i32 + k_offset) as f64 * self.sampling_period())
.collect())
}
pub fn f(&self) -> Vec<f64> {
let mut freqs: Vec<f64> = if self.fft_mode.is_onesided() {
(0..self.f_pts())
.map(|i| i as f64 / (self.mfft as f64 * self.sampling_period()))
.collect()
} else {
(0..self.mfft)
.map(|i| {
let freq = i as f64 / (self.mfft as f64 * self.sampling_period());
if i >= self.mfft.div_ceil(2) {
freq - 1.0 / self.sampling_period()
} else {
freq
}
})
.collect()
};
if matches!(self.fft_mode, FftMode::Centered) {
let shift = self.mfft.div_ceil(2);
freqs.rotate_left(shift);
}
freqs
}
pub fn debug_ifft_func(&self, x: Vec<Complex<f64>>) -> Vec<Complex<f64>> {
self.ifft_func(x)
}
}