recording_tools.py

from copy import deepcopy
from typing import Literal
import warnings
from pathlib import Path
import os
import tqdm
import numpy.typing as npt

import numpy as np

from .core_tools import add_suffix, make_shared_array
from .job_tools import (
    ensure_chunk_size,
    divide_segment_into_chunks,
    fix_job_kwargs,
    TimeSeriesChunkExecutor,
    _shared_job_kwargs_doc,
    split_job_kwargs,
)

from .time_series_tools import get_random_sample_slices, get_chunks, get_time_series_chunk_with_margin
from .time_series_tools import write_binary as _write_binary
from .time_series_tools import write_memory as _write_memory
from .time_series_tools import _write_time_series_to_zarr


def write_binary_recording(
    recording,
    file_paths,
    file_timestamps_paths=None,
    dtype=None,
    add_file_extension=True,
    byte_offset=0,
    verbose=False,
    **job_kwargs,
):
    """
    Save the traces of a recording to binary format.

    Parameters
    ----------
    recording : BaseRecording
        The recording to save to binary file.
    file_paths : list[Path | str] | Path | str
        The path to the files to save data for each segment.
    file_timestamps_paths : list[Path | str] | Path | str | None, default: None
        The path to the timestamps file. If None, timestamps are not saved.
    dtype : dtype or None, default: None
        Type of the saved data.
    add_file_extension : bool, default: True
        If True, and the file path does not end in "raw", "bin", or "dat" then "raw" is added as an extension.
    byte_offset : int, default: 0
        Offset in bytes for the binary file (e.g. to write a header).
    verbose : bool, default: False
        Verbosity of the chunk executor.
    {}
    """
    return _write_binary(
        recording,
        file_paths=file_paths,
        file_timestamps_paths=file_timestamps_paths,
        dtype=dtype,
        add_file_extension=add_file_extension,
        byte_offset=byte_offset,
        verbose=verbose,
        **job_kwargs,
    )


write_binary_recording.__doc__ = write_binary_recording.__doc__.format(_shared_job_kwargs_doc)


def write_memory_recording(
    recording,
    dtype=None,
    verbose=False,
    buffer_type="auto",
    job_name="write_memory",
    **job_kwargs,
):
    """
    Save the traces of a recording into numpy arrays in memory.

    Uses SharedMemory when ``n_jobs > 1``.

    Parameters
    ----------
    recording : BaseRecording
        The recording to save to memory.
    dtype : dtype, default: None
        Type of the saved data.
    verbose : bool, default: False
        If True, output is verbose (when chunks are used).
    buffer_type : "auto" | "numpy" | "sharedmem", default: "auto"
        The type of buffer to use for storing the data.
    job_name : str, default: "write_memory"
        Name of the job.
    {}

    Returns
    -------
    arrays : list
        One array per segment.
    """
    return _write_memory(
        recording,
        dtype=dtype,
        verbose=verbose,
        buffer_type=buffer_type,
        job_name=job_name,
        **job_kwargs,
    )


write_memory_recording.__doc__ = write_memory_recording.__doc__.format(_shared_job_kwargs_doc)


def write_recording_to_zarr(
    recording,
    zarr_group,
    dataset_paths,
    dataset_timestamps_paths=None,
    extra_chunks=None,
    dtype=None,
    compressor_data=None,
    filters_data=None,
    compressor_times=None,
    filters_times=None,
    verbose=False,
    **job_kwargs,
):
    """
    Save the traces of a recording to zarr format.

    Parameters
    ----------
    recording : BaseRecording
        The recording to save in zarr format.
    zarr_group : zarr.Group
        The zarr group to add traces to.
    dataset_paths : list
        List of paths to traces datasets in the zarr group.
    dataset_timestamps_paths : list or None, default: None
        List of paths to timestamps datasets in the zarr group. If None, timestamps are not saved.
    extra_chunks : tuple or None, default: None
        Extra chunking dimensions to use for the zarr dataset. The first dimension is always time and
        controlled by the job_kwargs. Useful to chunk by channel, with ``extra_chunks=(channel_chunk_size,)``.
    dtype : dtype, default: None
        Type of the saved data.
    compressor_data : zarr compressor or None, default: None
        Zarr compressor for data.
    filters_data : list, default: None
        List of zarr filters for data.
    compressor_times : zarr compressor or None, default: None
        Zarr compressor for timestamps.
    filters_times : list, default: None
        List of zarr filters for timestamps.
    verbose : bool, default: False
        If True, output is verbose (when chunks are used).
    {}
    """
    return _write_time_series_to_zarr(
        recording,
        zarr_group=zarr_group,
        dataset_paths=dataset_paths,
        dataset_timestamps_paths=dataset_timestamps_paths,
        extra_chunks=extra_chunks,
        dtype=dtype,
        compressor_data=compressor_data,
        filters_data=filters_data,
        compressor_times=compressor_times,
        filters_times=filters_times,
        verbose=verbose,
        **job_kwargs,
    )


