TheVirtualBrain (TVB)

https://docs.thevirtualbrain.org/
Admin
TheVirtualBrain (TVB) is a scientific simulation platform for modeling and simulating the dynamics...
Tokens:949,114
Snippets:3,694
Trust Score:9.1
Update:1 year ago
Show doc for...
Tokens:
Raw
### Start TheVirtualBrain Console Interface

Source: https://docs.thevirtualbrain.org/manuals/UserGuide/UserGuide-Installation

Starts TheVirtualBrain distribution in either COMMAND_PROFILE or LIBRARY_PROFILE mode, launching a Python shell. This is done by executing the distribution script with the 'start' command followed by the desired profile.

```shell
./distribution.sh start COMMAND_PROFILE
```

```shell
./distribution.sh start LIBRARY_PROFILE
```

--------------------------------

### Start TVB Console Interface

Source: https://docs.thevirtualbrain.org/_sources/manuals/UserGuide/UserGuide-Installation

Starts the TVB console interface by initiating a command or library profile. This action will open a Python shell.

```bash
$ ./distribution.sh start COMMAND_PROFILE
```

--------------------------------

### Launch TVB GUI (Linux/Windows)

Source: https://docs.thevirtualbrain.org/_sources/manuals/UserGuide/UserGuide-Installation

This script starts the TVB application, typically launching its GUI in a web browser. It requires navigating to the TVB_Distribution/bin directory. Ensure the script has execute permissions.

```bash
cd TVB_Distribution/bin
./tvb_start.sh
```

--------------------------------

### Install Latest TVB Release from PyPI

Source: https://docs.thevirtualbrain.org/_sources/manuals/ContributorsManual/ContributorsManual

This bash script installs the latest released version of TVB from PyPI into an active Anaconda environment. It first installs the core library, then the framework, and finally demonstrates how to launch the TVB web server locally. This is an alternative to building from source.

```bash
$ source activate $envname
$ pip install tvb-library               # In case you do not need the web interface. not storage, you can stop here
$ pip install tvb-framework
$ python -m tvb.interfaces.web.run WEB_PROFILE  # Launch TVB web server locally
```

--------------------------------

### Control TheVirtualBrain Distribution using a Script

Source: https://docs.thevirtualbrain.org/manuals/UserGuide/UserGuide-Installation

The distribution script is used from a terminal to manage TheVirtualBrain. Running it with the -h flag provides help information for its commands. It can be used to start different profiles.

```shell
./distribution.sh -h
```

--------------------------------

### Driver Setup and Execution Entry Point - Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/rateML/run/model_driver_zerlaut

This is the main execution block of the script. It initializes the Driver_Setup and Driver_Execute classes and then calls the run_all method to start the simulation process, storing the results in tavgGPU.

```python
if __name__ == '__main__':

	driver_setup = Driver_Setup()
	tavgGPU = Driver_Execute(driver_setup).run_all()
```

--------------------------------

### Setup PSE in Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/web/controllers/simulator/simulator_controller

Initializes the PSE (Problem Solving Environment) setup within the simulator wizard. This function retrieves common simulation parameters, burst configuration, and all range parameters needed for the setup process.

```python
    @expose_fragment('simulator_fragment')
    def setup_pse(self, **data):
        session_stored_simulator, is_simulation_copy, is_simulator_load, _ = self.context.get_common_params()
        burst_config = self.context.burst_config
        all_range_parameters = self.range_parameters.get_all_range_parameters()
```

--------------------------------

### Initialize and Start CherryPy Server

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/web/run

The `start_tvb` function initializes the CherryPy server for The Virtual Brain project. It handles optional database resets, ensures the storage directory exists, sets up displayer roots, initializes CherryPy with necessary configurations and tools, starts storage encryption if enabled, and finally starts the CherryPy engine. It can also optionally launch a browser window.

```python
[docs]
def start_tvb(arguments, browser=True):
    """
    Fire CherryPy server and listen on a free port
    """

    if PARAM_RESET_DB in arguments:
        # When specified, clean everything in DB
        reset()
        arguments.remove(PARAM_RESET_DB)

    if not os.path.exists(TvbProfile.current.TVB_STORAGE):
        try:
            os.makedirs(TvbProfile.current.TVB_STORAGE)
        except Exception:
            sys.exit("You do not have enough rights to use TVB storage folder:" + str(TvbProfile.current.TVB_STORAGE))

    try:
        initialize(arguments)
    except InvalidSettingsException as excep:
        LOGGER.exception(excep)
        sys.exit()

    # Mark that the interface is Web
    ABCDisplayer.VISUALIZERS_ROOT = TvbProfile.current.web.VISUALIZERS_ROOT

    init_cherrypy(arguments)
    if StorageInterface.encryption_enabled() and StorageInterface.app_encryption_handler():
        storage_interface = StorageInterface()
        storage_interface.start()
        storage_interface.startup_cleanup()

    # Fire a browser page at the end.

```

--------------------------------

### Start and Execute an Operation (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/services/backend_clients/standalone_client

Provides methods to start and execute operations. `start_operation` initiates an operation by updating its status, creating an `OperationExecutor` thread, and running or starting the thread based on the adapter's launch mode. `execute` handles the initial queuing logic, marking an operation as 'queue_full' if cloud deployment is active or the lock queue is empty, otherwise initiating the operation start.

