Source code for allensdk.brain_observatory.sync_utilities

from pathlib import Path
from typing import Tuple, Optional, List
import numpy as np
import pandas as pd

from allensdk.brain_observatory.sync_dataset import Dataset


def trim_discontiguous_times(times: np.ndarray, threshold=100) -> np.ndarray:
    """
    If the time sequence is discontigous,
    detect the first instance occurance and trim off the tail of the sequence

    Parameters
    ----------
    times : frame times

    Returns
    -------
    trimmed frame times
    """

    times = np.array(times)
    intervals = np.diff(times)

    med_interval = np.median(intervals)
    interval_threshold = med_interval * threshold

    gap_indices = np.where(intervals > interval_threshold)[0]

    # A special case for when the first element is a discontiguity
    if np.abs(intervals[0]) > interval_threshold:
        gap_indices = [0]

    if len(gap_indices) == 0:
        return times

    return times[:gap_indices[0] + 1]


def get_synchronized_frame_times(session_sync_file: Path,
                                 sync_line_label_keys: Tuple[str, ...],
                                 drop_frames: Optional[List[int]] = None,
                                 trim_after_spike: bool = True,
                                 ) -> pd.Series:
    """Get experimental frame times from an experiment session sync file.

    1. Get rising edges from the sync dataset
    2. Occasionally an extra set of frame times are acquired after the rest of
        the signals. These are manifested by a discontiguous time sequence.
        We detect and remove these.
    3. Remove dropped frames

    Parameters
    ----------
    session_sync_file : Path
        Path to an ephys session sync file.
        The sync file contains rising/falling edges from a daq system which
        indicates when certain events occur (so they can be related to
        each other).
    sync_line_label_keys : Tuple[str, ...]
        Line label keys to get times for. See class attributes of
        allensdk.brain_observatory.sync_dataset.Dataset for a listing of
        possible keys.
    drop_frames : List
        frame indices to be removed from frame times
    trim_after_spike : bool = True
        If True, will call trim_discontiguous_times on the frame times
        before returning them, which will detect any spikes in the data
        and remove all elements for the list which come after the spike.

    Returns
    -------
    pd.Series
        An array of times when eye tracking frames were acquired.
    """
    sync_dataset = Dataset(str(session_sync_file))

    times = sync_dataset.get_edges(
        "rising", sync_line_label_keys, units="seconds"
    )

    times = trim_discontiguous_times(times) if trim_after_spike else times
    if drop_frames is not None:
        times = [t for ix, t in enumerate(times) if ix not in drop_frames]

    return pd.Series(times)