Data#

class torch_brain.data.Data(*, domain=None, **kwargs)[source]#

Bases: object

A flexible container for other data objects such as ArrayDict, RegularTimeSeries, IrregularTimeSeries, and Interval objects, as well as nested Data objects and regular Python objects like scalars, strings, and numpy arrays.

Parameters:

**kwargs – Arbitrary attributes to attach to the data object (e.g. spikes, lfp, units, trials, metadata).
domain (Union[Interval, Literal['auto'], None]) – An Interval specifying time domain of the data object. If "auto", the domain is computed as the union of the domains of any time-based attributes. Defaults to None.

Example

>>> import numpy as np
>>> from torch_brain.data import (
...     ArrayDict,
...     IrregularTimeSeries,
...     RegularTimeSeries,
...     Interval,
...     Data,
... )

>>> data = Data(
...     session_id="session_0",
...     spikes=IrregularTimeSeries(
...         timestamps=[0.1, 0.2, 0.3, 2.1, 2.2, 2.3],
...         unit_index=[0, 0, 1, 0, 1, 2],
...         waveforms=np.zeros((6, 48)),
...         domain=Interval(0., 3.),
...     ),
...     lfp=RegularTimeSeries(
...         raw=np.zeros((1000, 3)),
...         sampling_rate=250.,
...     ),
...     units=ArrayDict(
...         id=["unit_0", "unit_1", "unit_2"],
...         brain_region=["M1", "M1", "PMd"],
...     ),
...     trials=Interval(
...         start=[0, 1, 2],
...         end=[1, 2, 3],
...         go_cue_time=[0.5, 1.5, 2.5],
...         drifting_gratings_dir=[0, 45, 90],
...     ),
...     drifting_gratings_imgs=np.zeros((8, 3, 32, 32)),
...     domain=Interval(0., 4.),
... )

>>> data
Data(
session_id='session_0',
spikes=IrregularTimeSeries(
  timestamps=[6],
  unit_index=[6],
  waveforms=[6, 48]
),
lfp=RegularTimeSeries(
  raw=[1000, 3]
),
units=ArrayDict(
  id=[3],
  brain_region=[3]
),
trials=Interval(
  start=[3],
  end=[3],
  go_cue_time=[3],
  drifting_gratings_dir=[3]
),
drifting_gratings_imgs=[8, 3, 32, 32],
)

>>> data.slice(1, 3)
Data(
session_id='session_0',
spikes=IrregularTimeSeries(
  timestamps=[3],
  unit_index=[3],
  waveforms=[3, 48]
),
lfp=RegularTimeSeries(
  raw=[500, 3]
),
units=ArrayDict(
  id=[3],
  brain_region=[3]
),
trials=Interval(
  start=[2],
  end=[2],
  go_cue_time=[2],
  drifting_gratings_dir=[2]
),
drifting_gratings_imgs=[8, 3, 32, 32],
_absolute_start=1.0,
)

property domain#: Returns the domain of the data object.

property start#: Returns the start time of the data object.

property end#: Returns the end time of the data object.

property absolute_start#

Returns the start time of this slice relative to the original start time. Should be 0. if the data object has not been sliced.

Example

>>> from torch_brain.data import Data
>>> data = Data(domain=Interval(0., 4.))

>>> data.absolute_start
0.0

>>> data = data.slice(1, 3)
>>> data.absolute_start
1.0

>>> data = data.slice(0.4, 1.4)
>>> data.absolute_start
1.4

slice(start, end, reset_origin=True)[source]#

Returns a new Data object that contains the data between the start and end times. This method will slice all time-based attributes that are present in the data object.

Parameters:

start (float) – Start time.
end (float) – End time.
reset_origin (bool) – If True, all time attributes will be updated to be relative to the new start time. Defaults to True.

select_by_interval(interval)[source]#

Return a new IrregularTimeSeries object where all timestamps are within the interval.

Parameters:: interval (Interval) – Interval object.

to_dict()[source]#

Returns a dictionary of stored key/value pairs.

Return type:: dict[str, Any]

to_hdf5(file, serialize_fn_map=None)[source]#

Saves the data object to an HDF5 file. This method will also call the to_hdf5 method of all contained data objects, so that the entire data object is saved to the HDF5 file, i.e. no need to call to_hdf5 for each contained data object.

Parameters:

file (File | Group) – An open h5py.File (or h5py.Group) to write into.
serialize_fn_map (Optional[dict[type, Callable]]) – Optional dictionary mapping a Python type to a callable that serializes values of that type into an HDF5-compatible object. Use this for attributes whose types are not natively supported by HDF5. If None, the map returned by get_default_serialize_fn_map() is used. Defaults to None.

Example ::

>>> import h5py
>>> from torch_brain.data import Data
>>> data = Data(...)  
>>> with h5py.File("data.h5", "w") as f:  
...     data.to_hdf5(f)

classmethod from_hdf5(file, lazy=True)[source]#

Loads the data object from an HDF5 file. This method will also call the from_hdf5 method of all contained data objects, so that the entire data object is loaded from the HDF5 file, i.e. no need to call from_hdf5 for each contained data object.