write_recording_to_zarr.__doc__ = write_recording_to_zarr.__doc__.format(_shared_job_kwargs_doc)


def read_binary_recording(file, num_channels, dtype, time_axis=0, offset=0):
    """
    Read binary .bin or .dat file.

    Parameters
    ----------
    file : str
        File name
    num_channels : int
        Number of channels
    dtype : dtype
        dtype of the file
    time_axis : 0 or 1, default: 0
        If 0 then traces are transposed to ensure (nb_sample, nb_channel) in the file.
        If 1, the traces shape (nb_channel, nb_sample) is kept in the file.
    offset : int, default: 0
        number of offset bytes

    """
    # TODO change this function to read_binary_traces() because this name is confusing
    num_channels = int(num_channels)
    with Path(file).open() as f:
        nsamples = (os.fstat(f.fileno()).st_size - offset) // (num_channels * np.dtype(dtype).itemsize)
    if time_axis == 0:
        samples = np.memmap(file, np.dtype(dtype), mode="r", offset=offset, shape=(nsamples, num_channels))
    else:
        samples = np.memmap(file, np.dtype(dtype), mode="r", offset=offset, shape=(num_channels, nsamples)).T
    return samples


def write_binary_file_handle(
    recording, file_handle=None, time_axis=0, dtype=None, byte_offset=0, verbose=False, **job_kwargs
):
    """
    Old variant version of write_binary with one file handle.
    Can be useful in some case ???
    Not used anymore at the moment.

    @ SAM useful for writing with time_axis=1!
    """
    assert file_handle is not None
    assert recording.get_num_segments() == 1, "If file_handle is given then only deals with one segment"

    if dtype is None:
        dtype = recording.get_dtype()

    job_kwargs = fix_job_kwargs(job_kwargs)
    chunk_size = ensure_chunk_size(recording, **job_kwargs)

    if chunk_size is not None and time_axis == 1:
        print("Chunking disabled due to 'time_axis' == 1")
        chunk_size = None

    if chunk_size is None:
        # no chunking
        traces = recording.get_traces(segment_index=0)
        if time_axis == 1:
            traces = traces.T
        if dtype is not None:
            traces = traces.astype(dtype, copy=False)
        traces.tofile(file_handle)
    else:
        num_frames = recording.get_num_samples(segment_index=0)
        chunks = divide_segment_into_chunks(num_frames, chunk_size)

        for start_frame, end_frame in chunks:
            traces = recording.get_traces(segment_index=0, start_frame=start_frame, end_frame=end_frame)
            if time_axis == 1:
                traces = traces.T
            if dtype is not None:
                traces = traces.astype(dtype, copy=False)
            file_handle.write(traces.tobytes())


