Source code for allensdk.brain_observatory.nwb.nwb_api

import pandas as pd
import pynwb
import SimpleITK as sitk

from allensdk.brain_observatory.behavior.data_objects.stimuli.presentations \
    import \
    Presentations
from allensdk.brain_observatory.running_speed import RunningSpeed
from allensdk.brain_observatory.behavior.image_api import ImageApi



[docs]
class NwbApi:

    __slots__ = ('path', '_nwbfile')

    @property
    def nwbfile(self):
        if hasattr(self, '_nwbfile'):
            return self._nwbfile

        io = pynwb.NWBHDF5IO(self.path, 'r', load_namespaces=True)
        return io.read()

    def __init__(self, path, **kwargs):
        ''' Reads data for a single Brain Observatory session from an NWB 2.0
        file
        '''

        self.path = path


[docs]
    @classmethod
    def from_nwbfile(cls, nwbfile, **kwargs):

        obj = cls(path=None, **kwargs)
        obj._nwbfile = nwbfile

        return obj



[docs]
    @classmethod
    def from_path(cls, path, **kwargs):
        with open(path, 'r'):
            pass

        return cls(path=path, **kwargs)



[docs]
    def get_running_speed(self, lowpass=True) -> RunningSpeed:
        """
        Gets the running speed
        Parameters
        ----------
        lowpass: bool
            Whether to return the running speed with lowpass filter applied
            or without

        Returns
        -------
        RunningSpeed:
            The running speed
        """

        interface_name = 'speed' if lowpass else 'speed_unfiltered'
        values = self.nwbfile.processing['running'].get_data_interface(
            interface_name).data[:]
        timestamps = self.nwbfile.processing['running'].get_data_interface(
            interface_name).timestamps[:]

        return RunningSpeed(
            timestamps=timestamps,
            values=values,
        )



[docs]
    def get_stimulus_presentations(self) -> pd.DataFrame:
        presentations = Presentations.from_nwb(nwbfile=self.nwbfile,
                                               add_is_change=False)
        return presentations.value



[docs]
    def get_invalid_times(self) -> pd.DataFrame:

        container = self.nwbfile.invalid_times
        if container:
            return container.to_dataframe()
        else:
            return pd.DataFrame()



[docs]
    def get_image(self, name, module, image_api=None) -> sitk.Image:

        if image_api is None:
            image_api = ImageApi

        nwb_img = self.nwbfile.processing[module].get_data_interface(
            'images')[name]
        data = nwb_img.data
        resolution = nwb_img.resolution  # px/cm
        spacing = [resolution * 10, resolution * 10]

        return ImageApi.serialize(data, spacing, 'mm')
Source code for allensdk.brain_observatory.nwb.nwb_api

Contents

Questions
