Skip to content

Environment

Environment

Methods:

Name Description
importModule

Imports the given module (if necessary) and returns a fake module object
install

Installs dependencies.
launch

Launch the environment, only available in ExternalEnvironment. Raises an exception in InternalEnvironment. See InternalEnvironment.launch and ExternalEnvironment.launch
executeCommands

Executes the given commands in this environment.
execute

Execute the given function in the given module. See ExternalEnvironment.execute and InternalEnvironment.execute
launched

Check if the environment is launched, important in ExternalEnvironment
exit

Exit the environment
Source code in wetlands/environment.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class Environment:
    modules: dict[str, ModuleType] = {}

    def __init__(self, name: str | None, environmentManager: "EnvironmentManager") -> None:
        self.name = name
        self.environmentManager = environmentManager

    def _isModFunction(self, mod, func):
        """Checks that func is a function defined in module mod"""
        return inspect.isfunction(func) and inspect.getmodule(func) == mod

    def _listFunctions(self, mod):
        """Returns the list of functions defined in module mod"""
        return [func.__name__ for func in mod.__dict__.values() if self._isModFunction(mod, func)]

    def _importModule(self, modulePath: Path | str):
        """Imports the given module (if necessary) and adds it to the module map."""
        modulePath = Path(modulePath)
        module = modulePath.stem
        if module not in self.modules:
            sys.path.append(str(modulePath.parent))
            self.modules[module] = import_module(module)
        return self.modules[module]

    def importModule(self, modulePath: Path | str) -> Any:
        """Imports the given module (if necessary) and returns a fake module object
        that contains the same methods of the module which will be executed within the environment."""
        module = self._importModule(modulePath)

        class FakeModule:
            pass

        for f in self._listFunctions(module):

            def fakeFunction(*args, _wetlands_imported_function=f, **kwargs):
                return self.execute(modulePath, _wetlands_imported_function, args, kwargs)

            setattr(FakeModule, f, fakeFunction)
        return FakeModule

    def install(self, dependencies: Dependencies, additionalInstallCommands: Commands = {}) -> list[str]:
        """Installs dependencies.
        See [`EnvironmentManager.create`][wetlands.environment_manager.EnvironmentManager.create] for more details on the ``dependencies`` and ``additionalInstallCommands`` parameters.

        Args:
                dependencies: Dependencies to install.
                additionalInstallCommands: Platform-specific commands during installation.
        Returns:
                Output lines of the installation commands.
        """
        return self.environmentManager.install(self.name, dependencies, additionalInstallCommands)

    @abstractmethod
    def launch(self, additionalActivateCommands: Commands = {}, logOutputInThread: bool = True) -> None:
        """Launch the environment, only available in [ExternalEnvironment][wetlands.external_environment.ExternalEnvironment]. Raises an exception in [InternalEnvironment][wetlands.internal_environment.InternalEnvironment]. See [`InternalEnvironment.launch`][wetlands.internal_environment.InternalEnvironment.launch] and [`ExternalEnvironment.launch`][wetlands.external_environment.ExternalEnvironment.launch]"""
        pass

    def executeCommands(
        self, commands: Commands, additionalActivateCommands: Commands = {}, popenKwargs: dict[str, Any] = {}
    ) -> subprocess.Popen:
        """Executes the given commands in this environment.

        Args:
                commands: The commands to execute in the environment.
                additionalActivateCommands: Platform-specific activation commands.
                popenKwargs: Keyword arguments for subprocess.Popen(). See [`EnvironmentManager.executeCommands`][wetlands.environment_manager.EnvironmentManager.executeCommands].

        Returns:
                The launched process.
        """
        return self.environmentManager.executeCommands(self.name, commands, additionalActivateCommands, popenKwargs)

    @abstractmethod
    def execute(self, modulePath: str | Path, function: str, args: tuple = (), kwargs: dict[str, Any] = {}) -> Any:
        """Execute the given function in the given module. See [`ExternalEnvironment.execute`][wetlands.external_environment.ExternalEnvironment.execute] and [`InternalEnvironment.execute`][wetlands.internal_environment.InternalEnvironment.execute]"""
        pass

    def _exit(self) -> None:
        """Exit the environment, important in ExternalEnvironment"""
        pass

    def launched(self) -> bool:
        """Check if the environment is launched, important in ExternalEnvironment"""
        return True

    def exit(self) -> None:
        """Exit the environment"""
        self._exit()
        self.environmentManager._removeEnvironment(self)

importModule(modulePath)

Imports the given module (if necessary) and returns a fake module object that contains the same methods of the module which will be executed within the environment.

