import warnings
import numpy as np
import pandas as pd
import logging
import matplotlib.pyplot as plt
from .stimulus_analysis import StimulusAnalysis
warnings.simplefilter(action='ignore', category=FutureWarning)
logger = logging.getLogger(__name__)
[docs]class DotMotion(StimulusAnalysis):
"""
A class for computing single-unit metrics from the dot motion stimulus of an ecephys session NWB file.
To use, pass in a EcephysSession object::
session = EcephysSession.from_nwb_path('/path/to/my.nwb')
dm_analysis = DotMotion(session)
or, alternatively, pass in the file path::
dm_analysis = DotMotion('/path/to/my.nwb')
You can also pass in a unit filter dictionary which will only select units with certain properties. For example
to get only those units which are on probe C and found in the VISp area::
dm_analysis = DotMotion(session, filter={'location': 'probeC', 'ecephys_structure_acronym': 'VISp'})
or a list of unit_ids:
dm_analysis = DotMotion(session, filter=[914580630, 914580280, 914580278])
To get a table of the individual unit metrics ranked by unit ID::
metrics_table_df = dm_analysis.metrics()
"""
def __init__(self, ecephys_session, col_dir='Dir', col_speeds='Speed', trial_duration=1.0, **kwargs):
super(DotMotion, self).__init__(ecephys_session, trial_duration=trial_duration, **kwargs)
self._dirvals = None
self._number_dir = None
self._speedvals = None
self._number_speed = None
self._col_dir = col_dir
self._col_speed = col_speeds
if self._params is not None:
self._params = self._params['dot_motion']
self._stimulus_key = self._params['stimulus_key']
#else:
# self._stimulus_key = 'motion_stimulus'
@property
def name(self):
return 'Dot Motion'
@property
def directions(self):
if self._dirvals is None:
self._get_stim_table_stats()
return self._dirvals
@property
def number_directions(self):
if self._number_dir is None:
self._get_stim_table_stats()
return self._number_dir
@property
def speeds(self):
if self._speedvals is None:
self._get_stim_table_stats()
return self._speedvals
@property
def number_speeds(self):
if self._number_speed is None:
self._get_stim_table_stats()
return self._number_speed
@property
def known_spontaneous_keys(self):
return ['dot_motion', "spontaneous_activity"]
@property
def null_condition(self):
""" Stimulus condition ID for null stimulus (not used, so set to -1) """
return -1
@property
def METRICS_COLUMNS(self):
return [('pref_speed_dm', np.float64),
('pref_speed_multi_dm', bool),
('pref_dir_dm', np.float64),
('pref_dir_multi_dm', bool),
('firing_rate_dm', np.float64),
('fano_dm', np.float64),
('time_to_peak_dm', np.float64),
('lifetime_sparseness_dm', np.float64),
('run_mod_dm', np.float64),
('run_pval_dm', np.float64)]
@property
def metrics(self):
if self._metrics is None:
logger.info('Calculating metrics for ' + self.name)
unit_ids = self.unit_ids
metrics_df = self.empty_metrics_table()
if len(self.stim_table) > 0:
metrics_df['pref_speed_dm'] = [self._get_pref_speed(unit) for unit in unit_ids]
metrics_df['pref_speed_multi_dm'] = [
self._check_multiple_pref_conditions(unit_id, self._col_speed, self.speeds) for unit_id in unit_ids
]
metrics_df['pref_dir_dm'] = [self._get_pref_dir(unit) for unit in unit_ids]
metrics_df['pref_dir_multi_dm'] = [
self._check_multiple_pref_conditions(unit_id, self._col_dir, self.directions) for unit_id in unit_ids
]
metrics_df['firing_rate_dm'] = [self._get_overall_firing_rate(unit) for unit in unit_ids]
metrics_df['fano_dm'] = [self._get_fano_factor(unit, self._get_preferred_condition(unit))
for unit in unit_ids]
# metrics_df['speed_tuning_idx_dm'] = [self._get_speed_tuning_index(unit) for unit in unit_ids]
metrics_df['time_to_peak_dm'] = [self._get_time_to_peak(unit, self._get_preferred_condition(unit)) for
unit in unit_ids]
metrics_df['lifetime_sparseness_dm'] = [self._get_lifetime_sparseness(unit) for unit in unit_ids]
metrics_df.loc[:, ['run_pval_dm', 'run_mod_dm']] = \
[self._get_running_modulation(unit, self._get_preferred_condition(unit)) for unit in unit_ids]
self._metrics = metrics_df
return self._metrics
[docs] @classmethod
def known_stimulus_keys(cls):
return ['motion_stimulus', 'dot_motion']
def _get_stim_table_stats(self):
""" Extract directions and speeds from the stimulus table """
self._dirvals = np.sort(self.stimulus_conditions.loc[self.stimulus_conditions[self._col_dir]
!= 'null'][self._col_dir].unique())
self._number_dir = len(self._dirvals)
self._speedvals = np.sort(self.stimulus_conditions.loc[self.stimulus_conditions[self._col_speed]
!= 'null'][self._col_speed].unique())
self._number_speed = len(self._speedvals)
def _get_pref_speed(self, unit_id):
""" Calculate the preferred speed condition for a given unit
Parameters
----------
unit_id : int
unique ID for the unit of interest
Returns
-------
pref_speed :
stimulus speed driving the maximal response
"""
# TODO: Most of the _get_pref_*() methods can be combined into one method and shared among the classes
similar_conditions = [self.stimulus_conditions.index[self.stimulus_conditions[self._col_speed]
== speed].tolist() for speed in self.speeds]
df = pd.DataFrame(
index=self.speeds,
data={'spike_mean': [self.conditionwise_statistics.loc[unit_id].loc[condition_inds]['spike_mean'].mean()
for condition_inds in similar_conditions]}
).rename_axis(self._col_speed)
return df.idxmax().iloc[0]
def _get_pref_dir(self, unit_id):
"""Calculate the preferred direction condition for a given unit
Parameters
----------
unit_id : int
unique ID for the unit of interest
Returns
-------
pref_dir : float
stimulus direction driving the maximal response
"""
similar_conditions = [self.stimulus_conditions.index[self.stimulus_conditions[self._col_dir]
== direction].tolist() for direction in self.directions]
df = pd.DataFrame(
index=self.directions,
data={'spike_mean': [self.conditionwise_statistics.loc[unit_id].loc[condition_inds]['spike_mean'].mean()
for condition_inds in similar_conditions]}
).rename_axis(self._col_dir)
return df.idxmax().iloc[0]
def _get_speed_tuning_index(self, unit_id):
""" Calculate the speed tuning for a given unit
SEE: https://github.com/AllenInstitute/ecephys_analysis_modules/blob/master/ecephys_analysis_modules/modules/tuning/tuning_speed.py
Parameters
----------
unit_id : int
unique ID for the unit of interest
Returns
-------
speed_tuning : float
degree to which the unit's responses are modulated by stimulus speed
"""
# TODO: Not implemented yet.
return np.nan