Fix raster plot handling of segment times where t_start is not zero.

src/spikeinterface/core/sorting_tools.py

@@ -400,8 +400,6 @@ def random_spikes_selection(
         random_spikes_indices = np.arange(spikes.size)
 
     elif method in rng_methods:
-        from spikeinterface.widgets.utils import get_segment_durations
-
         rng = np.random.default_rng(seed=seed)
 
         # since un concatenated
@@ -447,7 +445,7 @@ def random_spikes_selection(
                 if maximum_rate is None:
                     raise ValueError(f"maximum_rate must be defined")
 
-                t_duration = np.sum(get_segment_durations(sorting))
+                t_duration = np.sum(get_segment_durations(sorting)[0])
                 rng_size = min(int(t_duration * maximum_rate), max_spikes_per_unit, all_unit_indices.size)
                 selected_unit_indices = rng.choice(all_unit_indices, size=rng_size, replace=False, shuffle=False)
 
@@ -1213,3 +1211,85 @@ def remap_unit_indices_in_vector(vector, all_old_unit_ids, all_new_unit_ids, kee
     new_vector["unit_index"] = mapping[new_vector["unit_index"]]
 
     return new_vector, keep_mask_vector
+
+
+def get_segment_durations(sorting: BaseSorting, segment_indices: list[int] = None):
+    """
+    Calculate the duration and start, stop times of each segment in a sorting object.
+
+    Use first/last sample timestamps when a recording is available; otherwise
+    estimate from spikes.
+
+    Parameters
+    ----------
+    sorting : BaseSorting
+        The sorting object containing spike data
+    segment_indices : list[int] | None
+        List of the segment indices to process. Default to None.
+
+    Returns
+    -------
+    durations : np.ndarray
+        Segment durations in seconds, ordered like ``segment_indices``.
+    segment_start_stop_times : dict[int, tuple[float, float]]
+        Start and stop times of each segment, where the dict key is segment index.
+    """
+    if segment_indices is None:
+        segment_indices = range(sorting.get_num_segments())
+
+    if sorting.has_recording():
+        segment_start_stop_times = {
+            seg: (sorting.get_start_time(seg), sorting.get_end_time(seg)) for seg in segment_indices
+        }
+        ts = 1 / sorting.get_sampling_frequency()
+        durations = np.array([stop - start + ts for start, stop in segment_start_stop_times.values()])
+
+        return durations, segment_start_stop_times
+
+    spikes = sorting.to_spike_vector()
+
+    segment_t_starts = [sorting.get_start_time(seg_idx) for seg_idx in segment_indices]
+
+    return estimate_segment_durations_from_spike_vector(
+        spikes, segment_indices, sorting.get_sampling_frequency(), segment_t_starts
+    )
+
+
+def estimate_segment_durations_from_spike_vector(
+    spike_vector: np.ndarray,
+    segment_indices: list[int],
+    sampling_frequency: float,
+    segment_t_starts: list[float] | None = None,
+):
+    """
+    When we don't have the true segment times, we can estimate the
+    start and stop times for each segment from the spike times.
+
+    Starts are supplied or default to zero. Durations include the final sample interval.
+    Empty segments have zero duration and placeholder bounds from their start to one sample interval later.
+    """
+    segment_start_stop_times = {}
+    durations = []
+    ts = 1 / sampling_frequency
+    for i, seg_idx in enumerate(segment_indices):
+
+        if segment_t_starts is None:
+            segment_start = 0.0
+        else:
+            segment_start = segment_t_starts[i]
+
+        seg_end_idx = np.searchsorted(spike_vector["segment_index"], seg_idx + 1)
+
+        if seg_end_idx == 0 or spike_vector["segment_index"][seg_end_idx - 1] != seg_idx:
+            segment_start_stop_times[seg_idx] = (segment_start, segment_start + ts)
+            durations.append(0.0)
+            continue
+
+        segment_end = spike_vector["sample_index"][seg_end_idx - 1] / sampling_frequency + segment_start
+
+        segment_start_stop_times[seg_idx] = (segment_start, segment_end)
+        durations.append(segment_end - segment_start + ts)
+
+    durations = np.array(durations)
+
+    return durations, segment_start_stop_times

src/spikeinterface/core/tests/test_sorting_tools.py

@@ -2,7 +2,7 @@
 import pytest
 import numpy as np
 
-from spikeinterface.core import NumpySorting
+from spikeinterface.core import NumpySorting, NumpyRecording, generate_sorting
 
 from spikeinterface.core import generate_ground_truth_recording
 from spikeinterface.core.sorting_tools import (
@@ -17,6 +17,7 @@
     set_properties_after_merging,
     set_properties_after_splits,
     reorder_spike_vector_by_unit_and_segment,
+    get_segment_durations,
 )
 from spikeinterface.core.base import minimum_spike_dtype
 
@@ -530,6 +531,85 @@ def test_remap_unit_indices_in_vector():
     assert np.array_equal(mask, target_mask)
 
 
+def test_get_segment_durations_estimate():
+    """Test a sortings get_segment_durations when estimated from the spike time"""
+    from spikeinterface import generate_sorting
+
+    # Test with a normal multi-segment sorting
+    durations = [5.0, 10.0, 15.0]
+    t_starts = [10, 20, 30]
+
+    # Create sorting with high fr to ensure spikes near the end segments
+    sorting = generate_sorting(
+        durations=durations,
+        firing_rates=15.0,
+        t_starts=t_starts,
+    )
+
+    segment_indices = list(range(sorting.get_num_segments()))
+
+    # Calculate durations
+    calculated_durations, segment_start_stop_times = get_segment_durations(sorting, segment_indices)
+
+    # Check results
+    expected_start_stop_times = np.array([(10, 15), (20, 30), (30, 45)], dtype=float)
+    assert list(segment_start_stop_times) == segment_indices
+    start_stop_times = np.array(list(segment_start_stop_times.values()))
+    assert np.allclose(expected_start_stop_times, start_stop_times, rtol=0, atol=0.1)
+
+    # Check results
+    assert len(calculated_durations) == len(durations)
+    # Durations should be approximately correct
+    for calculated_duration, expected_duration in zip(calculated_durations, durations):
+        # Duration should be <= expected (spikes can't be after the end)
+        assert calculated_duration <= expected_duration
+        # And reasonably close
+        tolerance = max(0.1 * expected_duration, 0.1)
+        assert expected_duration - calculated_duration < tolerance
+
+    # Test with single-segment sorting
+    sorting_single = generate_sorting(
+        durations=[7.0],
+        firing_rates=15.0,
+        t_starts=[4],
+    )
+
+    single_duration, segment_start_stop_times = get_segment_durations(sorting_single, [0])
+
+    expected_start_stop_times = [(4, 11)]
+    assert list(segment_start_stop_times) == [0]
+    start_stop_times = np.array(list(segment_start_stop_times.values()))
+    assert np.allclose(expected_start_stop_times, start_stop_times, rtol=0, atol=0.1)
+
+    # Test that the calculated duration is reasonable
+    assert single_duration <= 7.0
+    assert 7.0 - single_duration < 0.7  # Within 10%
+
+
+def test_get_segment_durations_with_recording():
+    """Test a sortings get_segment_durations when a recording is
+    registered, meaning times should be exact and not estimated from spike times"""
+    durations = [5.0, 10.0, 15.0]
+    t_starts = [10, 20, 30]
+    sampling_frequency = 1024
+
+    sorting = generate_sorting(durations=durations, sampling_frequency=sampling_frequency, seed=42)
+
+    recording = NumpyRecording(
+        [np.zeros((int(duration * sampling_frequency), 1), dtype="float32") for duration in durations],
+        sampling_frequency=sampling_frequency,
+        t_starts=t_starts,
+    )
+    sorting.register_recording(recording)
+
+    actual_durations, bounds = get_segment_durations(sorting)
+    np.testing.assert_array_equal(actual_durations, durations)
+    assert bounds == {
+        seg: (start, start + duration - 1 / sampling_frequency)
+        for seg, (start, duration) in enumerate(zip(t_starts, durations))
+    }
+
+
 if __name__ == "__main__":
     # test_spike_vector_to_spike_trains()
     # test_spike_vector_to_indices()

src/spikeinterface/widgets/amplitudes.py

@@ -1,12 +1,12 @@
 import numpy as np
 from warnings import warn
 
+from spikeinterface.core import SortingAnalyzer
+from spikeinterface.core.sorting_tools import get_segment_durations
+
 from .rasters import BaseRasterWidget
 from .base import to_attr
-from .utils import validate_segment_indices, get_segment_durations
-
-from spikeinterface.core.sortinganalyzer import SortingAnalyzer
-from spikeinterface.core import SortingAnalyzer
+from .utils import validate_segment_indices
 
 
 class AmplitudesWidget(BaseRasterWidget):
@@ -121,15 +121,15 @@ def __init__(
         if plot_histograms and bins is None:
             bins = 100
 
-        # Calculate durations for all segments for x-axis limits
-        durations = get_segment_durations(sorting, segment_indices)
+        _, segment_start_stop_times = get_segment_durations(sorting, segment_indices)
 
         # Build the plot data with the full dict of dicts structure
         plot_data = dict(
             unit_colors=unit_colors,
             plot_histograms=plot_histograms,
             bins=bins,
-            durations=durations,
+            segment_start_stop_times=segment_start_stop_times,
+            sampling_frequency=sorting.get_sampling_frequency(),
             unit_ids=unit_ids,
             hide_unit_selector=hide_unit_selector,
             plot_legend=plot_legend,
@@ -180,9 +180,14 @@ def plot_figpack(self, data_plot, **backend_kwargs):
             for u in unit_ids
         ]
 
+        segment_times = dp.segment_start_stop_times.values()
+
+        start_time_sec = min(start for start, _ in segment_times)
+        end_time_sec = max(stop for _, stop in segment_times)
+
         self.view = vv_views.SpikeAmplitudes(
-            start_time_sec=0,
-            end_time_sec=np.sum(dp.durations),
+            start_time_sec=start_time_sec,
+            end_time_sec=end_time_sec,
             plots=sa_items,
             # hide_unit_selector=dp.hide_unit_selector,
         )

src/spikeinterface/widgets/motion.py

@@ -3,8 +3,9 @@
 from .base import BaseWidget, to_attr
 
 from spikeinterface.core import BaseRecording, SortingAnalyzer
-from .rasters import BaseRasterWidget
 from spikeinterface.core.motion import Motion
+from spikeinterface.core.sorting_tools import estimate_segment_durations_from_spike_vector
+from .rasters import BaseRasterWidget
 
 
 class MotionWidget(BaseWidget):
@@ -155,7 +156,6 @@ def __init__(
         color: str = "Gray",
         clim: tuple[float, float] | None = None,
         alpha: float = 1,
-        segment_index: int | list[int] | None = None,
         backend: str | None = None,
         **backend_kwargs,
     ):
@@ -229,7 +229,10 @@ def __init__(
             segment_locations = filtered_locations[segment_mask]
 
             # Convert peak times to seconds
-            spike_times = segment_peaks["sample_index"] / sampling_frequency
+            if recording is None:
+                spike_times = segment_peaks["sample_index"] / sampling_frequency
+            else:
+                spike_times = recording.sample_index_to_time(segment_peaks["sample_index"], segment_index=seg_idx)
 
             # Store in dict of dicts format (using 0 as the "unit" id)
             spike_train_data[seg_idx] = {0: spike_times}
@@ -257,19 +260,16 @@ def __init__(
         else:
             color_kwargs = dict(color=color, c=None, alpha=alpha)
 
-        # Calculate segment durations for x-axis limits
+        # Calculate segment start/stop times for x-axis limits
         if recording is not None:
-            durations = [recording.get_duration(seg_idx) for seg_idx in segment_indices]
+            segment_start_stop_times = {
+                seg: (recording.get_start_time(seg), recording.get_end_time(seg)) for seg in segment_indices
+            }
         else:
             # Find boundaries between segments using searchsorted
-            segment_boundaries = [
-                np.searchsorted(filtered_peaks["segment_index"], [seg_idx, seg_idx + 1]) for seg_idx in segment_indices
-            ]
-
-            # Calculate durations from max sample in each segment
-            durations = [
-                (filtered_peaks["sample_index"][end - 1] + 1) / sampling_frequency for (_, end) in segment_boundaries
-            ]
+            _, segment_start_stop_times = estimate_segment_durations_from_spike_vector(
+                filtered_peaks, segment_indices, sampling_frequency
+            )
 
         plot_data = dict(
             spike_train_data=spike_train_data,
@@ -280,7 +280,8 @@ def __init__(
             scatter_decimate=scatter_decimate,
             title="Peak depth",
             y_label="Depth [um]",
-            durations=durations,
+            segment_start_stop_times=segment_start_stop_times,
+            sampling_frequency=sampling_frequency,
         )
 
         BaseRasterWidget.__init__(self, **plot_data, backend=backend, **backend_kwargs)