Updated zscore for in-place operations (NeuralEnsemble#440)

* Updated zscore to change the units to dimensionless when performing in-place operations, and also to return the original object instead of a new AnalogSignal object pointing to the original array.
* Changed to raise ValueError when the in-place operation is not valid.
* New AnalogSignal object is created using Neo function.
* Updated unit tests to check for same/different objects when doing in-place operation in zscore.
* Changed test_zscore_single_inplace_int as the function now raises ValueError.

Co-authored-by: Cristiano Köhler <c.koehler@fz-juelich.de>

elephant/signal_processing.py

@@ -27,6 +27,8 @@
 
 from elephant.utils import deprecated_alias, check_same_units
 
+import warnings
+
 __all__ = [
     "zscore",
     "cross_correlation_function",
@@ -66,16 +68,22 @@ def zscore(signal, inplace=True):
     inplace : bool, optional
         If True, the contents of the input `signal` is replaced by the
         z-transformed signal, if possible, i.e when the signal type is float.
+        If the signal type is not float, an error is raised.
         If False, a copy of the original `signal` is returned.
         Default: True
 
     Returns
     -------
-    signal_ztransofrmed : neo.AnalogSignal or list of neo.AnalogSignal
+    signal_ztransformed : neo.AnalogSignal or list of neo.AnalogSignal
         The output format matches the input format: for each input
         `neo.AnalogSignal`, a corresponding `neo.AnalogSignal` is returned,
         containing the z-transformed signal with dimensionless unit.
 
+    Raises
+    ------
+    ValueError
+        If `inplace` is True and the type of `signal` is not float.
+
     Notes
     -----
     You may supply a list of `neo.AnalogSignal` objects, where each object in
@@ -153,29 +161,38 @@ def zscore(signal, inplace=True):
     mean = signal_stacked.mean(axis=0)
     std = signal_stacked.std(axis=0)
 
-    signal_ztransofrmed = []
+    signal_ztransformed = []
     for sig in signal:
+        # Perform inplace operation only if array is of dtype float.
+        # Otherwise, raise an error.
+        if inplace and not np.issubdtype(np.float, sig.dtype):
+            raise ValueError(f"Cannot perform inplace operation as the "
+                             f"signal dtype is not float. Source: {sig.name}")
+
         sig_normalized = sig.magnitude.astype(mean.dtype, copy=not inplace)
         sig_normalized -= mean
+
         # items where std is zero are already zero
         np.divide(sig_normalized, std, out=sig_normalized, where=std != 0)
-        sig_dimless = neo.AnalogSignal(signal=sig_normalized,
-                                       units=pq.dimensionless,
-                                       dtype=sig_normalized.dtype,
-                                       copy=False,
-                                       t_start=sig.t_start,
-                                       sampling_rate=sig.sampling_rate,
-                                       name=sig.name,
-                                       file_origin=sig.file_origin,
-                                       description=sig.description,
-                                       array_annotations=sig.array_annotations,
-                                       **sig.annotations)
-        signal_ztransofrmed.append(sig_dimless)
+
+        if inplace:
+            # Replace unit in the original array by dimensionless
+            sig._dimensionality = pq.dimensionless.dimensionality
+            sig_dimless = sig
+        else:
+            # Create new object
+            sig_dimless = sig.duplicate_with_new_data(sig_normalized,
+                                                      units=pq.dimensionless)
+            # todo use flag once is fixed
+            #      https://github.com/NeuralEnsemble/python-neo/issues/752
+            sig_dimless.array_annotate(**sig.array_annotations)
+
+        signal_ztransformed.append(sig_dimless)
 
     # Return single object, or list of objects
-    if len(signal_ztransofrmed) == 1:
-        signal_ztransofrmed = signal_ztransofrmed[0]
-    return signal_ztransofrmed
+    if len(signal_ztransformed) == 1:
+        signal_ztransformed = signal_ztransformed[0]
+    return signal_ztransformed
 
 
 @deprecated_alias(ch_pairs='channel_pairs', nlags='n_lags',

elephant/test/test_signal_processing.py

@@ -195,6 +195,9 @@ def test_zscore_single_dup(self):
         # Assert original signal is untouched
         self.assertEqual(signal[0].magnitude, self.test_seq1[0])
 
+        # Assert original and returned objects are different
+        self.assertIsNot(result, signal)
+
     def test_zscore_single_inplace(self):
         """
         Test z-score on a single AnalogSignal, asking for an inplace
@@ -218,6 +221,9 @@ def test_zscore_single_inplace(self):
         # Assert original signal is overwritten
         self.assertEqual(signal[0].magnitude, target[0])
 
+        # Assert original and returned objects are the same
+        self.assertIs(result, signal)
+
     def test_zscore_single_multidim_dup(self):
         """
         Test z-score on a single AnalogSignal with multiple dimensions, asking
@@ -232,13 +238,15 @@ def test_zscore_single_multidim_dup(self):
         s = np.std(signal.magnitude, axis=0, keepdims=True)
         target = (signal.magnitude - m) / s
 
-        assert_array_almost_equal(
-            elephant.signal_processing.zscore(
-                signal, inplace=False).magnitude, target, decimal=9)
+        result = elephant.signal_processing.zscore(signal, inplace=False)
+        assert_array_almost_equal(result.magnitude, target, decimal=9)
 
         # Assert original signal is untouched
         self.assertEqual(signal[0, 0].magnitude, self.test_seq1[0])
 
+        # Assert original and returned objects are different
+        self.assertIsNot(result, signal)
+
     def test_zscore_array_annotations(self):
         signal = neo.AnalogSignal(
             self.test_seq1, units='mV',
@@ -269,6 +277,9 @@ def test_zscore_single_multidim_inplace(self):
         # Assert original signal is overwritten
         self.assertAlmostEqual(signal[0, 0].magnitude, ground_truth[0, 0])
 
+        # Assert original and returned objects are the same
+        self.assertIs(result, signal)
+
     def test_zscore_single_dup_int(self):
         """
         Test if the z-score is correctly calculated even if the input is an
@@ -283,28 +294,28 @@ def test_zscore_single_dup_int(self):
         s = np.std(self.test_seq1)
         target = (self.test_seq1 - m) / s
 
-        assert_array_almost_equal(
-            elephant.signal_processing.zscore(signal, inplace=False).magnitude,
-            target.reshape(-1, 1), decimal=9)
+        result = elephant.signal_processing.zscore(signal, inplace=False)
+        assert_array_almost_equal(result.magnitude, target.reshape(-1, 1),
+                                  decimal=9)
 
         # Assert original signal is untouched
         self.assertEqual(signal.magnitude[0], self.test_seq1[0])
 
+        # Assert original and returned objects are different
+        self.assertIsNot(result, signal)
+
     def test_zscore_single_inplace_int(self):
         """
-        Test if the z-score is correctly calculated even if the input is an
-        AnalogSignal of type int, asking for an inplace operation.
+        Test if the z-score operation fails if the input is an
+        AnalogSignal of type int, when asking for an inplace operation.
         """
-        m = np.mean(self.test_seq1)
-        s = np.std(self.test_seq1)
-        target = (self.test_seq1 - m) / s
 
         signal = neo.AnalogSignal(
             self.test_seq1, units='mV',
             t_start=0. * pq.ms, sampling_rate=1000. * pq.Hz, dtype=int)
-        zscored = elephant.signal_processing.zscore(signal, inplace=True)
 
-        assert_array_almost_equal(zscored.magnitude.squeeze(), target)
+        with self.assertRaises(ValueError):
+            elephant.signal_processing.zscore(signal, inplace=True)
 
     def test_zscore_list_dup(self):
         """
@@ -344,6 +355,10 @@ def test_zscore_list_dup(self):
         self.assertEqual(signal1.magnitude[0, 0], self.test_seq1[0])
         self.assertEqual(signal2.magnitude[0, 1], self.test_seq2[0])
 
+        # Assert original and returned objects are different
+        self.assertIsNot(result[0], signal_list[0])
+        self.assertIsNot(result[1], signal_list[1])
+
     def test_zscore_list_inplace(self):
         """
         Test zscore on a list of AnalogSignal objects, asking for an
@@ -382,6 +397,10 @@ def test_zscore_list_inplace(self):
         self.assertEqual(signal1[0, 0].magnitude, target11[0])
         self.assertEqual(signal2[0, 0].magnitude, target21[0])
 
+        # Assert original and returned objects are the same
+        self.assertIs(result[0], signal_list[0])
+        self.assertIs(result[1], signal_list[1])
+
     def test_wrong_input(self):
         # wrong type
         self.assertRaises(TypeError, elephant.signal_processing.zscore,