```python
from logging import LOGGER
from .dao import dao
from .models import Operation
from .profile import TvbProfile
from queue import Queue
from .exceptions import TVBException
from .adapter import ABCAdapter, AdapterLaunchModeEnum
from .services import BurstService
from .constants import STATUS_ERROR, STATUS_CANCELED

# Placeholder for global variables and other classes used
# LOCKS_QUEUE = Queue()
# CURRENT_ACTIVE_THREADS = []
# class OperationExecutor: # Dummy class for type hinting
#     def __init__(self, operation_id):
#         self.operation_id = operation_id
#     def run(self):
#         pass
#     def start(self):
#         pass
# class KubeNotifier: # Dummy class
#     @staticmethod
#     def do_rest_call_to_pod(ip, method, op_id):
#         pass

class StandAloneClient:
    @staticmethod
    def start_operation(operation_id):
        LOGGER.info("Start processing operation id:{}".format(operation_id))

        operation = dao.get_operation_by_id(operation_id)
        operation.queue_full = False
        dao.store_entity(operation)

        thread = OperationExecutor(operation_id)
        CURRENT_ACTIVE_THREADS.append(thread)

        adapter_instance = ABCAdapter.build_adapter(operation.algorithm)
        if adapter_instance.launch_mode is AdapterLaunchModeEnum.SYNC_DIFF_MEM:
            thread.run()
            operation = dao.get_operation_by_id(operation_id)
            if operation.additional_info and operation.status == STATUS_ERROR:
                raise TVBException(operation.additional_info)
        else:
            thread.start()

    @staticmethod
    def execute(operation_id, user_name_label, adapter_instance):
        """Start asynchronous operation locally"""
        if TvbProfile.current.web.IS_CLOUD_DEPLOY or LOCKS_QUEUE.qsize() == 0:
            operation = dao.get_operation_by_id(operation_id)
            operation.queue_full = True
            dao.store_entity(operation)
            return

        LOCKS_QUEUE.get(True)
        StandAloneClient.start_operation(operation_id)

```

--------------------------------

### Set up Anaconda Environment for TVB

Source: https://docs.thevirtualbrain.org/_sources/manuals/ContributorsManual/ContributorsManual

This bash script sets up a virtual environment using Anaconda, installs core Python packages including NumPy, and then installs TVB dependencies from a requirements file. Finally, it builds and installs the full TVB framework. This is a prerequisite for developing with TVB.

```bash
$ envname="tvb-run"
$ conda create -y --name $envname python=3.10 numpy
$ source activate $envname
$ pip install -r [tvb-root]/tvb_framework/requirements.txt
$ pip install --no-build-isolation tvb-gdist
$ cd [tvb-root]/tvb_build/
$ bash install_full_tvb.sh
```

--------------------------------

### Install TVB Data from Zenodo

Source: https://docs.thevirtualbrain.org/_sources/manuals/ContributorsManual/ContributorsManual

This sequence of commands downloads the TVB data package from Zenodo, unpacks it into a 'tvb_data' directory, and then installs it in development mode. This ensures that the necessary data files are available for TVB to function correctly.

```bash
$ wget https://zenodo.org/record/10128131/files/tvb_data.zip?download=1 -O tvb_data.zip
$ mkdir tvb_data
$ unzip tvb_data.zip -d tvb_data
$ rm tvb_data.zip
$ cd tvb_data
$ python setup.py develop
```

--------------------------------

### Initialize and Start the WSGI HTTP Server

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/rest/server/run

This code initializes and starts the WSGI HTTP server. It checks for encryption enablement and then proceeds to start the storage interface, register the Keycloak authorization manager, and finally launches the server to listen on all interfaces on a configured port. This requires `WSGIServer`, `TvbProfile`, `StorageInterface`, and `AuthorizationManager` to be accessible.

```python
if StorageInterface.encryption_enabled() and StorageInterface.app_encryption_handler():
        storage_interface = StorageInterface()
        storage_interface.start()
        storage_interface.startup_cleanup()

    # Register keycloak authorization manager
    AuthorizationManager(TvbProfile.current.KEYCLOAK_CONFIG)

    http_server = WSGIServer(("0.0.0.0", TvbProfile.current.web.REST_PORT), app)
    http_server.serve_forever()
```

--------------------------------

### Get Operation Status

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/rest/client/tvb_client

Retrieves the status of an operation using its GID. It returns predefined status codes such as FINISHED, PENDING, STARTED, CANCELED, or ERROR.

```python
def get_operation_status(self, operation_gid):
        """
        Given an operation gid, this function returns the status of that operation in TVB. The status of operations can be:
        STATUS_FINISHED = "5-FINISHED"
        STATUS_PENDING = "4-PENDING"
        STATUS_STARTED = "3-STARTED"
        STATUS_CANCELED = "2-CANCELED"
        STATUS_ERROR = "1-ERROR"
        """
        return self.operation_api.get_operation_status(operation_gid)
```

--------------------------------

### Initialize Simulation Main Page - Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/web/controllers/simulator/simulator_controller

Renders the main page for the simulation cockpit, handling initial setup and configuration loading. It determines simulation states (copy, load, branch) and prepares rendering rules for the first fragment. Dependencies include context management and service layers for simulation data.

