Merge pull request #1707 from alejoe91/astype-unsigned

Deal with unsigned int to int conversion

doc/api.rst

@@ -141,13 +141,17 @@ spikeinterface.preprocessing
 
 .. automodule:: spikeinterface.preprocessing
 
+    .. autofunction:: astype
+    .. autofunction:: average_across_direction
     .. autofunction:: bandpass_filter
     .. autofunction:: blank_staturation
     .. autofunction:: center
     .. autofunction:: clip
     .. autofunction:: common_reference
     .. autofunction:: correct_lsb
+    .. autofunction:: depth_order
     .. autofunction:: detect_bad_channels
+    .. autofunction:: directional_derivative
     .. autofunction:: filter
     .. autofunction:: gaussian_bandpass_filter
     .. autofunction:: highpass_filter
@@ -158,7 +162,10 @@ spikeinterface.preprocessing
     .. autofunction:: phase_shift
     .. autofunction:: rectify
     .. autofunction:: remove_artifacts
+    .. autofunction:: resample
     .. autofunction:: scale
+    .. autofunction:: silence_periods
+    .. autofunction:: unsigned_to_signed
     .. autofunction:: whiten
     .. autofunction:: zero_channel_pad
     .. autofunction:: zscore

doc/modules/preprocessing.rst

@@ -269,6 +269,29 @@ strategies:
 * :py:func:`~spikeinterface.preprocessing.remove_artifacts()`
 
 
+astype() / unsigned_to_signed()
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Similarly to :code:`numpy.astype()`, the :code:`astype()` casts the traces to the desired :code:`dtype`:
+
+.. code-block:: python
+
+    rec_int16 = astype(rec_float, "int16")
+
+
+For recordings whose traces are unsigned (e.g. Maxwell Biosystems), the :code:`unsigned_to_signed()` function makes them
+signed by removing the unsigned "offset". For example, :code:`uint16` traces will be first upcast to :code:`uint32`, 2**15
+is subtracted, and the traces are finally cast to :code:`int16`:
+
+
+.. code-block:: python
+
+    rec_int16 = unsigned_to_signed(rec_uint16)
+
+* :py:func:`~spikeinterface.preprocessing.astype()`
+* :py:func:`~spikeinterface.preprocessing.unsigned_to_signed()`
+
+
 zero_channel_pad()
 ^^^^^^^^^^^^^^^^^^
 

src/spikeinterface/preprocessing/astype.py

@@ -7,6 +7,8 @@ class AstypeRecording(BasePreprocessor):
     """The spikeinterface analog of numpy.astype
 
     Converts a recording to another dtype on the fly.
+
+    For recording with an unsigned dtype, please use the `unsigned_to_signed` preprocessing function.
     """
 
     name = "astype"

src/spikeinterface/preprocessing/preprocessinglist.py

@@ -36,6 +36,7 @@
 from .directional_derivative import DirectionalDerivativeRecording, directional_derivative
 from .depth_order import DepthOrderRecording, depth_order
 from .astype import AstypeRecording, astype
+from .unsigned_to_signed import UnsignedToSignedRecording, unsigned_to_signed
 
 
 preprocessers_full_list = [
@@ -70,6 +71,7 @@
     AverageAcrossDirectionRecording,
     DirectionalDerivativeRecording,
     AstypeRecording,
+    UnsignedToSignedRecording,
 ]
 
 installed_preprocessers_list = [pp for pp in preprocessers_full_list if pp.installed]

src/spikeinterface/preprocessing/tests/test_astype.py

@@ -0,0 +1,24 @@
+import pytest
+from pathlib import Path
+import shutil
+
+from spikeinterface import set_global_tmp_folder, NumpyRecording
+from spikeinterface.core import generate_recording
+
+from spikeinterface.preprocessing import astype
+
+import numpy as np
+
+
+def test_astype():
+    rng = np.random.RandomState(0)
+    traces = (rng.randn(10000, 4) * 100).astype("float32")
+    rec_float32 = NumpyRecording(traces, sampling_frequency=30000)
+    traces_int16 = traces.astype("int16")
+    np.testing.assert_array_equal(traces_int16, astype(rec_float32, "int16").get_traces())
+    traces_float64 = traces.astype("float64")
+    np.testing.assert_array_equal(traces_float64, astype(rec_float32, "float64").get_traces())
+
+
+if __name__ == "__main__":
+    test_astype()