def write_to_h5_dataset_format(
    recording,
    dataset_path,
    segment_index,
    save_path=None,
    file_handle=None,
    time_axis=0,
    single_axis=False,
    dtype=None,
    chunk_size=None,
    chunk_memory="500M",
    verbose=False,
    return_scaled=None,
    return_in_uV=False,
):
    """
    Save the traces of a recording extractor in an h5 dataset.

    Parameters
    ----------
    recording : RecordingExtractor
        The recording extractor object to be saved in .dat format
    dataset_path : str
        Path to dataset in the h5 file (e.g. "/dataset")
    segment_index : int
        index of segment
    save_path : str, default: None
        The path to the file.
    file_handle : file handle, default: None
        The file handle to dump data. This can be used to append data to an header. In case file_handle is given,
        the file is NOT closed after writing the binary data.
    time_axis : 0 or 1, default: 0
        If 0 then traces are transposed to ensure (nb_sample, nb_channel) in the file.
        If 1, the traces shape (nb_channel, nb_sample) is kept in the file.
    single_axis : bool, default: False
        If True, a single-channel recording is saved as a one dimensional array
    dtype : dtype, default: None
        Type of the saved data
    chunk_size : None or int, default: None
        Number of chunks to save the file in. This avoids too much memory consumption for big files.
        If None and "chunk_memory" is given, the file is saved in chunks of "chunk_memory" MB
    chunk_memory : None or str, default: "500M"
        Chunk size in bytes must end with "k", "M" or "G"
    verbose : bool, default: False
        If True, output is verbose (when chunks are used)
    return_scaled : bool | None, default: None
        DEPRECATED. Use return_in_uV instead.
    return_in_uV : bool, default: False
        If True and the recording has scaling (gain_to_uV and offset_to_uV properties),
        traces are dumped to uV
    """
    import h5py

    # ~ assert HAVE_H5, "To write to h5 you need to install h5py: pip install h5py"
    assert save_path is not None or file_handle is not None, "Provide 'save_path' or 'file handle'"

    if save_path is not None:
        save_path = Path(save_path)
        if save_path.suffix == "":
            # when suffix is already raw/bin/dat do not change it.
            save_path = save_path.parent / (save_path.name + ".h5")

    num_channels = recording.get_num_channels()
    num_frames = recording.get_num_frames(segment_index=0)

    if file_handle is not None:
        assert isinstance(file_handle, h5py.File)
    else:
        file_handle = h5py.File(save_path, "w")

    if dtype is None:
        dtype_file = recording.get_dtype()
    else:
        dtype_file = dtype

    if single_axis:
        shape = (num_frames,)
    else:
        if time_axis == 0:
            shape = (num_frames, num_channels)
        else:
            shape = (num_channels, num_frames)

    dset = file_handle.create_dataset(dataset_path, shape=shape, dtype=dtype_file)

    chunk_size = ensure_chunk_size(recording, chunk_size=chunk_size, chunk_memory=chunk_memory, n_jobs=1)

    if chunk_size is None:
        # Handle deprecated return_scaled parameter
        if return_scaled is not None:
            warnings.warn(
                "`return_scaled` is deprecated and will be removed in version 0.105.0. Use `return_in_uV` instead.",
                category=DeprecationWarning,
            )
            return_in_uV = return_scaled

        traces = recording.get_traces(return_in_uV=return_in_uV)
        if dtype is not None:
            traces = traces.astype(dtype_file, copy=False)
        if time_axis == 1:
            traces = traces.T
        if single_axis:
            dset[:] = traces[:, 0]
        else:
            dset[:] = traces
    else:
        chunk_start = 0
        # chunk size is not None
        n_chunk = num_frames // chunk_size
        if num_frames % chunk_size > 0:
            n_chunk += 1
        if verbose:
            chunks = tqdm(range(n_chunk), ascii=True, desc="Writing to .h5 file")
        else:
            chunks = range(n_chunk)
        for i in chunks:
            traces = recording.get_traces(
                segment_index=segment_index,
                start_frame=i * chunk_size,
                end_frame=min((i + 1) * chunk_size, num_frames),
                return_in_uV=return_in_uV,
            )
            chunk_frames = traces.shape[0]
            if dtype is not None:
                traces = traces.astype(dtype_file, copy=False)
            if single_axis:
                dset[chunk_start : chunk_start + chunk_frames] = traces[:, 0]
            else:
                if time_axis == 0:
                    dset[chunk_start : chunk_start + chunk_frames, :] = traces
                else:
                    dset[:, chunk_start : chunk_start + chunk_frames] = traces.T

            chunk_start += chunk_frames

    if save_path is not None:
        file_handle.close()
    return save_path


def get_random_data_chunks(
    recording, return_scaled=None, return_in_uV=False, concatenated=True, **random_slices_kwargs
):
    """
    Extract random chunks across segments.

    Internally, it uses `get_random_sample_slices()` and retrieves the traces chunk as a list
    or a concatenated unique array.

    Please read `get_random_sample_slices()` for more details on parameters.


    Parameters
    ----------
    recording : BaseRecording
        The recording to get random chunks from
    return_scaled : bool | None, default: None
        DEPRECATED. Use return_in_uV instead.
    return_in_uV : bool, default: False
        If True and the recording has scaling (gain_to_uV and offset_to_uV properties),
        traces are scaled to uV
    num_chunks_per_segment : int, default: 20
        Number of chunks per segment
    concatenated : bool, default: True
        If True chunk are concatenated along time axis
    **random_slices_kwargs : dict
        Options transmited to  get_random_sample_slices(), please read documentation from this
        function for more details.

    Returns
    -------
    chunk_list : np.ndarray | list of np.array
        Array of concatenate chunks per segment
    """
    # Handle deprecated return_scaled parameter
    if return_scaled is not None:
        warnings.warn(
            "`return_scaled` is deprecated and will be removed in version 0.105.0. Use `return_in_uV` instead.",
            category=DeprecationWarning,
            stacklevel=2,
        )
        return_in_uV = return_scaled

    return get_chunks(
        recording,
        concatenated=concatenated,
        get_data_kwargs=dict(return_in_uV=return_in_uV),
        **random_slices_kwargs,
    )