```python
@context_selected
def index(self):
        """Get on burst main page"""
        template_specification = dict(mainContent="burst/main_burst", title="Simulation Cockpit",
                                      includedResources='project/included_resources')

        if not self.context.last_loaded_fragment_url:
            self.context.add_last_loaded_form_url_to_session(SimulatorWizzardURLs.SET_CONNECTIVITY_URL)

        self.context.set_burst_config()

        _, is_simulation_copy, is_simulation_load, is_branch = self.context.get_common_params()
        if self.context.burst_config.start_time is not None:
            is_simulation_load = True
            self.context.add_simulator_load_to_session(True)

        template_specification['burstConfig'] = self.context.burst_config
        template_specification['burst_list'] = self.burst_service.get_available_bursts(self.context.project.id)

        form = self.prepare_first_fragment()
        rendering_rules = SimulatorFragmentRenderingRules(
            form, SimulatorWizzardURLs.SET_CONNECTIVITY_URL, None, is_simulation_copy, is_simulation_load,
            last_form_url=self.context.last_loaded_fragment_url,
            last_request_type=cherrypy.request.method, is_first_fragment=True, is_branch=is_branch)
        template_specification.update(**rendering_rules.to_dict())

        cherrypy.response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
        cherrypy.response.headers['Pragma'] = 'no-cache'
        cherrypy.response.headers['Expires'] = '0'

        return self.fill_default_attributes(template_specification)

```

--------------------------------

### Get Operation Numbers (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/services/algorithm_service

Counts the total number of operations that have been started for a given project. It uses the DAO to perform the count.

```python
    @staticmethod
    def get_operation_numbers(proj_id):
        """ Count total number of operations started for current project. """
        return dao.get_operation_numbers(proj_id)
```

--------------------------------

### Get Tract by Index

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/datatypes/h5/tracts_h5

Retrieves a specific tract from the HDF5 file using its index. It calculates the start and end indices from `tract_start_idx` and then fetches the corresponding vertices from the `vertices` DataSet.

```python
def get_tract(self, i):
    """
    get a tract by index
    """
    start, end = self.tract_start_idx[i:i + 2]
    return self.vertices[start:end]
```

--------------------------------

### Get Host IP Address (Python)

Source: https://docs.thevirtualbrain.org/api/tvb.core.services

A utility function to retrieve the current host's IP address. This is useful for network configurations and identifying the local machine in distributed or standalone setups.

```Python
def get_host_current_host_ip() -> str:
    # Returns: current host ip address
    pass
```

--------------------------------

### Launch TVB Jupyter Notebook

Source: https://docs.thevirtualbrain.org/_sources/manuals/UserGuide/UserGuide-Installation

Launches the TVB Jupyter Notebook environment. Ensure you are in the TVB_Distribution/bin folder before executing.

```bash
$ cd TVB_Distribution/bin
$ ./jupyter_notebook.sh
```

--------------------------------

### Launch Jupyter Notebook Interface for TheVirtualBrain

Source: https://docs.thevirtualbrain.org/manuals/UserGuide/UserGuide-Installation

This script launches the user-friendly Jupyter notebook interface for TheVirtualBrain. It requires navigating to the TVB_Distribution/bin folder and executing the jupyter_notebook.sh script. This is a LIBRARY_PROFILE interface.

```shell
cd TVB_Distribution/bin
./jupyter_notebook.sh
```

--------------------------------

### Configure Simulator Adapter

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/simulator/simulator_adapter

Prepares the simulator adapter for launch by configuring the algorithm and handling potential configuration errors. It loads history if a simulation state GID is provided. Dependencies include the SimulatorAdapterModel and LaunchException.

```python
def configure(self, view_model):
    # type: (SimulatorAdapterModel) -> None
    """
    Make preparations for the adapter launch.
    """
    self.log.debug("%s: Configuring simulator adapter..." % str(self))
    self.algorithm = self.view_model_to_has_traits(view_model)
    self.branch_simulation_state_gid = view_model.history_gid

    try:
        self.algorithm.preconfigure()
    except ValueError as err:
        raise LaunchException("Failed to configure simulator due to invalid Input Values. It could be because "
                              "of an incompatibility between different version of TVB code.", err)
```

--------------------------------

### Launch Simulation

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/simulator/simulator_adapter

Initiates a simulation run. It configures the algorithm, loads historical data if available, and attempts to load region mappings. It then iterates through monitors to prepare for the simulation execution. Dependencies include SimulatorAdapterModel, SimulationHistory, TimeSeriesIndex, and SimulationHistoryIndex.

```python
def launch(self, view_model):
    # type: (SimulatorAdapterModel) -> [TimeSeriesIndex, SimulationHistoryIndex]
    """
    Called from the GUI to launch a simulation.
      *: string class name of chosen model, etc...
      *_parameters: dictionary of parameters for chosen model, etc...
      connectivity: tvb.datatypes.connectivity.Connectivity object.
      surface: tvb.datatypes.surfaces.CorticalSurface: or None.
      stimulus: tvb.datatypes.patters.* object
    """
    result_h5 = dict()
    result_indexes = dict()
    start_time = self.algorithm.current_step * self.algorithm.integrator.dt

    self.algorithm.configure(full_configure=False)
    if self.branch_simulation_state_gid is not None:
        history = self.load_traited_by_gid(self.branch_simulation_state_gid)
        assert isinstance(history, SimulationHistory)
        history.fill_into(self.algorithm)

    region_map, region_volume_map = self._try_load_region_mapping()
    for monitor in self.algorithm.monitors:

```

--------------------------------

### Run Simulation

Source: https://docs.thevirtualbrain.org/_modules/tvb/simulator/backend/np

Executes a simulation using the NpBackend. It first checks compatibility, calculates the number of steps, prepares the initial state and time points, renders the simulation kernel using a template, and then runs the simulation with the prepared arguments.

