Source code for tvb.core.entities.model.model_datatype

# -*- coding: utf-8 -*-
# TheVirtualBrain-Framework Package. This package holds all Data Management, and 
# Web-UI helpful to run brain-simulations. To use it, you also need to download
# TheVirtualBrain-Scientific Package (for simulators). See content of the
# documentation-folder for more details. See also
# (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 <>.
# When using The Virtual Brain for scientific publications, please cite it as explained here:

Entities for Generic DataTypes, Links and Groups of DataTypes are defined here.

.. moduleauthor:: Lia Domide <>
.. moduleauthor:: Bogdan Neacsa <>
.. moduleauthor:: Yann Gordon <yann@tvb.invalid>
import json
import numpy
import typing
from datetime import datetime
from copy import copy
from sqlalchemy.orm import relationship, backref
from sqlalchemy import Boolean, Integer, String, Float, Column, ForeignKey
from tvb.basic.logger.builder import get_logger
from tvb.basic.neotraits.api import HasTraits
from tvb.core.entities.generic_attributes import GenericAttributes
from tvb.core.entities.model.model_project import Project
from tvb.core.entities.model.model_operation import Operation, OperationGroup
from tvb.core.entities.model.model_burst import BurstConfiguration
from tvb.core.neotraits.db import HasTraitsIndex, Base, from_ndarray

LOG = get_logger(__name__)

FILTER_CATEGORIES = {'DataType.subject': {'display': 'Subject', 'type': 'string',
                                          'operations': ['!=', '==', 'like', 'in', 'not in']},
                     'DataType.state': {'display': 'State', 'type': 'string',
                                        'operations': ['!=', '==', 'in', 'not in']},
                     'DataType.disk_size': {'display': 'Disk Size (KB)', 'type': 'int',
                                            'operations': ['<', '==', '>']},
                     'DataType.user_tag_1': {'display': 'Tag 1', 'type': 'string',
                                             'operations': ['!=', '==', 'like']},
                     'DataType.user_tag_2': {'display': 'Tag 2', 'type': 'string',
                                             'operations': ['!=', '==', 'like']},
                     'DataType.user_tag_3': {'display': 'Tag 3', 'type': 'string',
                                             'operations': ['!=', '==', 'like']},
                     'DataType.user_tag_4': {'display': 'Tag 4', 'type': 'string',
                                             'operations': ['!=', '==', 'like']},
                     'DataType.user_tag_5': {'display': 'Tag 5', 'type': 'string',
                                             'operations': ['!=', '==', 'like']},
                     '': {'display': 'Simulation name', 'type': 'string',
                                                 'operations': ['==', '!=', 'like']},
                     'Operation.user_group': {'display': 'Operation Tag', 'type': 'string',
                                              'operations': ['==', '!=', 'like']},
                     'Operation.start_date': {'display': 'Start date', 'type': 'date',
                                              'operations': ['!=', '<', '>']},
                     'Operation.completion_date': {'display': 'Completion date', 'type': 'date',
                                                   'operations': ['!=', '<', '>']}}

