Source code for tvb.core.entities.file.simulator.simulation_history_h5

# -*- 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:

from tvb.basic.neotraits.api import HasTraits, NArray, Int, List, Attr, Float
from tvb.core.neotraits.h5 import H5File, DataSet, Scalar, Json
from tvb.simulator.integrators import IntegratorStochastic

[docs]class SimulationHistory(HasTraits): """ Simulation State, prepared for H5 file storage. """ history = NArray(required=False) # State array, all state variables current_state = NArray(required=False) # Simulator's current step number (in time) current_step = Int() # Array with _stock array for every monitor configured in current simulation. # As the monitors are dynamic, we prepare a bunch of arrays for storage in H5 file. monitor_stock_1 = NArray(required=False) monitor_stock_2 = NArray(required=False) monitor_stock_3 = NArray(required=False) monitor_stock_4 = NArray(required=False) monitor_stock_5 = NArray(required=False) monitor_stock_6 = NArray(required=False) monitor_stock_7 = NArray(required=False) monitor_stock_8 = NArray(required=False) monitor_stock_9 = NArray(required=False) monitor_stock_10 = NArray(required=False) monitor_stock_11 = NArray(required=False) monitor_stock_12 = NArray(required=False) monitor_stock_13 = NArray(required=False) monitor_stock_14 = NArray(required=False) monitor_stock_15 = NArray(required=False) # For matching the stocks above on reload monitor_names = List(of=str) # In case of noisy integrator, remember the Random State generator status integrator_noise_rng_state_algo = Attr(field_type=str, required=False) integrator_noise_rng_state_keys = NArray(dtype='uint32', required=False) integrator_noise_rng_state_pos = Int(required=False) integrator_noise_rng_state_has_gauss = Int(required=False) integrator_noise_rng_state_cached_gauss = Float(required=False) def __init__(self, **kwargs): """ Constructor for Simulator State """ super(SimulationHistory, self).__init__(**kwargs) self.visible = False
[docs] def populate_from(self, simulator_algorithm): """ Prepare a state for storage from a Simulator object. """ self.history = simulator_algorithm.history.buffer.copy() self.current_step = simulator_algorithm.current_step self.current_state = simulator_algorithm.current_state monitor_names = [] for i, monitor in enumerate(simulator_algorithm.monitors): field_name = "monitor_stock_" + str(i + 1) setattr(self, field_name, monitor._stock) monitor_names.append(type(monitor).__name__) if isinstance(simulator_algorithm.integrator, IntegratorStochastic): rng_state = simulator_algorithm.integrator.noise.random_stream.get_state() self.integrator_noise_rng_state_algo = rng_state[0] self.integrator_noise_rng_state_keys = rng_state[1] self.integrator_noise_rng_state_pos = rng_state[2] self.integrator_noise_rng_state_has_gauss = rng_state[3] self.integrator_noise_rng_state_cached_gauss = rng_state[4]
[docs] def fill_into(self, simulator_algorithm): """ Populate a Simulator object from current stored-state. """ simulator_algorithm.history.initialize(self.history) simulator_algorithm.current_step = self.current_step simulator_algorithm.current_state = self.current_state for i, monitor in enumerate(simulator_algorithm.monitors): monitor._stock = getattr(self, "monitor_stock_" + str(i + 1)) if self.integrator_noise_rng_state_algo is not None: rng_state = ( self.integrator_noise_rng_state_algo, self.integrator_noise_rng_state_keys, self.integrator_noise_rng_state_pos, self.integrator_noise_rng_state_has_gauss, self.integrator_noise_rng_state_cached_gauss ) simulator_algorithm.integrator.noise.random_stream.set_state(rng_state)
[docs]class SimulationHistoryH5(H5File): def __init__(self, path): super(SimulationHistoryH5, self).__init__(path) self.history = DataSet(SimulationHistory.history, self) self.current_state = DataSet(SimulationHistory.current_state, self) self.current_step = Scalar(SimulationHistory.current_step, self) self.monitor_names = Json(SimulationHistory.monitor_names, self) for i in range(1, 16): stock_name = 'monitor_stock_%i' % i setattr(self, stock_name, DataSet(getattr(SimulationHistory, stock_name), self)) self.integrator_noise_rng_state_algo = Scalar(SimulationHistory.integrator_noise_rng_state_algo, self) self.integrator_noise_rng_state_keys = DataSet(SimulationHistory.integrator_noise_rng_state_keys, self) self.integrator_noise_rng_state_pos = Scalar(SimulationHistory.integrator_noise_rng_state_pos, self) self.integrator_noise_rng_state_has_gauss = Scalar(SimulationHistory.integrator_noise_rng_state_has_gauss, self) self.integrator_noise_rng_state_cached_gauss = Scalar(SimulationHistory.integrator_noise_rng_state_cached_gauss, self)