Skip to content

Phy

pynapple.io.phy

⚠ DEPRECATED: This will be removed in version 1.0.0. Check nwbmatic or neuroconv instead.

Class and functions for loading data processed with Phy2

@author: Sara Mahallati, Guillaume Viejo

Phy

Bases: BaseLoader

Loader for Phy data

Source code in pynapple/io/phy.py
class Phy(BaseLoader):
    """
    Loader for Phy data
    """

    def __init__(self, path):
        """
        Instantiate the data class from a Phy folder.

        Parameters
        ----------
        path : str or Path object
            The path to the data.
        """
        path = Path(path)

        self.basename = path.name
        self.time_support = None

        super().__init__(path)

        self.load_nwb_spikes()

    def load_nwb_spikes(self):
        """Read the NWB spikes to extract the spike times.

        Returns
        -------
        TYPE
            Description
        """

        io = NWBHDF5IO(self.nwbfilepath, "r")
        nwbfile = io.read()

        if nwbfile.units is None:
            io.close()
            return None
        else:
            units = nwbfile.units.to_dataframe()
            spikes = {
                n: nap.Ts(t=units.loc[n, "spike_times"], time_units="s")
                for n in units.index
            }

            self.spikes = nap.TsGroup(
                spikes,
                time_support=self.time_support,
                time_units="s",
                group=units["group"],
            )

            if ~np.all(units["location"] == ""):
                self.spikes.set_info(location=units["location"])

            io.close()
            return True

    def load_lfp(
        self,
        filename=None,
        channel=None,
        extension=".eeg",
        frequency=1250.0,
        precision="int16",
        bytes_size=2,
    ):
        """
        Load the LFP.

        Parameters
        ----------
        filename : str, optional
            The filename of the lfp file.
            It can be useful it multiple dat files are present in the data directory
        channel : int or list of int, optional
            The channel(s) to load. If None return a memory map of the dat file to avoid memory error
        extension : str, optional
            The file extenstion (.eeg, .dat, .lfp). Make sure the frequency match
        frequency : float, optional
            Default 1250 Hz for the eeg file
        precision : str, optional
            The precision of the binary file
        bytes_size : int, optional
            Bytes size of the lfp file

        Raises
        ------
        RuntimeError
            If can't find the lfp/eeg/dat file

        Returns
        -------
        Tsd or TsdFrame
            The lfp in a time series format
        """
        if filename is not None:
            filepath = self.path / filename
        else:
            try:
                filepath = list(self.path.glob(f"*{extension}"))[0]
            except IndexError:
                raise RuntimeError(f"Path {self.path} contains no {extension} files;")

        # is it possible that this is a leftover from neurosuite data?
        # This is not implemented for this class.
        self.load_neurosuite_xml(self.path)

        n_channels = int(self.nChannels)

        f = open(filepath, "rb")
        startoffile = f.seek(0, 0)
        endoffile = f.seek(0, 2)
        bytes_size = 2
        n_samples = int((endoffile - startoffile) / n_channels / bytes_size)
        duration = n_samples / frequency
        f.close()
        fp = np.memmap(filepath, np.int16, "r", shape=(n_samples, n_channels))
        timestep = np.arange(0, n_samples) / frequency

        time_support = nap.IntervalSet(start=0, end=duration, time_units="s")

        if channel is None:
            return nap.TsdFrame(
                t=timestep, d=fp, time_units="s", time_support=time_support
            )
        elif type(channel) is int:
            return nap.Tsd(
                t=timestep, d=fp[:, channel], time_units="s", time_support=time_support
            )
        elif type(channel) is list:
            return nap.TsdFrame(
                t=timestep,
                d=fp[:, channel],
                time_units="s",
                time_support=time_support,
                columns=channel,
            )

load_data

load_data()

Load NWB data saved with pynapple in the pynapplenwb folder

Parameters:

Name Type Description Default
path str

Path to the session folder