def get_channel_distances(recording):
    """
    Distance between channel pairs
    """
    locations = recording.get_channel_locations()
    channel_distances = np.linalg.norm(locations[:, np.newaxis] - locations[np.newaxis, :], axis=2)

    return channel_distances


def get_closest_channels(recording, channel_ids=None, num_channels=None):
    """Get closest channels + distances

    Parameters
    ----------
    recording : RecordingExtractor
        The recording extractor to get closest channels
    channel_ids : list
        List of channels ids to compute there near neighborhood
    num_channels : int, default: None
        Maximum number of neighborhood channels to return

    Returns
    -------
    closest_channels_inds : array (2d)
        Closest channel indices in ascending order for each channel id given in input
    dists : array (2d)
        Distance in ascending order for each channel id given in input
    """
    if channel_ids is None:
        channel_ids = recording.get_channel_ids()
    if num_channels is None:
        num_channels = len(channel_ids) - 1

    locations = recording.get_channel_locations(channel_ids=channel_ids)

    closest_channels_inds = []
    dists = []
    for i in range(locations.shape[0]):
        distances = np.linalg.norm(locations[i, :] - locations, axis=1)
        order = np.argsort(distances)
        closest_channels_inds.append(order[1 : num_channels + 1])
        dists.append(distances[order][1 : num_channels + 1])

    return np.array(closest_channels_inds), np.array(dists)


def _noise_level_chunk(segment_index, start_frame, end_frame, worker_ctx):
    recording = worker_ctx["recording"]

    one_chunk = recording.get_traces(
        start_frame=start_frame,
        end_frame=end_frame,
        segment_index=segment_index,
        return_in_uV=worker_ctx["return_in_uV"],
    )

    if worker_ctx["method"] == "mad":
        med = np.median(one_chunk, axis=0, keepdims=True)
        # hard-coded so that core doesn't depend on scipy
        noise_levels = np.median(np.abs(one_chunk - med), axis=0) / 0.6744897501960817
    elif worker_ctx["method"] == "std":
        noise_levels = np.std(one_chunk, axis=0)
    elif worker_ctx["method"] == "rms":
        # sqrt(1/n * (x0^2 + ... + xn^2))
        squared_voltages = np.square(one_chunk)
        summed_squared_voltages = np.sum(squared_voltages, axis=0)
        noise_levels = np.sqrt(summed_squared_voltages / one_chunk.shape[0])

    return noise_levels


def _noise_level_chunk_init(recording, return_in_uV, method):
    worker_ctx = {}
    worker_ctx["recording"] = recording
    worker_ctx["return_in_uV"] = return_in_uV
    worker_ctx["method"] = method
    return worker_ctx