```python
def run_sim(self, sim, nstep=None, simulation_length=None, print_source=False):
    assert nstep is not None or simulation_length is not None or sim.simulation_length is not None

    self.check_compatibility(sim)

    if nstep is None:
        if simulation_length is None:
            simulation_length = sim.simulation_length
        nstep = int(np.ceil(simulation_length/sim.integrator.dt))

    buf = sim.history.buffer[...,0]
    rbuf = np.concatenate((buf[0:1], buf[1:][::-1]), axis=0)
    state = np.transpose(rbuf, (1, 0, 2)).astype('f')
    t = np.arange(1, nstep+1 ) * sim.integrator.dt 

    template = '<%include file="np-sim.py.mako"/>'
    content = dict(sim=sim, np=np, nstep=nstep)
    kernel = self.build_py_func(template, content, print_source=print_source)
    dX = state.copy()
    n_svar, _, n_node = state.shape
    state = state.reshape((n_svar, sim.connectivity.horizon, n_node))

    weights = sim.connectivity.weights.copy()
    yh = np.empty((len(t),)+state[:,0].shape)

    parmat = sim.model.spatial_parameter_matrix
    args = state, weights, yh, parmat
    if isinstance(sim.integrator, integrators.IntegratorStochastic):
        np.random.seed(sim.integrator.noise.noise_seed) 
        if len(sim.integrator.noise.nsig.shape) > 1:
```

--------------------------------

### Get Line Start Indices for Rendering

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/datatypes/h5/tracts_h5

Generates a compact representation of element buffer indices required for drawing tracts using `gl.drawElements`. It calculates offsets for each tract within a chunk, which are then used to reconstruct the tracts during rendering.

```python
def get_line_starts(self, region_id):
    """
    Returns a compact representation of the element buffers required to draw the streams via gl.drawElements
    A list of indices that describe where the first vertex for a tract is in the vertex array returned by
    get_tract_vertices_starting_in_region
    """
    region_id = int(region_id)
    chunks = self._get_track_ids_webgl_chunks(region_id)
    chunk_line_starts = []
    tract_start_idx = self.tract_start_idx  # traits make the . expensive

    for tract_ids in chunks:
        offset = 0
        tract_offsets = [0]

        for tid in tract_ids:
            start, end = tract_start_idx[tid:tid + 2]
            track_len = end - start
            offset += track_len
            tract_offsets.append(offset)

        chunk_line_starts.append(tract_offsets)

    return chunk_line_starts
```

--------------------------------

### Get Visualizers for a Group (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/services/algorithm_service

Retrieves visualizer categories and then fetches launchable algorithms associated with a specific group GUID. It relies on internal helper methods and data access objects.

```python
def get_visualizers_for_group(self, dt_group_gid):

        categories = dao.get_visualisers_categories()
        return self._get_launchable_algorithms(dt_group_gid, categories)[1]
```

--------------------------------

### H5File Initialization and Core Attributes (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/neotraits/_h5core

This snippet shows the initialization of the H5File class, setting up the storage manager and defining common scalar headers and generic attributes. It also handles the creation of a new file if it doesn't exist, writing initial metadata like 'written_by'.

```python
from datetime import datetime
import importlib
import typing
import uuid
import numpy
import scipy.sparse

from tvb.basic.logger.builder import get_logger
from tvb.basic.neotraits.api import HasTraits, TupleEnum, Attr, List, NArray, Range, EnumAttr, Final, TVBEnum
from tvb.basic.neotraits.ex import TraitFinalAttributeError
from tvb.core.entities.generic_attributes import GenericAttributes
from tvb.core.neotraits.h5 import EquationScalar, SparseMatrix, ReferenceList
from tvb.core.neotraits.h5 import Uuid, Scalar, Accessor, DataSet, Reference, JsonFinal, Json, JsonRange, Enum
from tvb.core.neotraits.view_model import DataTypeGidAttr
from tvb.core.utils import string2date, date2string
from tvb.datatypes.equations import Equation, EquationsEnum
from tvb.storage.h5.file.exceptions import MissingDataSetException
from tvb.storage.storage_interface import StorageInterface

LOGGER = get_logger(__name__)


class H5File(object):
    """
    A H5 based file format.
    This class implements reading and writing to a *specific* h5 based file format.
    A subclass of this defines a new file format.
    """
    KEY_WRITTEN_BY = 'written_by'
    is_new_file = False

    def __init__(self, path):
        # type: (str) -> None
        self.path = path
        self.storage_manager = StorageInterface.get_storage_manager(self.path)
        # would be nice to have an opened state for the chunked api instead of the close_file=False

        # common scalar headers
        self.gid = Uuid(HasTraits.gid, self)
        self.written_by = Scalar(Attr(str), self, name=self.KEY_WRITTEN_BY)
        self.create_date = Scalar(Attr(str), self, name='create_date')
        self.type = Scalar(Attr(str), self, name='type')

        # Generic attributes descriptors
        self.generic_attributes = GenericAttributes()
        self.invalid = Scalar(Attr(bool), self, name='invalid')
        self.is_nan = Scalar(Attr(bool), self, name='is_nan')
        self.subject = Scalar(Attr(str), self, name='subject')
        self.state = Scalar(Attr(str), self, name='state')
        self.user_tag_1 = Scalar(Attr(str), self, name='user_tag_1')
        self.user_tag_2 = Scalar(Attr(str), self, name='user_tag_2')
        self.user_tag_3 = Scalar(Attr(str), self, name='user_tag_3')
        self.user_tag_4 = Scalar(Attr(str), self, name='user_tag_4')
        self.user_tag_5 = Scalar(Attr(str), self, name='user_tag_5')
        self.operation_tag = Scalar(Attr(str, required=False), self, name='operation_tag')
        self.parent_burst = Uuid(Attr(uuid.UUID, required=False), self, name='parent_burst')
        self.visible = Scalar(Attr(bool), self, name='visible')
        self.metadata_cache = None
        # Keep a list with datasets for which we should write metadata before closing the file
        self.expandable_datasets = []

        if not self.storage_manager.is_valid_tvb_file():
            self.written_by.store(self.get_class_path())
            self.is_new_file = True

```

