Source code for schrodinger.application.steps.docking
import shlex
from rdkit import Chem
from schrodinger import stepper
from schrodinger.application.glide import http_client
from schrodinger.application.inputconfig import InputConfig
from schrodinger.models import parameters
from schrodinger.structutils.smiles import STEREO_FROM_ANNOTATION_AND_GEOM
from schrodinger.structutils.smiles import SmilesGenerator
from schrodinger.tasks import tasks
from schrodinger.utils import license
from . import utils
from .basesteps import MaeMapStep
from .basesteps import MolMapStep
from .dataclasses import ScoredMol
from .dataclasses import ScoredSmiles
from .dataclasses import ScoredSmilesSerializer
try:
from schrodinger.application.glide.packages.startup import DockingJob
except ImportError:
DockingJob = None
LICENSE_BY_NAME = {v: k for k, v in license.LICENSE_NAMES.items()}
INF = float('inf')
DOCKING_SCORE_KEY = 'r_i_docking_score'
GLIDE_SERVER_WAIT_TIME = 2.0
GLIDE_SERVER_STARTUP_WAIT_TIME = 300 # seconds
[docs]class GlideSettings(parameters.CompoundParam):
"""
The `glide_in_file` input file should not have a `LIGANDFILE` keyword.
The `POSES_PER_LIG` will be overridden with the value of 1, since we need to
know whether the molecule can dock, and what the best docking score is.
"""
glide_grid_file: stepper.StepperFile # required
glide_ref_ligand_file: stepper.StepperFile # optional
glide_in_file: stepper.StepperFile # optional
[docs] def validate(self, step):
"""
Validate the settings for use in `step`.
:param step: stepper._BaseStep
:rtype: list[TaskError or TaskWarning]
"""
issues = []
for attr, required in (('glide_grid_file', True),
('glide_ref_ligand_file', False),
('glide_in_file', False)): # yapf:disable
issues += utils.validate_file(step, attr, required=required)
return issues
class _GlideServerStartTask(tasks.BlockingFunctionTask):
"""
A task that starts up a glide server that will only return one pose for
every ligand that is docked
"""
input: GlideSettings
output: http_client.GlideServerManager = None
def _getKeywords(self):
keywords = InputConfig(self.input.glide_in_file or {})
keywords['GRIDFILE'] = self.input.glide_grid_file
keywords['POSES_PER_LIG'] = 1
if self.input.glide_ref_ligand_file:
keywords['CORE_RESTRAIN'] = True
keywords['REF_LIGAND_FILE'] = self.input.glide_ref_ligand_file
return keywords
def getRequiredLicenses(self):
docking_job = DockingJob(self._getKeywords(), 'foo')
required_licenses = {}
for rec in docking_job.licenseRequirements():
name, count = rec.split(':')
required_licenses[LICENSE_BY_NAME[name]] = int(count)
return required_licenses
def mainFunction(self):
keywords = self._getKeywords()
server = http_client.GlideServerManager(keywords=keywords, use_jc=False)
server.start(wait=GLIDE_SERVER_STARTUP_WAIT_TIME)
self.output = server
class _MolDockerMixin:
"""
Mixin for a `stepper._BaseStep` providing functionality for glide docking.
To use:
define `GLIDE_SERVER_START_TASK_CLASS`
settings should be `GlideSettings`, so that it has a `validation` method
Override `_dock` to do the actual docking for the `Mol` object.
"""
GLIDE_SERVER_START_TASK_CLASS = None
Settings = GlideSettings
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._docker = None
def getLicenseRequirements(self):
startup_task = _GlideServerStartTask()
startup_task.input.glide_in_file = self.settings.glide_in_file
return startup_task.getRequiredLicenses()
def validateSettings(self):
return self.settings.validate(self)
def getDocker(self):
"""
:return: the singleton glide server manager
:rtype: http_client.GlideServerManager
"""
if self._docker is None:
startup_task = self.GLIDE_SERVER_START_TASK_CLASS()
utils.update_params(startup_task.input, self.settings)
startup_task.start()
startup_task.wait()
if startup_task.status != tasks.Status.DONE:
raise RuntimeError('Glide docking server could not be started')
self._docker = startup_task.output
return self._docker
def _dock(self, mol):
"""
A generator of ScoredMol objects of docked molecules with a score value
less than or equal to `max_score` in the settings.
:param mol: the molecule to dock
:type mol: Chem.Mol
:return: the generator of docked ScoredMol objects
:rtype: collections.Generator[ScoredMol]
"""
raise NotImplementedError
def cleanUp(self):
super().cleanUp()
if self._docker:
self._docker.stop(wait=GLIDE_SERVER_WAIT_TIME)
self._docker = None
[docs]class GlideDocker(_MolDockerMixin, MolMapStep):
"""
Perform a glide docking step.
Only yields the original input molecule if at least 1 pose is found with
a score less than or equal to the settings' `max_score`.
Note: it is not the docked `Mol` in the ScoredMol that is yielded.
"""
GLIDE_SERVER_START_TASK_CLASS = _GlideServerStartTask
[docs] class Settings(GlideSettings):
max_score: float = INF
def _dock(self, mol):
st = utils.mol_to_structure(mol, self, generate_coordinates=True)
if st is not None:
docker = self.getDocker()
docked_sts = list(docker.dock(st))
if docked_sts:
score = docked_sts[0].property[DOCKING_SCORE_KEY]
if score <= self.settings.max_score:
yield ScoredMol(mol=mol, score=score)
[docs] def mapFunction(self, mol):
for scored_mol in self._dock(mol):
yield scored_mol.mol
[docs]class MaeGlideDocker(_MolDockerMixin, MaeMapStep):
"""
Perform a glide docking step, yielding the best scored pose for the input
structure.
"""
GLIDE_SERVER_START_TASK_CLASS = _GlideServerStartTask
[docs] class Settings(GlideSettings):
max_score: float = INF
[docs] def mapFunction(self, struc):
docker = self.getDocker()
docked_sts = list(docker.dock(struc))
if docked_sts:
docked_st = docked_sts[0]
if docked_st.property[DOCKING_SCORE_KEY] <= self.settings.max_score:
yield docked_st
[docs]class SmilesDockerSettings(GlideSettings):
arg_string: str = '-bff 16 -epik -s 32'
ligprep_filter_file: stepper.StepperFile
[docs] def validate(self, step):
issues = utils.validate_file(step, 'ligprep_filter_file')
return issues + super().validate(step)
class _SmilesGlideServerStartTask(_GlideServerStartTask):
"""
A task that starts up a glide server that will only return one pose for
every ligand that is docked using ligprep and SMILES docking.
"""
input: SmilesDockerSettings
def _getKeywords(self):
keywords = super()._getKeywords()
keywords['LIGPREP'] = 'yes'
flt_file = self.input.ligprep_filter_file
lp_args = self.input.arg_string
if flt_file:
lp_args += f' -f {shlex.quote(flt_file)}'
keywords['LIGPREP_ARGS'] = lp_args
return keywords
[docs]class SmilesDocker(_MolDockerMixin, MolMapStep):
"""
Perform a ligprep with glide docking step. This step will yield the
molecules and their glide score value as `ScoredMol` objects only for
molecules that had a score that is less than or equal to the `max_score`
in the settings.
Since to ligprep may generate different tautomers the same molecule may be
yielded more than once.
"""
GLIDE_SERVER_START_TASK_CLASS = _SmilesGlideServerStartTask
[docs] class Settings(SmilesDockerSettings):
max_score: float = INF
[docs] def setUp(self):
super().setUp()
self._smiles_generator = SmilesGenerator(
STEREO_FROM_ANNOTATION_AND_GEOM, unique=True)
def _dock(self, mol):
smi = Chem.MolToSmiles(mol)
docker = self.getDocker()
for st in docker.dockSmiles(smi):
score = st.property[DOCKING_SCORE_KEY]
if score <= self.settings.max_score:
mol = utils.structure_to_mol(st, self, mol)
yield ScoredMol(mol=mol, score=score)
[docs] def mapFunction(self, mol):
for scored_mol in self._dock(mol):
yield scored_mol.mol
[docs]class ScoredSmilesDocker(SmilesDocker):
"""
A SmilesDocker that returns ScoredSmiles objects.
"""
Output = ScoredSmiles
OutputSerializer = ScoredSmilesSerializer
[docs] def mapFunction(self, mol):
for scored_mol in self._dock(mol):
yield ScoredSmiles(smiles=Chem.MolToSmiles(scored_mol.mol),
score=scored_mol.score)