def get_noise_levels(
    recording: "BaseRecording",
    return_scaled: bool | None = None,
    return_in_uV: bool = True,
    method: Literal["mad", "std", "rms"] = "mad",
    force_recompute: bool = False,
    random_slices_kwargs: dict = {},
    **kwargs,
) -> np.ndarray:
    """
    Estimate noise for each channel using MAD methods.
    You can use standard deviation with `method="std"`

    Internally it samples some chunk across segment.
    And then, it uses the MAD estimator (more robust than STD) or the STD on each chunk.
    Finally the average of all MAD/STD values is performed.

    The result is cached in a property of the recording, so that the next call on the same
    recording will use the cached result unless `force_recompute=True`.

    Parameters
    ----------

    recording : BaseRecording
        The recording extractor to get noise levels
    return_scaled : bool | None, default: None
        DEPRECATED. Use return_in_uV instead.
    return_in_uV : bool, default: True
        If True, returned noise levels are scaled to uV
    method : "mad" | "std" | "rms", default: "mad"
        The method to use to estimate noise levels
    force_recompute : bool
        If True, noise levels are recomputed even if they are already stored in the recording extractor
    random_slices_kwargs : dict
        Options transmitted to  get_random_sample_slices(), please read documentation from this
        function for more details.

    {}

    Returns
    -------
    noise_levels : array
        Noise levels for each channel
    """

    # Handle deprecated return_scaled parameter
    if return_scaled is not None:
        warnings.warn(
            "`return_scaled` is deprecated and will be removed in version 0.105.0. Use `return_in_uV` instead.",
            category=DeprecationWarning,
        )
        return_in_uV = return_scaled

    if return_in_uV:
        key = f"noise_level_{method}_scaled"
    else:
        key = f"noise_level_{method}_raw"

    if key in recording.get_property_keys() and not force_recompute:
        noise_levels = recording.get_property(key=key)
    else:
        # This is to keep backward compatibility
        # lets keep for a while and remove this maybe in 0.103.0
        # chunk_size used to be in the signature and now is ambiguous
        random_slices_kwargs_, job_kwargs = split_job_kwargs(kwargs)
        if len(random_slices_kwargs_) > 0 or "chunk_size" in job_kwargs:
            msg = (
                "get_noise_levels(recording, num_chunks_per_segment=20) is deprecated\n"
                "Now, you need to use get_noise_levels(recording, random_slices_kwargs=dict(num_chunks_per_segment=20, chunk_size=1000))\n"
                "Please read get_random_sample_slices() documentation for more options."
            )
            # if the user use both the old and the new behavior then an error is raised
            assert len(random_slices_kwargs) == 0, msg
            warnings.warn(msg)
            random_slices_kwargs = random_slices_kwargs_
            if "chunk_size" in job_kwargs:
                random_slices_kwargs["chunk_size"] = job_kwargs["chunk_size"]

        slices = get_random_sample_slices(recording, **random_slices_kwargs)

        noise_levels_chunks = []

        def append_noise_chunk(res):
            noise_levels_chunks.append(res)

        func = _noise_level_chunk
        init_func = _noise_level_chunk_init
        init_args = (recording, return_in_uV, method)
        executor = TimeSeriesChunkExecutor(
            recording,
            func,
            init_func,
            init_args,
            job_name="noise_level",
            verbose=False,
            gather_func=append_noise_chunk,
            **job_kwargs,
        )
        executor.run(slices=slices)
        noise_levels_chunks = np.stack(noise_levels_chunks)
        noise_levels = np.mean(noise_levels_chunks, axis=0)

        # set property
        recording.set_property(key, noise_levels)

    return noise_levels


get_noise_levels.__doc__ = get_noise_levels.__doc__.format(_shared_job_kwargs_doc)


def order_channels_by_depth(recording, channel_ids=None, dimensions=("x", "y"), flip=False):
    """
    Order channels by depth, by first ordering the x-axis, and then the y-axis.

    Parameters
    ----------
    recording : BaseRecording
        The input recording
    channel_ids : list/array or None
        If given, a subset of channels to order locations for
    dimensions : str, tuple, or list, default: ('x', 'y')
        If str, it needs to be 'x', 'y', 'z'.
        If tuple or list, it sorts the locations in two dimensions using lexsort.
        This approach is recommended since there is less ambiguity
    flip : bool, default: False
        If flip is False then the order is bottom first (starting from tip of the probe).
        If flip is True then the order is upper first.

    Returns
    -------
    order_f : np.array
        Array with sorted indices
    order_r : np.array
        Array with indices to revert sorting
    """
    locations = recording.get_channel_locations()
    ndim = locations.shape[1]
    channel_inds = recording.ids_to_indices(ids=channel_ids, prefer_slice=True)
    locations = locations[channel_inds, :]

    if isinstance(dimensions, str):
        dim = ["x", "y", "z"].index(dimensions)
        assert dim < ndim, "Invalid dimensions!"
        order_f = np.argsort(locations[:, dim], kind="stable")
    else:
        assert isinstance(dimensions, (tuple, list)), "dimensions can be str, tuple, or list"
        locations_to_sort = ()
        for dim in dimensions:
            dim = ["x", "y", "z"].index(dim)
            assert dim < ndim, "Invalid dimensions!"
            locations_to_sort += (locations[:, dim],)
        order_f = np.lexsort(locations_to_sort)
    if flip:
        order_f = order_f[::-1]
    order_r = np.argsort(order_f, kind="stable")

    return order_f, order_r


