feat: More robust interpolation

Author: prouast

pyproject.toml

@@ -25,7 +25,7 @@ dependencies = [
     "importlib_resources>=5.12",
     "numpy>=1.24",
     "onnxruntime>=1.15.0",
-    "prpy[ffmpeg,numpy_min]>=0.2.17",
+    "prpy[ffmpeg,numpy_min]>=0.2.25",
     "python-dotenv>=1.0",
     "pyyaml>=6.0.1",
     "requests>=2.32.0",

vitallens/methods/simple_rppg_method.py

@@ -22,7 +22,7 @@
 import numpy as np
 from prpy.numpy.face import get_roi_from_det
 from prpy.numpy.image import reduce_roi, parse_image_inputs
-from prpy.numpy.signal import interpolate_cubic_spline
+from prpy.numpy.signal import interpolate_filtered
 from typing import Union, Tuple
 
 from vitallens.buffer import SignalBuffer
@@ -116,8 +116,10 @@ def __call__(
     # Perform rppg algorithm step (n_frames_ds,)
     sig_ds = self.algorithm(rgb_ds, fps_ds)
     # Interpolate to original sampling rate (n_frames,)
-    sig = interpolate_cubic_spline(
-      x=np.arange(inputs_shape[0])[0::ds_factor], y=sig_ds, xs=np.arange(inputs_shape[0]), axis=1)
+    sig = interpolate_filtered(t_in=np.arange(inputs_shape[0])[0::ds_factor],
+                               s_in=sig_ds,
+                               t_out=np.arange(inputs_shape[0]),
+                               axis=1, extrapolate=True)
     # Filter and add dim (1, n_frames)
     sig = self.pulse_filter(sig, fps)
     sig = np.expand_dims(sig, axis=0)

vitallens/methods/vitallens.py

@@ -25,7 +25,7 @@
 from prpy.numpy.face import get_roi_from_det
 from prpy.numpy.image import probe_image_inputs, parse_image_inputs
 from prpy.numpy.signal import detrend, moving_average, standardize
-from prpy.numpy.signal import interpolate_cubic_spline
+from prpy.numpy.signal import interpolate_filtered
 from prpy.numpy.utils import enough_memory_for_ndarray
 import json
 import logging
@@ -139,12 +139,9 @@ def __call__(
     conf_ds, _ = reassemble_from_windows(x=conf_results, idxs=idxs_results)
     live_ds = reassemble_from_windows(x=np.asarray(live_results)[:,np.newaxis], idxs=idxs_results)[0][0]
     # Interpolate to original sampling rate
-    sig = interpolate_cubic_spline(
-      x=idxs, y=sig_ds, xs=np.arange(inputs_n), axis=1)
-    conf = interpolate_cubic_spline(
-      x=idxs, y=conf_ds, xs=np.arange(inputs_n), axis=1)
-    live = interpolate_cubic_spline(
-      x=idxs, y=live_ds, xs=np.arange(inputs_n), axis=0)
+    sig = interpolate_filtered(t_in=idxs, s_in=sig_ds, t_out=np.arange(inputs_n), axis=1, extrapolate=True)
+    conf = interpolate_filtered(t_in=idxs, s_in=conf_ds, t_out=np.arange(inputs_n), axis=1, extrapolate=True)
+    live = interpolate_filtered(t_in=idxs, s_in=live_ds, t_out=np.arange(inputs_n), axis=0, extrapolate=True)
     # Filter only in batch mode (2, n_frames)
     if self.op_mode == Mode.BATCH:
       sig = np.asarray([self.postprocess(p, fps, type=name) for p, name in zip(sig, ['ppg', 'resp'])])