Deal with unsigned int to int conversion

src/spikeinterface/preprocessing/astype.py

@@ -1,3 +1,5 @@
+import numpy as np
+
 from ..core.core_tools import define_function_from_class
 from .basepreprocessor import BasePreprocessor, BasePreprocessorSegment
 from .filter import fix_dtype
@@ -45,6 +47,14 @@ 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
+        if traces_dtype.kind == "u" and np.dtype(self.dtype).kind == "i":
+            itemsize = traces_dtype.itemsize
+            assert itemsize < 8, "Cannot upcast uint64!"
+            nbits = traces_dtype.itemsize * 8
+            # upcast to int with double itemsize
+            traces = traces.astype(f"int{2 * (traces_dtype.itemsize) * 8}") - 2 ** (nbits - 1)
         return traces.astype(self.dtype, copy=False)
 
 

src/spikeinterface/preprocessing/tests/test_astype.py

@@ -0,0 +1,42 @@
+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
+
+
+if hasattr(pytest, "global_test_folder"):
+    cache_folder = pytest.global_test_folder / "preprocessing"
+else:
+    cache_folder = Path("cache_folder") / "preprocessing"
+
+set_global_tmp_folder(cache_folder)
+
+
+def test_astype():
+    traces = (np.random.rand(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())
+
+
+def test_astype_unsigned():
+    traces = np.random.rand(10000, 4) * 100 + 500
+    traces_uint16 = traces.astype("uint16")
+    rec_uint16 = NumpyRecording(traces_uint16, sampling_frequency=30000)
+    traces_int16 = (traces.astype("int32") - 2**15).astype("int16")
+    np.testing.assert_array_equal(traces_int16, astype(rec_uint16, "int16").get_traces())
+    traces_int32 = (traces.astype("int32") - 2**15).astype("int32")
+    np.testing.assert_array_equal(traces_int32, astype(rec_uint16, "int32").get_traces())
+
+
+if __name__ == "__main__":
+    test_astype()
+    test_astype_unsigned()