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).
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
# 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()
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))
# 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
# 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
# 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)
# 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
# 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
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]
# 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
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
# 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()