--------------------------------

### Get Operations by Status and Classname - Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/entities/storage/operation_dao

Retrieves operations based on their status and the classname of the associated algorithm. By default, it filters for 'PENDING' and 'STARTED' statuses and 'SimulatorAdapter' classname. It handles SQLAlchemyErrors by logging and returning None.

```python
def get_operations(self, status=None, algorithm_classname="SimulatorAdapter"):
        if status is None:
            status = [STATUS_PENDING, STATUS_STARTED]
        try:
            result = self.session.query(Operation).join(Algorithm) \
                .filter(Algorithm.classname == algorithm_classname) \
                .filter(Operation.status.in_(status)).all()
            return result
        except SQLAlchemyError as excep:
            self.logger.exception(excep)
            return None
```

--------------------------------

### Get URL After Monitor Equation in Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/web/controllers/simulator/simulator_controller

Determines the URL to navigate to after setting a monitor's equation. If there is no next monitor, it returns the setup PSE URL; otherwise, it constructs the URL for the next monitor's parameters.

```python
    def get_url_after_monitor_equation(self, next_monitor):
        if next_monitor is None:
            return SimulatorWizzardURLs.SETUP_PSE_URL

        last_loaded_fragment_url = self.build_monitor_url(SimulatorWizzardURLs.SET_MONITOR_PARAMS_URL,
                                                          type(next_monitor).__name__)
        return last_loaded_fragment_url
```

--------------------------------

### Python: Get BIDS Directory from Arguments

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/rest/bids_monitor/launch_bids_monitor

This Python function retrieves the BIDS directory path from command-line arguments. It iterates through `sys.argv` to find an argument starting with '--bids-dir' and extracts the directory path. It requires the `sys` module.

```python
# -*- coding: utf-8 -*-
import sys


def get_bids_dir():
    bids_dir = None
    if len(sys.argv) > 0:
        for arg in sys.argv:
            if arg.startswith('--bids-dir'):
                bids_dir = arg.split('=')[1]
    return bids_dir

```

--------------------------------

### Get Bursts for Project (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/entities/storage/burst_dao

Retrieves a paginated list of BurstConfiguration entities for a given project, ordered by start time in descending order. It supports counting total bursts or fetching a specific page. It handles SQLAlchemy errors by logging them and returning None.

```python
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

```

--------------------------------

### Launch Simulation - Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/command/lab

Launches a simulation within a project using a provided simulator model. It configures the profile, retrieves the project and simulator algorithm, and then uses `SimulatorService().async_launch_and_prepare_simulation()` to start the simulation.

```python
def fire_simulation(project_id, simulator_model):
    TvbProfile.set_profile(TvbProfile.COMMAND_PROFILE)
    project = dao.get_project_by_id(project_id)
    assert isinstance(simulator_model, SimulatorAdapterModel)
    # Load the SimulatorAdapter algorithm from DB
    cached_simulator_algorithm = AlgorithmService().get_algorithm_by_module_and_class(
        IntrospectionRegistry.SIMULATOR_MODULE,
        IntrospectionRegistry.SIMULATOR_CLASS)

    # Instantiate a SimulatorService and launch the configured simulation
    simulator_service = SimulatorService()
    burst = BurstConfiguration(project.id)
    burst.name = "Sim " + str(datetime.now())
    burst.start_time = datetime.now()
    dao.store_entity(burst)

    launched_operation = simulator_service.async_launch_and_prepare_simulation(burst, project.administrator, project,
                                                                               cached_simulator_algorithm,
                                                                               simulator_model)
    LOG.info("Operation launched ....")
    return launched_operation
```

--------------------------------

### Specify Variables of Interest for TVB Monitor

Source: https://docs.thevirtualbrain.org/api/tvb.core.entities.file

Defines the indices of model variables that a specific monitor should record. Indices start at zero. For example, if a model has VOIs V, W, and V+W, and W is selected, the correct index is 0.

```python
tvb.simulator.monitors.Monitor.variables_of_interest = NArray(label=’Model variables to watch’, dtype=int64, default=None, dim_names=(), ndim=None, required=False)
```

--------------------------------

### Prepare First Simulation Fragment - Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/web/controllers/simulator/simulator_controller

Prepares the initial form for the simulator fragment, setting up simulator context, branch conditions, and adapter form. It validates the fragment and fills it from the current simulator traits. Input is primarily the project context and simulator service.

```python
def prepare_first_fragment(self):
        self.context.set_simulator()
        simulator, _, _, is_branch = self.context.get_common_params()
        branch_conditions = self.simulator_service.compute_conn_branch_conditions(is_branch, simulator)
        form = self.algorithm_service.prepare_adapter_form(form_instance=SimulatorAdapterForm(),
                                                           project_id=self.context.project.id,
                                                           extra_conditions=branch_conditions)

        self.simulator_service.validate_first_fragment(form, self.context.project.id, ConnectivityIndex)
        form.fill_from_trait(self.context.simulator)
        return form

```

--------------------------------

### Launch EEG Monitor Visualization (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/visualizers/eeg_monitor

The `launch` method orchestrates the EEG monitor visualization. It first loads the necessary input data indexes using `_load_input_indexes`, then computes the visualization parameters via `compute_parameters`, and finally builds the display result with specified page templates.

```python
    def launch(self, view_model):
        # type: (EegMonitorModel) -> dict
        """
        Compute visualizer's page
        """
        main_time_series_index, time_series_index2, time_series_index3 = self._load_input_indexes(view_model)
        params = self.compute_parameters(main_time_series_index, time_series_index2, time_series_index3)
        pages = dict(controlPage="eeg/controls", channelsPage="commons/channel_selector.html")
        return self.build_display_result("eeg/view", params, pages=pages)

```

