Source code for schrodinger.job.jobwriter
"""
Functions for writing job command lists into shell scripts.
"""
import os
import pathlib
import shlex
import shutil
import stat
import sys
from schrodinger.infra import mm
from schrodinger.utils import fileutils
from schrodinger.utils import subprocess as subprocess_utils
DOLLAR_SCHRODINGER = '${SCHRODINGER}'
SCHRODINGER = os.environ['SCHRODINGER']
SCHRODINGER_ABSPATH = os.path.abspath(os.environ['SCHRODINGER'])
def _SCHRODINGER_env_var_in_str(s):
return DOLLAR_SCHRODINGER in s or '$SCHRODINGER' in s
[docs]def write_job_cmd(cmdlist, sh_filename, job_dir, append=False):
"""
Write a shell script that can be used to run the given `cmdlist`.
:param cmdlist: The command the script should run.
:type cmdlist: list[str]
:param sh_filename: The name of the file the script should be written to
:type sh_filename: sh
:param job_dir: The desired job directory to be used for the command
:type job_dir: str
:param append: Whether to append to the given file or overwrite it. This
is useful for writing a single shell script that launches multiple jobs.
:type append: bool
"""
_remove_maestro_args(cmdlist)
_replace_OPLS_dir(cmdlist, job_dir)
cmdlist = _make_locale_independent(cmdlist, job_dir)
cmd = cmdlist_to_cmd(cmdlist)
if append:
cmd = f'\n{cmd}'
mode = 'a'
else:
mode = 'w'
with open(sh_filename, mode) as f:
f.write(cmd)
set_sh_file_flags(sh_filename)
[docs]def cmdlist_to_cmd(cmdlist):
"""
Converts a command list to a command string. Don't do this if you can
possibly avoid it.
:param cmdlist: a list of commands
:type cmdlist: list
:return: str
"""
def _quote_arg(arg):
if _SCHRODINGER_env_var_in_str(arg):
# Double quote the argument to allow shell to expand when
# written to a sh script and run outside of maestro.
return f'"{arg}"'
# Otherwise, just use normal shlex.quote in case there are spaces
return shlex.quote(arg)
quoted_args = map(_quote_arg, cmdlist)
return " ".join(quoted_args)
[docs]def set_sh_file_flags(filename):
st = os.stat(filename) # Get default file permissions
try:
os.chmod(filename, st.st_mode | stat.S_IEXEC)
except OSError: # Ignore if OS won't let it be set to executable
pass
def _make_locale_independent(cmdlist, jobdir):
"""
This takes a job written for a specific machine / Schrodinger
installation, and makes it compatible wherever it's run. This is
necessary for STU. To do this it must:
1) Copy input files outside of $SCHRODINGER and jobdir into the jobdir
2) For files that are already in jobdir but are referenced with absolute
paths, convert them to local paths.
3) Change specific references to the current SCHRODINGER to the env
variable ${SCHRODINGER}
4) If on Windows, replace any backslashes in any paths (including
$SCHRODINGER) with forward slashes. Note that such SH files will
be usable on both Linux/Mac and Windows.
:param cmdlist: The list of args to be written out as a job.
:type cmdlist: list[str]
:return: The cmdlist, made locale independent
:rtype: list[str]
"""
# If the exec wasn't in $SCHRODINGER or $SCHRODINGER/utilities, we
# should run it using $SCHRODINGER/run.
cmdlist[0] = _normalize_schrodinger_exec(cmdlist[0])
schrodinger_run = '"${SCHRODINGER}/run"'
if not _SCHRODINGER_env_var_in_str(cmdlist[0]):
cmdlist.insert(0, schrodinger_run)
# Apply transformations for every argument
def _process_arg(arg):
arg = _strip_double_quotes(arg)
# relpath file args to make the command portable
if arg.startswith(os.path.abspath(jobdir)):
arg = os.path.relpath(arg, jobdir)
arg = _replace_SCHRODINGER_abspaths(arg)
arg = _process_input_file(arg, jobdir)
return arg
return [_process_arg(arg) for arg in cmdlist]
def _normalize_schrodinger_exec(executable):
"""
Find the executable if it's in `$SCHRODINGER` or `$SCHRODINGER/utilities`
and prepend with the correct path. For example::
_normalize_schrodinger_exec('testapp') -> ${SCHRODINGER}/testapp
_normalize_schrodinger_exec('py.test') -> ${SCHRODINGER}/utilities/py.test
_normalize_schrodinger_exec('/path/to/schrodinger/build/testapp') -> ${SCHRODINGER}/testapp
_normalize_schrodinger_exec('foo_exec') -> foo_exec
"""
executable = subprocess_utils.abs_schrodinger_path(executable)
executable = _replace_SCHRODINGER_abspaths(executable)
executable = _convert_to_posix_executable(executable)
return executable
def _convert_to_posix_executable(arg):
if sys.platform == "win32":
# On Windows, replace any backslashes in file paths
# and executable with forward slashes, to make SH
# files portable. Note that executable path can be
# non-existing, e.g. "${SCHRODINGER}\epik"
path = pathlib.PureWindowsPath(arg)
arg = path.as_posix()
# Script will work on both windows and linux without exe
if arg.endswith('.exe'):
arg = arg[:-4]
return arg
def _strip_double_quotes(arg):
if len(arg) >= 2 and arg[0] == '"' and arg[-1] == '"':
arg = arg[1:-1]
return arg
def _process_input_file(arg, jobdir):
"""
Move any files not in the jobdir or $SCHRODINGER into the
jobdir. Returns base name of the the arg.
"""
abs_arg = os.path.abspath(arg)
if (os.path.isfile(abs_arg) and not abs_arg.startswith(
(jobdir, SCHRODINGER, SCHRODINGER_ABSPATH))):
shutil.copy(arg, jobdir)
arg = os.path.basename(arg)
return arg
def _replace_SCHRODINGER_abspaths(arg):
"""
Make the shell script schrodinger-location neutral
"""
arg = arg.replace('$SCHRODINGER', DOLLAR_SCHRODINGER)
arg = arg.replace(SCHRODINGER, DOLLAR_SCHRODINGER)
arg = arg.replace(SCHRODINGER_ABSPATH, DOLLAR_SCHRODINGER)
return arg
def _remove_maestro_args(cmdlist):
"""
Removes options from the command that only work within Maestro.
:param cmdlist: the job invocation command
:type cmdlist: list
"""
remove_options = ['-PROJ', '-DISP', '-VIEWNAME']
for option in remove_options:
try:
index = cmdlist.index(option)
except ValueError:
pass
else:
cmdlist.pop(index + 1)
cmdlist.pop(index)
def _replace_OPLS_dir(cmdlist, job_dir):
"""
If necessary, replace the -OPLSDIR argument to use a version of the
original -OPLSDIR argument copied to the job directory.
:param cmdlist: the job invocation command
:type cmdlist: list of str
"""
try:
index = cmdlist.index('-OPLSDIR') + 1
except ValueError:
return
src = os.path.abspath(mm.get_archive_path(cmdlist[index]))
if not os.path.isfile(src):
return
base_name = os.path.basename(src)
dest = os.path.abspath(os.path.join(job_dir, base_name))
if src != dest:
fileutils.force_remove(dest)
shutil.copy(src, dest)
# update the OPSLDIR argument to its relative path in the job directory
cmdlist[index] = base_name