feature: make the scaling absolute across all the trials

annubes/task.py

@@ -72,7 +72,8 @@ class Task(TaskSettingsMixin):
             Note that values are read relative to each other, such that e.g. `{"v": 0.25, "a": 0.75}` is equivalent to
             `{"v": 1, "a": 3}`.
             Defaults to {"v": 0.5, "a": 0.5}.
-        stim_intensities: List of possible intensity values of each stimulus.
+        stim_intensities: List of possible intensity values of each stimulus, when the stimulus is present. Note that
+            when the stimulus is not present, the intensity is set to 0.
             Defaults to [0.8, 0.9, 1].
         stim_time: Duration of each stimulus in ms.
             Defaults to 1000.
@@ -169,6 +170,18 @@ def generate_trials(
         self._inputs = self._build_trials_inputs()
         self._outputs = self._build_trials_outputs()
 
+        # Scaling
+        if self.scaling:
+            flattened = np.concatenate(
+                (np.concatenate(self._inputs).reshape(-1), np.concatenate(self._outputs).reshape(-1)),
+            )
+            abs_min = np.min(flattened)
+            abs_max = np.max(flattened)
+
+            for n in range(self._ntrials):
+                self._inputs[n] = self._minmaxscaler(self._inputs[n], abs_min, abs_max)
+                self._outputs[n] = self._minmaxscaler(self._outputs[n], abs_min, abs_max)
+
         # Store trials settings and data
         return {
             "task_settings": self._task_settings,
@@ -434,30 +447,35 @@ def _setup_trial_phases(
 
     def _minmaxscaler(
         self,
-        input_: NDArray[np.float64],
+        array_: NDArray[np.float64],
+        abs_min: float,
+        abs_max: float,
         rescale_range: tuple[float, float] = (0, 1),
     ) -> NDArray[np.float64]:
-        """Rescale `input_` array to a given range.
+        """Rescale `array_` to a given range.
 
         Rescaling happens as follows:
 
-            `X_std = (input_ - input_.min()) / (input_.max() - input_.min())`
+            `X_std = (array_ - abs_min) / (abs_max - abs_min)`
             `X_scaled = X_std * (max - min) + min`
             where min, max = range.
         The logic is the same as that of `sklearn.preprocessing.MinMaxScaler` estimator. Each array is rescaled to the
-        given range, for each trial contained in `input_`.
+        given range.
 
 
         Args:
-            input_: Input array of shape (self._ntrials, len(self._time), self._n_inputs).
+            array_: Array to be rescaled. It can be either input or output array, respectively with shape
+                (self._time, self._n_inputs) or (self._time, self.n_outputs).
+            abs_min: Minimum value of the input and output arrays, considering all trials.
+            abs_max: Maximum value of the input and output arrays, considering all trials.
             rescale_range: Desired range of transformed data. Defaults to (0, 1).
 
         Returns:
             Rescaled input array.
         """
-        input_std = (input_ - input_.min()) / (input_.max() - input_.min())
+        array_std = (array_ - abs_min) / (abs_max - abs_min)
 
-        return np.array(input_std * (max(rescale_range) - min(rescale_range)) + min(rescale_range))
+        return np.array(array_std * (max(rescale_range) - min(rescale_range)) + min(rescale_range))
 
     def _build_trials_inputs(self) -> NDArray[np.float64]:
         """Generate trials time and inputs ndarrays."""
@@ -482,9 +500,6 @@ def _build_trials_inputs(self) -> NDArray[np.float64]:
             # add noise
             x[n] += noise_factor * self._rng.normal(loc=0, scale=1, size=x[n].shape)
 
-            if self.scaling:
-                x[n] = self._minmaxscaler(x[n])
-
         return x
 
     def _build_trials_outputs(self) -> NDArray[np.float64]:
@@ -496,7 +511,4 @@ def _build_trials_outputs(self) -> NDArray[np.float64]:
             y[n][self._phases[n]["input"], choice[n]] = max(self.output_behavior)
             y[n][self._phases[n]["input"], 1 - choice[n]] = min(self.output_behavior)
 
-            if self.scaling:
-                y[n] = self._minmaxscaler(y[n])
-
         return y

tests/test_task.py

@@ -426,17 +426,16 @@ def test_setup_trial_phases(fix_time: int | tuple[int, int], iti: int | tuple[in
     )
 
 
-def test_minmaxscaler():
-    task = Task(name=NAME, scaling=True)
+@pytest.mark.parametrize(("stim_intensities", "fix_intensity"), [(STIM_INTENSITIES, FIX_INTENSITY), ([2, 4, 6], 10)])
+def test_minmaxscaler(stim_intensities: list[float], fix_intensity: float):
+    task = Task(name=NAME, stim_intensities=stim_intensities, fix_intensity=fix_intensity, scaling=True)
     _ = task.generate_trials(ntrials=NTRIALS)
     trial_indices = range(NTRIALS)
-    # Check that the signals are scaled between 0 and 1, and that min is 0 and max is 1
+    # Check that the signals are scaled between 0 and 1
     ## Inputs
     assert all((task._inputs[n_trial] >= 0).all() and (task._inputs[n_trial] <= 1).all() for n_trial in trial_indices)
-    assert all(task._inputs[n_trial].min() == 0 and task._inputs[n_trial].max() == 1 for n_trial in trial_indices)
     # Outputs
     assert all((task._outputs[n_trial] >= 0).all() and (task._outputs[n_trial] <= 1).all() for n_trial in trial_indices)
-    assert all(task._outputs[n_trial].min() == 0 and task._outputs[n_trial].max() == 1 for n_trial in trial_indices)
 
 
 @pytest.mark.parametrize("random_seed", [RND_SEED, 100])