Interract with TVB REST server using the TVBClient API

TVB REST server is part of tvb-framework release. If you want to run this example you will have to start a REST server or you should have access to a public TVB REST server.

Note: TVB REST server is part of the tvb-framework release. If you want to run this example, you will have to start a REST server or you should have access to a public TVB REST server.

WARNING: Your jupyter notebook server probably runs on localhost:8888 which is used by TVBClient. You should run jupyter on a different port (e.g. jupyter notebook --port 9999).

In [ ]:
import os
import time

from keycloak import KeycloakOpenID
from tvb.adapters.datatypes.db.connectivity import ConnectivityIndex
from tvb.adapters.simulator.simulator_adapter import SimulatorAdapterModel
from tvb.core.entities.model.model_operation import STATUS_ERROR, STATUS_CANCELED, STATUS_FINISHED
from tvb.interfaces.rest.client.tvb_client import TVBClient
In [ ]:
# Enter the URL to your REST server
tvb_client = TVBClient("http://localhost:9090")

keycloak_instance = KeycloakOpenID("https://keycloak.codemart.ro/auth/", "TVB", "tvb-tests")

tvb_client._update_token(keycloak_instance.token("tvb_user", "pass"))
tvb_client.is_data_encrypted = tvb_client.datatype_api.is_data_encrypted()
In [ ]:
def monitor_operation(tvb_client, operation_gid):
    while True:
        status = tvb_client.get_operation_status(operation_gid)
        if status in [STATUS_FINISHED, STATUS_CANCELED, STATUS_ERROR]:
            break
        print("Current operation GID {} => STATUS {}".format(operation_gid, status))
        time.sleep(5)
    print("Operation {} has finished with status: {}".format(operation_gid, status))
In [ ]:
# Requesting projects for the logged user
projects_of_user = tvb_client.get_project_list()
default_project = projects_of_user[0]
default_project_gid = default_project.gid
default_project.name
In [ ]:
# Requesting datatypes for the default project
data_in_project = tvb_client.get_data_in_project(default_project_gid)
connectivity_gid = None
datatypes_type = []
for datatype in data_in_project:
    datatypes_type.append(datatype.type)
    if datatype.type == ConnectivityIndex().display_type:
        connectivity_gid = datatype.gid

datatypes_type
In [ ]:
# Preparing the simulator
simulator = SimulatorAdapterModel()
simulator.connectivity = connectivity_gid
simulator.simulation_length = 100

# Starting the simulation
operation_gid = tvb_client.fire_simulation(default_project_gid, simulator)

# Monitoring the simulation operation
monitor_operation(tvb_client, operation_gid)
In [ ]:
# Requesting the results of the simulation
simulation_results = tvb_client.get_operation_results(operation_gid)
datatype_names = []
for datatype in simulation_results:
    datatype_names.append(datatype.name)
    
datatype_names
In [ ]:
# Download the resulted time series file

time_series_gid = simulation_results[1].gid
time_series_path = tvb_client.retrieve_datatype(time_series_gid, os.getcwd())

time_series_path
In [ ]:
from tvb.adapters.datatypes.h5.time_series_h5 import TimeSeriesH5

# Loading a chuck from the time series H5 file, as this can be very large
with TimeSeriesH5(time_series_path) as time_series_h5:
    data_shape = time_series_h5.read_data_shape()
    chunk = time_series_h5.read_data_slice(tuple([slice(20), slice(data_shape[1]), slice(data_shape[2]), slice(data_shape[3])]))

assert chunk.shape[0] == 20
assert chunk.shape[1] == data_shape[1]
assert chunk.shape[2] == data_shape[2]
assert chunk.shape[3] == data_shape[3]
In [ ]:
# Requesting algorithms to run on time series
algos = tvb_client.get_operations_for_datatype(time_series_gid)
algos_pp = [(algo.module+'.'+algo.classname) for algo in algos]

algos_pp
In [ ]:
from tvb.adapters.analyzers.fourier_adapter import FFTAdapterModel, FourierAdapter
from tvb.datatypes.spectral import WindowingFunctionsEnum

# Prepare Fourier model
fourier_model = FFTAdapterModel()
fourier_model.time_series = time_series_gid
fourier_model.window_function = WindowingFunctionsEnum.HAMMING

# Launch Fourier Analyzer
operation_gid = tvb_client.launch_operation(default_project_gid, FourierAdapter, fourier_model)

operation_gid
In [ ]:
# Download the connectivity file
connectivity_path = tvb_client.retrieve_datatype(connectivity_gid, os.getcwd())
print("Connectivity path: {}".format(connectivity_path))

# Loading an entire Connectivity datatype in memory
connectivity = tvb_client.load_datatype_from_file(connectivity_path)
connectivity.summary_info()