def check_probe_do_not_overlap(probes):
    """
    When several probes this check that that they do not overlap in space
    and so channel positions can be safely concatenated.

    Raises
    ------
    Exception :
        If probes are overlapping

    Returns
    -------
    None : None
        If the check is successful
    """
    for i in range(len(probes)):
        probe_i = probes[i]
        # check that all positions in probe_j are outside probe_i boundaries
        x_bounds_i = [
            np.min(probe_i.contact_positions[:, 0]),
            np.max(probe_i.contact_positions[:, 0]),
        ]
        y_bounds_i = [
            np.min(probe_i.contact_positions[:, 1]),
            np.max(probe_i.contact_positions[:, 1]),
        ]

        for j in range(i + 1, len(probes)):
            probe_j = probes[j]
            if np.any(
                np.array(
                    [
                        x_bounds_i[0] <= cp[0] <= x_bounds_i[1] and y_bounds_i[0] <= cp[1] <= y_bounds_i[1]
                        for cp in probe_j.contact_positions
                    ]
                )
            ):
                raise Exception("Probes are overlapping! Retrieve locations of single probes separately")


def get_rec_attributes(recording):
    """
    Construct rec_attributes from recording object

    Parameters
    ----------
    recording : BaseRecording
        The recording object

    Returns
    -------
    dict
        The rec_attributes dictionary
    """
    properties_to_attrs = deepcopy(recording._properties)
    if "contact_vector" in properties_to_attrs:
        del properties_to_attrs["contact_vector"]
    rec_attributes = dict(
        channel_ids=recording.channel_ids,
        sampling_frequency=recording.get_sampling_frequency(),
        num_channels=recording.get_num_channels(),
        num_samples=[recording.get_num_samples(seg_index) for seg_index in range(recording.get_num_segments())],
        is_filtered=recording.is_filtered(),
        properties=properties_to_attrs,
        dtype=recording.get_dtype(),
    )
    return rec_attributes


def do_recording_attributes_match(
    recording1: "BaseRecording", recording2_attributes: bool, check_dtype: bool = True
) -> tuple[bool, str]:
    """
    Check if two recordings have the same attributes

    Parameters
    ----------
    recording1 : BaseRecording
        The first recording object
    recording2_attributes : dict
        The recording attributes to test against
    check_dtype : bool, default: True
        If True, check if the recordings have the same dtype

    Returns
    -------
    bool
        True if the recordings have the same attributes
    str
        A string with the exception message with the attributes that do not match
    """
    recording1_attributes = get_rec_attributes(recording1)
    recording2_attributes = deepcopy(recording2_attributes)
    recording1_attributes.pop("properties")
    recording2_attributes.pop("properties")

    attributes_match = True
    non_matching_attrs = []

    if not np.array_equal(recording1_attributes["channel_ids"], recording2_attributes["channel_ids"]):
        non_matching_attrs.append("channel_ids")
    if not recording1_attributes["sampling_frequency"] == recording2_attributes["sampling_frequency"]:
        non_matching_attrs.append("sampling_frequency")
    if not recording1_attributes["num_channels"] == recording2_attributes["num_channels"]:
        non_matching_attrs.append("num_channels")
    if not recording1_attributes["num_samples"] == recording2_attributes["num_samples"]:
        non_matching_attrs.append("num_samples")
    # dtype is optional
    if "dtype" in recording1_attributes and "dtype" in recording2_attributes:
        if check_dtype:
            if not recording1_attributes["dtype"] == recording2_attributes["dtype"]:
                non_matching_attrs.append("dtype")

    if len(non_matching_attrs) > 0:
        attributes_match = False
        exception_str = f"Recordings do not match in the following attributes: {non_matching_attrs}"
    else:
        attributes_match = True
        exception_str = ""

    return attributes_match, exception_str


def get_chunk_with_margin(
    rec_segment,
    start_frame,
    end_frame,
    channel_indices,
    margin,
    add_zeros=False,
    add_reflect_padding=False,
    window_on_margin=False,
    dtype=None,
):
    """
    Helper to get chunk with margin

    The margin is extracted from the recording when possible. If
    at the edge of the recording, no margin is used unless one
    of `add_zeros` or `add_reflect_padding` is True. In the first
    case zero padding is used, in the second case np.pad is called
    with mod="reflect".
    """

    # wrapper to keep backward compatibility with the previous naming and signature
    # this help dartsort for instance
    return get_time_series_chunk_with_margin(
        rec_segment,
        start_frame,
        end_frame,
        channel_indices,
        margin,
        add_zeros=add_zeros,
        add_reflect_padding=add_reflect_padding,
        window_on_margin=window_on_margin,
        dtype=dtype,
    )