Source code for tvb.datatypes.time_series

# -*- coding: utf-8 -*-
#
#
# TheVirtualBrain-Scientific Package. This package holds all simulators, and
# analysers necessary to run brain-simulations. You can use it stand alone or
# in conjunction with TheVirtualBrain-Framework Package. See content of the
# documentation-folder for more details. See also http://www.thevirtualbrain.org
#
# (c) 2012-2023, Baycrest Centre for Geriatric Care ("Baycrest") and others
#
# This program is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software Foundation,
# either version 3 of the License, or (at your option) any later version.
# This program is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
# PARTICULAR PURPOSE.  See the GNU General Public License for more details.
# You should have received a copy of the GNU General Public License along with this
# program.  If not, see <http://www.gnu.org/licenses/>.
#
#
#   CITATION:
# When using The Virtual Brain for scientific publications, please cite it as explained here:
# https://www.thevirtualbrain.org/tvb/zwei/neuroscience-publications
#
#

"""
The TimeSeries datatypes.

.. moduleauthor:: Stuart A. Knock <Stuart@tvb.invalid>

"""

from io import BytesIO
from tvb.datatypes import sensors, surfaces, volumes, region_mapping, connectivity
from tvb.basic.neotraits.api import HasTraits, Attr, NArray, List, Float, narray_summary_info
from tvb.basic.readers import H5Reader
import numpy
from copy import deepcopy


