@schema
class OptogeneticProtocol(SpyglassIngestion, dj.Manual):
definition = """
# Describes the optogenetic stimulation protocol used within an epoch
-> TaskEpoch
---
description: varchar(255) # description of the optogenetic stimulation
pulse_length: float # pulse length in ms
pulses_per_train: int # number of pulses per train
period: float # period in ms
intertrain_interval: float # intertrain interval in ms
stimulus_power: float # stimulus power in mW
stimulus_object_id: varchar(64) # object id of the dio corresponding to the optogenetic stimulation
"""
_nwb_table = Nwbfile
_source_nwb_object_name = "optogenetic_epochs"
# Exact class name: ndx_franklab_novela may not be importable
_source_nwb_object_type = "FrankLabOptogeneticEpochsTable"
# Master columns, read straight off a row of the epochs table. Each part
# declares its own columns the same way, so every mapping is visible to
# the ingestion-mapping doc generator.
table_key_to_obj_attr = {
"self": {
"epoch": "epoch_number",
"description": "convenience_code",
"pulse_length": "pulse_length_in_ms",
"pulses_per_train": "number_pulses_per_pulse_train",
"period": "period_in_ms",
"intertrain_interval": "intertrain_interval_in_ms",
"stimulus_power": "power_in_mW",
},
"stimulus_signal": {"stimulus_object_id": "object_id"},
}
def generate_entries_from_nwb_object(self, nwb_obj, base_key=None):
"""Generate a master entry plus whichever conditional parts apply.
Called once per row of the optogenetic epochs table. Each row yields
one master entry and a part entry for every trigger or condition the
row switches on.
"""
entries = super().generate_entries_from_nwb_object(nwb_obj, base_key)
if hasattr(nwb_obj, "to_dataframe"):
# Handed the whole epochs table: super() has already expanded it
# row by row, back through this method. Nothing to add here.
return entries
epoch_key = dict(
nwb_file_name=entries[self][0]["nwb_file_name"],
epoch=nwb_obj.epoch_number,
)
# Classes, not instances: entries from several rows are merged by
# dict key, and a fresh instance would be a fresh key each time.
conditions = [
("ripple_filter_on", self.RippleTrigger),
("theta_filter_on", self.ThetaTrigger),
("speed_filter_on", self.SpeedConditional),
]
for flag, part in conditions:
if getattr(nwb_obj, flag, None):
entries.setdefault(part, []).append(
dict(epoch_key, **self._part_fields(nwb_obj, part))
)
if (
nodes := getattr(
nwb_obj,
"spatial_filter_region_node_coordinates_in_pixels",
None,
)
) is not None:
entries.setdefault(self.SpatialConditional, []).append(
dict(
epoch_key,
nodes=nodes * nwb_obj.spatial_filter_cameras_cm_per_pixel,
)
)
return entries
def _part_fields(self, row, part) -> dict:
"""Read a part table's own columns off a row of the epochs table.
The part declares which row attribute each of its columns comes from,
so the mapping lives on the table it describes rather than in a shared
dict on the master.
Parameters
----------
row : namedtuple
One row of the optogenetic epochs dataframe.
part : dj.Part
The part table whose secondary attributes are wanted.
Returns
-------
dict
Attribute values keyed by part table column name.
"""
return {
name: getattr(row, attr)
for name, attr in part.table_key_to_obj_attr["self"].items()
}
def make(self, key):
"""Deprecated in favor of insert_from_nwbfile."""
raise NotImplementedError(
"OptogeneticProtocol.make is deprecated. Use insert_from_nwbfile."
)
def get_stimulus_on_intervals(self, key):
self.ensure_single_entry(key)
nwb = (self & key).fetch_nwb()[0]
stimulus = nwb["stimulus"]
stim_time = stimulus.get_timestamps()
# restrict data to the epoch
epoch_interval = (IntervalList & (TaskEpoch & key)).fetch_interval()
epoch_ind = epoch_interval.contains(stim_time, as_indices=True)
stim_time = stim_time[epoch_ind]
stim_data = stimulus.data[epoch_ind]
# make intervals between when the stimulus turns on and off
t_on = stim_time[stim_data == 1]
t_off = stim_time[stim_data == 0]
# if the first t_on is after the first t_off, remove the first t_off
if t_off[0] < t_on[0]:
t_off = t_off[1:]
# if the last t_on is after the last t_off, add an end time
if t_on[-1] > t_off[-1]:
t_off = np.append(t_off, stim_time[-1])
stim_on_interval = np.array([t_on, t_off]).T
return stim_on_interval
class RippleTrigger(SpyglassIngestion, dj.Part):
definition = """
# Parameters for detecting LFP ripples to trigger optogenetic stimulation
-> master
---
threshold_sd: float # standard deviation threshold for ripple detection
n_above_threshold: int # number of samples above threshold for ripple detection
ripple_lockout_period: int # minimum number of samples between ripple-triggered stimulations
"""
table_key_to_obj_attr = {
"self": {
"threshold_sd": "ripple_filter_threshold_sd",
"n_above_threshold": "ripple_filter_num_above_threshold",
"ripple_lockout_period": (
"ripple_filter_lockout_period_in_samples"
),
}
}
class ThetaTrigger(SpyglassIngestion, dj.Part):
definition = """
# Parameters for detecting LFP theta-phase to trigger optogenetic stimulation
-> master
---
filter_phase: float # target phase of the trigger
reference_ntrode: int # reference ntrode for the trigger
theta_lockout_period: int # lockout period in sample steps
"""
table_key_to_obj_attr = {
"self": {
"filter_phase": "theta_filter_phase_in_deg",
"reference_ntrode": "theta_filter_reference_ntrode",
"theta_lockout_period": (
"theta_filter_lockout_period_in_samples"
),
}
}
class SpeedConditional(SpyglassIngestion, dj.Part):
definition = """
# Speed-related condition gating optogenetic stimulation
-> master
---
speed_threshold: float # speed threshold for optogenetic stimulation (cm/s)
active_above_threshold: bool # whether the stimulation is active above or below the threshold
"""
table_key_to_obj_attr = {
"self": {
"speed_threshold": "speed_filter_threshold_in_cm_per_s",
"active_above_threshold": "speed_filter_on_above_threshold",
}
}
class SpatialConditional(SpyglassIngestion, dj.Part):
definition = """
# Spatial region where animal must be for optogenetic stimulation to be applied
-> master
---
nodes: mediumblob # list of nodes defining polygonal area for optogenetic stimulation
"""
def get_stimulus_timeseries(self):
"""Get the stimulus timeseries for the optogenetic protocol."""
self.ensure_single_entry()
return self.fetch_nwb()[0]["stimulus"]