"""
FEP+ generator
Version 0.1
Copyright Schrodinger, LLC. All rights reserved.
"""
import os
import subprocess
import sys
from typing import List
from schrodinger import structure
from schrodinger.application.desmond import cmdline
from schrodinger.application.desmond import constants
from schrodinger.application.desmond import launch_utils
from schrodinger.application.desmond import license
from schrodinger.application.desmond import struc
from schrodinger.application.desmond import mapper_msj_generator as mmg
from schrodinger.application.desmond.starter.generator import common
from schrodinger.application.desmond.starter.generator.common import \
find_fmpdb_file
from schrodinger.application.desmond.starter.generator.common import is_fmp
from schrodinger.application.desmond.starter.ui.fep_plus import FepPlusArgs
from schrodinger.application.desmond.util import fatal
from schrodinger.application.desmond.util import parse_edge_file
try:
from schrodinger.application.scisol.packages.fep import fepmae
from schrodinger.application.scisol.packages.fep import graph
except ImportError:
fepmae = None
graph = None
[docs]def is_covalent(filename):
'''
Determine if the given file is for covalent FEP. The file is either a .mae
or a .fmp file, but for covalent FEP, it has to be the latter (a .mae file
will cause this function to return False directly), and this function will
check the fep_type graph property and return True if its value is
'COVALENT_LIGAND', or False if not.
:type filename: `str`
:rtype: bool
'''
if not filename.endswith('fmp'):
return False
g = graph.Graph.deserialize(filename)
return g.fep_type == constants.FEP_TYPES.COVALENT_LIGAND
[docs]def validate_covalent_fmp(filename):
"""Make sure the fmp file is generated by >= 19-2 release that adds
i_fep_cov_protein_bb_atom to atom's property.
"""
g = graph.Graph.deserialize(filename)
node = next(g.nodes_iter())
has_tag = any(constants.FEP_COVALENT_PROTEIN_BB_ATOM in atom.property
for atom in node.struc.atom)
if not has_tag:
fatal(
"The Covalent FEP workflow has been modified and the input fmp file must be regenerated."
)
[docs]def generate(args: FepPlusArgs) -> List[str]:
"""
Generate the files and command line to run multisim for the
FEP+ workflow.
:param args: Object with input arguments.
:return: Command line to launch multisim
:raise SystemExit: If preparing the job without launching it.
"""
additional_licenses = []
stage_data_fnames = []
fmpdb_fn = find_fmpdb_file(args)
if fmpdb_fn:
stage_data_fnames.append(fmpdb_fn)
if args.mode == constants.UiMode.NEW:
cd_params = {
"processors_per_replica": 1,
"cpus": args.ppj,
"mps_factor": args.mps_factor,
}
generator_args = [args.JOBNAME, cd_params]
generator_kwargs = dict(
forcefield=args.forcefield,
sim_time=args.time,
sim_time_complex=args.time_complex,
sim_time_solvent=args.time_solvent,
water_model=args.water_model,
buffer_width=args.buffer,
salt_concentration=args.salt_concentration,
rand_seed=args.seed,
lambda_windows=args.lambda_windows,
charged_lambda_windows=args.charged_lambda_windows,
core_hopping_lambda_windows=args.core_hopping_lambda_windows,
custom_charge_mode=args.custom_charge_mode,
modify_dihe=args.modify_dihe,
h_mass=args.h_mass,
minimize_volume=args.minimize_volume,
concatenate=args.concat,
skip_legs=args.skip_legs,
membrane=args.membrane and args.inp_file,
ensemble=args.ensemble,
atom_mapping=args.atom_mapping,
max_walltime=args.max_walltime,
ffbuilder=args.ffbuilder,
ff_host=args.ff_host)
if args.ffbuilder:
structure_fname = f'{args.JOBNAME}_ligands.maegz'
generator_kwargs['ff_structure_file'] = structure_fname
fmp_fname = args.inp_file if is_fmp(args.inp_file) else ""
# non-vacuum
if args.mp:
# metalloprotein
if args.ffbuilder:
ligand_sts = fepmae.get_structures_from_file(args.inp_file)
generator = mmg.MetalMsjGenerator(*generator_args, args.mp,
**generator_kwargs)
elif is_covalent(args.inp_file):
# covalent ligand
validate_covalent_fmp(args.inp_file)
if args.ffbuilder:
ligand_sts = fepmae.get_structures_from_file(args.inp_file)
generator = mmg.CovalentMsjGenerator(*generator_args, args.inp_file,
**generator_kwargs)
elif not args.protein:
# small molecule
if not args.vacuum and 'vacuum' not in generator_kwargs["skip_legs"]:
# vacuum leg gets skipped unless -vacuum is passed in
generator_kwargs["skip_legs"].append('vacuum')
generator_kwargs["sim_time_vacuum"] = args.time_vacuum
if args.ffbuilder:
ligand_sts = fepmae.get_structures_from_file(args.inp_file)
generator = mmg.SmallMoleculeWithVacuumMsjGenerator(
*generator_args,
align_core_only=args.align_core_only,
ats=args.ats,
**generator_kwargs)
elif os.path.isfile(args.protein):
from schrodinger.application.scisol.packages.fep import \
protein_mutation_generator as pmg
# For the ProteinMutationGenerator stage
additional_licenses += license.protein_mutation_lics()
# determine FEP_TYPES (protein stability, protein
# selectivity, or ligand_selectivity)
ligand_sts = []
if is_fmp(args.inp_file):
g = graph.Graph.deserialize(args.inp_file)
fep_type = g.fep_type
if fep_type == constants.FEP_TYPES.LIGAND_SELECTIVITY:
_, _, _, add_ligand_sts = fepmae.filter_receptors_and_ligands(
g.environment_struc)
ligand_sts.extend(add_ligand_sts)
else:
sts = list(structure.StructureReader(args.inp_file))
fep_type = fepmae.get_prm_fep_type(sts, args.solvent_asl)
if fep_type == constants.FEP_TYPES.LIGAND_SELECTIVITY:
filtered_sts = [
st for st in sts
if not fepmae.is_ct_solvent_or_membrane(st)
]
ligand_sts.append(
fepmae.get_protein_fep_receptor(filtered_sts[0],
args.solvent_asl))
# DESMOND-8513, throws error if not valid
with open(args.protein, 'r') as f:
pmg.validate_mutations(f.readlines(), fep_type)
if args.expand_protein:
if is_fmp(args.expand_protein):
raise RuntimeError(
"To expand a protein fep map, provide the original wildtype "
"structure as the argument to -expand_protein.")
if not is_fmp(args.inp_file):
raise RuntimeError(
"To expand a protein fep map, provide the output fmp file "
"as input.")
# Add NSAA as 'ligands' to run ffbuilder on
if args.ffbuilder and args.residue_structure:
ligand_sts += list(
structure.StructureReader(args.residue_structure))
# If there are no NSAA/ligands, turn off the ffbuilder stage
if args.ffbuilder and not ligand_sts:
generator_kwargs['ffbuilder'] = False
generator = mmg.ProteinMapperMsjGenerator(
*generator_args,
fep_type,
args.solvent_asl,
args.protein,
args.residue_structure,
structure_file=args.expand_protein,
**generator_kwargs)
else:
fatal('ERROR: Unknown inputs, see help message for more info')
if args.ffbuilder and ligand_sts:
struc.write_structures(ligand_sts,
generator_kwargs['ff_structure_file'])
generator.fmp_fname = fmp_fname
generator.write_subjob_msjs()
main_msj_fname = generator.write_main_msj()
cmd = launch_utils.prepare_command_for_launch(args.HOST,
args.SUBHOST,
args.JOBNAME,
main_msj_fname,
args.maxjob,
input_fname=args.inp_file,
lics=additional_licenses)
else:
edges = None
if args.mode == constants.UiMode.EXTEND:
edges = [
f'{a[0][:7]}_{a[1][:7]}' for a in parse_edge_file(args.extend)
]
try:
cmd, more_fnames = common.prepare_files_and_command_for_fep_restart_extend(
args, edges)
except common.RestartException as ex:
sys.exit(str(ex))
stage_data_fnames.extend(more_fnames)
cmd += launch_utils.additional_command_arguments(
stage_data_fnames, args.RETRIES, args.WAIT, args.LOCAL, args.DEBUG,
args.TMPDIR, args.forcefield, args.OPLSDIR, args.NICE, args.SAVE)
print(
"Launch command:",
subprocess.list2cmdline(cmd).replace(os.environ["SCHRODINGER"],
"$SCHRODINGER"))
cmd.extend([
"-encoded_description",
cmdline.get_b64encoded_str(cmdline.get_job_command_in_startup()),
])
return cmd