Source code for tvb.interfaces.web.controllers.spatial.local_connectivity_controller

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

.. moduleauthor:: Paula Popa <>
.. moduleauthor:: Bogdan Neacsa <>
.. moduleauthor:: Ionel Ortelecan <>

import json
import uuid
import cherrypy
from tvb.adapters.creators.local_connectivity_creator import *
from tvb.adapters.datatypes.h5.local_connectivity_h5 import LocalConnectivityH5
from tvb.adapters.datatypes.h5.surface_h5 import SurfaceH5
from tvb.adapters.forms.equation_forms import get_form_for_equation
from tvb.core.adapters.abcadapter import ABCAdapter
from tvb.core.entities.load import try_get_last_datatype, load_entity_by_gid
from tvb.core.neocom import h5
from tvb.interfaces.web.controllers import common
from tvb.interfaces.web.controllers.autologging import traced
from tvb.interfaces.web.controllers.base_controller import BaseController
from tvb.interfaces.web.controllers.common import MissingDataException
from tvb.interfaces.web.controllers.decorators import check_user, handle_error
from tvb.interfaces.web.controllers.decorators import expose_fragment, expose_page, expose_json
from tvb.interfaces.web.controllers.spatial.base_spatio_temporal_controller import SpatioTemporalController
from tvb.interfaces.web.structure import WebStructure


LOAD_EXISTING_URL = SpatioTemporalController.build_path('/spatial/localconnectivity/load_local_connectivity')
RELOAD_DEFAULT_PAGE_URL = SpatioTemporalController.build_path('/spatial/localconnectivity/reset_local_connectivity')

# Between steps/pages we keep a LocalConnectivityCreatorModel in session at this key
KEY_LCONN = "local-conn"