--------------------------------

### Get Operations for HPC Job - Python

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/entities/storage/operation_dao

Retrieves operations that are 'PENDING' or 'STARTED', use the 'SimulatorAdapter' class, and are not marked as 'queue_full'. This is intended for fetching tasks suitable for High-Performance Computing (HPC) job scheduling. Errors are logged, and None is returned on exception.

```python
def get_operations_for_hpc_job(self):
        status = [STATUS_PENDING, STATUS_STARTED]
        algorithm_classname = "SimulatorAdapter"
        queue_full = False
        try:
            result = self.session.query(Operation).join(Algorithm) \
                .filter(Algorithm.classname == algorithm_classname) \
                .filter(Operation.status.in_(status)) \
                .filter(Operation.queue_full == queue_full).all()
            return result
        except SQLAlchemyError as excep:
            self.logger.exception(excep)
            return None
```

--------------------------------

### Get Launchable Algorithms for Datatype (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/services/algorithm_service

Fetches launchable algorithms compatible with a given datatype GUID. It retrieves categories, filters adapters based on datatype compatibility, and groups them by category. Includes error handling for filter evaluation.

```python
def get_launchable_algorithms(self, datatype_gid):
        """
        :param datatype_gid: Filter only algorithms compatible with this GUID
        :return: dict(category_name: List AlgorithmTransientGroup)
        """
        categories = dao.get_launchable_categories()
        datatype_instance, filtered_adapters, has_operations_warning = self._get_launchable_algorithms(datatype_gid,
                                                                                                       categories)

        categories_dict = dict()
        for c in categories:
            categories_dict[c.id] = c.displayname

        return self._group_adapters_by_category(filtered_adapters, categories_dict), has_operations_warning
```

--------------------------------

### Initialize NpBackend

Source: https://docs.thevirtualbrain.org/_modules/tvb/simulator/backend/np

Initializes the NpBackend by creating a temporary directory for generated code and adding it to the system's path.

```python
self.cgdir = tempfile.TemporaryDirectory()
sys.path.append(self.cgdir.name)
```

--------------------------------

### Get All Datatypes (Paginated) (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/entities/storage/datatype_dao

Retrieves a list of all data types available in the system, with support for pagination. It allows specifying a starting page index and the number of items per page. This function uses SQLAlchemy and handles potential database errors.

```python
def get_all_datatypes(self, page_start=0, page_size=DEFAULT_PAGE_SIZE):
        """
        Return a list with all of the datatypes currently available in TVB. Is used by 
        the file storage update manager to upgrade from version to the next. 
        
        :param page_start: the index from which to start adding datatypes to the result list
        :param page_size: maximum number of entities to retrieve
        """
        resulted_data = []
        try:
            resulted_data = self.session.query(DataType).order_by(DataType.id).offset(
                max(page_start, 0)).limit(max(page_size, 0)).all()
        except SQLAlchemyError as excep:
            self.logger.exception(excep)
        return resulted_data
```

--------------------------------

### Attribute with Factory Default

Source: https://docs.thevirtualbrain.org/_sources/manuals/neotraits/neotraits

This example demonstrates how to define an attribute 'a' with a mutable default value (a dictionary) using a factory function. This prevents unexpected behavior where all instances share the same mutable default object. The factory function ensures each instance gets a fresh dictionary.

```python
class A(HasTraits):
    a = Attr(field_type=dict, default=lambda: {'factory': 'default'})
```

--------------------------------

### Get Triangle Slice Boundaries for a Given Slice Index (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/datatypes/h5/surface_h5

Retrieves the start and end indices for triangle data within a specified slice. Similar to vertex boundary retrieval, it includes error handling for missing slice indices and returns default values, ensuring robustness.

```Python
def _get_slice_triangle_boundaries(self, slice_idx):
        if str(slice_idx) in self._split_slices:
            start_idx = max(0, self._split_slices[str(slice_idx)][KEY_TRIANGLES][KEY_START])
            end_idx = min(self._split_slices[str(slice_idx)][KEY_TRIANGLES][KEY_END], self._number_of_triangles)
            return start_idx, end_idx
        else:
            LOG.warning("Could not access slice indices, possibly due to an incompatibility with code update!")
            return 0, self._number_of_triangles
```

--------------------------------

### Async Launch and Prepare PSE

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/services/simulator_service

Initiates and prepares a 'Prepare Simulation Environment' (PSE) process asynchronously. This function orchestrates the preparation of multiple operations based on burst configurations and simulation parameters. It includes logic to handle cancellation and logs the success rate of launched workflows. Dependencies include various services like OperationService, BurstService, and data access objects.

```python
    def async_launch_and_prepare_pse(self, burst_config, user, project, simulator_algo, range_param1, range_param2,
                                     session_stored_simulator):
        try:
            algo_category = simulator_algo.algorithm_category
            operation_group = burst_config.operation_group
            metric_operation_group = burst_config.metric_operation_group
            range_param2_values = [None]
            if range_param2:
                range_param2_values = range_param2.get_range_values()
            GROUP_BURST_PENDING[burst_config.id] = True
            operations, pse_canceled = self._prepare_operations(algo_category, burst_config,
                                                                metric_operation_group,
                                                                operation_group, project,
                                                                range_param1,
                                                                range_param2, range_param2_values,
                                                                session_stored_simulator,
                                                                simulator_algo, user)

            GROUP_BURST_PENDING[burst_config.id] = False
            if pse_canceled:
                return

            wf_errs = self._launch_operations(operations, burst_config)
            self.logger.debug("Finished launching workflows. " + str(len(operations) - wf_errs) +
                              " were launched successfully, " + str(wf_errs) + " had error on pre-launch steps")
            return operations[0] if len(operations) > 0 else None

        except Exception as excep:
            self.logger.error(excep)
            self.burst_service.mark_burst_finished(burst_config, error_message=str(excep))
```