[docs] class TimeSeries(HasTraits): """ Base time-series dataType. """ title = Attr(str) data = NArray( label="Time-series data", doc="""An array of time-series data, with a shape of [tpts, :], where ':' represents 1 or more dimensions""") labels_ordering = List( default=("Time", "State Variable", "Space", "Mode"), label="Dimension Names", doc="""List of strings representing names of each data dimension""") labels_dimensions = Attr( field_type=dict, default={}, label="Specific labels for each dimension for the data stored in this timeseries.", doc=""" A dictionary containing mappings of the form {'dimension_name' : [labels for this dimension] }""") time = NArray( label="Time-series time", required=False, doc="""An array of time values for the time-series, with a shape of [tpts,]. This is 'time' as returned by the simulator's monitors.""") start_time = Float(label="Start Time:") sample_period = Float(label="Sample period", default=1.0) # Specify the measure unit for sample period (e.g sec, msec, usec, ...) sample_period_unit = Attr( field_type=str, label="Sample Period Measure Unit", default="ms") @property def nr_dimensions(self): return self.data.ndim @property def sample_rate(self): """:returns samples per second [Hz] """ if self.sample_period_unit in ("s", "sec"): return 1.0 / self.sample_period elif self.sample_period_unit in ("ms", "msec"): return 1000.0 / self.sample_period elif self.sample_period_unit in ("us", "usec"): return 1000000.0 / self.sample_period else: raise ValueError(f"{self.sample_period_unit} is not a recognized time unit") @property def sample_period_ms(self): """:returns sample_period is ms """ if self.sample_period_unit in ("s", "sec"): return 1000 * self.sample_period elif self.sample_period_unit in ("ms", "msec"): return self.sample_period elif self.sample_period_unit in ("us", "usec"): return self.sample_period / 1000.0 else: raise ValueError(f"{self.sample_period_unit} is not a recognized time unit")
[docs] def summary_info(self): """ Gather scientifically interesting summary information from an instance of this datatype. """ summary = { "Time-series type": self.__class__.__name__, "Time-series name": self.title, "Dimensions": self.labels_ordering, "Time units": self.sample_period_unit, "Sample period": self.sample_period, "Start time": self.start_time, "Length": self.sample_period * self.data.shape[0] } summary.update(narray_summary_info(self.data)) return summary
[docs] def duplicate(self, **kwargs): duplicate = super(TimeSeries, self).duplicate() for attr, value in kwargs.items(): setattr(duplicate, attr, value) duplicate.configure() return duplicate
def _get_index_of_state_variable(self, sv_label): try: sv_index = numpy.where(self.variables_labels == sv_label)[0][0] except KeyError: self.logger.error("There are no state variables defined for this instance. Its shape is: %s", self.data.shape) raise except IndexError: self.logger.error("Cannot access index of state variable label: %s. Existing state variables: %s" % ( sv_label, self.variables_labels)) raise return sv_index
[docs] def get_state_variable(self, sv_label): sv_data = self.data[:, self._get_index_of_state_variable(sv_label), :, :] subspace_labels_dimensions = deepcopy(self.labels_dimensions) subspace_labels_dimensions[self.labels_ordering[1]] = [sv_label] if sv_data.ndim == 3: sv_data = numpy.expand_dims(sv_data, 1) return self.duplicate(data=sv_data, labels_dimensions=subspace_labels_dimensions)
def _get_indices_for_labels(self, list_of_labels): list_of_indices_for_labels = [] for label in list_of_labels: try: space_index = numpy.where(self.space_labels == label)[0][0] except ValueError: self.logger.error("Cannot access index of space label: %s. Existing space labels: %s" % (label, self.space_labels)) raise list_of_indices_for_labels.append(space_index) return list_of_indices_for_labels
[docs] def get_subspace_by_index(self, list_of_index, **kwargs): self._check_space_indices(list_of_index) subspace_data = self.data[:, :, list_of_index, :] subspace_labels_dimensions = deepcopy(self.labels_dimensions) subspace_labels_dimensions[self.labels_ordering[2]] = self.space_labels[list_of_index].tolist() if subspace_data.ndim == 3: subspace_data = numpy.expand_dims(subspace_data, 2) return self.duplicate(data=subspace_data, labels_dimensions=subspace_labels_dimensions, **kwargs)
[docs] def get_subspace_by_labels(self, list_of_labels): list_of_indices_for_labels = self._get_indices_for_labels(list_of_labels) return self.get_subspace_by_index(list_of_indices_for_labels)
def __getattr__(self, attr_name): if self.labels_ordering[1] in self.labels_dimensions.keys(): if attr_name in self.variables_labels: return self.get_state_variable(attr_name) if self.labels_ordering[2] in self.labels_dimensions.keys(): if attr_name in self.space_labels: return self.get_subspace_by_labels([attr_name]) raise AttributeError("%r object has no attribute %r" % (self.__class__.__name__, attr_name)) def _get_index_for_slice_label(self, slice_label, slice_idx): if slice_idx == 1: return self._get_indices_for_labels([slice_label])[0] if slice_idx == 2: return self._get_index_of_state_variable(slice_label) @property def shape(self): return self.data.shape @property def time_unit(self): return self.sample_period_unit @property def space_labels(self): return numpy.array(self.labels_dimensions.get(self.labels_ordering[2], [])) @property def variables_labels(self): return numpy.array(self.labels_dimensions.get(self.labels_ordering[1], [])) def _check_space_indices(self, list_of_index): for index in list_of_index: if index < 0 or index > self.data.shape[1]: self.logger.error("Some of the given indices are out of space range: [0, %s]", self.data.shape[1]) raise IndexError
[docs] @classmethod def from_bytes_stream(cls, bytes_stream, content_type=".npz"): result = TimeSeries() if content_type == '.npz': ts_data = numpy.load(BytesIO(bytes_stream)) result.data = ts_data['data'] result.time = ts_data['time'] return result reader = H5Reader(BytesIO(bytes_stream)) result.data = reader.read_field("data") result.time = reader.read_optional_field("time") return result
[docs] class SensorsTSBase(TimeSeries):
[docs] def summary_info(self): """ Gather scientifically interesting summary information from an instance of this datatype. """ summary = super(SensorsTSBase, self).summary_info() summary.update({"Source Sensors": self.sensors.title}) return summary
[docs] class TimeSeriesEEG(SensorsTSBase): """ A time series associated with a set of EEG sensors. """ sensors = Attr(field_type=sensors.SensorsEEG) labels_ordering = List(of=str, default=("Time", "SV", "EEG Sensor", "Mode"))
[docs] class TimeSeriesMEG(SensorsTSBase): """ A time series associated with a set of MEG sensors. """ sensors = Attr(field_type=sensors.SensorsMEG) labels_ordering = List(of=str, default=("Time", "SV", "MEG Sensor", "Mode"))
[docs] class TimeSeriesSEEG(SensorsTSBase): """ A time series associated with a set of Internal sensors. """ sensors = Attr(field_type=sensors.SensorsInternal) labels_ordering = List(of=str, default=("Time", "SV", "sEEG Sensor", "Mode"))
[docs] class TimeSeriesRegion(TimeSeries): """ A time-series associated with the regions of a connectivity. """ connectivity = Attr(field_type=connectivity.Connectivity) region_mapping_volume = Attr(field_type=region_mapping.RegionVolumeMapping, required=False) region_mapping = Attr(field_type=region_mapping.RegionMapping, required=False) labels_ordering = List(of=str, default=("Time", "State Variable", "Region", "Mode"))
[docs] def summary_info(self): """ Gather scientifically interesting summary information from an instance of this datatype. """ summary = super(TimeSeriesRegion, self).summary_info() summary.update({ "Source Connectivity": self.connectivity.title, "Region Mapping": self.region_mapping.title if self.region_mapping else "None", "Region Mapping Volume": (self.region_mapping_volume.title if self.region_mapping_volume else "None") }) return summary
[docs] class TimeSeriesSurface(TimeSeries): """ A time-series associated with a Surface. """ surface = Attr(field_type=surfaces.CorticalSurface) labels_ordering = List(of=str, default=("Time", "State Variable", "Vertex", "Mode"))
[docs] def summary_info(self): """ Gather scientifically interesting summary information from an instance of this datatype. """ summary = super(TimeSeriesSurface, self).summary_info() summary.update({"Source Surface": self.surface.title}) return summary
[docs] class TimeSeriesVolume(TimeSeries): """ A time-series associated with a Volume. """ volume = Attr(field_type=volumes.Volume) labels_ordering = List(of=str, default=("Time", "X", "Y", "Z"))
[docs] def summary_info(self): """ Gather scientifically interesting summary information from an instance of this datatype. """ summary = super(TimeSeriesVolume, self).summary_info() summary.update({"Source Volume": self.volume.title}) return summary