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
tvb_client = TVBClient("http://192.168.123.93:9090")
keycloak_instance = KeycloakOpenID("https://keycloak.codemart.ro/auth/", "TVB", "tvb-tests")
tvb_client._update_token(keycloak_instance.token("tvb_user", "pass"))
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
# Prepare Fourier model
fourier_model = FFTAdapterModel()
fourier_model.time_series = time_series_gid
fourier_model.window_function = '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()