Source code for allensdk.brain_observatory.sync_utilities
from pathlib import Path
from typing import Tuple
import numpy as np
import pandas as pd
from allensdk.brain_observatory.sync_dataset import Dataset
[docs]def trim_discontiguous_times(times, threshold=100):
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]
[docs]def get_synchronized_frame_times(session_sync_file: Path,
sync_line_label_keys: Tuple[str, ...],
trim_after_spike: bool = True) -> pd.Series:
"""Get experimental frame times from an experiment session sync file.
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.
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 frames for the eye tracking camera were acquired.
"""
sync_dataset = Dataset(str(session_sync_file))
frame_times = sync_dataset.get_edges(
"rising", sync_line_label_keys, units="seconds"
)
# Occasionally an extra set of frame times are acquired after the rest of
# the signals. We detect and remove these.
frame_times = trim_discontiguous_times(frame_times) if trim_after_spike else frame_times
return pd.Series(frame_times)