diff --git a/docs/api/python/decoders/index.rst b/docs/api/python/decoders/index.rst
new file mode 100644
index 0000000..91760de
--- /dev/null
+++ b/docs/api/python/decoders/index.rst
@@ -0,0 +1,17 @@
+.. _decoding:
+
+:octicon:`file-directory;0.9em;sd-mr-1 fill-primary` decoders
+=============================================================
+
+The ``decoders`` module within the spikify library contains the essential components for converting spike trains back into continuous signals or meaningful representations. This section is organized to reflect the primary structure of the decoding algorithms included in the library:
+
+- **Temporal Coding**: Encloses algorithms that decode data based on the precise timing of spikes, reconstructing signals from spike timing information.
+
+Each submodule is dedicated to a specific family of decoding techniques, making it easy to navigate and understand the purpose of each algorithm within the library structure.
+
+Below, you will find links to the specific modules library for each decoding method:
+
+.. toctree::
+ :maxdepth: 1
+
+ temporal_decoding/index
diff --git a/docs/api/python/decoders/temporal_decoding/contrast/decoder.rst b/docs/api/python/decoders/temporal_decoding/contrast/decoder.rst
new file mode 100644
index 0000000..fbc9934
--- /dev/null
+++ b/docs/api/python/decoders/temporal_decoding/contrast/decoder.rst
@@ -0,0 +1,8 @@
+.. _contrast_decoder_function:
+
+.. title:: Contrast Decoder
+
+.. automodule:: spikify.decoders.temporal.contrast.decoder_algorithm
+ :members: contrast_decoder
+ :undoc-members:
+ :show-inheritance:
\ No newline at end of file
diff --git a/docs/api/python/decoders/temporal_decoding/contrast/index.rst b/docs/api/python/decoders/temporal_decoding/contrast/index.rst
new file mode 100644
index 0000000..5069dd6
--- /dev/null
+++ b/docs/api/python/decoders/temporal_decoding/contrast/index.rst
@@ -0,0 +1,17 @@
+.. _contrast_decoder:
+
+:octicon:`file-directory;0.9em;sd-mr-1 fill-primary` contrast
+================================================================
+
+The ``contrast`` folder within the temporal decoding module includes the algorithm responsible for reconstructing the original continuous signal from spike trains generated by any of the Contrast family encoders. Since the encoding methods in this family share a common spike representation based on threshold increments, a single unified decoder is able to handle all of them.
+
+Contents of the ``contrast`` folder:
+
+- **Contrast Decoder**: A unified decoding algorithm compatible with spike trains produced by the Moving Window (MW), Step-Forward (SF), Threshold-Based (TB) encoders ans Zero-Crossing Step Forward (ZCSF).
+
+Below, you will find links to the specific modules for the contrast decoding algorithm:
+
+.. toctree::
+ :maxdepth: 1
+
+ decoder
diff --git a/docs/api/python/decoders/temporal_decoding/deconvolution/decoder.rst b/docs/api/python/decoders/temporal_decoding/deconvolution/decoder.rst
new file mode 100644
index 0000000..2ef626e
--- /dev/null
+++ b/docs/api/python/decoders/temporal_decoding/deconvolution/decoder.rst
@@ -0,0 +1,8 @@
+.. _deconvolution_decoder_function:
+
+.. title:: Deconvolution Decoder
+
+.. automodule:: spikify.decoders.temporal.deconvolution.decoder_algorithm
+ :members: deconvolution_decoder
+ :undoc-members:
+ :show-inheritance:
\ No newline at end of file
diff --git a/docs/api/python/decoders/temporal_decoding/deconvolution/index.rst b/docs/api/python/decoders/temporal_decoding/deconvolution/index.rst
new file mode 100644
index 0000000..32aa45c
--- /dev/null
+++ b/docs/api/python/decoders/temporal_decoding/deconvolution/index.rst
@@ -0,0 +1,22 @@
+.. _deconvolution_decoder:
+
+:octicon:`file-directory;0.9em;sd-mr-1 fill-primary` deconvolution
+====================================================================
+
+The ``deconvolution`` module within the temporal decoding module includes the algorithm responsible for
+reconstructing the original continuous signal from spike trains generated by any of the Deconvolution
+family encoders. Since the encoding methods in this family share a common reconstruction approach based
+on causal linear filtering with the same FIR filter used during encoding, a single unified decoder is
+able to handle all of them.
+
+Contents of the ``deconvolution`` folder:
+
+- **Deconvolution Decoder**: A unified decoding algorithm compatible with spike trains produced by the
+ Hough Spiker (HSA), Modified Hough Spiker (MHSA), and Ben's Spiker (BSA) encoders.
+
+Below, you will find links to the specific modules for the deconvolution decoding algorithm:
+
+.. toctree::
+ :maxdepth: 1
+
+ decoder
diff --git a/docs/api/python/decoders/temporal_decoding/index.rst b/docs/api/python/decoders/temporal_decoding/index.rst
new file mode 100644
index 0000000..0717a7a
--- /dev/null
+++ b/docs/api/python/decoders/temporal_decoding/index.rst
@@ -0,0 +1,19 @@
+.. _temporal_decoding:
+
+:octicon:`file-directory;0.9em;sd-mr-1 fill-primary` temporal
+================================================================
+
+The ``temporal`` module within the decoding module contains algorithms that reconstruct signals based on the timing of individual spikes. This method is especially effective for scenarios where the timing of events conveys more information than their frequency.
+
+Contents of the ``temporal`` module:
+
+- **Contrast**: Algorithms that decode based on variations in the signal over time.
+- **Deconvolution**: Approaches that reconstruct the original signal from a spike train using deconvolution techniques.
+
+Below, you will find links to the specific modules for each temporal decoding algorithm:
+
+.. toctree::
+ :maxdepth: 1
+
+ contrast/index
+ deconvolution/index
diff --git a/docs/api/python/index.rst b/docs/api/python/index.rst
index c464368..fa0e921 100644
--- a/docs/api/python/index.rst
+++ b/docs/api/python/index.rst
@@ -21,3 +21,4 @@ Below, you will find links to the specific modules within the Python API:
filters/index
encoders/index
+ decoders/index
diff --git a/spikify/decoders/__init__.py b/spikify/decoders/__init__.py
new file mode 100644
index 0000000..149d474
--- /dev/null
+++ b/spikify/decoders/__init__.py
@@ -0,0 +1 @@
+"""Decoders package."""
diff --git a/spikify/decoders/temporal/__init__.py b/spikify/decoders/temporal/__init__.py
new file mode 100644
index 0000000..e61d8fd
--- /dev/null
+++ b/spikify/decoders/temporal/__init__.py
@@ -0,0 +1 @@
+"""Temporal package."""
diff --git a/spikify/decoders/temporal/contrast/__init__.py b/spikify/decoders/temporal/contrast/__init__.py
new file mode 100644
index 0000000..aa03780
--- /dev/null
+++ b/spikify/decoders/temporal/contrast/__init__.py
@@ -0,0 +1,5 @@
+"""Temporal Contrast package."""
+
+from .decoder_algorithm import contrast_decoder
+
+__all__ = ["contrast_decoder"]
diff --git a/spikify/decoders/temporal/contrast/decoder_algorithm.py b/spikify/decoders/temporal/contrast/decoder_algorithm.py
new file mode 100644
index 0000000..d08fd35
--- /dev/null
+++ b/spikify/decoders/temporal/contrast/decoder_algorithm.py
@@ -0,0 +1,107 @@
+"""
+.. raw:: html
+
+ Contrast Decoder
+"""
+
+import numpy as np
+
+
+def contrast_decoder(
+ spikes: np.ndarray,
+ thresholds: np.ndarray,
+ start_point: float | int | list[float | int] | np.ndarray,
+) -> np.ndarray:
+ """
+ Perform Contrast family decoding on the input spike train.
+
+ This function takes a spike train encoded with any of the Contrast family algorithms,
+ Threshold-Based Representation (TBR), Step-Forward (SF), Moving Window (MW) or Zero-Crossing Step-Forward (ZCSF),
+ and reconstructs the original continuous signal. The reconstruction is performed by starting from the initial
+ signal value and applying the threshold incrementally based on the spike values.
+
+ .. note::
+ - The ``thresholds`` parameter should be the ``thresholds`` array returned directly by the encoder
+ (TBR, SF, or MW). All three encoders return this array as part of their output to ensure the step
+ size used for reconstruction exactly matches the one used during encoding.
+ - The ``start_point`` parameter should be set to the first sample of the original signal prior to
+ encoding (e.g. ``signal[0]`` for a 1D signal, or ``signal[0, :]`` for a multi-feature/channel signal).
+ Since the Contrast family encodes only differences between consecutive samples, the absolute signal
+ level is lost during encoding and must be restored by anchoring reconstruction to this initial value.
+
+
+ **Code Example:**
+
+ .. code-block:: python
+
+ import numpy as np
+ from spikify.encoders.temporal.contrast import moving_window
+ from spikify.decoders.temporal.contrast import contrast_decoder
+
+ signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0])
+ window_length = 3
+ spikes, thresholds = moving_window(signal, window_length, threshold=0.2)
+ reconstructed_signal = contrast_decoder(spikes, thresholds, start_point=signal[0])
+
+ .. doctest::
+ :hide:
+
+ >>> import numpy as np
+ >>> from spikify.encoders.temporal.contrast import moving_window
+ >>> from spikify.decoders.temporal.contrast import contrast_decoder
+ >>> signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0])
+ >>> spikes, thresholds = moving_window(signal, window_length=3, threshold=0.2)
+ >>> reconstructed_signal = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ >>> reconstructed_signal.flatten()
+ array([0.1, 0.1, 0.1, 0.3, 0.5, 0.7])
+
+ :param spikes: The input spike train to be decoded. This should be a numpy ndarray with values in {-1, 0, +1},
+ as produced by any of the Contrast family encoders (TBR, SF, MW, ZCSF).
+ :type spikes: numpy.ndarray
+ :param threshold: Per-feature or channels threshold values used during encoding, as returned directly by the TBR,
+ SF, MW or ZCSF encoder. Passing the encoder's returned ``thresholds`` array
+ ensures the reconstruction step size exactly matches the one used during encoding.
+ :type threshold: numpy.ndarray
+ :param start_point: Initial signal value(s) for reconstruction (scalar or 1D sequence matching features).
+ Should be set to the first sample of the original signal before encoding (e.g. ``signal[0]``), as the
+ Contrast family encodes only signal differences and the absolute offset must be restored manually.
+ :type start_point: float | int | list[float | int] | numpy.ndarray
+ :return: A numpy array representing the reconstructed continuous signal approximation.
+ :rtype: numpy.ndarray
+ :raises ValueError: If the input spike train is empty or if the start_point dimensions do not match
+ the spike train feature dimensions.
+
+ """
+ # Check for empty spike train
+ if len(spikes) == 0:
+ raise ValueError("Spike train cannot be empty.")
+
+ # Ensure 2D processing (T, F)
+ if spikes.ndim == 1:
+ spikes = spikes.reshape(-1, 1)
+
+ T, F = spikes.shape
+
+ # Handle startpoint
+ if np.isscalar(start_point):
+ start_points = np.full(F, float(start_point))
+ else:
+ start_points = np.asarray(start_point, dtype=float)
+ if start_points.ndim != 1:
+ raise ValueError("Startpoint must be a scalar or a 1D sequence of numbers.")
+ if start_points.size != F:
+ raise ValueError("Startpoint must match the number of features in the spike train.")
+
+ signal = np.zeros((T, F), dtype=float)
+ signal[0] = start_points
+
+ for t in range(1, T):
+ for f in range(F):
+ if spikes[t, f] == 1:
+ signal[t, f] = signal[t - 1, f] + thresholds[f]
+ elif spikes[t, f] == -1:
+ signal[t, f] = signal[t - 1, f] - thresholds[f]
+ else:
+ signal[t, f] = signal[t - 1, f]
+
+ return signal
diff --git a/spikify/decoders/temporal/deconvolution/__init__.py b/spikify/decoders/temporal/deconvolution/__init__.py
new file mode 100644
index 0000000..e9fca18
--- /dev/null
+++ b/spikify/decoders/temporal/deconvolution/__init__.py
@@ -0,0 +1,5 @@
+"""Temporal Deconvolution package."""
+
+from .decoder_algorithm import deconvolution_decoder
+
+__all__ = ["deconvolution_decoder"]
diff --git a/spikify/decoders/temporal/deconvolution/decoder_algorithm.py b/spikify/decoders/temporal/deconvolution/decoder_algorithm.py
new file mode 100644
index 0000000..c7be8a1
--- /dev/null
+++ b/spikify/decoders/temporal/deconvolution/decoder_algorithm.py
@@ -0,0 +1,102 @@
+"""
+.. raw:: html
+
+ Deconvolution Decoder
+"""
+
+import numpy as np
+from scipy.signal import lfilter
+
+
+def deconvolution_decoder(
+ spikes: np.ndarray,
+ fir_bank: np.ndarray,
+ shift: np.ndarray,
+ norm: np.ndarray | None = None,
+) -> np.ndarray:
+ """
+ Perform signal reconstruction via deconvolution decoding of a spike train.
+
+ The Deconvolution Decoder reverses the encoding performed by spike-based deconvolution algorithms such as
+ the Hough Spiker (HSA), its Modified variant (MHSA) and Bens Spiker (BSA). Given a binary spike train and the FIR
+ filter bank produced during encoding, it reconstructs an approximation of the original continuous signal
+ by applying each feature's filter to its corresponding spike train using causal linear filtering.
+
+ .. note::
+ - The ``fir_bank``, ``shift``, and ``norm`` parameters should be the values returned directly by the
+ encoder algorithm to ensure accurate inversion of all pre-processing steps.
+ - If ``norm`` is ``None``, no amplitude rescaling is applied; only the per-feature shift is restored.
+ Pass ``norm`` explicitly whenever the encoder performed normalization. Only MHSA and HSA have normalization
+ steps, while BSA does not.
+
+ **Code Example:**
+
+ .. code-block:: python
+
+ import numpy as np
+ from spikify.encoders.temporal.deconvolution import modified_hough_spiker
+ from spikify.decoders.temporal.deconvolution import deconvolution_decoder
+
+ signal = np.array([0.1, 0.2, 0.3, 1.0, 0.5, 0.3, 0.1])
+ window_length = 3
+ threshold = 0.5
+ cutoff = 0.1
+ spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ reconstructed = deconvolution_decoder(spikes, fir_bank, shift, norm)
+
+ .. doctest::
+ :hide:
+
+ >>> import numpy as np
+ >>> from spikify.encoders.temporal.deconvolution import modified_hough_spiker
+ >>> from spikify.decoders.temporal.deconvolution import deconvolution_decoder
+ >>> signal = np.array([0.0, 0.0, 0.5, 0.9, 0.7, 0.4, 0.1, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0])
+ >>> window_length = 5
+ >>> threshold = 0.3
+ >>> cutoff = 0.1
+ >>> spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ >>> reconstructed = deconvolution_decoder(spikes, fir_bank, shift, norm)
+ >>> reconstructed.flatten()
+ array([0. , 0.24793707, 0.50412586, 0.49587414, 0.50412586,
+ 0.24793707, 0. , 0. , 0. , 0. ,
+ 0. , 0. , 0. , 0. , 0.24793707])
+
+ :param spikes: Binary spike train to decode (values in {0, 1}), as produced by
+ any of the deconvolution family encoders (HSA, MHSA, BSA).
+ :type spikes: numpy.ndarray
+ :param fir_bank: FIR filter coefficients used during encoding.
+ Each column corresponds to the filter applied to one feature or channel. This is the ``fir_bank`` value
+ returned directly by the encoder.
+ :type fir_bank: numpy.ndarray
+ :param shift: Per-feature shift values subtracted during encoding to ensure signal non-negativity.
+ This is the ``shift`` value returned directly by the encoder.
+ :type shift: numpy.ndarray
+ :param norm: Per-feature normalization values used during encoding to scale the signal to [0, 1].
+ If ``None``, no amplitude rescaling is applied and only the shift is restored.
+ This is the ``norm`` value returned directly by the encoder.
+ :type norm: numpy.ndarray | None
+ :return: A numpy array representing the reconstructed continuous signal approximation.
+ :rtype: numpy.ndarray
+ :raises ValueError: If the input spike train is empty.
+
+ """
+
+ # Check for empty spike train
+ if len(spikes) == 0:
+ raise ValueError("Spike train cannot be empty.")
+
+ # Ensure 2D processing (T, F)
+ if spikes.ndim == 1:
+ spikes = spikes.reshape(-1, 1)
+
+ T, F = spikes.shape
+ signal = np.zeros((T, F))
+
+ for f in range(F):
+ x = lfilter(fir_bank[:, f], 1, spikes[:, f], axis=0)
+ if norm is not None:
+ signal[:, f] = x * norm[f] + shift[f]
+ else:
+ signal[:, f] = x + shift[f]
+
+ return signal
diff --git a/spikify/encoders/temporal/contrast/moving_window_algorithm.py b/spikify/encoders/temporal/contrast/moving_window_algorithm.py
index 549361d..f67ce17 100644
--- a/spikify/encoders/temporal/contrast/moving_window_algorithm.py
+++ b/spikify/encoders/temporal/contrast/moving_window_algorithm.py
@@ -9,7 +9,7 @@
def moving_window(
signal: np.ndarray, window_length: int, threshold: float | int | list[float | int] | np.ndarray
-) -> np.ndarray:
+) -> tuple[np.ndarray, np.ndarray]:
"""
Perform Moving Window (MW) encoding on the input signal.
@@ -29,8 +29,7 @@ def moving_window(
signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0])
window_length = 3
threshold = 0.2
- encoded_signal = moving_window(signal, window_length, threshold)
- encoded_signal
+ encoded_signal, thresholds = moving_window(signal, window_length, threshold)
.. doctest::
:hide:
@@ -40,20 +39,23 @@ def moving_window(
>>> signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0])
>>> window_length = 3
>>> threshold = 0.2
- >>> encoded_signal = moving_window(signal, window_length, threshold)
- >>> encoded_signal
+ >>> encoded_signal, _ = moving_window(signal, window_length, threshold)
+ >>> encoded_signal.flatten()
array([0, 0, 0, 1, 1, 1], dtype=int8)
- :param signal: The input signal to be encoded. This should be a numpy ndarray.
+ :param signal: Input signal to encode (1D or 2D: time × features or channels).
:type signal: numpy.ndarray
:param window_length: The size of the sliding window for calculating the signal base mean.
:type window_length: int
:param threshold: Threshold(s) for spike generation; scalar or 1D sequence matching features.
:type threshold: float | int | list[float | int] | numpy.ndarray
- :return: A numpy array representing the encoded spike train.
- :rtype: numpy.ndarray
- :raises ValueError: If the input signal is empty or if the threshold dimensions do not match the signal features.
- :raises TypeError: If the threshold parameter is of invalid dimension.
+ :return:
+ - spikes: A numpy array representing the encoded spike train. (values in {-1, 0, +1})
+ - thresholds: Per-feature or channel thresholds used for encoding, returned for use in decoding,
+ shape (features or channels,).
+ :rtype: tuple[numpy.ndarray, numpy.ndarray]
+ :raises ValueError: If the input signal is empty or if the threshold dimensions do not match the signal f
+ eature dimensions.
"""
@@ -73,7 +75,7 @@ def moving_window(
else:
thresholds = np.asarray(threshold, dtype=float)
if thresholds.ndim != 1:
- raise TypeError("Threshold must be a scalar or a 1D sequence of numbers.")
+ raise ValueError("Threshold must be a scalar or a 1D sequence of numbers.")
if thresholds.size != F:
raise ValueError("Threshold must match the number of features in the signal.")
@@ -100,8 +102,4 @@ def moving_window(
elif signal[t, f] < base - thresholds[f]:
spikes[t, f] = -1
- # Flatten if input was 1D
- if F == 1:
- spikes = spikes.flatten()
-
- return spikes
+ return spikes, thresholds
diff --git a/spikify/encoders/temporal/contrast/step_forward_algorithm.py b/spikify/encoders/temporal/contrast/step_forward_algorithm.py
index 8dfc6ef..7131946 100644
--- a/spikify/encoders/temporal/contrast/step_forward_algorithm.py
+++ b/spikify/encoders/temporal/contrast/step_forward_algorithm.py
@@ -7,7 +7,9 @@
import numpy as np
-def step_forward(signal: np.ndarray, threshold: float | int | list[float | int] | np.ndarray) -> np.ndarray:
+def step_forward(
+ signal: np.ndarray, threshold: float | int | list[float | int] | np.ndarray
+) -> tuple[np.ndarray, np.ndarray]:
"""
Perform Step-Forward (SF) encoding on the input signal.
@@ -25,7 +27,7 @@ def step_forward(signal: np.ndarray, threshold: float | int | list[float | int]
from spikify.encoders.temporal.contrast import step_forward
signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6])
threshold = 0.2
- encoded_signal = step_forward(signal, threshold)
+ encoded_signal, thresholds = step_forward(signal, threshold)
.. doctest::
:hide:
@@ -34,18 +36,21 @@ def step_forward(signal: np.ndarray, threshold: float | int | list[float | int]
>>> from spikify.encoders.temporal.contrast import step_forward
>>> signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6])
>>> threshold = 0.2
- >>> encoded_signal = step_forward(signal, threshold)
- >>> encoded_signal
+ >>> encoded_signal, _ = step_forward(signal, threshold)
+ >>> encoded_signal.flatten()
array([0, 0, 1, 0, 0, 1], dtype=int8)
- :param signal: The input signal to be encoded. This should be a numpy ndarray.
+ :param signal: Input signal to encode (1D or 2D: time × features or channels).
:type signal: numpy.ndarray
:param threshold: Threshold(s) for spike generation; scalar or 1D sequence matching features.
:type threshold: float | int | list[float | int] | numpy.ndarray
- :return: A numpy array representing the encoded spike train.
- :rtype: numpy.ndarray
- :raises ValueError: If the input signal is empty or if the threshold dimensions do not match the signal features.
- :raises TypeError: If the threshold parameter is of invalid dimension.
+ :return:
+ - spikes: A numpy array representing the encoded spike train. (values in {-1, 0, +1})
+ - thresholds: Per-feature or channel thresholds used for encoding, returned for use in decoding,
+ shape (features or channels,).
+ :rtype: tuple[numpy.ndarray, numpy.ndarray]
+ :raises ValueError: If the input signal is empty or if the threshold dimensions do not match the signal
+ features dimensions.
"""
@@ -65,7 +70,7 @@ def step_forward(signal: np.ndarray, threshold: float | int | list[float | int]
else:
thresholds = np.asarray(threshold, dtype=float)
if thresholds.ndim != 1:
- raise TypeError("Threshold must be a scalar or a 1D sequence of numbers.")
+ raise ValueError("Threshold must be a scalar or a 1D sequence of numbers.")
if thresholds.size != F:
raise ValueError("Threshold must match the number of features in the signal.")
@@ -86,8 +91,4 @@ def step_forward(signal: np.ndarray, threshold: float | int | list[float | int]
spike[t, feat] = -1
base -= thresholds[feat]
- # Flatten if input was 1D
- if F == 1:
- spike = spike.flatten()
-
- return spike
+ return spike, thresholds
diff --git a/spikify/encoders/temporal/contrast/threshold_based_algorithm.py b/spikify/encoders/temporal/contrast/threshold_based_algorithm.py
index 1862b06..d3cbb09 100644
--- a/spikify/encoders/temporal/contrast/threshold_based_algorithm.py
+++ b/spikify/encoders/temporal/contrast/threshold_based_algorithm.py
@@ -36,19 +36,21 @@ def threshold_based_representation(
>>> from spikify.encoders.temporal.contrast import threshold_based_representation
>>> signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6])
>>> factor = 0.5
- >>> encoded_signal, threshold = threshold_based_representation(signal, factor)
- >>> encoded_signal
+ >>> encoded_signal, _ = threshold_based_representation(signal, factor)
+ >>> encoded_signal.flatten()
array([ 1, 0, -1, 1, 0, 0], dtype=int8)
- :param signal: The input signal to be encoded. This should be a numpy ndarray.
+ :param signal: Input signal to encode (1D or 2D: time × features or channels).
:type signal: numpy.ndarray
:param factor: The factor value (`factor`) that controls the noise-reduction threshold.
Can be a float, an integer, or a list of floats or integers.
:type factor: float | int | list[float | int] | numpy.ndarray
- :return: A tuple containing the encoded spike train and the computed threshold for each feature.
+ :return:
+ - spikes: A numpy array representing the encoded spike train. (values in {-1, 0, +1})
+ - thresholds: Per-feature or channel thresholds used for encoding, returned for use in decoding,
+ shape (features or channels,).
:rtype: tuple[numpy.ndarray, numpy.ndarray]
:raises ValueError: If the input signal is empty or if the factor length does not match the number of features.
- :raises TypeError: If the factor parameter is of invalid dimension.
"""
@@ -68,7 +70,7 @@ def threshold_based_representation(
else:
factors = np.asarray(factor, dtype=float)
if factors.ndim != 1:
- raise TypeError("Factor must be a scalar or a 1D sequence of numbers.")
+ raise ValueError("Factor must be a scalar or a 1D sequence of numbers.")
if factors.size != F:
raise ValueError("Factor must match the number of features in the signal.")
spike = np.zeros_like(signal, dtype=np.int8)
@@ -87,8 +89,7 @@ def threshold_based_representation(
spike[diff > threshold] = 1
spike[diff < -threshold] = -1
- # Flatten if input was 1D
- if F == 1:
- spike = spike.flatten()
+ # Reshape threshold to 1D for return
+ threshold = threshold.flatten()
- return spike, threshold.flatten()
+ return spike, threshold
diff --git a/spikify/encoders/temporal/contrast/zero_cross_step_forward_algorithm.py b/spikify/encoders/temporal/contrast/zero_cross_step_forward_algorithm.py
index 1ce4b36..d141380 100644
--- a/spikify/encoders/temporal/contrast/zero_cross_step_forward_algorithm.py
+++ b/spikify/encoders/temporal/contrast/zero_cross_step_forward_algorithm.py
@@ -7,7 +7,9 @@
import numpy as np
-def zero_cross_step_forward(signal: np.ndarray, threshold: float | int | list[float | int] | np.ndarray) -> np.ndarray:
+def zero_cross_step_forward(
+ signal: np.ndarray, threshold: float | int | list[float | int] | np.ndarray
+) -> tuple[np.ndarray, np.ndarray]:
"""
Perform Zero-Crossing Step-Forward (ZCSF) encoding on the input signal.
@@ -26,7 +28,7 @@ def zero_cross_step_forward(signal: np.ndarray, threshold: float | int | list[fl
from spikify.encoders.temporal.contrast import zero_cross_step_forward
signal = np.array([-0.2, 0.1, 0.5, 0.0, 1.2, 0.3])
threshold = 0.4
- encoded_signal = zero_cross_step_forward(signal, threshold)
+ encoded_signal, thresholds = zero_cross_step_forward(signal, threshold)
.. doctest::
:hide:
@@ -35,18 +37,20 @@ def zero_cross_step_forward(signal: np.ndarray, threshold: float | int | list[fl
>>> from spikify.encoders.temporal.contrast import zero_cross_step_forward
>>> signal = np.array([-0.2, 0.1, 0.5, 0.0, 1.2, 0.3])
>>> threshold = 0.4
- >>> encoded_signal = zero_cross_step_forward(signal, threshold)
- >>> encoded_signal
+ >>> encoded_signal, _ = zero_cross_step_forward(signal, threshold)
+ >>> encoded_signal.flatten()
array([0, 0, 1, 0, 1, 0], dtype=int8)
- :param signal: The input signal to be encoded. This should be a numpy ndarray.
+ :param signal: Input signal to encode (1D or 2D: time × features or channels).
:type signal: numpy.ndarray
:param threshold: Threshold(s) for spike generation; scalar or 1D sequence matching features.
:type threshold: float | int | list[float | int] | numpy.ndarray
- :return: A numpy array representing the encoded spike train.
- :rtype: numpy.ndarray
+ :return:
+ - spikes: A numpy array representing the encoded spike train. (values in {0, +1})
+ - thresholds: Per-feature or channel thresholds used for encoding, returned for use in decoding,
+ shape (features or channels,).
+ :rtype: tuple[numpy.ndarray, numpy.ndarray]
:raises ValueError: If the input signal is empty or if the threshold dimensions do not match the signal features.
- :raises TypeError: If the threshold parameter is of invalid dimension.
"""
@@ -66,7 +70,7 @@ def zero_cross_step_forward(signal: np.ndarray, threshold: float | int | list[fl
else:
thresholds = np.asarray(threshold, dtype=float)
if thresholds.ndim != 1:
- raise TypeError("Threshold must be a scalar or a 1D sequence of numbers.")
+ raise ValueError("Threshold must be a scalar or a 1D sequence of numbers.")
if thresholds.size != F:
raise ValueError("Threshold must match the number of features in the signal.")
@@ -78,8 +82,4 @@ def zero_cross_step_forward(signal: np.ndarray, threshold: float | int | list[fl
# Apply threshold condition
spike[signal > thresholds] = 1
- # Flatten if input was 1D
- if F == 1:
- spike = spike.flatten()
-
- return spike
+ return spike, thresholds
diff --git a/spikify/encoders/temporal/deconvolution/bens_spiker_algorithm.py b/spikify/encoders/temporal/deconvolution/bens_spiker_algorithm.py
index 943f384..07c6913 100644
--- a/spikify/encoders/temporal/deconvolution/bens_spiker_algorithm.py
+++ b/spikify/encoders/temporal/deconvolution/bens_spiker_algorithm.py
@@ -55,7 +55,7 @@ def bens_spiker(
window_length = 3
threshold = 0.1
cutoff = 0.1
- encoded_signal, shift, fir_coeffs = bens_spiker(signal, window_length, cutoff, threshold)
+ encoded_signal, fir_coeffs, shift = bens_spiker(signal, window_length, cutoff, threshold)
.. doctest::
:hide:
@@ -66,11 +66,11 @@ def bens_spiker(
>>> window_length = 3
>>> threshold = 0.1
>>> cutoff = 0.1
- >>> encoded_signal, shift, fir_coeffs = bens_spiker(signal, window_length, cutoff, threshold)
- >>> encoded_signal
+ >>> encoded_signal, _, _ = bens_spiker(signal, window_length, cutoff, threshold)
+ >>> encoded_signal.flatten()
array([0, 1, 1, 0, 0, 0, 0], dtype=int8)
- :param signal: Input signal to encode (1D or 2D: time × features).
+ :param signal: Input signal to encode (1D or 2D: time × features or channels).
:type signal: numpy.ndarray
:param window_length: Length of the FIR filter (number of coefficients).
:type window_length: int
@@ -92,13 +92,12 @@ def bens_spiker(
:param fs: Sampling frequency (used for physical frequency units in cutoff; optional).
:type fs: float | None
:return:
- - spikes: Spike train (same shape as input, dtype=int8, values 0 or 1)
- - shift: Per-feature shift values subtracted to make signal non-negative (1D array)
- - fir_bank: Final filter coefficients used (after scaling), shape (window_length, n_features)
+ - spikes: A numpy array representing the encoded spike train. (values in {0, +1})
+ - fir_bank: Final filter coefficients used, shape (window_length, features or channels).
+ - shift: Per-feature shift values subtracted to make signal non-negative, shape (features or channels,).
:rtype: tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray]
:raises ValueError: If the input signal is empty or if the threshold dimensions do not match the signal
- features or if the window_length is greater than the signal lenght.
- :raises TypeError: If the threshold parameter is of invalid dimension.
+ feature dimensions or if the window_length is greater than the signal lenght.
"""
@@ -122,7 +121,7 @@ def bens_spiker(
else:
thresholds = np.asarray(threshold, dtype=float)
if thresholds.ndim != 1:
- raise TypeError("Threshold must be a scalar or a 1D sequence of numbers.")
+ raise ValueError("Threshold must be a scalar or a 1D sequence of numbers.")
if thresholds.size != F:
raise ValueError("Threshold must match the number of features in the signal.")
@@ -161,8 +160,4 @@ def bens_spiker(
spikes[t, f] = 1
signal_copy[t : t + window_length, f] -= fir_bank[:, f] # update signal by removing filter effect
- # Flatten if input was 1D
- if F == 1:
- spikes = spikes.flatten()
-
- return spikes, shift, fir_bank
+ return spikes, fir_bank, shift
diff --git a/spikify/encoders/temporal/deconvolution/hough_spiker_algorithm.py b/spikify/encoders/temporal/deconvolution/hough_spiker_algorithm.py
index b2f6438..fb1d662 100644
--- a/spikify/encoders/temporal/deconvolution/hough_spiker_algorithm.py
+++ b/spikify/encoders/temporal/deconvolution/hough_spiker_algorithm.py
@@ -46,7 +46,7 @@ def hough_spiker(
signal = np.array([0.1, 0.2, 4.1, 1.0, 3.0, 0.3, 0.1])
window_length = 3
cutoff = 0.1
- encoded_signal, shift, norm, fir_coeffs = hough_spiker(signal, window_length, cutoff)
+ encoded_signal, fir_coeffs, shift, norm = hough_spiker(signal, window_length, cutoff)
.. doctest::
:hide:
@@ -56,11 +56,11 @@ def hough_spiker(
>>> signal = np.array([0.1, 0.2, 4.1, 1.0, 3.0, 0.3, 0.1])
>>> window_length = 3
>>> cutoff = 0.1
- >>> encoded_signal, shift, norm, fir_coeffs = hough_spiker(signal, window_length, cutoff)
- >>> encoded_signal
+ >>> encoded_signal, _, _, _ = hough_spiker(signal, window_length, cutoff)
+ >>> encoded_signal.flatten()
array([0, 1, 0, 0, 0, 0, 0], dtype=int8)
- :param signal: Input signal to encode (1D or 2D: time × features).
+ :param signal: Input signal to encode (1D or 2D: time × features or channels).
:type signal: numpy.ndarray
:param window_length: Length of the FIR filter (number of coefficients).
:type window_length: int
@@ -79,10 +79,10 @@ def hough_spiker(
:param fs: Sampling frequency (used for physical frequency units in cutoff; optional).
:type fs: float | None
:return:
- - spikes: Spike train (same shape as input, dtype=int8, values 0 or 1)
- - shift: Per-feature shift values subtracted to make signal non-negative (1D array)
- - norm: Per-feature normalization values used to scale signal to [0, 1] (1D array)
- - fir_bank: Final filter coefficients used, shape (window_length, n_features)
+ - spikes: A numpy array representing the encoded spike train. (values in {0, +1})
+ - fir_bank: Final filter coefficients used (window_length, features or channels).
+ - shift: Per-feature shift values subtracted to make signal non-negative, shape (features or channels,).
+ - norm: Per-feature normalization values used to scale signal to [0, 1], shape (features or channels,).
:rtype: tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]
:raises ValueError: If the input signal is empty or if the window_length is greater than the signal lenght.
@@ -129,8 +129,4 @@ def hough_spiker(
signal_copy[t : t + window_length, f] -= fir_bank[:, f]
spikes[t, f] = 1
- # Flatten if input was 1D
- if F == 1:
- spikes = spikes.flatten()
-
- return spikes, shift, norm, fir_bank
+ return spikes, fir_bank, shift, norm
diff --git a/spikify/encoders/temporal/deconvolution/modified_hough_spiker_algorithm.py b/spikify/encoders/temporal/deconvolution/modified_hough_spiker_algorithm.py
index fdc78e7..b568daa 100644
--- a/spikify/encoders/temporal/deconvolution/modified_hough_spiker_algorithm.py
+++ b/spikify/encoders/temporal/deconvolution/modified_hough_spiker_algorithm.py
@@ -54,7 +54,7 @@ def modified_hough_spiker(
window_length = 3
threshold = 0.5
cutoff = 0.1
- encoded_signal, shift, norm, fir_coeffs = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ encoded_signal, fir_coeffs, shift, norm, = modified_hough_spiker(signal, window_length, cutoff, threshold)
.. doctest::
:hide:
@@ -65,11 +65,11 @@ def modified_hough_spiker(
>>> window_length = 3
>>> threshold = 0.5
>>> cutoff = 0.1
- >>> encoded_signal, shift, norm, fir_coeffs = modified_hough_spiker(signal, window_length, cutoff, threshold)
- >>> encoded_signal
+ >>> encoded_signal, _, _, _ = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ >>> encoded_signal.flatten()
array([0, 0, 1, 1, 0, 0, 1], dtype=int8)
- :param signal: Input signal to encode (1D or 2D: time × features).
+ :param signal: Input signal to encode (1D or 2D: time × features or channels).
:type signal: numpy.ndarray
:param window_length: Length of the FIR filter (number of coefficients).
:type window_length: int
@@ -91,14 +91,13 @@ def modified_hough_spiker(
:param fs: Sampling frequency (used for physical frequency units in cutoff; optional).
:type fs: float | None
:return:
- - spikes: Spike train (same shape as input, dtype=int8, values 0 or 1)
- - shift: Per-feature shift values subtracted to make signal non-negative (1D array)
- - norm: Per-feature normalization values used to scale signal to [0, 1] (1D array)
- - fir_bank: Final filter coefficients used, shape (window_length, n_features)
+ - spikes: A numpy array representing the encoded spike train. (values in {0, +1})
+ - fir_bank: Final filter coefficients used, shape (window_length, features or channels).
+ - shift: Per-feature shift values subtracted to make signal non-negative, shape (features or channels,).
+ - norm: Per-feature normalization values used to scale signal to [0, 1], shape (features or channels,).
:rtype: tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray, numpy.ndarray]
:raises ValueError: If the input signal is empty or if the threshold dimensions do not match the signal
features or if the window_length is greater than the signal lenght.
- :raises TypeError: If the threshold parameter is of invalid dimension.
"""
# Check for empty signal
@@ -121,7 +120,7 @@ def modified_hough_spiker(
else:
thresholds = np.asarray(threshold, dtype=float)
if thresholds.ndim != 1:
- raise TypeError("Threshold must be a scalar or a 1D sequence of numbers.")
+ raise ValueError("Threshold must be a scalar or a 1D sequence of numbers.")
if thresholds.size != F:
raise ValueError("Threshold must match the number of features in the signal.")
@@ -163,8 +162,4 @@ def modified_hough_spiker(
signal_copy[t:end_index, f] -= filter_segment
spikes[t, f] = 1
- # Flatten if input was 1D
- if F == 1:
- spikes = spikes.flatten()
-
- return spikes, shift, norm, fir_bank
+ return spikes, fir_bank, shift, norm
diff --git a/tests/decoders/__init__.py b/tests/decoders/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/decoders/temporal/contrast/test_contrast_decoder.py b/tests/decoders/temporal/contrast/test_contrast_decoder.py
new file mode 100644
index 0000000..73cfc6a
--- /dev/null
+++ b/tests/decoders/temporal/contrast/test_contrast_decoder.py
@@ -0,0 +1,263 @@
+import unittest
+import numpy as np
+from spikify.decoders.temporal.contrast.decoder_algorithm import contrast_decoder
+from spikify.encoders.temporal.contrast import (
+ threshold_based_representation,
+ step_forward,
+ moving_window,
+ zero_cross_step_forward,
+)
+
+
+class TestContrastDecoderInput(unittest.TestCase):
+
+ def test_empty_spike_train_raises(self):
+ with self.assertRaises(ValueError):
+ contrast_decoder(np.array([]), thresholds=np.array([0.2]), start_point=0.1)
+
+ def test_invalid_start_point_dimension_raises(self):
+ spikes = np.array([0, 1, -1, 0], dtype=np.int8)
+ with self.assertRaises(ValueError):
+ contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=np.array([[0.1, 0.2], [0.3, 0.4]]))
+
+ def test_start_point_size_mismatch_raises(self):
+ spikes = np.array([[0, 1], [-1, 0], [1, -1]], dtype=np.int8)
+ with self.assertRaises(ValueError):
+ contrast_decoder(spikes, thresholds=np.array([0.2, 0.3]), start_point=[0.1, 0.2, 0.3])
+
+ def test_output_is_ndarray(self):
+ spikes = np.array([0, 1, -1, 0], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.1)
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_1d_input_returns_2d_output(self):
+ spikes = np.array([0, 1, -1, 0], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.1)
+ self.assertEqual(result.ndim, 2)
+
+ def test_2d_input_returns_2d_output(self):
+ spikes = np.array([[0, 1], [-1, 0], [1, -1]], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2, 0.2]), start_point=0.1)
+ self.assertEqual(result.ndim, 2)
+
+ def test_output_length_matches_input(self):
+ spikes = np.array([0, 1, -1, 0, 1], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.1)
+ self.assertEqual(len(result), len(spikes))
+
+ def test_output_shape_matches_2d_input(self):
+ spikes = np.array([[0, 1], [-1, 0], [1, -1]], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2, 0.2]), start_point=0.1)
+ self.assertEqual(result.shape, spikes.shape)
+
+ def test_all_zeros_returns_constant_signal(self):
+ spikes = np.array([0, 0, 0, 0], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.5)
+ expected = np.array([0.5, 0.5, 0.5, 0.5]).reshape(-1, 1)
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_all_positive_spikes_increments(self):
+ spikes = np.array([0, 1, 1, 1], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.1)
+ expected = np.array([0.1, 0.3, 0.5, 0.7]).reshape(-1, 1)
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_all_negative_spikes_decrements(self):
+ spikes = np.array([0, -1, -1, -1], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.7)
+ expected = np.array([0.7, 0.5, 0.3, 0.1]).reshape(-1, 1)
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_mixed_spikes_reconstruction(self):
+ spikes = np.array([0, 0, 0, 1, 1, 1], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.1)
+ expected = np.array([0.1, 0.1, 0.1, 0.3, 0.5, 0.7]).reshape(-1, 1)
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_alternating_spikes_returns_to_start(self):
+ spikes = np.array([0, 1, -1, 1, -1], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2]), start_point=0.5)
+ expected = np.array([0.5, 0.7, 0.5, 0.7, 0.5]).reshape(-1, 1)
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_scalar_start_point(self):
+ spikes = np.array([0, 1, 0, -1], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.1]), start_point=0.0)
+ expected = np.array([0.0, 0.1, 0.1, 0.0]).reshape(-1, 1)
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_multi_feature_ndarray_threshold_and_list_start_point(self):
+ spikes = np.array([[0, 0], [1, -1], [0, 1]], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2, 0.3]), start_point=[0.1, 0.5])
+ expected = np.array([[0.1, 0.5], [0.3, 0.2], [0.3, 0.5]])
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_multi_feature_ndarray_threshold_and_ndarray_start_point(self):
+ spikes = np.array([[0, 0], [1, 1]], dtype=np.int8)
+ result = contrast_decoder(spikes, thresholds=np.array([0.2, 0.4]), start_point=np.array([0.0, 0.0]))
+ expected = np.array([[0.0, 0.0], [0.2, 0.4]])
+ np.testing.assert_array_almost_equal(result, expected)
+
+
+class TestContrastDecoderRoundTripTBR(unittest.TestCase):
+
+ def test_round_trip_output_shape(self):
+ signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6]).reshape(-1, 1)
+ spikes, thresholds = threshold_based_representation(signal, factor=0.5)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_output_is_ndarray(self):
+ signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6])
+ spikes, thresholds = threshold_based_representation(signal, factor=0.5)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_round_trip_starts_at_signal_start_point(self):
+ signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6])
+ spikes, thresholds = threshold_based_representation(signal, factor=0.5)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertAlmostEqual(result[0], signal[0])
+
+ def test_round_trip_multi_feature(self):
+ signal = np.array([[0.1, 0.5], [0.3, 0.4], [0.4, 0.6], [0.2, 0.3], [0.5, 0.7], [0.6, 0.8]])
+ spikes, thresholds = threshold_based_representation(signal, factor=0.5)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_zero_factor_all_zeros(self):
+ signal = np.array([0.5, 0.5, 0.5, 0.5, 0.5]).reshape(-1, 1)
+ spikes, thresholds = threshold_based_representation(signal, factor=0.0)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, signal[0]))
+
+
+class TestContrastDecoderRoundTripSF(unittest.TestCase):
+
+ def test_round_trip_output_shape(self):
+ signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6]).reshape(-1, 1)
+ spikes, thresholds = step_forward(signal, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_output_is_ndarray(self):
+ signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6])
+ spikes, thresholds = step_forward(signal, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_round_trip_starts_at_signal_start_point(self):
+ signal = np.array([0.1, 0.3, 0.4, 0.2, 0.5, 0.6])
+ spikes, thresholds = step_forward(signal, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertAlmostEqual(result[0], signal[0])
+
+ def test_round_trip_flat_signal_no_spikes(self):
+ signal = np.array([0.5, 0.5, 0.5, 0.5, 0.5]).reshape(-1, 1)
+ spikes, thresholds = step_forward(signal, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, signal[0]))
+
+ def test_round_trip_monotonic_increase(self):
+ signal = np.array([0.0, 0.2, 0.4, 0.6, 0.8, 1.0]).reshape(-1, 1)
+ spikes, thresholds = step_forward(signal, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+ self.assertTrue(np.all(result >= signal[0]))
+
+ def test_round_trip_multi_feature(self):
+ signal = np.array([[0.1, 0.5], [0.3, 0.4], [0.4, 0.6], [0.2, 0.3], [0.5, 0.7], [0.6, 0.8]])
+ spikes, thresholds = step_forward(signal, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+
+class TestContrastDecoderRoundTripMW(unittest.TestCase):
+
+ def test_round_trip_output_shape(self):
+ signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0]).reshape(-1, 1)
+ spikes, thresholds = moving_window(signal, window_length=3, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_output_is_ndarray(self):
+ signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0])
+ spikes, thresholds = moving_window(signal, window_length=3, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_round_trip_starts_at_signal_start_point(self):
+ signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0])
+ spikes, thresholds = moving_window(signal, window_length=3, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertAlmostEqual(result[0], signal[0])
+
+ def test_round_trip_flat_signal_no_spikes(self):
+ signal = np.array([0.5, 0.5, 0.5, 0.5, 0.5]).reshape(-1, 1)
+ spikes, thresholds = moving_window(signal, window_length=3, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, signal[0]))
+
+ def test_round_trip_multi_feature(self):
+ signal = np.array([[0.1, 0.5], [0.3, 0.4], [0.4, 0.6], [0.2, 0.3], [0.5, 0.7], [0.6, 0.8]])
+ spikes, thresholds = moving_window(signal, window_length=3, threshold=0.2)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_window_length_one(self):
+ signal = np.array([0.1, 0.3, 0.2, 0.5, 0.8, 1.0]).reshape(-1, 1)
+ spikes, thresholds = moving_window(signal, window_length=1, threshold=0.1)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+
+class TestContrastDecoderRoundTripZCSF(unittest.TestCase):
+
+ def test_round_trip_output_shape(self):
+ signal = np.array([-0.2, 0.1, 0.5, 0.0, 1.2, 0.3]).reshape(-1, 1)
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=0.4)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_output_is_ndarray(self):
+ signal = np.array([-0.2, 0.1, 0.5, 0.0, 1.2, 0.3])
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=0.4)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_round_trip_starts_at_signal_start_point(self):
+ signal = np.array([-0.2, 0.1, 0.5, 0.0, 1.2, 0.3])
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=0.4)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertAlmostEqual(result[0], signal[0])
+
+ def test_round_trip_all_negative_signal_no_spikes(self):
+ signal = np.array([-0.5, -0.3, -0.1, -0.8, -0.2]).reshape(-1, 1)
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=0.4)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, signal[0]))
+
+ def test_round_trip_all_below_threshold_no_spikes(self):
+ signal = np.array([0.1, 0.2, 0.1, 0.3, 0.2]).reshape(-1, 1)
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=0.4)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, signal[0]))
+
+ def test_round_trip_only_positive_spikes(self):
+ signal = np.array([-0.2, 0.1, 0.5, 0.0, 1.2, 0.3])
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=0.4)
+ self.assertTrue(np.all(spikes >= 0))
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertTrue(np.all(np.diff(result) >= 0))
+
+ def test_round_trip_multi_feature(self):
+ signal = np.array([[-0.2, 0.5], [0.1, 0.0], [0.5, 1.2], [0.0, 0.3], [1.2, 0.8], [0.3, 0.6]])
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=0.4)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_high_threshold_no_spikes(self):
+ signal = np.array([0.1, 0.3, 0.2, 0.5, 0.4]).reshape(-1, 1)
+ spikes, thresholds = zero_cross_step_forward(signal, threshold=999.0)
+ result = contrast_decoder(spikes, thresholds, start_point=signal[0])
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, signal[0]))
diff --git a/tests/decoders/temporal/deconvolution/test_deconvolution_decoder.py b/tests/decoders/temporal/deconvolution/test_deconvolution_decoder.py
new file mode 100644
index 0000000..1fcb7ef
--- /dev/null
+++ b/tests/decoders/temporal/deconvolution/test_deconvolution_decoder.py
@@ -0,0 +1,283 @@
+import unittest
+import numpy as np
+from spikify.decoders.temporal.deconvolution.decoder_algorithm import deconvolution_decoder
+from spikify.encoders.temporal.deconvolution import (
+ bens_spiker,
+ hough_spiker,
+ modified_hough_spiker,
+)
+
+
+class TestDeconvolutionDecoderInput(unittest.TestCase):
+
+ def test_empty_spike_train_raises(self):
+ with self.assertRaises(ValueError):
+ deconvolution_decoder(
+ np.array([]),
+ fir_bank=np.array([[0.25, 0.5, 0.25]]).T,
+ shift=np.array([0.0]),
+ )
+
+ def test_output_is_ndarray(self):
+ spikes = np.array([0, 1, 0, 0], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift = np.array([0.0])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_1d_input_returns_2d_output(self):
+ spikes = np.array([0, 1, 0, 0], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift = np.array([0.0])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(result.ndim, 2)
+
+ def test_2d_input_returns_2d_output(self):
+ spikes = np.array([[0, 0], [1, 0], [0, 1], [0, 0]], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25], [0.25, 0.5, 0.25]]).T
+ shift = np.array([0.0, 0.0])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(result.ndim, 2)
+
+ def test_output_length_matches_input(self):
+ spikes = np.array([0, 1, 0, 0, 1], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift = np.array([0.0])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(len(result), len(spikes))
+
+ def test_output_shape_matches_2d_input(self):
+ spikes = np.array([[0, 0], [1, 0], [0, 1], [0, 0]], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25], [0.25, 0.5, 0.25]]).T
+ shift = np.array([0.0, 0.0])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(result.shape, spikes.shape)
+
+ def test_all_zeros_with_no_norm_returns_only_shift(self):
+ spikes = np.array([0, 0, 0, 0], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift = np.array([0.1])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=None)
+ expected = np.full((4, 1), 0.1)
+ np.testing.assert_array_almost_equal(result, expected)
+
+ def test_norm_none_applies_only_shift(self):
+ spikes = np.array([0, 0, 0, 0], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift = np.array([0.5])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=None)
+ self.assertTrue(np.all(result == 0.5))
+
+ def test_norm_provided_scales_output(self):
+ spikes = np.array([1, 0, 0, 0], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift = np.array([0.0])
+ norm = np.array([2.0])
+ result_with_norm = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ result_no_norm = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=None)
+ # with norm=2.0, output should be 2x larger (before shift)
+ np.testing.assert_array_almost_equal(result_with_norm, result_no_norm * 2.0)
+
+ def test_single_spike_produces_nonzero_output(self):
+ spikes = np.array([1, 0, 0, 0, 0], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift = np.array([0.0])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertTrue(np.any(result != 0.0))
+
+ def test_shift_offsets_output(self):
+ spikes = np.array([0, 0, 0, 0], dtype=np.int8)
+ fir_bank = np.array([[0.25, 0.5, 0.25]]).T
+ shift_a = np.array([0.0])
+ shift_b = np.array([1.0])
+ result_a = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift_a)
+ result_b = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift_b)
+ np.testing.assert_array_almost_equal(result_b - result_a, np.full((4, 1), 1.0))
+
+ def test_multi_feature_output_shape(self):
+ spikes = np.array([[0, 1], [1, 0], [0, 0], [0, 1]], dtype=np.int8)
+ fir = np.array([0.25, 0.5, 0.25])
+ fir_bank = np.stack([fir, fir], axis=0).T
+ shift = np.array([0.0, 0.1])
+ norm = np.array([1.0, 2.0])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape, spikes.shape)
+
+ def test_multi_feature_independent_shift(self):
+ spikes = np.zeros((4, 2), dtype=np.int8)
+ fir = np.array([0.25, 0.5, 0.25])
+ fir_bank = np.stack([fir, fir], axis=0).T
+ shift = np.array([0.2, 0.8])
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=None)
+ np.testing.assert_array_almost_equal(result[:, 0], np.full(4, 0.2))
+ np.testing.assert_array_almost_equal(result[:, 1], np.full(4, 0.8))
+
+
+class TestDeconvolutionDecoderRoundTripBSA(unittest.TestCase):
+
+ def test_round_trip_output_shape(self):
+ signal = np.array([0.1, 0.2, 0.8, 0.95, 0.5, 0.3, 0.1]).reshape(-1, 1)
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_output_is_ndarray(self):
+ signal = np.array([0.1, 0.2, 0.8, 0.95, 0.5, 0.3, 0.1])
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_round_trip_flat_signal_no_spikes(self):
+ signal = np.array([0.5, 0.5, 0.5, 0.5, 0.5]).reshape(-1, 1)
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ self.assertTrue(np.all(spikes == 0))
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, shift[0]))
+
+ def test_round_trip_positive_signal_no_shift(self):
+ signal = np.array([0.1, 0.3, 0.6, 0.8, 0.5])
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ np.testing.assert_array_almost_equal(shift, np.array([0.0]))
+
+ def test_round_trip_negative_signal_applies_shift(self):
+ signal = np.array([-0.5, 0.1, 0.3, 0.6, 0.2])
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ self.assertAlmostEqual(shift[0], -0.5)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(result.shape[0], len(signal))
+
+ def test_round_trip_bsa_does_not_return_norm(self):
+ signal = np.array([0.1, 0.3, 0.6, 0.8, 0.5, 0.2, 0.4])
+ result_tuple = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ # BSA returns (spikes, fir_bank, shift) — no norm
+ self.assertEqual(len(result_tuple), 3)
+
+ def test_round_trip_multi_feature(self):
+ signal = np.array([[0.1, 0.5], [0.3, 0.4], [0.6, 0.9], [0.8, 0.7], [0.5, 0.3], [0.2, 0.1]])
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_high_threshold_no_spikes(self):
+ signal = np.array([0.1, 0.3, 0.6, 0.8, 0.5, 0.2, 0.4]).reshape(-1, 1)
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=999.0)
+ self.assertTrue(np.all(spikes == 0))
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ np.testing.assert_array_almost_equal(result, np.full_like(signal, shift[0]))
+
+ def test_round_trip_output_non_negative_for_non_negative_signal(self):
+ signal = np.array([0.1, 0.4, 0.7, 0.9, 0.6, 0.3, 0.2])
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.05)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertTrue(np.all(result >= 0))
+
+ def test_round_trip_large_amplitude_signal(self):
+ signal = np.array([1.0, 2.5, 5.0, 3.0, 1.5, 0.5, 0.2])
+ spikes, fir_bank, shift = bens_spiker(signal, window_length=3, cutoff=0.1, threshold=0.1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift)
+ self.assertEqual(result.shape[0], len(signal))
+
+
+class TestDeconvolutionDecoderRoundTripMHSA(unittest.TestCase):
+
+ def test_round_trip_output_shape(self):
+ signal = np.array([0.1, 0.2, 0.3, 1.0, 0.5, 0.3, 0.1]).reshape(-1, 1)
+ spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.5)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_output_is_ndarray(self):
+ signal = np.array([0.1, 0.2, 0.3, 1.0, 0.5, 0.3, 0.1])
+ spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.5)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_round_trip_mhsa_returns_norm(self):
+ signal = np.array([0.1, 0.3, 0.6, 0.8, 0.5, 0.2, 0.4])
+ result_tuple = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.5)
+ # MHSA returns (spikes, fir_bank, shift, norm)
+ self.assertEqual(len(result_tuple), 4)
+
+ def test_round_trip_negative_signal_applies_shift(self):
+ signal = np.array([-0.5, 0.1, 0.3, 0.6, 0.2, 0.4, 0.1])
+ spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.5)
+ self.assertAlmostEqual(shift[0], -0.5)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape[0], len(signal))
+
+ def test_round_trip_multi_feature(self):
+ signal = np.array([[0.1, 0.5], [0.3, 0.4], [0.4, 0.6], [0.2, 0.3], [0.5, 0.7], [0.6, 0.8]])
+ spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.5)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_large_amplitude_applies_norm(self):
+ signal = np.array([1.0, 2.5, 5.0, 3.0, 1.5, 0.5, 0.2])
+ spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.5)
+ # norm should be > 1 since signal exceeds 1 after shift
+ self.assertTrue(norm[0] > 1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape[0], len(signal))
+
+ def test_round_trip_only_unipolar_spikes(self):
+ signal = np.array([0.1, 0.2, 0.3, 1.0, 0.5, 0.3, 0.1])
+ spikes, fir_bank, shift, norm = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.5)
+ self.assertTrue(np.all(spikes >= 0))
+
+
+class TestDeconvolutionDecoderRoundTripHSA(unittest.TestCase):
+
+ def test_round_trip_output_shape(self):
+ signal = np.array([0.1, 0.2, 4.1, 1.0, 3.0, 0.3, 0.1]).reshape(-1, 1)
+ spikes, fir_bank, shift, norm = hough_spiker(signal, window_length=3, cutoff=0.1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_output_is_ndarray(self):
+ signal = np.array([0.1, 0.2, 4.1, 1.0, 3.0, 0.3, 0.1])
+ spikes, fir_bank, shift, norm = hough_spiker(signal, window_length=3, cutoff=0.1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertIsInstance(result, np.ndarray)
+
+ def test_round_trip_hsa_returns_norm(self):
+ signal = np.array([0.1, 0.3, 0.6, 0.8, 0.5, 0.2, 0.4])
+ result_tuple = hough_spiker(signal, window_length=3, cutoff=0.1)
+ # HSA returns (spikes, fir_bank, shift, norm)
+ self.assertEqual(len(result_tuple), 4)
+
+ def test_round_trip_only_unipolar_spikes(self):
+ signal = np.array([0.1, 0.2, 4.1, 1.0, 3.0, 0.3, 0.1])
+ spikes, fir_bank, shift, norm = hough_spiker(signal, window_length=3, cutoff=0.1)
+ self.assertTrue(np.all(spikes >= 0))
+
+ def test_round_trip_negative_signal_applies_shift(self):
+ signal = np.array([-0.5, 0.1, 0.3, 0.6, 0.2, 0.4, 0.1])
+ spikes, fir_bank, shift, norm = hough_spiker(signal, window_length=3, cutoff=0.1)
+ self.assertAlmostEqual(shift[0], -0.5)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape[0], len(signal))
+
+ def test_round_trip_large_amplitude_applies_norm(self):
+ signal = np.array([1.0, 2.5, 5.0, 3.0, 1.5, 0.5, 0.2])
+ spikes, fir_bank, shift, norm = hough_spiker(signal, window_length=3, cutoff=0.1)
+ self.assertTrue(norm[0] > 1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape[0], len(signal))
+
+ def test_round_trip_multi_feature(self):
+ signal = np.array([[0.1, 0.5], [0.3, 0.4], [0.4, 0.6], [0.2, 0.3], [0.5, 0.7], [0.6, 0.8]])
+ spikes, fir_bank, shift, norm = hough_spiker(signal, window_length=3, cutoff=0.1)
+ result = deconvolution_decoder(spikes, fir_bank=fir_bank, shift=shift, norm=norm)
+ self.assertEqual(result.shape, signal.shape)
+
+ def test_round_trip_hsa_vs_mhsa_same_signal_different_sparsity(self):
+ # HSA is stricter than MHSA, so it should produce fewer or equal spikes
+ signal = np.array([0.1, 0.2, 0.5, 0.9, 0.7, 0.4, 0.2])
+ hsa_spikes, _, _, _ = hough_spiker(signal, window_length=3, cutoff=0.1)
+ mhsa_spikes, _, _, _ = modified_hough_spiker(signal, window_length=3, cutoff=0.1, threshold=0.3)
+ self.assertLessEqual(hsa_spikes.sum(), mhsa_spikes.sum())
+
+ def test_round_trip_window_length_equals_signal_length_raises(self):
+ signal = np.array([0.1, 0.3, 0.5])
+ with self.assertRaises(ValueError):
+ hough_spiker(signal, window_length=4, cutoff=0.1)
diff --git a/tests/encoders/temporal/contrast/test_moving_window.py b/tests/encoders/temporal/contrast/test_moving_window.py
index 2e8476a..ad26de7 100644
--- a/tests/encoders/temporal/contrast/test_moving_window.py
+++ b/tests/encoders/temporal/contrast/test_moving_window.py
@@ -12,7 +12,8 @@ def test_basic_functionality(self):
window_length = 3
threshold = 0.5
expected_spikes = np.array([-1, 0, 1, 1, 1, 0, -1, -1, -1])
- result = moving_window(signal, window_length, threshold)
+ result, _ = moving_window(signal, window_length, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_empty_signal(self):
@@ -37,7 +38,8 @@ def test_all_zero_signal(self):
window_length = 3
threshold = 0.5
expected_spikes = np.array([0, 0, 0, 0, 0])
- result = moving_window(signal, window_length, threshold)
+ result, _ = moving_window(signal, window_length, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_large_signal(self):
@@ -45,7 +47,8 @@ def test_large_signal(self):
signal = np.random.randn(1000)
window_length = 50
threshold = 0.3
- result = moving_window(signal, window_length, threshold)
+ result, _ = moving_window(signal, window_length, threshold)
+ result = result.flatten()
self.assertEqual(len(result), len(signal))
def test_no_spikes(self):
@@ -54,7 +57,8 @@ def test_no_spikes(self):
window_length = 3
threshold = 5
expected_spikes = np.array([0, 0, 0, 0, 0])
- result = moving_window(signal, window_length, threshold)
+ result, _ = moving_window(signal, window_length, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_variable_window_length(self):
@@ -65,14 +69,16 @@ def test_variable_window_length(self):
window_length_3 = 3
threshold_3 = 0.5
expected_spikes_3 = np.array([-1, 0, 1, 1, 1, 1, 1, 0, -1, -1, -1, -1, -1])
- result_3 = moving_window(signal, window_length_3, threshold_3)
+ result_3, _ = moving_window(signal, window_length_3, threshold_3)
+ result_3 = result_3.flatten()
np.testing.assert_array_equal(result_3, expected_spikes_3)
# Test case for window length of 5
window_length_5 = 5
threshold_5 = 0.5
expected_spikes_5 = np.array([-1, -1, 0, 1, 1, 1, 1, 1, -1, -1, -1, -1, -1])
- result_5 = moving_window(signal, window_length_5, threshold_5)
+ result_5, _ = moving_window(signal, window_length_5, threshold_5)
+ result_5 = result_5.flatten()
np.testing.assert_array_equal(result_5, expected_spikes_5)
def test_with_multiple_features(self):
@@ -80,12 +86,14 @@ def test_with_multiple_features(self):
np.random.seed(42)
signal = np.random.rand(10, 2)
window_length = 2
- encoded_signal = moving_window(signal, window_length, threshold=0.1)
+ encoded_signal, _ = moving_window(signal, window_length, threshold=0.1)
self.assertEqual(encoded_signal.shape, signal.shape)
signal_f1 = signal[:, 0]
signal_f2 = signal[:, 1]
- encoded_signal_f1 = moving_window(signal_f1, window_length, threshold=0.1)
- encoded_signal_f2 = moving_window(signal_f2, window_length, threshold=0.1)
+ encoded_signal_f1, _ = moving_window(signal_f1, window_length, threshold=0.1)
+ encoded_signal_f1 = encoded_signal_f1.flatten()
+ encoded_signal_f2, _ = moving_window(signal_f2, window_length, threshold=0.1)
+ encoded_signal_f2 = encoded_signal_f2.flatten()
np.testing.assert_array_equal(encoded_signal[:, 0], encoded_signal_f1)
np.testing.assert_array_equal(encoded_signal[:, 1], encoded_signal_f2)
@@ -94,7 +102,7 @@ def test_threshold_dimension(self):
signal = np.random.rand(10, 2)
window_length = 2
threshold = np.array([[0.1, 0.2], [0.3, 0.4]])
- with self.assertRaises(TypeError):
+ with self.assertRaises(ValueError):
moving_window(signal, window_length, threshold)
def test_threshold_dims_different_from_features(self):
diff --git a/tests/encoders/temporal/contrast/test_step_forward.py b/tests/encoders/temporal/contrast/test_step_forward.py
index 7e81030..6732356 100644
--- a/tests/encoders/temporal/contrast/test_step_forward.py
+++ b/tests/encoders/temporal/contrast/test_step_forward.py
@@ -11,7 +11,8 @@ def test_basic_functionality(self):
signal = np.array([0, 2, 4, 6, 4, 2, 0], dtype=float)
threshold = 2.0
expected_spikes = np.array([0, 0, 1, 1, 0, 0, -1])
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_empty_signal(self):
@@ -26,7 +27,8 @@ def test_no_spikes(self):
signal = np.array([1, 1.5, 1.8, 1.6, 1.4, 1.2])
threshold = 2.0
expected_spikes = np.array([0, 0, 0, 0, 0, 0])
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_single_positive_spike(self):
@@ -34,7 +36,8 @@ def test_single_positive_spike(self):
signal = np.array([0, 0, 5, 0, 0], dtype=float)
threshold = 4.0
expected_spikes = np.array([0, 0, 1, 0, 0])
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_single_negative_spike(self):
@@ -42,7 +45,8 @@ def test_single_negative_spike(self):
signal = np.array([0, 0, -5, 0, 0], dtype=float)
threshold = 4.0
expected_spikes = np.array([0, 0, -1, 0, 0])
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_alternating_spikes(self):
@@ -50,7 +54,8 @@ def test_alternating_spikes(self):
signal = np.array([0, 5, 0, -5, 0], dtype=float)
threshold = 4.0
expected_spikes = np.array([0, 1, 0, -1, 0])
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_signal_with_noise(self):
@@ -58,14 +63,16 @@ def test_signal_with_noise(self):
signal = np.array([0, 0.5, 2.5, 1.5, 0, -1, -2.5, -1.5, 0])
threshold = 2.0
expected_spikes = np.array([0, 0, 1, 0, 0, -1, -1, 0, 0])
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_large_signal(self):
"""Test the function's performance and correctness on a large signal."""
signal = np.random.randn(1000) * 10 # Random signal with mean 0 and standard deviation 10
threshold = 15.0
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
self.assertEqual(len(result), len(signal))
def test_boundary_conditions(self):
@@ -73,7 +80,8 @@ def test_boundary_conditions(self):
signal = np.array([10, 0, -10], dtype=float)
threshold = 5.0
expected_spikes = np.array([0, -1, -1])
- result = step_forward(signal, threshold)
+ result, _ = step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_with_multiple_features(self):
@@ -84,11 +92,13 @@ def test_with_multiple_features(self):
signal_f2 = signal[:, 1]
threshold = [0.1, 0.3]
- encoded_signal = step_forward(signal, threshold)
+ encoded_signal, _ = step_forward(signal, threshold)
self.assertEqual(encoded_signal.shape, signal.shape)
- encoded_signal_f1 = step_forward(signal_f1, threshold[0])
- encoded_signal_f2 = step_forward(signal_f2, threshold[1])
+ encoded_signal_f1, _ = step_forward(signal_f1, threshold[0])
+ encoded_signal_f1 = encoded_signal_f1.flatten()
+ encoded_signal_f2, _ = step_forward(signal_f2, threshold[1])
+ encoded_signal_f2 = encoded_signal_f2.flatten()
np.testing.assert_array_equal(encoded_signal[:, 0], encoded_signal_f1)
np.testing.assert_array_equal(encoded_signal[:, 1], encoded_signal_f2)
@@ -105,5 +115,5 @@ def test_threshold_dimension(self):
"""Test that the function raises TypeError when threshold is of invalid dimension."""
signal = np.random.rand(10, 2)
threshold = np.array([[0.1, 0.2], [0.3, 0.4]])
- with self.assertRaises(TypeError):
+ with self.assertRaises(ValueError):
step_forward(signal, threshold)
diff --git a/tests/encoders/temporal/contrast/test_threshold_based.py b/tests/encoders/temporal/contrast/test_threshold_based.py
index 7a2d2ec..df64c62 100644
--- a/tests/encoders/temporal/contrast/test_threshold_based.py
+++ b/tests/encoders/temporal/contrast/test_threshold_based.py
@@ -12,6 +12,7 @@ def test_basic_functionality(self):
factor = 1.0
expected_spikes = np.array([1, 1, 1, -1, -1, -1, -1])
result, _ = threshold_based_representation(signal, factor)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_empty_signal(self):
@@ -27,6 +28,7 @@ def test_no_variation(self):
factor = 1.0
expected_spikes = np.array([0, 0, 0, 0, 0])
result, _ = threshold_based_representation(signal, factor)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_signal_with_noise(self):
@@ -52,12 +54,14 @@ def test_varying_factor(self):
factor_high = 2.0
expected_spikes_high = np.array([0, 0, 0, 0, 0, 0, 0])
result_high, _ = threshold_based_representation(signal, factor_high)
+ result_high = result_high.flatten()
np.testing.assert_array_equal(result_high, expected_spikes_high)
# Lower threshold factor, more sensitivity
factor_low = 0.1
expected_spikes_low = np.array([1, 1, 1, -1, -1, -1, -1])
result_low, _ = threshold_based_representation(signal, factor_low)
+ result_low = result_low.flatten()
np.testing.assert_array_equal(result_low, expected_spikes_low)
def test_with_multiple_features(self):
@@ -70,7 +74,9 @@ def test_with_multiple_features(self):
signal_f1 = signal[:, 0]
signal_f2 = signal[:, 1]
encoded_signal_f1, _ = threshold_based_representation(signal_f1, threshold[0])
+ encoded_signal_f1 = encoded_signal_f1.flatten()
encoded_signal_f2, _ = threshold_based_representation(signal_f2, threshold[1])
+ encoded_signal_f2 = encoded_signal_f2.flatten()
np.testing.assert_array_equal(encoded_signal[:, 0], encoded_signal_f1)
np.testing.assert_array_equal(encoded_signal[:, 1], encoded_signal_f2)
@@ -93,5 +99,5 @@ def test_factor_dimension(self):
"""Test that the function raises TypeError when factor is of invalid dimension."""
signal = np.random.rand(10, 2)
factor = np.array([[1.0, 2.0], [3.0, 4.0]])
- with self.assertRaises(TypeError):
+ with self.assertRaises(ValueError):
threshold_based_representation(signal, factor)
diff --git a/tests/encoders/temporal/contrast/test_zero_cross_step_forward.py b/tests/encoders/temporal/contrast/test_zero_cross_step_forward.py
index 866c515..d0cf3d6 100644
--- a/tests/encoders/temporal/contrast/test_zero_cross_step_forward.py
+++ b/tests/encoders/temporal/contrast/test_zero_cross_step_forward.py
@@ -11,7 +11,8 @@ def test_basic_functionality(self):
signal = np.array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
threshold = 5.0
expected_spikes = np.array([0, 0, 0, 0, 0, 0, 1, 1, 1, 1])
- result = zero_cross_step_forward(signal, threshold)
+ result, _ = zero_cross_step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_empty_signal(self):
@@ -26,7 +27,8 @@ def test_all_below_threshold(self):
signal = np.array([0, 1, 2, 3, 4])
threshold = 5.0
expected_spikes = np.array([0, 0, 0, 0, 0])
- result = zero_cross_step_forward(signal, threshold)
+ result, _ = zero_cross_step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_all_above_threshold(self):
@@ -34,7 +36,8 @@ def test_all_above_threshold(self):
signal = np.array([6, 7, 8, 9, 10])
threshold = 5.0
expected_spikes = np.array([1, 1, 1, 1, 1])
- result = zero_cross_step_forward(signal, threshold)
+ result, _ = zero_cross_step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_signal_with_negative_values(self):
@@ -42,7 +45,8 @@ def test_signal_with_negative_values(self):
signal = np.array([-1, -2, -3, 4, 5, -6, 7])
threshold = 5.0
expected_spikes = np.array([0, 0, 0, 0, 0, 0, 1])
- result = zero_cross_step_forward(signal, threshold)
+ result, _ = zero_cross_step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_no_spikes(self):
@@ -50,14 +54,16 @@ def test_no_spikes(self):
signal = np.array([0, 0, 0, 0, 0])
threshold = 1.0
expected_spikes = np.array([0, 0, 0, 0, 0])
- result = zero_cross_step_forward(signal, threshold)
+ result, _ = zero_cross_step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_large_signal(self):
"""Test the function's performance and correctness on a large signal."""
signal = np.random.randint(-10, 20, size=1000)
threshold = 10.0
- result = zero_cross_step_forward(signal, threshold)
+ result, _ = zero_cross_step_forward(signal, threshold)
+ result = result.flatten()
self.assertEqual(len(result), len(signal))
def test_threshold_at_zero(self):
@@ -65,7 +71,8 @@ def test_threshold_at_zero(self):
signal = np.array([-1, 0, 1, 2, 3])
threshold = 0.0
expected_spikes = np.array([0, 0, 1, 1, 1])
- result = zero_cross_step_forward(signal, threshold)
+ result, _ = zero_cross_step_forward(signal, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_shape_with_multiple_features(self):
@@ -73,12 +80,14 @@ def test_shape_with_multiple_features(self):
np.random.seed(42)
signal = np.random.rand(10, 2)
threshold = [0.5, 0.3]
- encoded_signal = zero_cross_step_forward(signal, threshold)
+ encoded_signal, _ = zero_cross_step_forward(signal, threshold)
self.assertEqual(encoded_signal.shape, signal.shape)
signal_f1 = signal[:, 0]
signal_f2 = signal[:, 1]
- encoded_signal_f1 = zero_cross_step_forward(signal_f1, threshold[0])
- encoded_signal_f2 = zero_cross_step_forward(signal_f2, threshold[1])
+ encoded_signal_f1, _ = zero_cross_step_forward(signal_f1, threshold[0])
+ encoded_signal_f1 = encoded_signal_f1.flatten()
+ encoded_signal_f2, _ = zero_cross_step_forward(signal_f2, threshold[1])
+ encoded_signal_f2 = encoded_signal_f2.flatten()
np.testing.assert_array_equal(encoded_signal[:, 0], encoded_signal_f1)
np.testing.assert_array_equal(encoded_signal[:, 1], encoded_signal_f2)
@@ -94,5 +103,5 @@ def test_threshold_dimension(self):
"""Test that the function raises TypeError when threshold is of invalid dimension."""
signal = np.random.rand(10, 2)
threshold = np.array([[0.1, 0.2], [0.3, 0.4]])
- with self.assertRaises(TypeError):
+ with self.assertRaises(ValueError):
zero_cross_step_forward(signal, threshold)
diff --git a/tests/encoders/temporal/deconvolution/test_ben_spiker.py b/tests/encoders/temporal/deconvolution/test_ben_spiker.py
index efada42..9f570d5 100644
--- a/tests/encoders/temporal/deconvolution/test_ben_spiker.py
+++ b/tests/encoders/temporal/deconvolution/test_ben_spiker.py
@@ -16,6 +16,7 @@ def test_basic_functionality(self):
cutoff = 0.2
expected_spikes = np.array([0, 1, 0, 1, 0, 1, 0, 0, 0, 0, 0])
result, _, _ = bens_spiker(signal, window_length, cutoff, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_threshold_sensitivity(self):
@@ -27,9 +28,7 @@ def test_threshold_sensitivity(self):
threshold_high = 10.0
cutoff = 0.2
result_low, _, _ = bens_spiker(signal, window_length, cutoff, threshold_low)
- print(result_low)
result_high, _, _ = bens_spiker(signal, window_length, cutoff, threshold_high)
- print(result_high)
self.assertTrue(np.any(result_low))
self.assertFalse(np.any(result_high))
@@ -71,7 +70,7 @@ def test_threshold_dimension(self):
threshold = np.array([[0.1, 0.2], [0.3, 0.4]])
window_length = 3
cutoff = 0.2
- with self.assertRaises(TypeError):
+ with self.assertRaises(ValueError):
bens_spiker(signal, window_length, cutoff, threshold)
def test_no_matching_pattern(self):
@@ -84,6 +83,7 @@ def test_no_matching_pattern(self):
cutoff = 0.2
expected_spikes = np.array([0, 0, 0, 0, 0])
result, _, _ = bens_spiker(signal, window_length, cutoff, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_varying_filter_window_size(self):
@@ -97,12 +97,14 @@ def test_varying_filter_window_size(self):
window_length_3 = 3
expected_spikes_3 = np.array([0, 0, 0, 0, 0, 0, 0, 0, 0])
result_3, _, _ = bens_spiker(signal, window_length_3, cutoff, threshold)
+ result_3 = result_3.flatten()
np.testing.assert_array_equal(result_3, expected_spikes_3)
# Test case for filter window size of 5
window_length_5 = 5
expected_spikes_5 = np.array([0, 1, 0, 1, 0, 0, 0, 0, 0])
result_5, _, _ = bens_spiker(signal, window_length_5, cutoff, threshold)
+ result_5 = result_5.flatten()
np.testing.assert_array_equal(result_5, expected_spikes_5)
# Test case for filter window size of 7
@@ -110,6 +112,7 @@ def test_varying_filter_window_size(self):
window_length_7 = 7
expected_spikes_7 = np.array([1, 0, 1, 0, 0, 0, 0, 0, 0])
result_7, _, _ = bens_spiker(signal, window_length_7, cutoff, threshold)
+ result_7 = result_7.flatten()
np.testing.assert_array_equal(result_7, expected_spikes_7)
def test_large_signal(self):
@@ -120,6 +123,7 @@ def test_large_signal(self):
threshold = 5.0
cutoff = 0.2
result, _, _ = bens_spiker(signal, window_length, cutoff, threshold)
+ result = result.flatten()
self.assertEqual(len(result), len(signal))
def test_with_multiple_features(self):
@@ -134,7 +138,9 @@ def test_with_multiple_features(self):
signal_f1 = signal[:, 0]
signal_f2 = signal[:, 1]
encoded_signal_f1, _, _ = bens_spiker(signal_f1, window_length, cutoff, threshold[0])
+ encoded_signal_f1 = encoded_signal_f1.flatten()
encoded_signal_f2, _, _ = bens_spiker(signal_f2, window_length, cutoff, threshold[1])
+ encoded_signal_f2 = encoded_signal_f2.flatten()
np.testing.assert_array_equal(encoded_signal[:, 0], encoded_signal_f1)
np.testing.assert_array_equal(encoded_signal[:, 1], encoded_signal_f2)
@@ -155,7 +161,7 @@ def test_signal_with_negative_values(self):
window_length = 3
cutoff = 0.4
threshold = 0.95
- _, shift, fir_coeff = bens_spiker(signal, window_length, cutoff, threshold)
+ _, fir_coeff, shift = bens_spiker(signal, window_length, cutoff, threshold)
expected_shift = np.array([-0.1], dtype=np.float32)
expected_fir_sum = 1.0
@@ -170,7 +176,7 @@ def test_signal_with_positive_values_lower_than_one(self):
window_length = 3
cutoff = 0.4
threshold = 0.95
- _, shift, fir_coeff = bens_spiker(signal, window_length, cutoff, threshold)
+ _, fir_coeff, shift = bens_spiker(signal, window_length, cutoff, threshold)
expected_shift = np.array([0.0], dtype=np.float32)
expected_fir_sum = 1.0
@@ -185,7 +191,7 @@ def test_signal_with_positive_values_greater_than_one(self):
window_length = 3
cutoff = 0.4
threshold = 0.95
- _, shift, fir_coeff = bens_spiker(signal, window_length, cutoff, threshold)
+ _, fir_coeff, shift = bens_spiker(signal, window_length, cutoff, threshold)
expected_shift = np.array([0.0], dtype=np.float32)
expected_fir_sum = 3.8
diff --git a/tests/encoders/temporal/deconvolution/test_hough_spiker.py b/tests/encoders/temporal/deconvolution/test_hough_spiker.py
index b60075a..8e1cacd 100644
--- a/tests/encoders/temporal/deconvolution/test_hough_spiker.py
+++ b/tests/encoders/temporal/deconvolution/test_hough_spiker.py
@@ -15,6 +15,7 @@ def test_basic_functionality(self):
cutoff = 0.1
expected_spikes = np.array([0, 0, 1, 0, 1, 0, 0, 0, 0, 0, 0])
result, _, _, _ = hough_spiker(signal, window_length, cutoff)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_empty_signal(self):
@@ -54,6 +55,7 @@ def test_single_point_spike(self):
cutoff = 0.2
expected_spikes = np.array([0, 1, 0, 0, 0, 0, 0, 0])
result, _, _, _ = hough_spiker(signal, window_length, cutoff)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_varying_filter_window_size(self):
@@ -66,18 +68,21 @@ def test_varying_filter_window_size(self):
window_length_3 = 3
expected_spikes_3 = np.array([0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0])
result_3, _, _, _ = hough_spiker(signal, window_length_3, cutoff)
+ result_3 = result_3.flatten()
np.testing.assert_array_equal(result_3, expected_spikes_3)
# Test case for filter window size of 5
window_length_5 = 5
expected_spikes_5 = np.array([1, 0, 1, 1, 0, 0, 0, 0, 0, 0, 0])
result_5, _, _, _ = hough_spiker(signal, window_length_5, cutoff)
+ result_5 = result_5.flatten()
np.testing.assert_array_equal(result_5, expected_spikes_5)
# Test case for filter window size of 7
window_length_7 = 7
expected_spikes_7 = np.array([1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0])
result_7, _, _, _ = hough_spiker(signal, window_length_7, cutoff)
+ result_7 = result_7.flatten()
np.testing.assert_array_equal(result_7, expected_spikes_7)
def test_large_signal(self):
@@ -87,6 +92,7 @@ def test_large_signal(self):
window_length = 10
cutoff = 0.1
result, _, _, _ = hough_spiker(signal, window_length, cutoff)
+ result = result.flatten()
self.assertEqual(len(result), len(signal))
def test_with_multiple_features(self):
@@ -100,7 +106,9 @@ def test_with_multiple_features(self):
signal_f1 = signal[:, 0]
signal_f2 = signal[:, 1]
encoded_signal_f1 = hough_spiker(signal_f1, window_length, cutoff)[0]
+ encoded_signal_f1 = encoded_signal_f1.flatten()
encoded_signal_f2 = hough_spiker(signal_f2, window_length, cutoff)[0]
+ encoded_signal_f2 = encoded_signal_f2.flatten()
np.testing.assert_array_equal(encoded_signal[:, 0], encoded_signal_f1)
np.testing.assert_array_equal(encoded_signal[:, 1], encoded_signal_f2)
@@ -110,7 +118,7 @@ def test_signal_with_negative_values(self):
signal = np.array([-0.1, 0.1, 0.2, 0.3, 0.4, 0.3, 0.2, 0.1], dtype=np.float32)
window_length = 3
cutoff = 0.4
- _, shift, norm, fir_coeff = hough_spiker(signal, window_length, cutoff)
+ _, fir_coeff, shift, norm = hough_spiker(signal, window_length, cutoff)
expected_shift = np.array([-0.1], dtype=np.float32)
expected_norm = np.array([1.0], dtype=np.float32)
@@ -126,7 +134,7 @@ def test_signal_with_positive_values_lower_than_one(self):
signal = np.array([0.1, 0.3, 0.5, 0.7, 0.9, 0.6, 0.4, 0.2], dtype=np.float32)
window_length = 3
cutoff = 0.4
- _, shift, norm, fir_coeff = hough_spiker(signal, window_length, cutoff)
+ _, fir_coeff, shift, norm = hough_spiker(signal, window_length, cutoff)
expected_shift = np.array([0.0], dtype=np.float32)
expected_norm = np.array([1.0], dtype=np.float32)
@@ -142,7 +150,7 @@ def test_signal_with_positive_values_greater_than_one(self):
signal = np.array([0.1, 0.3, 0.5, 0.7, 1.9, 1.6, 1.4, 0.2], dtype=np.float32)
window_length = 3
cutoff = 0.4
- _, shift, norm, fir_coeff = hough_spiker(signal, window_length, cutoff)
+ _, fir_coeff, shift, norm = hough_spiker(signal, window_length, cutoff)
expected_shift = np.array([0.0], dtype=np.float32)
expected_norm = np.array([1.9], dtype=np.float32)
diff --git a/tests/encoders/temporal/deconvolution/test_modified_hough_spiker.py b/tests/encoders/temporal/deconvolution/test_modified_hough_spiker.py
index 1f88642..3c8cce5 100644
--- a/tests/encoders/temporal/deconvolution/test_modified_hough_spiker.py
+++ b/tests/encoders/temporal/deconvolution/test_modified_hough_spiker.py
@@ -16,6 +16,7 @@ def test_basic_functionality(self):
cutoff = 0.2
expected_spikes = np.array([0, 0, 0, 0, 1, 1, 0, 0, 0, 0, 1])
result, _, _, _ = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_empty_signal(self):
@@ -49,6 +50,7 @@ def test_no_matching_pattern(self):
threshold = -0.1
expected_spikes = np.array([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0])
result, _, _, _ = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_single_point_spike(self):
@@ -60,6 +62,7 @@ def test_single_point_spike(self):
threshold = 0.1
expected_spikes = np.array([0, 0, 1, 0, 0])
result, _, _, _ = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ result = result.flatten()
np.testing.assert_array_equal(result, expected_spikes)
def test_varying_filter_window_size(self):
@@ -72,18 +75,21 @@ def test_varying_filter_window_size(self):
window_length_3 = 3
expected_spikes_3 = np.array([0, 0, 0, 1, 1, 0, 0, 0, 0, 0, 1])
result_3, _, _, _ = modified_hough_spiker(signal, window_length_3, cutoff, threshold)
+ result_3 = result_3.flatten()
np.testing.assert_array_equal(result_3, expected_spikes_3)
# Test case for filter window size of 5
window_length_5 = 5
expected_spikes_5 = np.array([1, 1, 0, 1, 1, 0, 0, 0, 0, 0, 1])
result_5, _, _, _ = modified_hough_spiker(signal, window_length_5, cutoff, threshold)
+ result_5 = result_5.flatten()
np.testing.assert_array_equal(result_5, expected_spikes_5)
# Test case for filter window size of 7
window_length_7 = 7
expected_spikes_7 = np.array([1, 1, 1, 1, 0, 0, 0, 0, 0, 1, 1])
result_7, _, _, _ = modified_hough_spiker(signal, window_length_7, cutoff, threshold)
+ result_7 = result_7.flatten()
np.testing.assert_array_equal(result_7, expected_spikes_7)
def test_large_signal(self):
@@ -94,6 +100,7 @@ def test_large_signal(self):
cutoff = 0.1
threshold = 1.0
result, _, _, _ = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ result = result.flatten()
self.assertEqual(len(result), len(signal))
def test_with_multiple_features(self):
@@ -108,7 +115,9 @@ def test_with_multiple_features(self):
signal_f1 = signal[:, 0]
signal_f2 = signal[:, 1]
encoded_signal_f1, _, _, _ = modified_hough_spiker(signal_f1, window_length, cutoff, threshold[0])
+ encoded_signal_f1 = encoded_signal_f1.flatten()
encoded_signal_f2, _, _, _ = modified_hough_spiker(signal_f2, window_length, cutoff, threshold[1])
+ encoded_signal_f2 = encoded_signal_f2.flatten()
np.testing.assert_array_equal(encoded_signal[:, 0], encoded_signal_f1)
np.testing.assert_array_equal(encoded_signal[:, 1], encoded_signal_f2)
@@ -129,7 +138,7 @@ def test_threshold_dimension(self):
threshold = np.array([[0.1, 0.2], [0.3, 0.4]])
window_length = 3
cutoff = 0.2
- with self.assertRaises(TypeError):
+ with self.assertRaises(ValueError):
modified_hough_spiker(signal, window_length, cutoff, threshold)
def test_signal_with_negative_values(self):
@@ -139,7 +148,7 @@ def test_signal_with_negative_values(self):
window_length = 3
cutoff = 0.4
threshold = 0.1
- _, shift, norm, fir_coeff = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ _, fir_coeff, shift, norm = modified_hough_spiker(signal, window_length, cutoff, threshold)
expected_shift = np.array([-0.1], dtype=np.float32)
expected_norm = np.array([1.0], dtype=np.float32)
@@ -156,7 +165,7 @@ def test_signal_with_positive_values_lower_than_one(self):
window_length = 3
cutoff = 0.4
threshold = 0.1
- _, shift, norm, fir_coeff = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ _, fir_coeff, shift, norm = modified_hough_spiker(signal, window_length, cutoff, threshold)
expected_shift = np.array([0.0], dtype=np.float32)
expected_norm = np.array([1.0], dtype=np.float32)
@@ -173,7 +182,7 @@ def test_signal_with_positive_values_greater_than_one(self):
window_length = 3
cutoff = 0.4
threshold = 0.1
- _, shift, norm, fir_coeff = modified_hough_spiker(signal, window_length, cutoff, threshold)
+ _, fir_coeff, shift, norm = modified_hough_spiker(signal, window_length, cutoff, threshold)
expected_shift = np.array([0.0], dtype=np.float32)
expected_norm = np.array([1.9], dtype=np.float32)