File: C:/Users/fred/anaconda3/Lib/site-packages/anaconda_navigator/utils/launch.py
# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright (c) 2016-2017 Anaconda, Inc.
#
# May be copied and distributed freely only as part of an Anaconda or
# Miniconda installation.
# -----------------------------------------------------------------------------
"""Launch applications utilities."""
import datetime
import itertools
import os
import posixpath
import subprocess # nosec
import sys
import typing
from anaconda_navigator.config import CONF_PATH, LAUNCH_SCRIPTS_PATH, WIN
from anaconda_navigator.utils.programs import is_program_installed
from anaconda_navigator.utils.win_elevate import run_as_admin
if WIN:
import ctypes
DEVNULL: typing.Final[str] = 'nul' if (sys.platform == 'win32') else '/dev/null'
LOG_FILE: typing.Final[str] = '{package}-{level}-{index}.txt'
class CommandDetails(typing.TypedDict):
"""Dictionary with extra details about executed command."""
args: str
id: int # pylint: disable=invalid-name
cmd: str
stdout: str
stderr: str
class GetCommand(typing.Protocol): # pylint: disable=too-few-public-methods
"""Common interface for all `get_command_*` functions."""
def __call__( # pylint: disable=too-many-arguments
self,
root_prefix: str,
prefix: str,
command: str,
extra_arguments: typing.Iterable[typing.Any] = (),
package_name: str = 'app',
environment: typing.Optional[typing.Mapping[str, str]] = None,
cwd: str = os.path.expanduser('~'),
default_scripts_path: str = LAUNCH_SCRIPTS_PATH,
) -> typing.Tuple[typing.Mapping[str, typing.Any], CommandDetails]:
"""
Generate script to launch application and return path to it.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param command: Actual command to launch application.
:param extra_arguments: Additional arguments to attach to command.
:param cwd: Custom working directory to launch application in.
:param package_name: Name of the conda package, or alias of the external application.
:param environment: Custom environment to launch application in.
:param default_scripts_path: Root directory to store launch scripts in.
:return: Path to generated launch script file.
"""
def to_posix_path(value: str) -> str:
"""Convert path to a posix-compatible."""
return value.replace(os.path.sep, posixpath.sep)
def safe_unix(value: str) -> str:
"""Prepare argument which is safe to use in unix command line."""
to_replace: typing.Final[typing.Set[str]] = {'"', '$', '\\', '`'}
to_escape: typing.Final[typing.Set[str]] = {' ', '&', ';', '<', '>', '|'} | to_replace
if set(value) & to_escape:
character: str
for character in to_replace:
value = value.replace(character, f'\\{character}')
value = f'"{value}"'
return value
def safe_windows(value: str) -> str:
"""Prepare argument which is safe to use in windows command line."""
to_escape: typing.Final[typing.Set[str]] = {' ', '%', '"', '&', ';', '<', '>', '|'}
if set(value) & to_escape:
value = value.replace('"', '""').replace('%', '"^%"')
value = f'"{value}"'
return value
def safe_argument(value: str) -> str:
"""Prepare argument which is safe to use in command line."""
if WIN:
return safe_windows(value)
return safe_unix(value)
def expand_environment(
command: str,
environment: typing.Mapping[str, typing.Any],
default: typing.Any = None,
recursive: bool = False,
) -> str:
"""
Update command with environment variables.
:param command: Original command to update with environment variable values.
:param environment: Environment variables to inject into command.
:code:`None` values won't be expanded.
:param default: Value to insert for all unknown variables in `command`.
:param recursive: Allow recursive expand of the values.
:return: String with expanded environment.
"""
cursor: int = 0
while True:
start: int = command.find('${', cursor)
if start < 0:
break
stop: int = command.find('}', start)
environment_key: str = command[start + 2:stop]
stop += 1
environment_value: typing.Any = environment.get(environment_key, default)
if environment_value is None:
cursor = stop
continue
environment_value = str(environment_value)
command = command[:start] + environment_value + command[stop:]
cursor = start
if not recursive:
cursor += len(environment_value)
return command
class RunningProcess:
"""
Minimal description of a process launched from the Navigator.
Common example of such process - anything started from the Home page tiles.
:param package: Name of the Conda package, or alias of a launched application.
:param process: :class:`~subprocess.Popen` instance, which is used to launch the application.
:param stdout: Path to a file with captured `stdout`.
:param stderr: Path to a file with captured `stderr`.
"""
__slots__ = ('__birth', '__package', '__process', '__return_code', '__stderr', '__stdout')
def __init__(
self,
package: str,
process: subprocess.Popen,
stdout: typing.Optional[str] = None,
stderr: typing.Optional[str] = None,
) -> None:
"""Initialize new :class:`~RunningProcess` instance."""
if stdout == DEVNULL:
stdout = None
if stderr == DEVNULL:
stderr = None
self.__package: typing.Final[str] = package
self.__process: typing.Final[subprocess.Popen] = process
self.__return_code: typing.Optional[int] = None
self.__stdout: typing.Final[typing.Optional[str]] = stdout
self.__stderr: typing.Final[typing.Optional[str]] = stderr
self.__birth: typing.Final[datetime.datetime] = datetime.datetime.utcnow()
@property
def age(self) -> datetime.timedelta:
"""Current age of the application."""
return datetime.datetime.utcnow() - self.__birth
@property
def package(self) -> str: # noqa: D401
"""Name of the Conda package, or alias of a launched application."""
return self.__package
@property
def pid(self) -> int: # noqa: D401
"""PID of the process."""
return self.__process.pid
@property
def return_code(self) -> typing.Optional[int]: # noqa: D401
"""Current status of the application."""
if self.__return_code is None:
self.__return_code = self.__process.poll()
return self.__return_code
@property
def stderr(self) -> typing.Optional[str]: # noqa: D401
"""Content of the `stdout` log."""
return self.__read(self.__stderr)
@property
def stdout(self) -> typing.Optional[str]: # noqa: D401
"""Content of the `stdout` log."""
return self.__read(self.__stdout)
@staticmethod
def __read(path: typing.Optional[str]) -> typing.Optional[str]:
"""Read content of the file."""
if path is None:
return None
try:
stream: typing.TextIO
with open(path, 'rt') as stream: # pylint: disable=unspecified-encoding
return stream.read()
except OSError:
return None
def cleanup(self) -> None:
"""Remove all temporary content."""
path: str
for path in typing.cast(typing.Iterable[str], filter(bool, [self.__stdout, self.__stderr])):
try:
os.remove(path)
except OSError:
pass
def get_scripts_path(
root_prefix: str,
prefix: str,
default_scripts_path: str = LAUNCH_SCRIPTS_PATH,
) -> str:
"""
Return path to directory, where all launch scripts should be placed.
This directory can also be used for the application logs.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param default_scripts_path: Root directory to store launch scripts in.
:return: Path to directory, where launch script should be placed.
"""
result: str = os.path.abspath(default_scripts_path)
root_prefix = os.path.abspath(root_prefix)
prefix = os.path.abspath(prefix)
if prefix != root_prefix:
result = os.path.join(result, os.path.basename(prefix))
os.makedirs(result, exist_ok=True)
return result
def remove_package_logs(
root_prefix: str,
prefix: str,
default_scripts_path: str = LAUNCH_SCRIPTS_PATH,
) -> None:
"""
Try to remove output, error logs for launched applications.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param default_scripts_path: Root directory to store launch scripts in.
"""
logs_root: str = get_scripts_path(
root_prefix=root_prefix,
prefix=prefix,
default_scripts_path=default_scripts_path,
)
file_path: str
for file_path in os.listdir(logs_root):
if not file_path.endswith('.txt'):
continue
try:
os.remove(os.path.join(logs_root, file_path))
except OSError:
pass
def get_package_logs(
root_prefix: str,
prefix: str,
package_name: str = 'app',
id_: typing.Optional[int] = None,
default_scripts_path: str = LAUNCH_SCRIPTS_PATH,
) -> typing.Tuple[str, str, int]:
"""
Return the package log names for launched applications.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param package_name: Name of the conda package, or alias of the external application.
:param id_: Application session identifier.
Used to create different names for different application launch sessions.
:param default_scripts_path: Root directory to store launch scripts in.
"""
logs_root: str = get_scripts_path(
root_prefix=root_prefix,
prefix=prefix,
default_scripts_path=default_scripts_path,
)
stdout_log_path: str = LOG_FILE.format(package=package_name, level='out', index=id_)
stderr_log_path: str = LOG_FILE.format(package=package_name, level='err', index=id_)
if id_ is None:
used: typing.Set[str] = set(os.listdir(logs_root))
for id_ in itertools.count(start=1): # pylint: disable=redefined-argument-from-local
stdout_log_path = LOG_FILE.format(package=package_name, level='out', index=id_)
stderr_log_path = LOG_FILE.format(package=package_name, level='err', index=id_)
if (stdout_log_path not in used) and (stderr_log_path not in used):
break
if prefix and root_prefix:
stdout_log_path = os.path.join(logs_root, stdout_log_path)
stderr_log_path = os.path.join(logs_root, stderr_log_path)
return stdout_log_path, stderr_log_path, id_
def create_app_run_script( # pylint: disable=too-many-arguments
root_prefix: str,
prefix: str,
command: str,
extension: str,
package_name: str = 'app',
default_scripts_path: str = LAUNCH_SCRIPTS_PATH,
) -> str:
"""
Create new application launching script file.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param command: Content of the script file.
:param extension: Extension for the script file (`.bat`, `.sh`)
:param package_name: Name of the conda package, or alias of the external application.
:param default_scripts_path: Root directory to store launch scripts in.
:return: Path to generated launch script file.
"""
scripts_root: str = get_scripts_path(
root_prefix=root_prefix,
prefix=prefix,
default_scripts_path=default_scripts_path,
)
encoding: str = 'utf-8'
if WIN:
encoding = f'cp{ctypes.cdll.kernel32.GetACP()}'
stream: typing.TextIO
file_path: str = os.path.join(scripts_root, f'{package_name}{extension}')
with open(file_path, 'wt', encoding=encoding) as stream:
stream.write(command)
os.chmod(file_path, 0o755) # nosec
return file_path
def get_command_on_win( # pylint: disable=too-many-arguments
root_prefix: str,
prefix: str,
command: str,
extra_arguments: typing.Iterable[typing.Any] = (),
package_name: str = 'app',
environment: typing.Optional[typing.Mapping[str, str]] = None,
cwd: str = os.path.expanduser('~'),
default_scripts_path: str = LAUNCH_SCRIPTS_PATH,
) -> typing.Tuple[typing.Mapping[str, typing.Any], 'CommandDetails']:
"""
Generate script to launch application and return path to it.
This function is optimized to run on Windows.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param command: Actual command to launch application.
:param extra_arguments: Additional arguments to attach to command.
:param cwd: Custom working directory to launch application in.
:param package_name: Name of the conda package, or alias of the external application.
:param environment: Custom environment to launch application in.
:param default_scripts_path: Root directory to store launch scripts in.
:return: Path to generated launch script file.
"""
stdout_log_path: str
stderr_log_path: str
id_: int
stdout_log_path, stderr_log_path, id_ = get_package_logs(
root_prefix=root_prefix,
prefix=prefix,
package_name=package_name,
default_scripts_path=default_scripts_path,
)
command = dict((
(
'start cmd.exe /K "${CONDA_ROOT_PREFIX}\\\\Scripts\\\\activate.bat" "${CONDA_PREFIX}"',
'start cmd.exe \\K "${CONDA_ROOT_PREFIX}\\Scripts\\activate.bat" "${CONDA_PREFIX}"',
),
(
str(
'start powershell.exe -ExecutionPolicy ByPass -NoExit -Command ' # pylint: disable=implicit-str-concat
'"& \'{CONDA_ROOT_PREFIX}\\\\shell\\\\condabin\\\\conda-hook.ps1\' ; conda activate \'{CONDA_PREFIX}\'"'
),
str(
'start powershell.exe -ExecutionPolicy ByPass -NoExit -Command ' # pylint: disable=implicit-str-concat
'"& \'${CONDA_ROOT_PREFIX}\\shell\\condabin\\conda-hook.ps1\' ; conda activate \'${CONDA_PREFIX}\'"'
),
)
)).get(command, command)
command = expand_environment(
command=command,
environment={
'PREFIX': prefix,
'CONDA_PREFIX': prefix,
'CONDA_ROOT_PREFIX': root_prefix,
},
)
activate: str = os.path.join(root_prefix, 'Scripts', 'activate')
extra_argument: typing.Any
for extra_argument in extra_arguments:
command += f' {safe_windows(str(extra_argument))}'
script: str = '\n'.join([
# disable echoing commands to the command line
'@echo off',
f'chcp {ctypes.cdll.kernel32.GetACP()}',
# "call" is needed to avoid the batch script from closing after running the first (environment activation) line
f'call {safe_windows(activate)} {safe_windows(prefix)}',
f'{command} >{safe_windows(stdout_log_path)} 2>{safe_windows(stderr_log_path)}',
'',
])
file_path: str = create_app_run_script(
root_prefix=root_prefix,
prefix=prefix,
command=script,
extension='.bat',
package_name=package_name,
default_scripts_path=default_scripts_path,
)
return { # popen_dict
'creationflags': subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP, # type: ignore
'cwd': cwd,
'env': environment,
}, {
'args': file_path,
'id': id_,
'cmd': script,
'stdout': stdout_log_path,
'stderr': stderr_log_path,
}
def get_command_on_unix( # pylint: disable=too-many-arguments
root_prefix: str,
prefix: str,
command: str,
extra_arguments: typing.Iterable[typing.Any] = (),
package_name: str = 'app',
environment: typing.Optional[typing.Mapping[str, str]] = None,
cwd: str = os.path.expanduser('~'),
default_scripts_path: str = LAUNCH_SCRIPTS_PATH,
) -> typing.Tuple[typing.Mapping[str, typing.Any], 'CommandDetails']:
"""
Generate script to launch application and return path to it.
This function is optimized to run on Linux/OS X.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param command: Actual command to launch application.
:param extra_arguments: Additional arguments to attach to command.
:param cwd: Custom working directory to launch application in.
:param package_name: Name of the conda package, or alias of the external application.
:param environment: Custom environment to launch application in.
:param default_scripts_path: Root directory to store launch scripts in.
:return: Path to generated launch script file.
"""
stdout_log_path: str
stderr_log_path: str
id_: int
stdout_log_path, stderr_log_path, id_ = get_package_logs(
root_prefix=root_prefix,
prefix=prefix,
package_name=package_name,
default_scripts_path=default_scripts_path,
)
command = expand_environment(
command=command,
environment={
'PREFIX': prefix,
'CONDA_PREFIX': prefix,
'CONDA_ROOT_PREFIX': root_prefix,
},
)
activate: str = os.path.join(root_prefix, 'bin', 'activate')
extra_argument: typing.Any
for extra_argument in extra_arguments:
command += f' {safe_unix(str(extra_argument))}'
script: str = '\n'.join([
'#!/usr/bin/env bash',
f'. {safe_unix(activate)} {safe_unix(prefix)}',
f'{command} >{safe_unix(stdout_log_path)} 2>{safe_unix(stderr_log_path)}',
''
])
file_path: str = create_app_run_script(
root_prefix=root_prefix,
prefix=prefix,
command=script,
extension='.sh',
package_name=package_name,
default_scripts_path=default_scripts_path
)
return {
'cwd': cwd,
'env': environment,
}, {
'args': file_path,
'id': id_,
'cmd': script,
'stdout': stdout_log_path,
'stderr': stderr_log_path,
}
def launch( # pylint: disable=too-many-arguments
root_prefix: str,
prefix: str,
command: str,
extra_arguments: typing.Iterable[typing.Any] = (),
working_directory: str = os.path.expanduser('~'),
package_name: str = 'app',
environment: typing.Optional[typing.Mapping[str, str]] = None,
leave_path_alone: bool = True, # pylint: disable=unused-argument
non_conda: bool = False, # pylint: disable=unused-argument
as_admin: bool = False,
) -> typing.Optional[RunningProcess]:
"""
Handle launching commands from projects.
:param root_prefix: Path to root (base) conda prefix. Used to activate `prefix`.
:param prefix: Conda prefix, which should be active.
:param command: Actual command to launch application.
:param extra_arguments: Additional arguments to attach to command.
:param working_directory: Custom working directory to launch application in.
If not provided - home directory will be used.
:param package_name: Name of the conda package, or alias of the external application.
:param environment: Custom environment to launch application in.
:param as_admin: Launch application with admin rights.
This breaks function result (changes it to :code:`True`/:code:`None`).
:return: Description of the launched process.
"""
if isinstance(extra_arguments, str):
extra_arguments = (extra_arguments,)
get_command: 'GetCommand' = get_command_on_unix
if WIN:
get_command = get_command_on_win
popen_dict: typing.Mapping[str, typing.Any]
extra_args: 'CommandDetails'
popen_dict, extra_args = get_command(
root_prefix=root_prefix,
prefix=prefix,
command=command,
extra_arguments=extra_arguments,
package_name=package_name,
environment=environment,
cwd=working_directory,
)
args: str = extra_args['args']
if WIN:
args = f'cmd.exe /C {safe_windows(args)}'
if as_admin:
if run_as_admin(args):
return typing.cast(typing.Any, True)
return None
return RunningProcess(
package=package_name,
process=subprocess.Popen(args, **popen_dict), # nosec
stdout=extra_args['stdout'],
stderr=extra_args['stderr'],
)
def console(activate=None, working_directory=os.path.expanduser('~'), term_command=''):
"""
Open command prompt console and optionally activate the environment.
optionally pass an application to be launched in terminal such as python,
ipython, or jupyter as when called from py_in_console()
"""
cwd = working_directory
if os.name == 'nt':
if activate:
# cmd = 'start cmd.exe /k activate ' + activate
cmd = f'start cmd.exe /k "activate "{activate}" & {term_command}"'
else:
cmd = 'start cmd.exe'
# pylint: disable=consider-using-with
subprocess.Popen(cmd, shell=True, cwd=cwd) # nosec
elif sys.platform == 'darwin':
if activate:
from anaconda_navigator.api.conda_api import CONDA_API # pylint: disable=import-outside-toplevel
rootprefix = CONDA_API.ROOT_PREFIX
cmd = f'''\
#!/usr/bin/osascript
tell application "Terminal"
activate
do script ". {rootprefix}/bin/activate && conda activate {activate}; {term_command}"
end tell
'''
else:
cmd = 'bash'
fname = os.path.join(CONF_PATH, 'a.tool')
with open(fname, 'w') as f: # pylint: disable=invalid-name,unspecified-encoding
f.write(cmd)
os.chmod(fname, 0o755) # nosec
subprocess.call([fname], shell=True, cwd=cwd) # nosec
else: # Linux, solaris, etc
if is_program_installed('gnome-terminal'):
if activate:
cmd = [
'gnome-terminal',
'-x',
'bash',
'-c',
f'bash --init-file <(echo ". activate {activate};")',
f'; {term_command}',
]
else:
cmd = ['gnome-terminal', '-e', 'bash']
# pylint: disable=consider-using-with
subprocess.Popen(cmd, cwd=cwd) # nosec
elif is_program_installed('xterm'):
if activate:
cmd = [
'xterm',
'-e',
f'bash --init-file <(echo ". activate {activate};")',
f'; {term_command}',
]
else:
cmd = ['xterm']
# pylint: disable=consider-using-with
subprocess.Popen(cmd, cwd=cwd) # nosec
def check_prog(prog: str, prefix: typing.Optional[str] = None) -> bool:
"""Check if program exists in prefix."""
from anaconda_navigator.api.anaconda_api import AnacondaAPI # pylint: disable=cyclic-import,import-outside-toplevel
api = AnacondaAPI()
prefix = prefix or api.conda_get_prefix_envname(name='root')
if prog in ['notebook', 'jupyter notebook']:
pkgs = ['notebook', 'ipython-notebook', 'jupyter-notebook']
elif prog in ['ipython', 'jupyter console']:
pkgs = ['ipython', 'jupyter']
else:
pkgs = [prog]
return any(
api.conda_package_version(prefix=prefix, pkg=pkg) is not None
for pkg in pkgs
)
def py_in_console(activate=None, prog='python'):
"""
Run (i)python in a new console.
It optionally run activate first on the given env name/path.
"""
if not check_prog(prog, activate):
raise RuntimeError(f'Program not available in environment: {prog}, {activate}')
if prog == 'python':
cmd = 'python -i'
elif prog == 'ipython':
cmd = 'ipython -i'
elif 'notebook' in prog:
cmd = 'jupyter notebook'
from anaconda_navigator.api.conda_api import CONDA_API # pylint: disable=import-outside-toplevel
# Jupyter notebook shouldn't be launched from a console
launch(
root_prefix=CONDA_API.ROOT_PREFIX,
prefix=activate,
command=cmd,
leave_path_alone=True,
)
return
else:
cmd = None
console(activate=activate, term_command=cmd)