Source code for allensdk.brain_observatory.ecephys.optotagging

import pandas as pd
import pynwb
from pynwb import NWBFile

from allensdk.brain_observatory.nwb import setup_table_for_epochs
from allensdk.core import DataObject, JsonReadableInterface, \
    NwbWritableInterface, NwbReadableInterface


class OptotaggingTable(DataObject, JsonReadableInterface,
                       NwbWritableInterface, NwbReadableInterface):
    """Optotagging table - optotagging stimulation"""
    def __init__(self, table: pd.DataFrame):
        # "name" is a pynwb reserved column name that older versions of the
        # pre-processed optotagging_table may use.
        table = \
            table.rename(columns={"name": "stimulus_name"})
        table.index = table.index.rename('id')
        super().__init__(name='optotaggging_table', value=table)

    @property
    def value(self) -> pd.DataFrame:
        """

        Returns
        -------
        A dataframe with columns:
            - start_time: onset of stimulation
            - condition: optical stimulation pattern
            - level: intensity (in volts output to the LED) of stimulation
            - stop_time: stop time of stimulation
            - stimulus_name: stimulus name
            - duration: duration of stimulation
        """
        return self._value

    @classmethod
    def from_json(cls, dict_repr: dict) -> "OptotaggingTable":
        table = pd.read_csv(dict_repr['optotagging_table_path'])
        table.index.name = 'id'
        return OptotaggingTable(table=table)

    @classmethod
    def from_nwb(cls, nwbfile: NWBFile) -> "OptotaggingTable":
        mod = nwbfile.get_processing_module('optotagging')
        table = mod.get_data_interface('optogenetic_stimulation')\
            .to_dataframe()
        table.drop(columns=['tags', 'timeseries'], inplace=True)
        return OptotaggingTable(table=table)

    def to_nwb(self, nwbfile: NWBFile) -> NWBFile:
        optotagging_table = self.value

        opto_ts = pynwb.base.TimeSeries(
            name="optotagging",
            timestamps=optotagging_table["start_time"].values,
            data=optotagging_table["duration"].values,
            unit="seconds"
        )

        opto_mod = pynwb.ProcessingModule("optotagging",
                                          "optogenetic stimulution data")
        opto_mod.add_data_interface(opto_ts)
        nwbfile.add_processing_module(opto_mod)

        optotagging_table = setup_table_for_epochs(optotagging_table, opto_ts,
                                                   'optical_stimulation')

        if len(optotagging_table) > 0:
            container = \
                pynwb.epoch.TimeIntervals.from_dataframe(
                    optotagging_table, "optogenetic_stimulation")
            opto_mod.add_data_interface(container)

        return nwbfile