Source code in wetlands/environment.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def importModule(self, modulePath: Path | str) -> Any:
    """Imports the given module (if necessary) and returns a fake module object
    that contains the same methods of the module which will be executed within the environment."""
    module = self._importModule(modulePath)

    class FakeModule:
        pass

    for f in self._listFunctions(module):

        def fakeFunction(*args, _wetlands_imported_function=f, **kwargs):
            return self.execute(modulePath, _wetlands_imported_function, args, kwargs)

        setattr(FakeModule, f, fakeFunction)
    return FakeModule

install(dependencies, additionalInstallCommands={})

Installs dependencies. See EnvironmentManager.create for more details on the dependencies and additionalInstallCommands parameters.

Parameters:

Name Type Description Default
dependencies Dependencies

Dependencies to install.

 required
additionalInstallCommands Commands

Platform-specific commands during installation.

 {}

Returns: Output lines of the installation commands.

Source code in wetlands/environment.py 
57
58
59
60
61
62
63
64
65
66
67
def install(self, dependencies: Dependencies, additionalInstallCommands: Commands = {}) -> list[str]:
    """Installs dependencies.
    See [`EnvironmentManager.create`][wetlands.environment_manager.EnvironmentManager.create] for more details on the ``dependencies`` and ``additionalInstallCommands`` parameters.

    Args:
            dependencies: Dependencies to install.
            additionalInstallCommands: Platform-specific commands during installation.
    Returns:
            Output lines of the installation commands.
    """
    return self.environmentManager.install(self.name, dependencies, additionalInstallCommands)

launch(additionalActivateCommands={}, logOutputInThread=True) abstractmethod

Launch the environment, only available in ExternalEnvironment. Raises an exception in InternalEnvironment. See InternalEnvironment.launch and ExternalEnvironment.launch

Source code in wetlands/environment.py 
69
70
71
72
@abstractmethod
def launch(self, additionalActivateCommands: Commands = {}, logOutputInThread: bool = True) -> None:
    """Launch the environment, only available in [ExternalEnvironment][wetlands.external_environment.ExternalEnvironment]. Raises an exception in [InternalEnvironment][wetlands.internal_environment.InternalEnvironment]. See [`InternalEnvironment.launch`][wetlands.internal_environment.InternalEnvironment.launch] and [`ExternalEnvironment.launch`][wetlands.external_environment.ExternalEnvironment.launch]"""
    pass

executeCommands(commands, additionalActivateCommands={}, popenKwargs={})

Executes the given commands in this environment.

Parameters:

Name Type Description Default
commands Commands

The commands to execute in the environment.

 required
additionalActivateCommands Commands

Platform-specific activation commands.

 {}
popenKwargs dict[str, Any]

Keyword arguments for subprocess.Popen(). See EnvironmentManager.executeCommands.

 {}

Returns:

Type Description
Popen

The launched process.
Source code in wetlands/environment.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def executeCommands(
    self, commands: Commands, additionalActivateCommands: Commands = {}, popenKwargs: dict[str, Any] = {}
) -> subprocess.Popen:
    """Executes the given commands in this environment.

    Args:
            commands: The commands to execute in the environment.
            additionalActivateCommands: Platform-specific activation commands.
            popenKwargs: Keyword arguments for subprocess.Popen(). See [`EnvironmentManager.executeCommands`][wetlands.environment_manager.EnvironmentManager.executeCommands].

    Returns:
            The launched process.
    """
    return self.environmentManager.executeCommands(self.name, commands, additionalActivateCommands, popenKwargs)

execute(modulePath, function, args=(), kwargs={}) abstractmethod

Execute the given function in the given module. See ExternalEnvironment.execute and InternalEnvironment.execute

Source code in wetlands/environment.py 
89
90
91
92
@abstractmethod
def execute(self, modulePath: str | Path, function: str, args: tuple = (), kwargs: dict[str, Any] = {}) -> Any:
    """Execute the given function in the given module. See [`ExternalEnvironment.execute`][wetlands.external_environment.ExternalEnvironment.execute] and [`InternalEnvironment.execute`][wetlands.internal_environment.InternalEnvironment.execute]"""
    pass

launched()

Check if the environment is launched, important in ExternalEnvironment

Source code in wetlands/environment.py 
 98
 99
100
def launched(self) -> bool:
    """Check if the environment is launched, important in ExternalEnvironment"""
    return True

exit()

Exit the environment

Source code in wetlands/environment.py 
102
103
104
105
def exit(self) -> None:
    """Exit the environment"""
    self._exit()
    self.environmentManager._removeEnvironment(self)