Source code for allensdk.brain_observatory.ecephys.align_timestamps.channel_states

import numpy as np

from . import barcode



[docs]def extract_barcodes_from_states(
    channel_states, timestamps, sampling_rate, **barcode_kwargs
):
    """Obtain barcodes from timestamped rising/falling edges.

    Parameters
    ----------
    channel_states : numpy.ndarray
        Rising and falling edges, denoted 1 and -1
    timestamps : numpy.ndarray
        Sample index of each event.
    sampling_rate : numeric
        Samples / second
    **barcode_kwargs :
        Additional parameters describing the barcodes.


    """

    on_events = np.where(channel_states == 1)
    off_events = np.where(channel_states == -1)

    T_on = timestamps[on_events] / float(sampling_rate)
    T_off = timestamps[off_events] / float(sampling_rate)

    return barcode.extract_barcodes_from_times(T_on, T_off, **barcode_kwargs)




[docs]def extract_splits_from_states(
    channel_states, timestamps, sampling_rate, **barcode_kwargs
):
    """Obtain data split times from timestamped rising/falling edges.

    Parameters
    ----------
    channel_states : numpy.ndarray
        Rising and falling edges, denoted 1 and -1
    timestamps : numpy.ndarray
        Sample index of each event.
    sampling_rate : numeric
        Samples / second
    **barcode_kwargs :
        Additional parameters describing the barcodes.


    """

    split_events = np.where(channel_states == 0)

    T_split = timestamps[split_events] / float(sampling_rate)

    if len(T_split) == 0:
        T_split = np.array([0])

    return T_split




[docs]def extract_splits_from_barcode_times(
    barcode_times,
    tolerance=0.0001
):
    """Determine locations of likely dropped data from barcode times
    Parameters
    ----------
    barcode_times : numpy.ndarray
        probe barcode times
    tolerance : float
        Timing tolerance (relative to median interval)
    """

    barcode_intervals = np.diff(barcode_times)

    median_interval = np.median(barcode_intervals)

    irregular_intervals = np.where(np.abs(barcode_intervals - median_interval)
                                   > tolerance * median_interval)[0]

    T_split = [0]

    for i in irregular_intervals:

        T_split.append(barcode_times[i-1])

        if i+1 < len(barcode_times):
            T_split.append(barcode_times[i+1])

    return np.array(T_split)

#