Source code for tvb.core.entities.storage.burst_dao
# -*- 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 http://www.thevirtualbrain.org
#
# (c) 2012-2024, 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
#
#
from sqlalchemy import desc, func, or_
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import aliased
from sqlalchemy.orm.exc import NoResultFound
from tvb.core.entities.model.model_burst import BurstConfiguration
from tvb.core.entities.model.model_datatype import DataType
from tvb.core.entities.storage.root_dao import RootDAO, DEFAULT_PAGE_SIZE
[docs]
class BurstDAO(RootDAO):
"""
DAO layer for Burst entities.
"""
[docs]
def get_bursts_for_project(self, project_id, page_start=0, page_size=DEFAULT_PAGE_SIZE, count=False):
"""Get latest 50 BurstConfiguration entities for the current project"""
try:
bursts = self.session.query(BurstConfiguration
).filter_by(fk_project=project_id
).order_by(desc(BurstConfiguration.start_time))
if count:
return bursts.count()
if page_size is not None:
bursts = bursts.offset(max(page_start, 0)).limit(page_size)
bursts = bursts.all()
except SQLAlchemyError as excep:
self.logger.exception(excep)
bursts = None
return bursts
[docs]
def get_max_burst_id(self):
"""
Return the maximum of the currently stored burst IDs to be used as the new burst name.
This is not a thread-safe value, but we use it just for a label.
"""
try:
max_id = self.session.query(func.max(BurstConfiguration.id)).one()
if max_id[0] is None:
return 0
return max_id[0]
except SQLAlchemyError as excep:
self.logger.exception(excep)
return 0
[docs]
def count_bursts_with_name(self, burst_name, project_id):
"""
Return the number of burst already named 'custom_b%' and NOT 'custom_b%_%' in current project.
"""
count = 0
try:
count = self.session.query(BurstConfiguration
).filter_by(fk_project=project_id
).filter(BurstConfiguration.name.like(burst_name + '_branch%')
).filter(BurstConfiguration.name.notlike(burst_name + '_branch%_branch%', escape='/')
).count()
except SQLAlchemyError as excep:
self.logger.exception(excep)
return count
[docs]
def get_burst_by_id(self, burst_id):
"""Get the BurstConfiguration entity with the given id"""
try:
burst = self.session.query(BurstConfiguration).filter_by(id=burst_id).one()
burst.project
return burst
except NoResultFound:
self.logger.debug("No burst found with id = {}".format(burst_id))
except SQLAlchemyError as excep:
self.logger.exception(excep)
return None
[docs]
def get_burst_for_operation_id(self, operation_id, is_group=False):
burst = None
try:
burst = self.get_burst_for_direct_operation_id(operation_id, is_group)
if not burst:
burst_alias = aliased(BurstConfiguration, flat=True)
burst = self.session.query(burst_alias
).join(DataType, DataType.fk_parent_burst == burst_alias.gid
).filter(DataType.fk_from_operation == operation_id).first()
except NoResultFound:
self.logger.debug("No burst found for operation id = %s" % (operation_id,))
except SQLAlchemyError as excep:
self.logger.exception(excep)
return burst
[docs]
def get_burst_for_direct_operation_id(self, operation_id, is_group=False):
burst = None
try:
if is_group:
burst = self.session.query(BurstConfiguration
).filter(BurstConfiguration.fk_operation_group == operation_id).first()
else:
burst = self.session.query(BurstConfiguration
).filter(BurstConfiguration.fk_simulation == operation_id).first()
except NoResultFound:
self.logger.debug("No direct burst found for operation id = %s" % (operation_id,))
except SQLAlchemyError as excep:
self.logger.exception(excep)
return burst