[docs]class DataType(HasTraitsIndex): """ Base class for DB storage of Types. DataTypes, are the common language between Visualizers, Simulator and Analyzers. """ id = Column(Integer, ForeignKey(, primary_key=True) type = Column(String) # Name of class inheriting from current type module = Column(String) subject = Column(String) state = Column(String) # RAW, INTERMEDIATE, FINAL visible = Column(Boolean, default=True) invalid = Column(Boolean, default=False) is_nan = Column(Boolean, default=False) disk_size = Column(Integer, default=0) user_tag_1 = Column(String) # Name used by framework and perpetuated from a DataType to derived entities. user_tag_2 = Column(String) user_tag_3 = Column(String) user_tag_4 = Column(String) user_tag_5 = Column(String) # GID of a burst in which current dataType was generated # Native burst-results are referenced from a workflowSet as well # But we also have results generated afterwards from TreeBurst tab. fk_parent_burst = Column(String(32), ForeignKey(BurstConfiguration.gid, ondelete="SET NULL"), nullable=True) _parent_burst = relationship(BurstConfiguration, foreign_keys=fk_parent_burst, primaryjoin=BurstConfiguration.gid == fk_parent_burst, cascade='none') # it should be a reference to a DataTypeGroup, but we can not create that FK # because this two tables (DATA_TYPES, DATA_TYPES_GROUPS) will reference each # other mutually and SQL-Alchemy complains about that. fk_datatype_group = Column(Integer, ForeignKey('')) fk_from_operation = Column(Integer, ForeignKey('', ondelete="CASCADE")) parent_operation = relationship(Operation, backref=backref("DATA_TYPES", order_by=id, cascade="all,delete")) # Transient info fixed_generic_attributes = False
[docs] def get_extra_info(self): return {}
def __init__(self, gid=None, **kwargs): self.gid = gid self.type = self.__class__.__name__ self.module = self.__class__.__module__ try: self.__initdb__(**kwargs) except Exception as exc: LOG.warning('Could not perform __initdb__: %r', exc) super(DataType, self).__init__() def __initdb__(self, subject='', state=None, operation_id=None, fk_parent_burst=None, disk_size=None, user_tag_1=None, user_tag_2=None, user_tag_3=None, user_tag_4=None, user_tag_5=None, **_): """Set attributes""" self.subject = subject self.state = state self.fk_from_operation = operation_id self.user_tag_1 = user_tag_1 self.user_tag_2 = user_tag_2 self.user_tag_3 = user_tag_3 self.user_tag_4 = user_tag_4 self.user_tag_5 = user_tag_5 self.disk_size = disk_size self.fk_parent_burst = fk_parent_burst
[docs] def fill_from_h5(self, h5_file): LOG.warning("fill_from_h5 for: {}".format(type(self))) self.gid = h5_file.gid.load().hex
[docs] def after_store(self): """ Put here code (as a trigger after storage) to be executed by ABCAdapter after the current DT is stored in DB """ pass
def __repr__(self): msg = "<DataType(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)>" return msg % (str(, self.gid, self.type, self.module, self.subject, self.state, str(self.fk_parent_burst), self.user_tag_1, self.user_tag_2, self.user_tag_3, self.user_tag_4, self.user_tag_5) @property def display_type(self): return self.type.replace("Index", "") @property def display_name(self): """ To be implemented in each sub-class which is about to be displayed in UI, and return the text to appear. """ display_name = self.display_type for tag in [self.user_tag_1, self.user_tag_2, self.user_tag_3, self.user_tag_4, self.user_tag_5]: if tag is not None and len(tag) > 0: display_name += " - " + str(tag) return display_name @property def summary_info(self): # type: () -> typing.Dict[str, str] ret = {} if self.title: ret['Title'] = str(self.title) columns = self._get_table_columns() for attr_name in columns: try: if "id" == attr_name: continue name = attr_name.title().replace("Fk_", "Linked ").replace("_", " ") attr_value = getattr(self, attr_name) ret[name] = str(attr_value) except Exception: pass return ret @property def is_ts(self): return hasattr(self, 'time_series_type') def _get_table_columns(self): columns = self.__table__.columns.keys() if type(self).__bases__[0] is DataType: return columns # Consider the immediate superclass only, as for now we have # - most of *Index classes directly inheriting from DataType # - except the ones with one intermediate: DataTypeMatrix or TimeSeriesIndex base_table_columns = type(self).__bases__[0].__table__.columns.keys() columns.extend(base_table_columns) return columns
[docs] @staticmethod def accepted_filters(): """ Return accepted UI filters for current DataType. """ return copy(FILTER_CATEGORIES)
[docs] def fill_from_has_traits(self, has_traits): # type: (HasTraits) -> None self.gid = has_traits.gid.hex
[docs] def fill_from_generic_attributes(self, attrs): # type: (GenericAttributes) -> None self.subject = attrs.subject self.state = attrs.state self.user_tag_1 = attrs.user_tag_1 self.user_tag_2 = attrs.user_tag_2 self.user_tag_3 = attrs.user_tag_3 self.user_tag_4 = attrs.user_tag_4 self.user_tag_5 = attrs.user_tag_5 self.fk_parent_burst = attrs.parent_burst self.is_nan = attrs.is_nan self.visible = attrs.visible self.create_date = attrs.create_date or
[docs]class DataTypeMatrix(DataType): id = Column(Integer, ForeignKey(, primary_key=True) subtype = Column(String, nullable=True) ndim = Column(Integer, default=0) shape = Column(String, nullable=True) array_data_min = Column(Float) array_data_max = Column(Float) array_data_mean = Column(Float) array_is_finite = Column(Boolean, default=True) array_has_complex = Column(Boolean, default=False) # has_volume_mapping will currently be changed for ConnectivityMeasureIndex subclass has_volume_mapping = Column(Boolean, nullable=False, default=False) # currently has_valid_time_series is False only when importing Correlation Coefficients via BIDS has_valid_time_series = Column(Boolean, nullable=False, default=True)
[docs] def fill_from_has_traits(self, datatype): super(DataTypeMatrix, self).fill_from_has_traits(datatype) self.subtype = datatype.__class__.__name__ if hasattr(datatype, "array_data"): self.array_has_complex = numpy.iscomplex(datatype.array_data).any().item() if not self.array_has_complex: self.array_data_min, self.array_data_max, self.array_data_mean = from_ndarray(datatype.array_data) self.array_is_finite = numpy.isfinite(datatype.array_data).all().item() self.shape = json.dumps(datatype.array_data.shape) self.ndim = len(datatype.array_data.shape)
@property def parsed_shape(self): try: return tuple(json.loads(self.shape)) except: return ()
[docs]class DataTypeGroup(DataType): """ All the DataTypes resulted from an operation group will be part from a DataType group. """ __tablename__ = 'DATA_TYPES_GROUPS' id = Column('id', Integer, ForeignKey('', ondelete="CASCADE"), primary_key=True) count_results = Column(Integer) no_of_ranges = Column(Integer, default=0) # Number of ranged parameters fk_operation_group = Column(Integer, ForeignKey('', ondelete="CASCADE")) parent_operation_group = relationship(OperationGroup, backref=backref("DATA_TYPES_GROUPS", cascade="delete")) def __init__(self, operation_group, **kwargs): super(DataTypeGroup, self).__init__(**kwargs) self.fk_operation_group = self.count_results = 0 if operation_group.range3 is not None: self.no_of_ranges = 3 elif operation_group.range2 is not None: self.no_of_ranges = 2 elif operation_group.range1 is not None: self.no_of_ranges = 1 else: self.no_of_ranges = 0
[docs] @staticmethod def is_data_a_group(data): """ Checks if the provided data, ready for export is a DataTypeGroup or not """ return isinstance(data, DataTypeGroup)
[docs]class MeasurePointsSelection(Base): """ Interest area. A subset of nodes from a Connectivity or Sensors. """ __tablename__ = "MEASURE_POINTS_SELECTIONS" id = Column(Integer, primary_key=True) # Unique name /DataType/Project, to be displayed in selector UI: ui_name = Column(String) # JSON with node indices in current selection (0-based): selected_nodes = Column(String) # A Connectivity of Sensor GID, Referring to the entity that this selection was produced for: fk_datatype_gid = Column(String(32), ForeignKey('HasTraitsIndex.gid', ondelete="CASCADE")) # Current Project the selection was defined in: fk_in_project = Column(Integer, ForeignKey('', ondelete="CASCADE")) def __init__(self, ui_name, selected_nodes, datatype_gid, project_id): self.ui_name = ui_name self.selected_nodes = selected_nodes self.fk_in_project = project_id self.fk_datatype_gid = datatype_gid def __repr__(self): return '<Selection(%s, %s, for %s)>' % (self.ui_name, self.selected_nodes, self.fk_datatype_gid)
[docs]class StoredPSEFilter(Base): """ Interest Area PSE Viewer Filter tool, specific filter configuration user inputs to be stored from multiple elements """ __tablename__ = "PSE_FILTERS" id = Column(Integer, primary_key=True) # Unique name /DataType, to be displayed in selector UI: ui_name = Column(String) # A DataType Group GID, Referring to the Group that this filter was stored for: fk_datatype_gid = Column(String(32), ForeignKey('HasTraitsIndex.gid', ondelete="CASCADE")) threshold_value = Column(Float) applied_on = Column(String) def __init__(self, ui_name, datatype_gid, threshold_value, applied_on): self.ui_name = ui_name self.fk_datatype_gid = datatype_gid self.threshold_value = threshold_value self.applied_on = applied_on def __repr__(self): return '<StoredPSEFilter(%s, for %s, %s, %s)>' % (self.ui_name, self.fk_datatype_gid, self.threshold_value, self.applied_on)