"""
File: qumphy/data/signal_preprocessing/resampling.py
Project: 22HLT01 QUMPHY
Contact: oskar.pfeffer@ptb.de
Gitlab: https://gitlab.com/qumphy
Description: Functions for resampling the rate of PPG signals.
"""
from __future__ import annotations
from dataclasses import dataclass
from fractions import Fraction
from typing import Optional, Tuple, Union
import numpy as np
import torch
from scipy.signal import resample_poly
ArrayLike = Union[np.ndarray, torch.Tensor]
[docs]
@dataclass(frozen=True)
class MatlabResampleConfig:
"""
Parameters chosen to mimic MATLAB's resample defaults:
- Kaiser beta default is 5
- n default is 10 (controls filter length in MATLAB)
"""
n: int = 10
beta: float = 5.0
padtype: str = "constant" # MATLAB-like: assume zeros beyond boundary
cval: float = 0.0 # constant padding value
def _as_numpy(x: ArrayLike) -> Tuple[np.ndarray, Optional[torch.dtype], bool]:
"""Return numpy array + (torch dtype, was_torch)."""
if isinstance(x, torch.Tensor):
# Keep it deterministic and safe: resampling on CPU in numpy/scipy
dtype = x.dtype
x_np = x.detach().cpu().numpy()
return x_np, dtype, True
return np.asarray(x), None, False
def _back_to_input_type(
y_np: np.ndarray,
was_torch: bool,
torch_dtype: Optional[torch.dtype],
device: Optional[torch.device] = None,
) -> ArrayLike:
if not was_torch:
return y_np
y_t = torch.from_numpy(y_np)
if torch_dtype is not None:
# Avoid surprises: cast back to original dtype (float32/float64/etc.)
y_t = y_t.to(dtype=torch_dtype)
if device is not None:
y_t = y_t.to(device=device)
return y_t
def _rational_factor(
fs_in: float, fs_out: float, max_den: int = 10_000
) -> Tuple[int, int]:
"""
Convert fs_out/fs_in to a rational up/down pair.
For integer Hz like 80->256, 80->20, etc. this is exact.
"""
if fs_in <= 0 or fs_out <= 0:
raise ValueError("fs_in and fs_out must be positive.")
frac = Fraction(fs_out, fs_in).limit_denominator(max_den)
return frac.numerator, frac.denominator
[docs]
def resample_like_matlab(
x: ArrayLike,
fs_in: float,
fs_out: float,
*,
axis: int = -1,
cfg: MatlabResampleConfig = MatlabResampleConfig(),
max_denominator: int = 10_000,
) -> ArrayLike:
"""
MATLAB-like resampling:
- polyphase FIR lowpass
- Kaiser window (beta default 5)
- zero padding beyond edges (padtype='constant', cval=0)
"""
if fs_in == fs_out:
return x
x_np, torch_dtype, was_torch = _as_numpy(x)
device = x.device if isinstance(x, torch.Tensor) else None
up, down = _rational_factor(fs_in, fs_out, max_den=max_denominator)
# SciPy's resample_poly default window is ('kaiser', 5.0),
# matching MATLAB default beta.
y_np = resample_poly(
x_np,
up=up,
down=down,
axis=axis,
window=("kaiser", cfg.beta),
padtype=cfg.padtype,
cval=cfg.cval,
)
return _back_to_input_type(y_np, was_torch, torch_dtype, device=device)