--------------------------------

### Get Vertex Slice Boundaries for a Given Slice Index (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/datatypes/h5/surface_h5

Retrieves the start and end indices for vertex data within a specified slice. It handles cases where the slice index might not be found, returning default values and logging a warning. This is crucial for efficient data retrieval for visualization.

```Python
def get_slice_vertex_boundaries(self, slice_idx):
        if str(slice_idx) in self._split_slices:
            start_idx = max(0, self._split_slices[str(slice_idx)][KEY_VERTICES][KEY_START])
            end_idx = min(self._split_slices[str(slice_idx)][KEY_VERTICES][KEY_END], self._number_of_vertices)
            return start_idx, end_idx
        else:
            LOG.warning("Could not access slice indices, possibly due to an incompatibility with code update!")
            return 0, min(SPLIT_BUFFER_SIZE, self._number_of_vertices)
```

--------------------------------

### Prepare and Render Simulator Configuration Snippet (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/web/controllers/simulator/simulator_controller

This snippet prepares a simulator fragment form and rendering rules for simulation copying. It handles adding branches and copying to the session, returning the rendering rules as a dictionary. Dependencies include SimulatorFragmentRenderingRules and SimulatorWizzardURLs.

```python
self.context.add_branch_and_copy_to_session(False, True)
        form = self._prepare_first_fragment_for_burst_copy(burst_config_id, self.COPY_NAME_FORMAT)
        rendering_rules = SimulatorFragmentRenderingRules(form, SimulatorWizzardURLs.SET_CONNECTIVITY_URL,
                                                          is_simulation_copy=True, is_simulation_readonly_load=True,
                                                          is_first_fragment=True)
        return rendering_rules.to_dict()
```

```python
self.context.add_branch_and_copy_to_session(True, False)
        form = self._prepare_first_fragment_for_burst_copy(burst_config_id, self.BRANCH_NAME_FORMAT)
        rendering_rules = SimulatorFragmentRenderingRules(form, SimulatorWizzardURLs.SET_CONNECTIVITY_URL,
                                                          is_simulation_copy=True, is_simulation_readonly_load=True,
                                                          is_first_fragment=True)

        return rendering_rules.to_dict()
```

--------------------------------

### Get Project Data Types with Filters (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/entities/storage/datatype_dao

Retrieves all DataTypes for a given project, supporting filtering by visibility and partial string matching across multiple attributes. It handles direct DataTypes and those within groups, ensuring comprehensive data fetching. This function requires SQLAlchemy and logger setup.

```python
def get_data_in_project(self, project_id, visibility_filter=None, filter_value=None):
        """
        Get all the DataTypes for a given project, including Linked Entities and DataType Groups.

        :param project_id: ID for the current project to filter for
        :param visibility_filter: when not None, will filter by DataTye fields
        :param filter_value: when not None, will filter with ilike multiple DataType string attributes
        """
        resulted_data = []
        try:
            # First Query DT, DT_gr, Lk_DT and Lk_DT_gr
            datatype_alias = aliased(DataType, flat=True)
            query = self.session.query(datatype_alias
                        ).join(Operation, Operation.id == datatype_alias.fk_from_operation
                        ).join(Algorithm).join(AlgorithmCategory
                        ).outerjoin(Links, and_(Links.fk_from_datatype == datatype_alias.id,
                                                Links.fk_to_project == project_id)
                        ).outerjoin(BurstConfiguration, datatype_alias.fk_parent_burst == BurstConfiguration.gid
                                    ).filter(datatype_alias.fk_datatype_group == None
                        ).filter(or_(Operation.fk_launched_in == project_id,
                                     Links.fk_to_project == project_id))

            if visibility_filter:
                filter_str = visibility_filter.get_sql_filter_equivalent('datatype_alias')
                if filter_str is not None:
                    query = query.filter(eval(filter_str))
            if filter_value is not None:
                query = query.filter(self._compose_filter_datatype_ilike(filter_value))

            resulted_data = query.all()

            # Now query what it was not covered before:
            # Links of DT which are part of a group, but the entire group is not linked
            links = aliased(Links)
            query2 = self.session.query(datatype_alias
                        ).join(Operation, Operation.id == datatype_alias.fk_from_operation
                        ).join(Algorithm).join(AlgorithmCategory
                        ).join(Links, and_(Links.fk_from_datatype == datatype_alias.id,
                                           Links.fk_to_project == project_id)
                        ).outerjoin(links, and_(links.fk_from_datatype == datatype_alias.fk_datatype_group,
                                                links.fk_to_project == project_id)
                        ).outerjoin(BurstConfiguration, datatype_alias.fk_parent_burst == BurstConfiguration.gid
                                    ).filter(datatype_alias.fk_datatype_group != None
                        ).filter(links.id == None)

            if visibility_filter:
                filter_str = visibility_filter.get_sql_filter_equivalent('datatype_alias')
                if filter_str is not None:
                    query2 = query2.filter(eval(filter_str))
            if filter_value is not None:
                query2 = query2.filter(self._compose_filter_datatype_ilike(filter_value))

            resulted_data.extend(query2.all())

            # Load lazy fields for future usage
            for dt in resulted_data:
                dt._parent_burst
                dt.parent_operation.algorithm
                dt.parent_operation.algorithm.algorithm_category
                dt.parent_operation.project
                dt.parent_operation.operation_group
                dt.parent_operation.user
                dt.display_name

        except Exception as excep:
            self.logger.exception(excep)

        return resulted_data

```

