### 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:]) ```