Add scale_to_uV preprocessing

src/spikeinterface/preprocessing/preprocessinglist.py

@@ -24,6 +24,8 @@
     CenterRecording,
     center,
 )
+from .scale import ScaleTouV, scale_to_uV
+
 from .whiten import WhitenRecording, whiten, compute_whitening_matrix
 from .rectify import RectifyRecording, rectify
 from .clip import BlankSaturationRecording, blank_staturation, ClipRecording, clip
@@ -54,6 +56,7 @@
     ScaleRecording,
     CenterRecording,
     ZScoreRecording,
+    ScaleTouV,
     # decorrelation stuff
     WhitenRecording,
     # re-reference

src/spikeinterface/preprocessing/scale.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from spikeinterface.core import BaseRecording
+from spikeinterface.preprocessing.basepreprocessor import BasePreprocessor
+
+
+class ScaleTouV(BasePreprocessor):
+    """
+    Scale raw traces to microvolts (µV).
+
+    This preprocessor uses the channel-specific gain and offset information
+    stored in the recording extractor to convert the raw traces to µV units.
+
+    Parameters
+    ----------
+    recording : BaseRecording
+        The recording extractor to be scaled. The recording extractor must
+        have gains and offsets otherwise an error will be raised.
+
+    Raises
+    ------
+    AssertionError
+        If the recording extractor does not have scaleable traces.
+    """
+
+    name = "scale_to_uV"
+
+    def __init__(self, recording: BaseRecording):
+        assert recording.has_scaleable_traces(), "Recording must have scaleable traces"
+        from spikeinterface.preprocessing.normalize_scale import ScaleRecordingSegment
+
+        dtype = recording.get_dtype()
+        BasePreprocessor.__init__(self, recording, dtype=dtype)
+
+        gain = recording.get_channel_gains()[None, :]
+        offset = recording.get_channel_offsets()[None, :]
+        for parent_segment in recording._recording_segments:
+            rec_segment = ScaleRecordingSegment(parent_segment, gain, offset, self._dtype)
+            self.add_recording_segment(rec_segment)
+
+        self._kwargs = dict(
+            recording=recording,
+        )
+
+
+scale_to_uV = ScaleTouV

src/spikeinterface/preprocessing/tests/test_scaling.py

@@ -0,0 +1,36 @@
+import pytest
+import numpy as np
+from spikeinterface.core.testing_tools import generate_recording
+from spikeinterface.preprocessing import ScaleTouV  # Replace 'your_module' with your actual module name
+
+
+def test_scale_to_uv():
+    # Create a sample recording extractor with fake gains and offsets
+    num_channels = 4
+    sampling_frequency = 30_000.0
+    durations = [1]  # seconds
+    recording = generate_recording(
+        num_channels=num_channels,
+        durations=durations,
+        sampling_frequency=sampling_frequency,
+    )
+
+    rng = np.random.default_rng(0)
+    gains = rng.random(size=(num_channels)).astype(np.float32)
+    offsets = rng.random(size=(num_channels)).astype(np.float32)
+    recording.set_channel_gains(gains)
+    recording.set_channel_offsets(offsets)
+
+    # Apply the preprocessor
+    scaled_recording = ScaleTouV(recording=recording)
+
+    # Check if the traces are indeed scaled
+    expected_traces = recording.get_traces(return_scaled=True)
+    scaled_traces = scaled_recording.get_traces()
+
+    np.testing.assert_allclose(scaled_traces, expected_traces)
+
+    # Test for the error when recording doesn't have scaleable traces
+    recording.set_channel_gains(None)  # Remove gains to make traces unscaleable
+    with pytest.raises(AssertionError):
+        ScaleTouV(recording)