--------------------------------

### Setup TVB Environment

Source: https://docs.thevirtualbrain.org/demos/Demos_Matlab

Ensures the TVB environment is properly set up for running simulations. This is a prerequisite for all subsequent simulation steps. It confirms that TVB modules are available.

```shell
tvb_setup

```

--------------------------------

### Tract Importer Base Class Logic (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/adapters/uploaders/tract_importer

Provides the base implementation for importing tract data. It handles file validation, region map caching, and initial setup for the tract data type. It defines methods for getting the form class, output type, and internal logic for mapping tract vertices to region IDs using HDF5 data.

```python
class _TrackImporterBase(ABCUploader, metaclass=ABCMeta):
    _ui_name = "Tracts TRK or TCK"
    _ui_subsection = "tracts_importer"
    _ui_description = "Import tracts"

    READ_CHUNK = 4 * 1024

    def get_form_class(self):
        return TrackImporterForm

    def get_output(self):
        return [TractsIndex]

    def _get_tract_region(self, start_vertex):
        # Map to voxel index space
        # Lacking any affine matrix between these, we assume they are in the same geometric space
        # What remains is to map geometry to the discrete region volume mapping indices
        x_plane, y_plane, z_plane = [int(i) for i in start_vertex]

        if not (0 <= x_plane < self.region_volume_shape[0] and
                0 <= y_plane < self.region_volume_shape[1] and
                0 <= z_plane < self.region_volume_shape[2]):
            # import random
            # return random.randint(0, 75)
            raise IndexError('There are vertices outside the region volume map cube!')

        # in memory data set
        if self.full_rmap_cache is not None:
            region_id = self.full_rmap_cache[x_plane, y_plane, z_plane]
            return region_id

        # not in memory have to go to disk
        slices = slice(x_plane, x_plane + 1), slice(y_plane, y_plane + 1), slice(z_plane, z_plane + 1)
        region_id = self.region_volume_h5.read_data_slice(slices)[0, 0, 0]
        return region_id

    def _attempt_to_cache_regionmap(self, region_volume):
        a, b, c = region_volume.read_data_shape()
        if a * b * c <= 256 * 256 * 256:
            # read all
            slices = slice(a), slice(b), slice(c)
            self.full_rmap_cache = region_volume.read_data_slice(slices)
        else:
            self.full_rmap_cache = None

    def _base_before_launch(self, data_file, region_volume_gid):
        if data_file is None:
            raise LaunchException("Please select a file to import!")

        if region_volume_gid is not None:
            rvm_h5 = h5.h5_file_for_gid(region_volume_gid)
            self._attempt_to_cache_regionmap(rvm_h5)
            self.region_volume_shape = rvm_h5.read_data_shape()
            self.region_volume_h5 = rvm_h5

        datatype = Tracts()
        dummy_rvm = RegionVolumeMapping()
        dummy_rvm.gid = region_volume_gid
        datatype.region_volume_map = dummy_rvm
        return datatype
```

--------------------------------

### Mark Operation as Started (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/core/services/hpc_operation_service

This static method marks an operation as having started. It sets the operation's start time to the current time and persists this change using the data access object (dao).

```python
def _operation_started(operation):
    operation.start_now()
    dao.store_entity(operation)
```

--------------------------------

### Launch Web Interface - TheVirtualBrain

Source: https://docs.thevirtualbrain.org/manuals/UserGuide/UserGuide-Shell

Launches the web interface of TheVirtualBrain using the distribution script. This command starts the WEB_PROFILE.

```bash
$ ./distribution.sh start WEB_PROFILE

```

--------------------------------

### Launch CherryPy Server and Browser (Python)

Source: https://docs.thevirtualbrain.org/_modules/tvb/interfaces/web/run

This Python code snippet demonstrates how to launch the CherryPy web server and automatically open the TVB web interface in the default browser. It checks the operating system to select the appropriate browser command and constructs the URL to open, including a fallback for initial setup. Dependencies include `cherrypy`, `webbrowser`, and custom TVB modules like `LOGGER`, `TvbProfile`, and `CONFIG_EXISTS`.

```python
if browser:
        run_browser()

    # Launch CherryPy loop forever.
    LOGGER.info("Finished starting TVB version %s in %.3f s",
                TvbProfile.current.version.CURRENT_VERSION, time.time() - STARTUP_TIC)
    cherrypy.engine.block()
    cherrypy.log.error_log


@user_environment_execution
def run_browser():
    try:
        if TvbProfile.env.is_windows():
            browser_app = webbrowser.get('windows-default')
        elif TvbProfile.env.is_mac():
            browser_app = webbrowser.get('macosx')
        else:
            browser_app = webbrowser

        url_to_open = TvbProfile.current.web.BASE_LOCAL_URL
        if not CONFIG_EXISTS:
            url_to_open += 'settings/settings'

        LOGGER.info("We will try to open in a browser: " + url_to_open)
        browser_app.open(url_to_open)

    except Exception:
        LOGGER.warning("Browser could not be fired!  Please manually type in your "
                       "preferred browser: %s" % TvbProfile.current.web.BASE_LOCAL_URL)


if __name__ == '__main__':
    # Prepare parameters and fire CherryPy
    # Remove not-relevant parameter, 0 should point towards this "run.py" file, 1 to the profile
    start_tvb(sys.argv[2:])
```
Tokens:
Raw