required
Source code in pynapple/io/loader.py
def load_data(self):
    """
    Load NWB data saved with pynapple in the pynapplenwb folder

    Parameters
    ----------
    path : str
        Path to the session folder
    """

    io = NWBHDF5IO(self.nwbfilepath, "r+")
    nwbfile = io.read()

    position = {}
    acq_keys = nwbfile.acquisition.keys()
    if "CompassDirection" in acq_keys:
        compass = nwbfile.acquisition["CompassDirection"]
        for k in compass.spatial_series.keys():
            position[k] = pd.Series(
                index=compass.get_spatial_series(k).timestamps[:],
                data=compass.get_spatial_series(k).data[:],
            )
    if "Position" in acq_keys:
        tracking = nwbfile.acquisition["Position"]
        for k in tracking.spatial_series.keys():
            position[k] = pd.Series(
                index=tracking.get_spatial_series(k).timestamps[:],
                data=tracking.get_spatial_series(k).data[:],
            )
    if len(position):
        position = pd.DataFrame.from_dict(position)

        # retrieveing time support position if in epochs
        if "position_time_support" in nwbfile.intervals.keys():
            epochs = nwbfile.intervals["position_time_support"].to_dataframe()
            time_support = nap.IntervalSet(
                start=epochs["start_time"], end=epochs["stop_time"], time_units="s"
            )

        self.position = nap.TsdFrame(
            position, time_units="s", time_support=time_support
        )

    if nwbfile.epochs is not None:
        epochs = nwbfile.epochs.to_dataframe()
        # NWB is dumb and cannot take a single string for labels
        epochs["label"] = [epochs.loc[i, "tags"][0] for i in epochs.index]
        epochs = epochs.drop(labels="tags", axis=1)
        epochs = epochs.rename(columns={"start_time": "start", "stop_time": "end"})
        self.epochs = self._make_epochs(epochs)

        self.time_support = self._join_epochs(epochs, "s")

    io.close()

    return

save_nwb_intervals

save_nwb_intervals(iset, name, description='')

Add epochs to the NWB file (e.g. ripples epochs) See pynwb.epoch.TimeIntervals

Parameters:

Name Type Description Default
iset IntervalSet

The intervalSet to save

required
name str

The name in the nwb file

required
Source code in pynapple/io/loader.py
def save_nwb_intervals(self, iset, name, description=""):
    """
    Add epochs to the NWB file (e.g. ripples epochs)
    See pynwb.epoch.TimeIntervals

    Parameters
    ----------
    iset : IntervalSet
        The intervalSet to save
    name : str
        The name in the nwb file
    """
    io = NWBHDF5IO(self.nwbfilepath, "r+")
    nwbfile = io.read()

    epochs = iset.as_units("s")
    time_intervals = TimeIntervals(name=name, description=description)
    for i in epochs.index:
        time_intervals.add_interval(
            start_time=epochs.loc[i, "start"],
            stop_time=epochs.loc[i, "end"],
            tags=str(i),
        )

    nwbfile.add_time_intervals(time_intervals)
    io.write(nwbfile)
    io.close()

    return

save_nwb_timeseries

save_nwb_timeseries(tsd, name, description='')

Save timestamps in the NWB file (e.g. ripples time) with the time support. See pynwb.base.TimeSeries

Parameters:

Name Type Description Default
tsd TsdFrame

_

required
name str

_

required
description str

_

''
Source code in pynapple/io/loader.py
def save_nwb_timeseries(self, tsd, name, description=""):
    """
    Save timestamps in the NWB file (e.g. ripples time) with the time support.
    See pynwb.base.TimeSeries


    Parameters
    ----------
    tsd : TsdFrame
        _
    name : str
        _
    description : str, optional
        _
    """
    io = NWBHDF5IO(self.nwbfilepath, "r+")
    nwbfile = io.read()

    ts = TimeSeries(
        name=name,
        unit="s",
        data=tsd.values,
        timestamps=tsd.as_units("s").index.values,
    )

    time_support = TimeIntervals(
        name=name + "_timesupport", description="The time support of the object"
    )

    epochs = tsd.time_support.as_units("s")
    for i in epochs.index:
        time_support.add_interval(
            start_time=epochs.loc[i, "start"],
            stop_time=epochs.loc[i, "end"],
            tags=str(i),
        )
    nwbfile.add_time_intervals(time_support)
    nwbfile.add_acquisition(ts)
    io.write(nwbfile)
    io.close()

    return

load_nwb_intervals

load_nwb_intervals(name)

Load epochs from the NWB file (e.g. 'ripples')

Parameters:

Name Type Description Default
name str

The name in the nwb file

required
Source code in pynapple/io/loader.py
def load_nwb_intervals(self, name):
    """
    Load epochs from the NWB file (e.g. 'ripples')

    Parameters
    ----------
    name : str
        The name in the nwb file
    """
    io = NWBHDF5IO(self.nwbfilepath, "r")
    nwbfile = io.read()

    if name in nwbfile.intervals.keys():
        epochs = nwbfile.intervals[name].to_dataframe()
        isets = nap.IntervalSet(
            start=epochs["start_time"], end=epochs["stop_time"], time_units="s"
        )
        io.close()
        return isets
    else:
        io.close()
    return

load_nwb_timeseries

load_nwb_timeseries(name)

Load timestamps in the NWB file (e.g. ripples time)

Parameters:

Name Type Description Default
name str

_

required

Returns:

Type Description
Tsd

_

Source code in pynapple/io/loader.py
def load_nwb_timeseries(self, name):
    """
    Load timestamps in the NWB file (e.g. ripples time)

    Parameters
    ----------
    name : str
        _

    Returns
    -------
    Tsd
        _
    """
    io = NWBHDF5IO(self.nwbfilepath, "r")
    nwbfile = io.read()

    ts = nwbfile.acquisition[name]

    time_support = self.load_nwb_intervals(name + "_timesupport")

    tsd = nap.Tsd(
        t=ts.timestamps[:], d=ts.data[:], time_units="s", time_support=time_support
    )

    io.close()

    return tsd