Parameters:: file (File | Group) – HDF5 file.

Example ::

>>> import h5py  
>>> from torch_brain.data import Data  
>>> with h5py.File("data.h5", "r") as f:  
...     data = Data.from_hdf5(f)

property file: File | None#: The underlying HDF5 file handle, or None if no file is open. Only set when the object was created via load() or from_hdf5() with lazy=True.

classmethod load(path, lazy=True)[source]#

Loads the Data object from an HDF5 file given its file path.

When lazy=True (default), the underlying HDF5 file remains open and data is loaded on demand. The caller is responsible for closing the file handle when done, either by calling close() or by using the context manager protocol.

When lazy=False, all data is read into memory immediately and the file is closed before returning.

Parameters:

path (Path | str) – The file path to the HDF5 file containing the Data object.
lazy (bool) – If True (default), load contained objects in lazy mode (using LazyArrayDict, LazyRegularTimeSeries, etc.); if False, read all data immediately into memory.

Returns:

The loaded Data object from the HDF5 file.

Return type:

Data

Example ::

>>> from torch_brain.data import Data
>>> # lazy with context manager (recommended)
>>> with Data.load("data.h5") as data:  
...     ...

>>> # lazy with explicit close
>>> data = Data.load("data.h5")  
>>> ...  
>>> data.close()  

>>> # non-lazy (no close needed)
>>> data = Data.load("data.h5", lazy=False)  

close(strict=False)[source]#

Close the file-handle that was opened for lazy-loading. Any lazy attributes that have not been materialized will become invalid.

Parameters:: strict (bool) – If True, raise an error when no open file handle is present. Default False.

save(path)[source]#

Saves the data object to an HDF5 file at the given path.

Parameters:: path (Path | str) – Destination file path

Example ::

>>> from torch_brain.data import Data  
>>> data = Data(...)  
>>> data.save("data.h5")  

set_train_domain(interval)[source]#: Deprecated no-op retained for backward compatibility.

set_valid_domain(interval)[source]#: Deprecated no-op retained for backward compatibility.

set_test_domain(interval)[source]#: Deprecated no-op retained for backward compatibility.

keys()[source]#

Returns a list of all attribute names.

Return type:: list[str]

__contains__(key)[source]#

Returns True if the attribute key is present in the data.

Return type:: bool

get_nested_attribute(path)[source]#

Return the value of a nested attribute specified by its dot-separated path.

Parameters:: path (str) – Nested attribute path (dot-separated).
Return type:: Any
Returns:: The value of the attribute at the end of the path.
Raises:: AttributeError – If any component of the path cannot be resolved.

Example ::

>>> import numpy as np
>>> from torch_brain.data import Data, IrregularTimeSeries, Interval
>>> data = Data(
...     spikes=IrregularTimeSeries(
...         timestamps=[0.1, 0.2, 0.3],
...         unit_index=[0, 0, 1],
...         waveforms=np.zeros((3, 2)),
...         domain=Interval(0., 1.),
...     ),
...     domain=Interval(0., 1.),
... )

>>> for attr in ["timestamps", "unit_index", "waveforms"]:
...     print(attr, data.get_nested_attribute(f"spikes.{attr}"))
timestamps [0.1 0.2 0.3]
unit_index [0 0 1]
waveforms [[0. 0.]
 [0. 0.]
 [0. 0.]]

set_nested_attribute(path, value)[source]#

Set a nested attribute specified by its dot-separated path, modifying the object in-place.

Example

>>> data.set_nested_attribute("session.subject.age", 5)  
>>> data.set_nested_attribute("spikes.unit_index", remapped_units)  

Parameters:

path (str) – Nested attribute path (dot-separated).
value (Any) – The value to set for the attribute.

Returns:

self with the updated nested attribute.

Return type:

Data

Raises:

AttributeError – If any component of the path cannot be resolved.

has_nested_attribute(path)[source]#

Return True if the dot-separated path resolves to an existing attribute.

Example ::

>>> if data.has_nested_attribute("spikes.waveforms"):  
...     process(data.spikes.waveforms)

Note

Uses __dict__ rather than getattr at each level of the path traversal to avoid two side-effects:

getattr on a lazy object (e.g. LazyIrregularTimeSeries) would materialise the underlying h5py dataset into a NumPy array as a side-effect of the existence check.
getattr would invoke computed properties such as RegularTimeSeries.timestamps, incorrectly reporting them as stored attributes when they are not.

Parameters:: path (str) – Nested attribute path (dot-separated).
Returns:: True if the attribute exists, False otherwise.
Return type:: bool

delete_nested_attribute(path)[source]#

Delete a nested attribute specified by its dot-separated path, modifying the object in-place.

Example ::

>>> for path in ["spikes.waveforms", "lfp.raw"]:  
...     if data.has_nested_attribute(path):
...         data.delete_nested_attribute(path)

Parameters:: path (str) – Nested attribute path (dot-separated).
Raises:: AttributeError – If any component of the path cannot be resolved, or if the target attribute is protected.

materialize()[source]#

Materializes the data object, i.e., loads into memory all of the data that is still referenced in the HDF5 file.

Return type:: Data