src/spikeinterface/preprocessing/tests/test_unsigned_to_signed.py

@@ -0,0 +1,29 @@
+import pytest
+from pathlib import Path
+import shutil
+
+from spikeinterface import set_global_tmp_folder, NumpyRecording
+from spikeinterface.core import generate_recording
+
+from spikeinterface.preprocessing import unsigned_to_signed
+
+import numpy as np
+
+
+def test_unsigned_to_signed():
+    rng = np.random.RandomState(0)
+    traces = rng.rand(10000, 4) * 100 + 2**15
+    traces_uint16 = traces.astype("uint16")
+    traces = rng.rand(10000, 4) * 100 + 2**31
+    traces_uint32 = traces.astype("uint32")
+    rec_uint16 = NumpyRecording(traces_uint16, sampling_frequency=30000)
+    rec_uint32 = NumpyRecording(traces_uint32, sampling_frequency=30000)
+
+    traces_int16 = (traces_uint16.astype("int32") - 2**15).astype("int16")
+    np.testing.assert_array_equal(traces_int16, unsigned_to_signed(rec_uint16).get_traces())
+    traces_int32 = (traces_uint32.astype("int64") - 2**31).astype("int32")
+    np.testing.assert_array_equal(traces_int32, unsigned_to_signed(rec_uint32).get_traces())
+
+
+if __name__ == "__main__":
+    test_unsigned_to_signed()

src/spikeinterface/preprocessing/unsigned_to_signed.py

@@ -0,0 +1,56 @@
+import numpy as np
+
+from ..core.core_tools import define_function_from_class
+from .basepreprocessor import BasePreprocessor, BasePreprocessorSegment
+from .filter import fix_dtype
+
+
+class UnsignedToSignedRecording(BasePreprocessor):
+    """
+    Converts a recording with unsigned traces to a signed one.
+    """
+
+    name = "unsigned_to_signed"
+
+    def __init__(
+        self,
+        recording,
+    ):
+        dtype = np.dtype(recording.dtype)
+        assert dtype.kind == "u", "Recording is not unsigned!"
+        itemsize = dtype.itemsize
+        assert itemsize < 8, "Cannot convert uint64 to int64."
+        dtype_signed = dtype.str.replace("uint", "int")
+
+        BasePreprocessor.__init__(self, recording, dtype=dtype_signed)
+
+        for parent_segment in recording._recording_segments:
+            rec_segment = UnsignedToSignedRecordingSegment(parent_segment, dtype_signed)
+            self.add_recording_segment(rec_segment)
+
+        self._kwargs = dict(
+            recording=recording,
+        )
+
+
+class UnsignedToSignedRecordingSegment(BasePreprocessorSegment):
+    def __init__(self, parent_recording_segment, dtype_signed):
+        BasePreprocessorSegment.__init__(self, parent_recording_segment)
+        self.dtype_signed = dtype_signed
+
+    def get_traces(self, start_frame, end_frame, channel_indices):
+        if channel_indices is None:
+            channel_indices = slice(None)
+        traces = self.parent_recording_segment.get_traces(start_frame, end_frame, channel_indices)
+        # if uint --> take care of offset
+        traces_dtype = traces.dtype
+        nbits = traces_dtype.itemsize * 8
+        signed_dtype = f"int{2 * (traces_dtype.itemsize) * 8}"
+        offset = 2 ** (nbits - 1)
+        # upcast to int with double itemsize
+        traces = traces.astype(signed_dtype, copy=False) - offset
+        return traces.astype(self.dtype_signed, copy=False)
+
+
+# function for API
+unsigned_to_signed = define_function_from_class(source_class=UnsignedToSignedRecording, name="unsigned_to_signed")