From 74cc8cf6f0246d7d96c7e7ff7fac49a103cb0a7c Mon Sep 17 00:00:00 2001 From: Eloy Retamino Date: Mon, 2 Dec 2019 12:05:17 +0000 Subject: [PATCH 1/3] Merged in NRRPLT-7700-tf-execution-order (pull request #66) [NRRPLT-7700] Implemented method to run tfs in order using their order attribute and using it in CLE run step * [NRRPLT-7700] Implemented method to run tfs in order using their order attribute and using it in CLE run step * [NRRPLT-7700] fixed bug when order is specified for no tf * [NRRPLT-7700] Added tests * [NRRPLT-7700] Added missing params to docstring * [NRRPLT-7700] Reversed order of tf execution. This makes more sense since it allows to reserve 0 for tfs withouth defined order and always allows the user to increase the execution order or a tf without having to change all the other tf order attributes * [NRRPLT-7700] Corrected unit test * [NRRPLT-7700] Fixed unsafe int() operation Approved-by: Michael Zechmair Approved-by: Ugo Albanese --- .../cle/DeterministicClosedLoopEngine.py | 6 ++- .../_MockTransferFunctionManager.py | 10 ++++ .../tests/tf_framework/test_tf_ok.py | 46 +++++++++++++++++++ .../tf_framework/_TransferFunction.py | 44 ++++++++++++++++++ .../tf_framework/_TransferFunctionManager.py | 25 +++++++++- .../hbp_nrp_cle/tf_framework/__init__.py | 7 ++- 6 files changed, 134 insertions(+), 4 deletions(-) diff --git a/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py b/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py index c1b5584..72a743b 100755 --- a/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py +++ b/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py @@ -224,8 +224,10 @@ def run_step(self, timestep): # transfer functions logger.debug("Run step: Transfer functions") - self.tfm.run_robot_to_neuron(clk) - self.tfm.run_neuron_to_robot(clk) + + # self.tfm.run_robot_to_neuron(clk) + # self.tfm.run_neuron_to_robot(clk) + self.tfm.run_tfs(clk) # update clock cle.clock += timestep diff --git a/hbp_nrp_cle/hbp_nrp_cle/mocks/tf_framework/_MockTransferFunctionManager.py b/hbp_nrp_cle/hbp_nrp_cle/mocks/tf_framework/_MockTransferFunctionManager.py index 5b5b40e..8129b7d 100644 --- a/hbp_nrp_cle/hbp_nrp_cle/mocks/tf_framework/_MockTransferFunctionManager.py +++ b/hbp_nrp_cle/hbp_nrp_cle/mocks/tf_framework/_MockTransferFunctionManager.py @@ -86,6 +86,16 @@ def run_robot_to_neuron(self, t): self.__r2nTimes.append(t) time.sleep(self.__sleepTime) + def run_tfs(self, t): + """ + Runs all the transfer functions mocks + + :param t: The simulation time + """ + self.__r2nTimes.append(t) + self.__n2rTimes.append(t) + time.sleep(self.__sleepTime) + @property def name(self): """ diff --git a/hbp_nrp_cle/hbp_nrp_cle/tests/tf_framework/test_tf_ok.py b/hbp_nrp_cle/hbp_nrp_cle/tests/tf_framework/test_tf_ok.py index 5a457f6..7f2975d 100644 --- a/hbp_nrp_cle/hbp_nrp_cle/tests/tf_framework/test_tf_ok.py +++ b/hbp_nrp_cle/hbp_nrp_cle/tests/tf_framework/test_tf_ok.py @@ -117,6 +117,44 @@ def transform_camera(t, camera, camera_device): self.assertIn(expected_source_r2n, loaded_source_n2r_and_r2n) self.assertEqual(2, len(loaded_source_n2r_and_r2n)) + def test_run_tfs(self): + nrp.start_new_tf_manager() + brain = MockBrainCommunicationAdapter() + robot = MockRobotCommunicationAdapter() + config.active_node.brain_adapter = brain + config.active_node.robot_adapter = robot + + tf_1 = """@nrp.MapVariable('shared_list', initial_value=[], scope=nrp.GLOBAL) +@nrp.Neuron2Robot() +def first_tf(t, shared_list): + shared_list.value.append('first_tf') +""" + + tf_2 = """@nrp.MapVariable('shared_list', initial_value=[], scope=nrp.GLOBAL) +@nrp.Neuron2Robot() +def second_tf(t, shared_list): + shared_list.value.append('second_tf') +""" + + tf_3 = """@nrp.MapVariable('shared_list', initial_value=[], scope=nrp.GLOBAL) +@nrp.Neuron2Robot() +def third_tf(t, shared_list): + shared_list.value.append('third_tf') +""" + # add the TFs + nrp.set_transfer_function(tf_2, tf_2, 'second_tf', activation=True, priority=1) + nrp.set_transfer_function(tf_3, tf_3, 'third_tf', activation=True) + nrp.set_transfer_function(tf_1, tf_1, 'first_tf', activation=True, priority=2) + + config.active_node.run_tfs(1.0) + + # check that the global variable has been manipulated as expected + shared_list = config.active_node.global_data['shared_list'] + + self.assertTrue(shared_list[0] == "first_tf") + self.assertTrue(shared_list[1] == "second_tf") + self.assertTrue(shared_list[2] == "third_tf") + def test_tf_set(self): nrp.start_new_tf_manager() brain = MockBrainCommunicationAdapter() @@ -185,6 +223,14 @@ def transform_camera(t, camera, camera_device): self.assertIn('line 4', e.message) self.assertEqual('it_cant_work', e.tf_name) + nrp.delete_transfer_function('right_arm') + nrp.set_transfer_function(tf_n2r, tf_n2r, 'right_arm') + tf = nrp.get_transfer_function('right_arm') + self.assertTrue(hasattr(tf, 'priority') and tf.priority == 0) + nrp.set_transfer_function(tf_n2r, tf_n2r, 'right_arm', priority=1) + nrp.set_transfer_function(tf_n2r, tf_n2r, 'right_arm') + self.assertTrue(tf.priority == 1) + @patch('hbp_nrp_cle.tf_framework.CSVRecorder.cleanup') def test_tf_delete(self, mock_cleanup): diff --git a/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunction.py b/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunction.py index e1121df..0b80158 100644 --- a/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunction.py +++ b/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunction.py @@ -60,6 +60,7 @@ def __init__(self, triggers=None, throttling_rate=None): self._params = [] self._func = None self.__active = False + self.__priority = 0 self.__local_data = {} self.__source = None self.__elapsed_time = 0.0 @@ -201,6 +202,31 @@ def active(self, bool_value): if bool_value is not None and type(bool_value) == bool: self.__active = bool_value + @property + def priority(self): + """ + Gets the execution priority of this transfer function + + :return: the execution priority of this transfer function + + """ + + return self.__priority + + @priority.setter + def priority(self, priority_value): + """ + Sets the execution priority of this transfer function. + + :param priority_value: positive integer denoting the execution priority of this transfer + function + + """ + try: + self.__priority = int(priority_value) + except (ValueError, TypeError): + self.__priority = 0 + @abstractmethod def __call__(self, func): """ @@ -348,6 +374,24 @@ def active(self, value): """ pass + @property + def priority(self): + """ + A flawed transfer function can't be executed. + + :return: False + + """ + return None + + # pylint: disable=no-self-use + @priority.setter + def priority(self, value): + """ + Since a flawed transfer function can't be executed, it doesn't do anything. + """ + pass + @property def params(self): """ diff --git a/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunctionManager.py b/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunctionManager.py index 13c5014..83e2c61 100644 --- a/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunctionManager.py +++ b/hbp_nrp_cle/hbp_nrp_cle/tf_framework/_TransferFunctionManager.py @@ -58,6 +58,7 @@ def __init__(self): # -> None: self.__nestAdapter = None self.__initialized = False self.__global_data = {} + self.__tf_priority = lambda tf: tf.priority if tf.priority else 0 @property def n2r(self): # -> list: @@ -135,6 +136,24 @@ def run_robot_to_neuron(self, t): # -> None: self.__flawed.append(FlawedTransferFunction(_r2n.name, _r2n.source, tf_exception)) self.__r2n.remove(_r2n) + def run_tfs(self, t): + """ + Runs all the transfer functions in the order specified in bibi file + + :param t: The simulation time + """ + _tfs = self.transfer_functions(sorted_=True) + + for tf in _tfs: + try: + TransferFunctionManager.run_tf(tf, t) + except TFRunningException as tf_exception: + self.__flawed.append(FlawedTransferFunction(tf.name, tf.source, tf_exception)) + if tf in self.__n2r: + self.__n2r.remove(tf) + elif tf in self.__r2n: + self.__r2n.remove(tf) + @property def robot_adapter(self): # -> IRobotCommunicationAdapter: """ @@ -188,15 +207,19 @@ def brain_adapter(self, nest_adapter): # -> None: raise Exception("The given object is not a valid brain communication adapter") self.__nestAdapter = nest_adapter - def transfer_functions(self, flawed=False): + def transfer_functions(self, flawed=False, sorted_=False): """ Gets a list of transfer functions managed by this instance :param flawed: if True the list will include also flawed TFs + :param sorted_: if True the list is sorted by transfer function priority attribute. + Flawed TFs, if any, come last. :return: A list of transfer functions """ proper_tfs = self.__n2r + self.__r2n + self.__silent + if sorted_: + proper_tfs.sort(key=self.__tf_priority, reverse=True) return proper_tfs if not flawed else proper_tfs + self.__flawed diff --git a/hbp_nrp_cle/hbp_nrp_cle/tf_framework/__init__.py b/hbp_nrp_cle/hbp_nrp_cle/tf_framework/__init__.py index 20b1132..c39c422 100755 --- a/hbp_nrp_cle/hbp_nrp_cle/tf_framework/__init__.py +++ b/hbp_nrp_cle/hbp_nrp_cle/tf_framework/__init__.py @@ -429,13 +429,16 @@ def delete_flawed_transfer_function(name): return result -def set_transfer_function(new_source, new_code, new_name, activation=True): +def set_transfer_function(new_source, new_code, new_name, activation=True, priority=None): """ Apply transfer function changes made by a client :param new_source: Transfer function's updated source :param new_code: Compiled code of the updated source :param new_name: Transfer function's updated name + :param activation: Activation state of the transfer function + :param priority: execution order of the transfer function. Transfer functions with higher + priority are executed first. """ # pylint: disable=broad-except @@ -457,6 +460,8 @@ def set_transfer_function(new_source, new_code, new_name, activation=True): # indeed inspect.getsource is based on a source file object # see findsource in http://www.opensource.apple.com/source/python/python-3/python/Lib/inspect.py tf.source = new_source + if priority is not None: + tf.priority = priority def set_flawed_transfer_function(source, name="NO_NAME", error=None): From a9e54fed63fe40108d132e5a246768b4410c9908 Mon Sep 17 00:00:00 2001 From: Michael Zechmair Date: Tue, 17 Dec 2019 16:03:13 +0000 Subject: [PATCH 2/3] IBA --- .../cle/DeterministicClosedLoopEngine.py | 6 ++ .../hbp_nrp_cle/externalsim/AsyncEmaCall.py | 27 +++++++ .../hbp_nrp_cle/externalsim/ExternalModule.py | 74 ++++++++++++++++++ .../externalsim/ExternalModuleManager.py | 76 +++++++++++++++++++ .../hbp_nrp_cle/externalsim/__init__.py | 5 ++ 5 files changed, 188 insertions(+) create mode 100644 hbp_nrp_cle/hbp_nrp_cle/externalsim/AsyncEmaCall.py create mode 100755 hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModule.py create mode 100644 hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModuleManager.py create mode 100755 hbp_nrp_cle/hbp_nrp_cle/externalsim/__init__.py diff --git a/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py b/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py index 72a743b..726d092 100755 --- a/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py +++ b/hbp_nrp_cle/hbp_nrp_cle/cle/DeterministicClosedLoopEngine.py @@ -60,6 +60,7 @@ def __init__(self, brain_control_adapter, brain_comm_adapter, transfer_function_manager, + external_module_array, dt ): """ @@ -78,6 +79,7 @@ def __init__(self, self.bca = brain_control_adapter self.bcm = brain_comm_adapter self.tfm = transfer_function_manager + self.ema = external_module_array # default timestep self.timestep = dt @@ -130,6 +132,7 @@ def initialize(self, brain_file=None, **configuration): self.rca.initialize() self.bca.initialize() self.tfm.initialize('tfnode') + self.ema.initialize() cle.clock = 0.0 self.start_time = 0.0 self.elapsed_time = 0.0 @@ -229,6 +232,8 @@ def run_step(self, timestep): # self.tfm.run_neuron_to_robot(clk) self.tfm.run_tfs(clk) + self.ema.run_step() + # update clock cle.clock += timestep @@ -245,6 +250,7 @@ def shutdown(self): self.bcm.shutdown() self.rca.shutdown() self.bca.shutdown() + self.ema.shutdown() def start(self): """ diff --git a/hbp_nrp_cle/hbp_nrp_cle/externalsim/AsyncEmaCall.py b/hbp_nrp_cle/hbp_nrp_cle/externalsim/AsyncEmaCall.py new file mode 100644 index 0000000..b8291b8 --- /dev/null +++ b/hbp_nrp_cle/hbp_nrp_cle/externalsim/AsyncEmaCall.py @@ -0,0 +1,27 @@ +from __future__ import print_function + +import rospy +from concurrent.futures import ThreadPoolExecutor + +class AsyncServiceProxy(object): + + def __init__(self, service_name, service_type, persistent=True, + headers=None, callback=None): + """Create an asynchronous service proxy.""" + + self.executor = ThreadPoolExecutor(max_workers=1) + self.service_proxy = rospy.ServiceProxy( + service_name, + service_type, + persistent, + headers) + self.callback = callback + + def __call__(self, *args, **kwargs): + """Get a Future corresponding to a call of this service.""" + + fut = self.executor.submit(self.service_proxy.call, *args, **kwargs) + if self.callback is not None: + fut.add_done_callback(self.callback) + + return fut \ No newline at end of file diff --git a/hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModule.py b/hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModule.py new file mode 100755 index 0000000..a1f7169 --- /dev/null +++ b/hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModule.py @@ -0,0 +1,74 @@ +""" +ExternalModule.py includes the corresponding CLE class for external ROS modules. +""" + +__author__ = 'Omer Yilmaz' + +import os +import threading +from multiprocessing import Value +import logging +import rospy +from hbp_nrp_cle.externalsim.AsyncEmaCall import AsyncServiceProxy +from cle_ros_msgs.srv import Initialize, RunStep, RunStepRequest, Shutdown + +logger = logging.getLogger('hbp_nrp_cle') + + +class ExternalModule(object): + """ + External ROS modules have initialize, run_step and shutdown methods. + This class has the corresponding initialize, run_step and shutdown + methods which triggers the external ones through ROS service proxies. + Objects of this class is synchronized with the Deterministic Closed + Loop Engine via ExternalModuleManager. + """ + + def __init__(self, module_name): + self.service_name = 'emi/' + module_name + '_module/' + self.resp = None + + rospy.wait_for_service(self.service_name + 'initialize') + self.initialize_proxy = AsyncServiceProxy( + self.service_name + 'initialize', Initialize, persistent=False) + + rospy.wait_for_service(self.service_name + 'run_step') + self.run_step_proxy = AsyncServiceProxy( + self.service_name + 'run_step', RunStep, persistent=True) + + rospy.wait_for_service(self.service_name + 'shutdown') + self.shutdown_proxy = AsyncServiceProxy( + self.service_name + 'shutdown', Shutdown, persistent=False) + + def initialize(self): + """ + This method triggers the initialize method served at the external module synchronously + with the CLE. + """ + try: + self.resp = self.initialize_proxy() + return self.resp + except rospy.ServiceException as e: + logger.exception(self.service_name + 'initialize call failed: %s' % e) + + def run_step(self): + """ + This method triggers the run_step method served at the external module synchronously + with the CLE. + """ + try: + fut = self.run_step_proxy() + return fut + except rospy.ServiceException as e: + logger.exception(self.service_name + 'run_step call failed: %s' % e) + + def shutdown(self): + """ + This method triggers the shutdown method served at the external module synchronously + with the CLE. + """ + try: + fut = self.shutdown_proxy() + return fut + except rospy.ServiceException as e: + logger.exception(self.service_name + 'shutdown call failed: %s' % e) diff --git a/hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModuleManager.py b/hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModuleManager.py new file mode 100644 index 0000000..f87fb25 --- /dev/null +++ b/hbp_nrp_cle/hbp_nrp_cle/externalsim/ExternalModuleManager.py @@ -0,0 +1,76 @@ +""" +The manager for the external modules which extend the NRP through ROS launch +mechanism. +""" + +__author__ = 'Omer Yilmaz' + +import concurrent.futures +from multiprocessing import Process, Pool, cpu_count +import re +import rosservice +from hbp_nrp_cle.externalsim.ExternalModule import ExternalModule + + +class ExternalModuleManager(object): + """ + This class automatically detects the external modules searching the ROS + services available at the ROS server. It keeps and array of the external + modules and calls initialize, run_step ans shutdown methods for each + external module. One object of this class is used by the Deterministic + Closed Loop Engine and is synchronized with it making every external module + on the array also synchronized. + """ + + def __init__(self): + self.module_names = [] + for service in rosservice.get_service_list(): + m = re.match(r"/emi/.*/initialize", str(service)) + if m: + module_name = m.group(0)[5:-18] + self.module_names.append(module_name) + + self.ema = [] + if len(self.module_names) is not 0: + with concurrent.futures.ThreadPoolExecutor(max_workers=max(len(self.module_names), 1)) as executor: + future_results = [executor.submit(ExternalModule, x) for x in self.module_names] + concurrent.futures.wait(future_results) + for future in future_results: + self.ema.append(future.result()) + + + def initialize(self): + """ + This method is used to run all initialize methods served at each external models at once. + """ + if len(self.module_names) is not 0: + with concurrent.futures.ThreadPoolExecutor(max_workers=len(self.ema)) as executor: + future_results = [executor.submit(x.initialize) for x in self.ema] + concurrent.futures.wait(future_results) + for future in future_results: + while not future.result().done(): + pass + + def run_step(self): + """ + This method is used to run all run_step methods served at each external models at once. + """ + if len(self.module_names) is not 0: + with concurrent.futures.ThreadPoolExecutor(max_workers=len(self.ema)) as executor: + future_results = [executor.submit(x.run_step) for x in self.ema] + concurrent.futures.wait(future_results) + for future in future_results: + while not future.result().done(): + pass + + def shutdown(self): + """ + This method is used to run all shutdown methods served at each external models at once. + """ + with concurrent.futures.ThreadPoolExecutor(max_workers=len(self.ema)) as executor: + future_results = [executor.submit(x.shutdown) for x in self.ema] + concurrent.futures.wait(future_results) + for future in future_results: + while not future.result().done(): + pass + diff --git a/hbp_nrp_cle/hbp_nrp_cle/externalsim/__init__.py b/hbp_nrp_cle/hbp_nrp_cle/externalsim/__init__.py new file mode 100755 index 0000000..db959ab --- /dev/null +++ b/hbp_nrp_cle/hbp_nrp_cle/externalsim/__init__.py @@ -0,0 +1,5 @@ +""" +CLE part of the External Ros Modules API version 2.0.0 +""" + +__author__ = 'Omer Yilmaz' From e24f280fdc54f0832d12e47defabbc772f5c7c4c Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Tue, 17 Dec 2019 16:18:18 +0000 Subject: [PATCH 3/3] Bump pyyaml from 3.11 to 5.1 in /hbp_nrp_cle Bumps [pyyaml](https://github.com/yaml/pyyaml) from 3.11 to 5.1. - [Release notes](https://github.com/yaml/pyyaml/releases) - [Changelog](https://github.com/yaml/pyyaml/blob/master/CHANGES) - [Commits](https://github.com/yaml/pyyaml/compare/3.11...5.1) Signed-off-by: dependabot[bot] --- hbp_nrp_cle/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) mode change 100755 => 100644 hbp_nrp_cle/requirements.txt diff --git a/hbp_nrp_cle/requirements.txt b/hbp_nrp_cle/requirements.txt old mode 100755 new mode 100644 index 56b8907..89a4b8c --- a/hbp_nrp_cle/requirements.txt +++ b/hbp_nrp_cle/requirements.txt @@ -7,7 +7,7 @@ h5py==2.6.0 lazyarray==0.2.9 neo==0.5.2 PyNN==0.9.1 -PyYAML==3.11 +PyYAML==5.1 rospkg==1.0.38 catkin_pkg==0.2.10 progressbar2==3.34.0