[docs]@traced class LocalConnectivityController(SpatioTemporalController): """ Controller layer for displaying/creating a LocalConnectivity entity. """ # These 4 strings are used on client-side to set onchange events on form fields SURFACE_FIELD = 'set_surface' EQUATION_FIELD = 'set_equation' CUTOFF_FIELD = 'set_cutoff_value' DISPLAY_NAME_FIELD = 'set_display_name' EQUATION_PARAMS_FIELD = 'set_equation_param' base_url = '/spatial/localconnectivity' def __init__(self): SpatioTemporalController.__init__(self) self.plotted_equation_prefixes = {}
[docs] @expose_page def step_1(self, do_reset=0, **kwargs): """ Generate the html for the first step of the local connectivity page. :param do_reset: Boolean telling to start from empty page or not :param kwargs: not actually used, but parameters are still submitted from UI since we just\ use the same js function for this. """ project_id = common.get_current_project().id if int(do_reset) == 1: new_lconn = LocalConnectivityCreatorModel() default_surface_index = try_get_last_datatype(project_id, SurfaceIndex, LocalConnectivityCreatorForm.get_filters()) if default_surface_index: new_lconn.surface = uuid.UUID(default_surface_index.gid) else: # Surface is required in model and we should keep it like this, but we also want to new_lconn.surface = uuid.uuid4() common.set_error_message(self.MSG_MISSING_SURFACE) common.add2session(KEY_LCONN, new_lconn) current_lconn = common.get_from_session(KEY_LCONN) existent_lcon_form = self.algorithm_service.prepare_adapter_form(form_instance=LocalConnectivitySelectorForm(), project_id=common.get_current_project().id) = current_lconn.gid.hex configure_lcon_form = self.algorithm_service.prepare_adapter_form( form_instance=LocalConnectivityCreatorForm(), project_id=common.get_current_project().id) configure_lcon_form.fill_from_trait(current_lconn) current_lconn.equation = configure_lcon_form.spatial.value() template_specification = dict(title="Surface - Local Connectivity") template_specification['mainContent'] = 'spatial/local_connectivity_step1_main' template_specification['inputList'] = self.render_spatial_form(configure_lcon_form) template_specification['displayCreateLocalConnectivityBtn'] = True template_specification['loadExistentEntityUrl'] = LOAD_EXISTING_URL template_specification['resetToDefaultUrl'] = RELOAD_DEFAULT_PAGE_URL template_specification['existentEntitiesInputList'] = self.render_spatial_form(existent_lcon_form) template_specification['submit_parameters_url'] = self.build_path('/spatial/localconnectivity/create_local_connectivity') template_specification['equationViewerUrl'] = self.build_path('/spatial/localconnectivity/get_equation_chart') template_specification['baseUrl'] = self.base_url self.plotted_equation_prefixes = {self.SURFACE_FIELD:, self.EQUATION_FIELD:, self.CUTOFF_FIELD:, self.DISPLAY_NAME_FIELD:, self.EQUATION_PARAMS_FIELD:[1:]} template_specification['equationsPrefixes'] = json.dumps(self.plotted_equation_prefixes) template_specification['next_step_url'] = '/spatial/localconnectivity/step_2' return self.fill_default_attributes(template_specification)
[docs] @cherrypy.expose def set_equation_param(self, **param): current_lconn = common.get_from_session(KEY_LCONN) eq_param_form_class = get_form_for_equation(type(current_lconn.equation)) eq_param_form = eq_param_form_class() eq_param_form.fill_from_trait(current_lconn.equation) eq_param_form.fill_from_post(param) eq_param_form.fill_trait(current_lconn.equation)
[docs] @cherrypy.expose def set_cutoff_value(self, **param): current_lconn = common.get_from_session(KEY_LCONN) cutoff_form_field = LocalConnectivityCreatorForm().cutoff cutoff_form_field.fill_from_post(param) current_lconn.cutoff = cutoff_form_field.value
[docs] @cherrypy.expose def set_surface(self, **param): current_lconn = common.get_from_session(KEY_LCONN) surface_form_field = LocalConnectivityCreatorForm().surface surface_form_field.fill_from_post(param) current_lconn.surface = surface_form_field.value
[docs] @cherrypy.expose def set_display_name(self, **param): display_name_form_field = LocalConnectivityCreatorForm().display_name display_name_form_field.fill_from_post(param) if display_name_form_field.value is not None: lconn = common.get_from_session(KEY_LCONN) lconn.display_name = display_name_form_field.value
[docs] @expose_page def step_2(self, **kwargs): """ Generate the html for the second step of the local connectivity page. :param kwargs: not actually used, but parameters are still submitted from UI since we just\ use the same js function for this. """ current_lconn = common.get_from_session(KEY_LCONN) left_side_form = self.algorithm_service.prepare_adapter_form(form_instance=LocalConnectivitySelectorForm(), project_id=common.get_current_project().id) = current_lconn.gid.hex template_specification = dict(title="Surface - Local Connectivity") template_specification['mainContent'] = 'spatial/local_connectivity_step2_main' template_specification['existentEntitiesInputList'] = self.render_adapter_form(left_side_form) template_specification['loadExistentEntityUrl'] = LOAD_EXISTING_URL template_specification['resetToDefaultUrl'] = RELOAD_DEFAULT_PAGE_URL template_specification['next_step_url'] = self.build_path('/spatial/localconnectivity/step_1') msg, _ = common.get_message_from_session() template_specification['displayedMessage'] = msg if current_lconn is not None: selected_local_conn = load_entity_by_gid(current_lconn.gid) template_specification.update(self.display_surface(selected_local_conn.fk_surface_gid)) template_specification['no_local_connectivity'] = False template_specification['minValue'] = selected_local_conn.matrix_non_zero_min template_specification['maxValue'] = selected_local_conn.matrix_non_zero_max else: template_specification['no_local_connectivity'] = True template_specification[common.KEY_PARAMETERS_CONFIG] = False return self.fill_default_attributes(template_specification)
[docs] @cherrypy.expose @handle_error(redirect=False) @check_user def create_local_connectivity(self, **kwargs): """ Used for creating and storing a local connectivity. """ current_lconn = common.get_from_session(KEY_LCONN) local_connectivity_creator = ABCAdapter.build_adapter_from_class(LocalConnectivityCreator) self.operation_service.fire_operation(local_connectivity_creator, common.get_logged_user(), common.get_current_project().id, view_model=current_lconn) common.set_important_message("The operation for creating the local connectivity was successfully launched.") return self.step_1()
[docs] @cherrypy.expose @handle_error(redirect=False) @check_user def load_local_connectivity(self, local_connectivity_gid, from_step=None): """ Loads an existing local connectivity. """ lconn_index = load_entity_by_gid(local_connectivity_gid) existent_lconn = LocalConnectivityCreatorModel() lconn_h5_path = h5.path_for_stored_index(lconn_index) with LocalConnectivityH5(lconn_h5_path) as lconn_h5: lconn_h5.load_into(existent_lconn) existent_lconn.surface = uuid.UUID(lconn_index.fk_surface_gid) common.add2session(KEY_LCONN, existent_lconn) existent_lconn.display_name = lconn_index.user_tag_1 if existent_lconn.equation: msg = "Successfully loaded existent entity gid=%s" % (local_connectivity_gid,) else: msg = "There is no equation specified for this local connectivity. " msg += "The default equation is displayed into the spatial field." common.set_message(msg) if int(from_step) == 1: return self.step_1() if int(from_step) == 2: return self.step_2()
[docs] @cherrypy.expose @handle_error(redirect=False) @check_user def reset_local_connectivity(self, from_step): """ Reset the context and reset to the first step. This method is called when the None entry is selected from the select. :param from_step: is not used in local connectivity case since we don't want to remain in step 2 in case none was selected. We are keeping it so far to remain compatible with the stimulus pages. """ return self.step_1(do_reset=1)
[docs] def fill_default_attributes(self, template_dictionary): """ Overwrite base controller to add required parameters for adapter templates. """ template_dictionary[common.KEY_SECTION] = WebStructure.SECTION_CONNECTIVITY template_dictionary[common.KEY_SUB_SECTION] = 'local' template_dictionary[common.KEY_SUBMENU_LIST] = self.connectivity_submenu template_dictionary[common.KEY_INCLUDE_RESOURCES] = 'spatial/included_resources' BaseController.fill_default_attributes(self, template_dictionary) return template_dictionary
[docs] @expose_json def compute_data_for_gradient_view(self, local_connectivity_gid, selected_triangle): """ When the user loads an existent local connectivity and he picks a vertex from the used surface, this method computes the data needed for drawing a gradient view corresponding to that vertex. Returns a json which contains the data needed for drawing a gradient view for the selected vertex. """ triangle_index = int(selected_triangle) lconn_h5 = h5.h5_file_for_gid(local_connectivity_gid) surface_h5 = h5.h5_file_for_gid(lconn_h5.surface.load()) assert isinstance(surface_h5, SurfaceH5) vertex_index = int(surface_h5.triangles[triangle_index][0]) assert isinstance(lconn_h5, LocalConnectivityH5) lconn_matrix = lconn_h5.matrix.load() picked_data = list(lconn_matrix[vertex_index].toarray().squeeze()) lconn_h5.close() result = [] number_of_split_slices = surface_h5.number_of_split_slices.load() if number_of_split_slices <= 1: result.append(picked_data) else: for slice_number in range(number_of_split_slices): start_idx, end_idx = surface_h5.get_slice_vertex_boundaries(slice_number) result.append(picked_data[start_idx:end_idx]) surface_h5.close() result = {'data': json.dumps(result)} return result
[docs] @staticmethod def get_series_json(ideal_case, average_case, worst_case, best_case, vertical_line): """ Gather all the separate data arrays into a single flot series. """ return json.dumps([ {"data": ideal_case, "lines": {"lineWidth": 1}, "label": "Theoretical case", "color": "rgb(52, 255, 25)"}, {"data": average_case, "lines": {"lineWidth": 1}, "label": "Most probable", "color": "rgb(148, 0, 179)"}, {"data": worst_case, "lines": {"lineWidth": 1}, "label": "Worst case", "color": "rgb(0, 0, 255)"}, {"data": best_case, "lines": {"lineWidth": 1}, "label": "Best case", "color": "rgb(122, 122, 0)"}, {"data": vertical_line, "points": {"show": True, "radius": 1}, "label": "Cut-off distance", "color": "rgb(255, 0, 0)"} ])
[docs] @expose_fragment('spatial/equation_displayer') def get_equation_chart(self): """ Returns the html which contains the plot with the equations specified into 'plotted_equations_prefixes' field. """ try: # This should be called once at first rendering and once for any change event on form fields used # in computation: equation, equation params, surface, cutoff current_lconn = common.get_from_session(KEY_LCONN) surface_gid = current_lconn.surface.hex surface = load_entity_by_gid(surface_gid) if surface is None: raise MissingDataException(self.MSG_MISSING_SURFACE + "!!!") max_x = current_lconn.cutoff if max_x <= 0: max_x = 50 equation = current_lconn.equation # What we want ideal_case_series, _ = equation.get_series_data(0, 2 * max_x) # What we'll mostly get avg_res = 2 * int(max_x / surface.edge_mean_length) step = max_x * 2 / (avg_res - 1) average_case_series, _ = equation.get_series_data(0, 2 * max_x, step) # It can be this bad worst_res = 2 * int(max_x / surface.edge_max_length) step = 2 * max_x / (worst_res - 1) worst_case_series, _ = equation.get_series_data(0, 2 * max_x, step) # This is as good as it gets... best_res = 2 * int(max_x / surface.edge_min_length) step = 2 * max_x / (best_res - 1) best_case_series, _ = equation.get_series_data(0, 2 * max_x, step) max_y = -1000000000 min_y = 10000000000 for case in ideal_case_series: if min_y > case[1]: min_y = case[1] if min_y > case[1]: min_y = case[1] if max_y < case[1]: max_y = case[1] if max_y < case[1]: max_y = case[1] vertical_line = [] vertical_step = (max_y - min_y) / NO_OF_CUTOFF_POINTS for i in range(NO_OF_CUTOFF_POINTS): vertical_line.append([max_x, min_y + i * vertical_step]) all_series = self.get_series_json(ideal_case_series, average_case_series, worst_case_series, best_case_series, vertical_line) return {'allSeries': all_series, 'prefix': 'spatial', "message": None} except NameError as ex: self.logger.exception(ex) return {'allSeries': None, 'errorMsg': "Incorrect parameters for equation passed."} except SyntaxError as ex: self.logger.exception(ex) return {'allSeries': None, 'errorMsg': "Some of the parameters hold invalid characters."} except Exception as ex: self.logger.exception(ex) return {'allSeries': None, 'errorMsg': ex}