Source code for schrodinger.application.glide.ligand_designer
"""
This module provides the APIs behind the Ligand Designer panel and workflow.
It combines binding site Phase hypothesis generation with R-group enumeration
and Glide grid generation and docking. The docking uses a Glide HTTP server for
speed.
"""
import glob
import hashlib
import json
import os
from pathlib import Path
import schrodinger
from schrodinger import structure
from schrodinger.application.glide import utils as glide_utils
from schrodinger.application.glide import glide
from schrodinger.application.glide import http_client
from schrodinger.models import parameters
from schrodinger.Qt import QtCore
from schrodinger.tasks import jobtasks
from schrodinger.tasks import tasks
from schrodinger.utils import fileutils
from schrodinger.utils import log
from schrodinger.utils import mmutil
# Maximum time in seconds to wait for Glide grid generation to finish
GRIDGEN_WAIT = 600
logger = log.get_output_logger(__file__)
if schrodinger.in_dev_env():
logger.setLevel(log.DEBUG)
REFLIG_NAME = 'reflig.maegz'
GRIDFILE_NAME = 'grid.grd'
JOBDIR_SEP = '-'
[docs]class GridgenRunningException(RuntimeError):
[docs] def __init__(self, message="Grid generation is still running"):
super().__init__(message)
[docs]class BuildServerTask(tasks.BlockingFunctionTask):
"""
Task to set up and start a glide server. The server performs
core-constrained Glide docking.
The server state and intermediate files are in a scratch
directory. A unique subdirectory is created for each ligand-receptor
complex; if another object is created for the same complex, it will share
the same directory. This allows the reuse of existing grid files, for
example. However, only one object at a time can be performing an
enumeration because the underlying Glide server process is single-threaded.
:ivar gridJobStarted: Signal when grid job has launched. Emitted with task
"""
gridJobStarted = QtCore.pyqtSignal(jobtasks.CmdJobTask)
gg_task = parameters.NonParamAttribute()
[docs] class Output(parameters.CompoundParam):
server: http_client.NonBlockingGlideServerManager = None
[docs] def initConcrete(self, tmpdir=None, *args, **kwargs):
"""
:param tmpdir: Base temporary directory for server directories
"""
super().initConcrete(*args, **kwargs)
self._tmpdir = tmpdir
self.gg_task = None
@tasks.preprocessor(order=tasks.BEFORE_TASKDIR)
def _initGridgenTask(self):
if not self.input.start_gridgen or self.gg_task is not None:
return
self.gg_task = GridgenTask()
self.gg_task.input.ligand_st = self.input.ligand_st
self.gg_task.input.receptor_st = self.input.receptor_st
@tasks.preprocessor(order=tasks.BEFORE_TASKDIR + 1)
def _findTaskDir(self):
"""
Check existing task dirs to see if any have a grid compatible with the
current ligand_st and receptor_st
"""
if self.taskDirSetting() is not self.DEFAULT_TASKDIR_SETTING:
# Skip if we have already set a taskdir
return
recepname = fileutils.get_jobname(self.input.receptor_st.title)
signature = get_structure_digest(self.input.receptor_st)
jobname = '-'.join(['ld', recepname, signature])
tmpdir = self._tmpdir or fileutils.get_directory_path(fileutils.TEMP)
jobdir_stem = Path(tmpdir) / 'ligand_designer' / jobname
jobdir_stem = jobdir_stem.absolute()
jobdir = self._checkExistingJobDirs(jobdir_stem)
if not jobdir:
jobdir = fileutils.get_next_filename(str(jobdir_stem), JOBDIR_SEP)
jobdir = Path(jobdir)
logger.debug('Jobdir: %s', jobdir)
self.specifyTaskDir(jobdir)
self.gg_task.specifyTaskDir(jobdir)
def _checkExistingJobDirs(self, jobdir_stem):
dir_pattern = f"{jobdir_stem}{JOBDIR_SEP}*"
for jobdir in glob.iglob(dir_pattern):
self.gg_task.specifyTaskDir(jobdir)
if self.gg_task.checkGridfile(match_reflig=True):
return jobdir
[docs] def mainFunction(self):
self.output.reset()
gg_task = self.gg_task
grid_ok = gg_task.checkGridfile()
if self.input.start_gridgen:
if not grid_ok:
gg_task.start()
self.gridJobStarted.emit(gg_task)
elif grid_ok:
self._startServer()
else:
# Fail if the gridfile isn't acceptable but we aren't supposed to
# run gridgen
if gg_task.status is gg_task.RUNNING:
exc = GridgenRunningException()
else:
exc = RuntimeError("Gridgen failed")
self._recordFailure(exc)
def _startServer(self):
gg_task = self.gg_task
docking_keywords = {
'PRECISION': 'HTVS',
'GRIDFILE': gg_task.getTaskFilename(GRIDFILE_NAME),
'REF_LIGAND_FILE': gg_task.getTaskFilename(REFLIG_NAME),
'CORE_DEFINITION': 'mcssmarts',
'CORE_RESTRAIN': True,
'CORECONS_FALLBACK': True,
}
docking_keywords.update(self.input.docking_keywords)
kwargs = dict(
jobdir=self.getTaskDir(),
use_jc=False,
)
if kwargs['use_jc']:
# Only set jobname for jobcontrol to allow reattaching
kwargs['jobname'] = 'glide_server'
if mmutil.feature_flag_is_enabled(mmutil.FAST_LIGAND_DESIGNER):
ServerCls = http_client.NonBlockingGlideServerManagerZmq
kwargs['nworkers'] = 2
else:
ServerCls = http_client.NonBlockingGlideServerManager
kwargs['timeout'] = 1200 # 20 minutes
server = ServerCls(docking_keywords, **kwargs)
ready = server.isReady() # Check whether server is already running
if not ready:
try:
server.start()
except Exception as exc:
self._recordFailure(exc)
return
else:
ready = True
if ready:
self.output.server = server
[docs]class GridgenTask(jobtasks.CmdJobTask):
"""
Task to run glide grid generation.
:cvar RUNNING_MESSAGE: Message for RUNNING status
:cvar FAILED_MESSAGE: Message for FAILED status
"""
RUNNING_MESSAGE = 'Generating grid...'
FAILED_MESSAGE = 'Grid generation failed'
infile = parameters.NonParamAttribute()
[docs] class Output(parameters.CompoundParam):
gridfile: str = None
[docs] def initConcrete(self):
super().initConcrete()
self.name = os.path.splitext(GRIDFILE_NAME)[0]
self.infile = self.name + ".in"
@tasks.preprocessor(order=tasks.AFTER_TASKDIR)
def _writeInputs(self):
logger.debug("Writing gridgen input files")
self._writeReflig()
pvfile = self.getTaskFilename(self.name + "_in.maegz")
with structure.StructureWriter(pvfile) as writer:
writer.append(self.input.receptor_st)
writer.append(self.input.ligand_st)
keywords = {
'RECEP_FILE': pvfile,
'LIGAND_INDEX': 2,
'GRIDFILE': self.getTaskFilename(GRIDFILE_NAME),
}
glide_job = glide.get_glide_job(keywords)
infile = self.getTaskFilename(self.infile)
glide_job.writeSimplified(infile)
[docs] def makeCmd(self):
infile = self.getTaskFilename(self.infile)
return ['$SCHRODINGER/glide', infile]
@tasks.postprocessor
def _checkOutput(self):
gridfile = self.getTaskFilename(GRIDFILE_NAME)
if not os.path.isfile(gridfile):
return False, "Gridfile not found"
self.output.gridfile = gridfile
[docs] def checkGridfile(self, match_reflig=False):
"""
Return whether the specified taskdir contains a gridfile or reference
ligand that match `input.ligand_st`
:param bool match_reflig: Whether to allow match based on reference
ligand only
"""
taskdir = self.taskDirSetting()
if not isinstance(taskdir, (str, Path)):
raise ValueError("Can only be used with a specified taskdir")
elif not os.path.exists(taskdir):
raise ValueError("Can only be used with an existing taskdir")
self._createTaskDir() # will be a no-op but allows calling getTaskDir
if self._checkGridfile():
return True
elif match_reflig and self._checkReflig():
return True
return False
def _checkGridfile(self):
"""
Check whether the taskdir contains a gridfile compatible with the input
ligand. If so, updates the reference ligand file.
"""
grid_file = self.getTaskFilename(GRIDFILE_NAME)
if (glide_utils.check_required_gridfiles(grid_file) and
glide_utils.is_grid_good_for_ligand(grid_file,
self.input.ligand_st)):
self._writeReflig()
return True
return False
def _checkReflig(self):
ref_lig_file = self.getTaskFilename(REFLIG_NAME)
if os.path.isfile(ref_lig_file):
ref_lig_st = structure.Structure.read(ref_lig_file)
if ref_lig_st.isEquivalent(self.input.ligand_st):
return True
return False
def _writeReflig(self):
ref_lig_file = self.getTaskFilename(REFLIG_NAME)
self.input.ligand_st.write(ref_lig_file)
[docs]def read_json_file(filename):
"""
Read a JSON file. If there are issues reading it (doesn't exist, syntax
errors...) quietly return an empty dict.
:type filename: str
:rtype: object
"""
try:
with open(filename) as fh:
return json.load(fh)
except (IOError, ValueError):
return {}
[docs]def md5sum(input_str):
"""
MD5 hex digest of a string.
:type input_str: str
:rtype: str
"""
m = hashlib.md5()
m.update(input_str.encode('utf-8'))
return m.hexdigest()
[docs]def get_structure_digest(st, length=8):
"""
Return an abbreviated MD5 hex digest given a Structure, considering only
element, formal charge, and XYZ coordinates.
:param st: input structure (not modified)
:type st: schrodinger.structure.Structure
:param length: digest length in hex digits
:type length: int
:return: hex digest
:rtype: str
"""
receptor_str = '\n'.join(
'{} {:d} {:.3} {:.3} {:.3}'.format(a.element, a.formal_charge, *a.xyz)
for a in st.atom)
return md5sum(receptor_str)[:length]