__init__

__init__(path)

Instantiate the data class from a Phy folder.

Parameters:

Name Type Description Default
path str or Path object

The path to the data.

required
Source code in pynapple/io/phy.py
def __init__(self, path):
    """
    Instantiate the data class from a Phy folder.

    Parameters
    ----------
    path : str or Path object
        The path to the data.
    """
    path = Path(path)

    self.basename = path.name
    self.time_support = None

    super().__init__(path)

    self.load_nwb_spikes()

load_nwb_spikes

load_nwb_spikes()

Read the NWB spikes to extract the spike times.

Returns:

Type Description
TYPE

Description

Source code in pynapple/io/phy.py
def load_nwb_spikes(self):
    """Read the NWB spikes to extract the spike times.

    Returns
    -------
    TYPE
        Description
    """

    io = NWBHDF5IO(self.nwbfilepath, "r")
    nwbfile = io.read()

    if nwbfile.units is None:
        io.close()
        return None
    else:
        units = nwbfile.units.to_dataframe()
        spikes = {
            n: nap.Ts(t=units.loc[n, "spike_times"], time_units="s")
            for n in units.index
        }

        self.spikes = nap.TsGroup(
            spikes,
            time_support=self.time_support,
            time_units="s",
            group=units["group"],
        )

        if ~np.all(units["location"] == ""):
            self.spikes.set_info(location=units["location"])

        io.close()
        return True

load_lfp

load_lfp(
    filename=None,
    channel=None,
    extension=".eeg",
    frequency=1250.0,
    precision="int16",
    bytes_size=2,
)

Load the LFP.

Parameters:

Name Type Description Default
filename str

The filename of the lfp file. It can be useful it multiple dat files are present in the data directory

None
channel int or list of int

The channel(s) to load. If None return a memory map of the dat file to avoid memory error

None
extension str

The file extenstion (.eeg, .dat, .lfp). Make sure the frequency match

'.eeg'
frequency float

Default 1250 Hz for the eeg file

1250.0
precision str

The precision of the binary file

'int16'
bytes_size int

Bytes size of the lfp file

2

Raises:

Type Description
RuntimeError

If can't find the lfp/eeg/dat file

Returns:

Type Description
Tsd or TsdFrame

The lfp in a time series format

Source code in pynapple/io/phy.py
def load_lfp(
    self,
    filename=None,
    channel=None,
    extension=".eeg",
    frequency=1250.0,
    precision="int16",
    bytes_size=2,
):
    """
    Load the LFP.

    Parameters
    ----------
    filename : str, optional
        The filename of the lfp file.
        It can be useful it multiple dat files are present in the data directory
    channel : int or list of int, optional
        The channel(s) to load. If None return a memory map of the dat file to avoid memory error
    extension : str, optional
        The file extenstion (.eeg, .dat, .lfp). Make sure the frequency match
    frequency : float, optional
        Default 1250 Hz for the eeg file
    precision : str, optional
        The precision of the binary file
    bytes_size : int, optional
        Bytes size of the lfp file

    Raises
    ------
    RuntimeError
        If can't find the lfp/eeg/dat file

    Returns
    -------
    Tsd or TsdFrame
        The lfp in a time series format
    """
    if filename is not None:
        filepath = self.path / filename
    else:
        try:
            filepath = list(self.path.glob(f"*{extension}"))[0]
        except IndexError:
            raise RuntimeError(f"Path {self.path} contains no {extension} files;")

    # is it possible that this is a leftover from neurosuite data?
    # This is not implemented for this class.
    self.load_neurosuite_xml(self.path)

    n_channels = int(self.nChannels)

    f = open(filepath, "rb")
    startoffile = f.seek(0, 0)
    endoffile = f.seek(0, 2)
    bytes_size = 2
    n_samples = int((endoffile - startoffile) / n_channels / bytes_size)
    duration = n_samples / frequency
    f.close()
    fp = np.memmap(filepath, np.int16, "r", shape=(n_samples, n_channels))
    timestep = np.arange(0, n_samples) / frequency

    time_support = nap.IntervalSet(start=0, end=duration, time_units="s")

    if channel is None:
        return nap.TsdFrame(
            t=timestep, d=fp, time_units="s", time_support=time_support
        )
    elif type(channel) is int:
        return nap.Tsd(
            t=timestep, d=fp[:, channel], time_units="s", time_support=time_support
        )
    elif type(channel) is list:
        return nap.TsdFrame(
            t=timestep,
            d=fp[:, channel],
            time_units="s",
            time_support=time_support,
            columns=channel,
        )