# ---LICENSE-BEGIN - DO NOT CHANGE OR MOVE THIS HEADER
# This file is part of the Neurorobotics Platform software
# Copyright (C) 2014,2015,2016,2017 Human Brain Project
# https://www.humanbrainproject.eu
#
# The Human Brain Project is a European Commission funded project
# in the frame of the Horizon2020 FET Flagship plan.
# http://ec.europa.eu/programmes/horizon2020/en/h2020-section/fet-flagships
#
# 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 2
# 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
# ---LICENSE-END
"""
This module represents the interfaces for the closed loop engine.
"""
from builtins import object
__author__ = 'LorenzoVannucci'
[docs]class BrainRuntimeException(Exception):
"""
Represents that a exception was raised from the execution of the step in the brain
control adapter.
"""
def __init__(self, message):
super(BrainRuntimeException, self).__init__(message)
[docs]class ForcedStopException(Exception):
"""
Represents that a closed loop has been forced to quit
"""
def __init__(self):
super(ForcedStopException, self).__init__("The simulation was forced to stop")
[docs]class BrainTimeoutException(Exception):
"""
Represents that a brain file could not be loaded due to a Timeout
"""
def __init__(self):
super(BrainTimeoutException, self).__init__(
"The brain cannot be imported due to timeout or an error in the brain file")
[docs]class IClosedLoopControl(object): # pragma: no cover
"""
Represents the closed loop engine synchronization mechanism.
"""
[docs] def load_brain(self, brain_file, brain_populations=None):
"""
Load (or reload) the brain model from a file the neuronal network file
:param brain_file: A python PyNN script or an h5 file
containing the neural network definition
:param brain_populations: A (optional) dictionary indexed by population names and
containing neuron indices. Neuron indices can be defined by
lists of integers or slices. Slices are either python slices or
dictionaries containing 'from', 'to' and 'step' values.
"""
raise NotImplementedError("Method not implemented")
[docs] def load_populations(self, populations):
"""
Load (or reload) populations into the brain
:param populations: A dictionary indexed by population names and
containing neuron indices. Neuron indices can be defined by a single integer,
list of integers or python slices. Python slices can be replaced by a
dictionary containing the 'from', 'to' and 'step' values.
"""
raise NotImplementedError("Method not implemented")
[docs] def initialize(self, brain_file, configuration): # -> None:
"""
Initializes the closed loop engine.
:param brain_file: A python PyNN script or an h5 file
containing the neural network definition
:param configuration: A set of populations
"""
raise NotImplementedError("Method not implemented")
@property
def is_initialized(self): # -> bool
"""
Returns True if the simulation is initialized, False otherwise.
"""
raise NotImplementedError("Method not implemented")
[docs] def run_step(self, timestep_s): # -> float64:
"""
Runs both simulations for the given time step in seconds.
:param timestep_s: The CLE time step in seconds
:return: Updated simulation time, otherwise -1
"""
raise NotImplementedError("Method not implemented")
[docs] def shutdown(self): # -> None:
"""
Shuts down both simulations.
"""
raise NotImplementedError("Method not implemented")
[docs] def start(self): # -> None:
"""
Starts the orchestrated simulations.
"""
raise NotImplementedError("Method not implemented")
[docs] def stop(self, forced=False, in_loop=False): # -> None:
"""
Stops the orchestrated simulations. Also waits for the current
simulation step to end.
:param forced: If set, the CLE instance cancels pending tasks
:param in_loop: If set, do not wait for multi-threading signals of pending tasks.
It must be set if and only if the function is called by the CLE main loop's thread.
"""
raise NotImplementedError("Method not implemented")
[docs] def reset(self): # -> None:
"""
Reset the orchestrated simulations.
"""
raise NotImplementedError("Method not implemented")
[docs] def reset_world(self, sdf_world_string):
"""
Reset the world to the configuration described by sdf_world_string.
"""
raise NotImplementedError("Method not implemented")
@property
def simulation_time(self): # -> float64
"""
Get the current simulation time.
"""
raise NotImplementedError("Method not implemented")
@property
def real_time(self): # -> float64
"""
Get the total execution time.
"""
raise NotImplementedError("Method not implemented")
[docs] def tf_elapsed_time(self):
"""
Gets the time share of the Transfer Functions
"""
raise NotImplementedError("Method not implemented")
[docs] def brainsim_elapsed_time(self):
"""
Gets the time share of the brain simulation
"""
raise NotImplementedError("Method not implemented")
[docs] def robotsim_elapsed_time(self):
"""
Gets the time share of the robot simulation
"""
raise NotImplementedError("Method not implemented")
[docs] def wait_step(self, timeout=None): # -> None
"""
Wait for the currently running simulation step to end.
:param timeout: The maximum amount of time (in seconds) to wait for the end of this step
"""
raise NotImplementedError("Method not implemented")