Source code for allensdk.brain_observatory.ecephys.file_io.ecephys_sync_dataset
from itertools import product
import functools
from collections import defaultdict
import logging
import warnings
import numpy as np
from allensdk.brain_observatory.sync_dataset import Dataset
from allensdk.brain_observatory.ecephys import stimulus_sync
from allensdk.brain_observatory import sync_utilities
[docs]class EcephysSyncDataset(Dataset):
@property
def sample_frequency(self):
return self.meta_data['ni_daq']['counter_output_freq']
@sample_frequency.setter
def sample_frequency(self, value):
if not hasattr(self, 'meta_data'):
self.meta_data = defaultdict(dict)
self.meta_data['ni_daq']['counter_output_freq'] = value
def __init__(self):
'''In-memory representation of a sync h5 file as produced by the sync package.
Notes
-----
base is from here: http://aibspi/mpe_apps/sync/blob/master/sync/dataset.py
Construction works slightly differently for this class as its base. In particular,
this class' __init__ method merely constructs the object. To make a new SyncDataset in client code, use the
factory classmethod. This is done for ease of testability.
'''
pass
[docs] def extract_led_times(self, keys=Dataset.OPTOGENETIC_STIMULATION_KEYS, fallback_line=18):
try:
led_times = self.get_edges(
kind="rising",
keys=keys,
units="seconds"
)
except KeyError:
warnings.warn(f"unable to find LED times using line labels {keys}, returning line {fallback_line}")
led_times = self.get_rising_edges(fallback_line, units="seconds")
return led_times
[docs] def extract_frame_times_from_photodiode(self, photodiode_cycle=60, frame_keys=Dataset.FRAME_KEYS, photodiode_keys=Dataset.PHOTODIODE_KEYS):
photodiode_times = self.get_edges('all', photodiode_keys)
vsync_times = self.get_edges('falling', frame_keys)
vsync_times = sync_utilities.trim_discontiguous_times(vsync_times)
logging.info(f"Total vsyncs: {len(vsync_times)}")
photodiode_times = stimulus_sync.trim_border_pulses(photodiode_times, vsync_times)
photodiode_times = stimulus_sync.correct_on_off_effects(photodiode_times)
photodiode_times = stimulus_sync.fix_unexpected_edges(photodiode_times, cycle=photodiode_cycle)
frame_duration = stimulus_sync.estimate_frame_duration(photodiode_times, cycle=photodiode_cycle)
irregular_interval_policy = functools.partial(stimulus_sync.allocate_by_vsync, np.diff(vsync_times))
frame_indices, frame_start_times, frame_end_times = stimulus_sync.compute_frame_times(
photodiode_times, frame_duration, len(vsync_times),
cycle=photodiode_cycle, irregular_interval_policy=irregular_interval_policy
)
return frame_start_times
[docs] def extract_frame_times_from_vsyncs(self, photodiode_cycle=60,
frame_keys=Dataset.FRAME_KEYS, photodiode_keys=Dataset.PHOTODIODE_KEYS
):
raise NotImplementedError()
[docs] def extract_frame_times(self, strategy, photodiode_cycle=60,
frame_keys=Dataset.FRAME_KEYS, photodiode_keys=Dataset.PHOTODIODE_KEYS
):
if strategy == 'use_photodiode':
return self.extract_frame_times_from_photodiode(
photodiode_cycle=photodiode_cycle, frame_keys=frame_keys, photodiode_keys=photodiode_keys
)
elif strategy == 'use_vsyncs':
return self.extract_frame_times_from_vsyncs(
photodiode_cycle=photodiode_cycle, frame_keys=frame_keys, photodiode_keys=photodiode_keys
)
else:
raise ValueError('unrecognized strategy: {}'.format(strategy))
[docs] @classmethod
def factory(cls, path):
''' Build a new SyncDataset.
Parameters
----------
path : str
Filesystem path to the h5 file containing sync information to be loaded.
'''
obj = cls()
obj.load(path)
return obj