[execution][subprocess] Allow changing execution environment

This patch implements a new parameter that allows to pass
the path to the python executable of the environment one
would like to use to execute an experiment. The various
beat/beat.backend.python from that environment will be
called in place of the current environment as it is
by default.

Fixes #90

beat/core/execution/subprocess.py

@@ -123,6 +123,12 @@ class SubprocessExecutor(RemoteExecutor):
         guarantee that the cache is refreshed as appropriate in case the
         underlying libraries change.
 
+      custom_root_folders (dict): A dictionary mapping databases name and
+        their location on disk
+      ip_address (str): IP address of the machine to connect to for the database
+        execution and message handlers.
+      python_path (str): Path to the python executable of the environment to use
+        for experiment execution.
 
     Attributes:
 
@@ -172,8 +178,8 @@ class SubprocessExecutor(RemoteExecutor):
         library_cache=None,
         custom_root_folders=None,
         ip_address="127.0.0.1",
+        python_path=None,
     ):
-
         super(SubprocessExecutor, self).__init__(
             prefix,
             data,
@@ -186,14 +192,30 @@ class SubprocessExecutor(RemoteExecutor):
             custom_root_folders=custom_root_folders,
         )
 
-        # We need three apps to run this function: databases_provider and execute
-        self.EXECUTE_BIN = _which(os.path.join(os.path.dirname(sys.argv[0]), "execute"))
-        self.LOOP_EXECUTE_BIN = _which(
-            os.path.join(os.path.dirname(sys.argv[0]), "loop_execute")
-        )
-        self.DBPROVIDER_BIN = _which(
-            os.path.join(os.path.dirname(sys.argv[0]), "databases_provider")
-        )
+        if python_path is None:
+            base_path = os.path.dirname(sys.argv[0])
+
+            # We need three apps to run this function: databases_provider and execute
+            self.EXECUTE_BIN = _which(os.path.join(base_path, "execute"))
+            self.LOOP_EXECUTE_BIN = _which(os.path.join(base_path, "loop_execute"))
+            self.DBPROVIDER_BIN = _which(os.path.join(base_path, "databases_provider"))
+        else:
+            base_path = os.path.dirname(python_path)
+            self.EXECUTE_BIN = os.path.join(base_path, "execute")
+            self.LOOP_EXECUTE_BIN = os.path.join(base_path, "loop_execute")
+            self.DBPROVIDER_BIN = os.path.join(base_path, "databases_provider")
+
+        if any(
+            [
+                not os.path.exists(executable)
+                for executable in [
+                    self.EXECUTE_BIN,
+                    self.LOOP_EXECUTE_BIN,
+                    self.DBPROVIDER_BIN,
+                ]
+            ]
+        ):
+            raise RuntimeError("Invalid environment")
 
     def __create_db_process(self, configuration_name=None):
         databases_process = None

beat/core/test/test_execution.py

@@ -40,6 +40,9 @@ import os
 import glob
 import logging
 import nose.tools
+import subprocess as sp  # nosec
+
+from shutil import rmtree
 
 from ..experiment import Experiment
 from ..execution import LocalExecutor
@@ -106,6 +109,8 @@ class BaseExecutionMixIn(object):
         machine. It borrows some code from the package ``beat.cmdline``.
         """
 
+        executor_parameters = kwargs.pop("executor_parameters", {})
+
         dataformat_cache = {}
         database_cache = {}
         algorithm_cache = {}
@@ -140,6 +145,7 @@ class BaseExecutionMixIn(object):
                 dataformat_cache,
                 database_cache,
                 algorithm_cache,
+                **executor_parameters
             )
             nose.tools.assert_true(
                 executor.valid, "\n  * %s" % "\n  * ".join(executor.errors)
@@ -391,6 +397,7 @@ class TestLocalExecution(BaseExecutionMixIn):
         dataformat_cache,
         database_cache,
         algorithm_cache,
+        **kwargs
     ):
         return LocalExecutor(
             prefix,
@@ -426,14 +433,16 @@ class TestSubprocessExecution(BaseExecutionMixIn):
         dataformat_cache,
         database_cache,
         algorithm_cache,
+        python_path=None,
     ):
         return SubprocessExecutor(
-            prefix,
-            configuration,
-            tmp_prefix,
-            dataformat_cache,
-            database_cache,
-            algorithm_cache,
+            prefix=prefix,
+            data=configuration,
+            cache=tmp_prefix,
+            dataformat_cache=dataformat_cache,
+            database_cache=database_cache,
+            algorithm_cache=algorithm_cache,
+            python_path=python_path,
         )
 
     @slow
@@ -453,3 +462,56 @@ class TestSubprocessExecution(BaseExecutionMixIn):
         nose.tools.eq_(
             result["user_error"], "'Could not setup algorithm (returned False)'"
         )
+
+    def create_conda_environment(self, additional_packages=[]):
+        environment_name = "subprocess_environment"
+        environment_prefix = os.path.join(tmp_prefix, environment_name)
+        packages = ["python=3"] + additional_packages
+        sp.run(
+            [
+                "conda",
+                "create",
+                "-y",
+                "-c",
+                "defaults",
+                "-c",
+                "http://www.idiap.ch/software/bob/conda/",
+                "--prefix",
+                environment_prefix,
+            ]
+            + packages,
+            check=True,
+            stdout=sp.PIPE,
+            stderr=sp.PIPE,
+        )
+        return environment_prefix
+
+    def clear_conda_environment(self, environment_prefix):
+        rmtree(environment_prefix)
+
+    @slow
+    def test_different_environment(self):
+        environment_prefix = self.create_conda_environment(["beat.backend.python"])
+        result = self.execute(
+            "user/user/loop/1/loop",
+            [{"sum": 135, "nb": 9}, {"sum": 9, "nb": 9}],
+            executor_parameters={
+                "python_path": os.path.join(environment_prefix, "bin", "python")
+            },
+        )
+        self.clear_conda_environment(environment_prefix)
+
+        nose.tools.assert_is_none(result)
+
+    @slow
+    def test_wrong_different_environment(self):
+        environment_prefix = self.create_conda_environment()
+        with nose.tools.assert_raises(RuntimeError):
+            self.execute(
+                "user/user/loop/1/loop",
+                [{"sum": 135, "nb": 9}, {"sum": 9, "nb": 9}],
+                executor_parameters={
+                    "python_path": os.path.join(environment_prefix, "bin", "python")
+                },
+            )
+        self.clear_conda_environment(environment_prefix)