Source code for schrodinger.application.desmond.rest_inp
"""
It prepares or runs a REST MD job.
Copyright Schrodinger, LLC. All rights reserved.
"""
import os
import schrodinger.application.desmond.cmj as cmj
import schrodinger.application.desmond.cms as cms
import schrodinger.application.desmond.envir as envir
import schrodinger.application.desmond.stage as stage # noqa: F401
import schrodinger.job.jobcontrol as jobcontrol
import schrodinger.structutils.analyze as analyze
from schrodinger.application.desmond import constants
from schrodinger.infra import mm
EXCHANGE_PROBABILITY = 0.3
[docs]def get_rest_input(model, jobname, host, **kwargs):
"""
Function to prepare rest input.
:type structure: `Structure` or `cms.Cms`
:param structure: Structure to apply rest hot region.
:type jobname: str
:param jobname: jobname
:type hostname: str
:param hostname: hostname
:type kwargs: dict
:param kwargs: dictionary to set options
:rtype: tuple
:return: A tuple composed of (structure, msj, command)
"""
msj_fname = jobname + ".msj"
mae_fname = jobname + ".mae"
if isinstance(model, cms.Cms):
mae_fname = jobname + ".cms"
rest = RestInput(model, **kwargs)
rest_model = rest.prepare_structure()
msj_string = rest.prepare_msj()
cmd = rest.prepare_command(jobname, host, msj_fname, mae_fname)
return (rest_model, msj_string, cmd)
[docs]def add_rest_region(model, **kwargs):
"""
Function to add rest region.
:type structure: `Structure` or `cms.Cms`
:param structure: Structure to apply rest hot region.
:type kwargs: dict
:param kwargs: dictionary to set options
:rtype: tuple
:return: A tuple composed of (structure, msj)
msj only has "replica_exchange" stage
"""
rest = RestInput(model, **kwargs)
rest_model = rest.prepare_structure()
msj_string = rest.prepare_msj()
all_stages = cmj.parse_msj(None, msj_content=msj_string)
all_stages = [x for x in all_stages if x.NAME == "replica_exchange"]
msj_string = cmj.write_msj(all_stages, to_str=True)
return (rest_model, msj_string)
[docs]def get_rest_region(model):
rest_asl = f"atom.{constants.REST_HOTREGION} > 0"
if isinstance(model, cms.Cms):
return model.select_atom(rest_asl)
else:
return analyze.evaluate_asl(model, rest_asl)
[docs]class RestInput:
[docs] def __init__(self, model, **kwargs):
self.model = model
self.asl = "all"
self.ref_temperature = 300
self.ensemble = "NPT" # or NVT
self.forcefield = mm.OPLS_NAME_F16 # or OPLS_2005
self.time = 5000
self.n_replica = 2
self.trajectory_interval = 24
self.np = 2
self.update(**kwargs)
# FIXME explicitly handle values!
[docs] def prepare_structure(self):
"""
Set hot region
"""
if isinstance(self.model, cms.Cms):
for a in self.model.fsys_ct.atom:
try:
del a.property[constants.REST_HOTREGION]
except:
pass
for ct in self.model.comp_ct:
for a in ct.atom:
try:
del a.property[constants.REST_HOTREGION]
except:
pass
# setup hot region atom property
for aindex in self.model.select_atom(self.asl):
self.model.atom[aindex].property[constants.REST_HOTREGION] = 1
for ct_index, atom_list in enumerate(
self.model.select_atom_comp(self.asl)):
for aindex in atom_list:
self.model.comp_ct[ct_index].atom[aindex].property[
constants.REST_HOTREGION] = 1
else:
for a in self.model.atom:
try:
del a.property[constants.REST_HOTREGION]
except:
pass
for a in analyze.get_atoms_from_asl(self.model, self.asl):
a.property[constants.REST_HOTREGION] = 1
return self.model
[docs] def prepare_msj(self):
"""
Prepare multisim file
"""
template_fname = os.path.join(envir.CONST.MMSHARE_DATA_DESMOND_DIR,
"rest_template.msj")
try:
fh = open(template_fname, "r")
s = fh.read()
fh.close()
except IOError:
raise RuntimeError("reading '%s' failed." % template_fname)
all_stages = cmj.parse_msj(None, msj_content=s)
for stg in all_stages:
if stg.NAME == "assign_forcefield":
stg.param.forcefield.val = self.forcefield
if stg.NAME == "simulate" and self.ensemble == "NVT":
stg.param.ensemble.class_.val = self.ensemble
if stg.NAME == "replica_exchange":
stg.param.temperature.val = self.ref_temperature
stg.param.ensemble.val = self.ensemble
stg.param.time.val = self.time
stg.param.replica.temperature.n_replica.val = self.n_replica
stg.param.replica.temperature.exchange_probability.val = EXCHANGE_PROBABILITY
stg.param.trajectory.interval.val = self.trajectory_interval
if (self.n_replica % self.np):
raise RuntimeError(
"# replica must be multiples of # of gpu: (gpu=%d, n_replia=%d)"
% (self.np, self.n_replica))
stg.param.total_proc.val = self.np
# cms structure does not need build_geometry stage
if isinstance(self.model, cms.Cms):
all_stages = [x for x in all_stages if x.NAME != "build_geometry"]
s = cmj.write_msj(all_stages, to_str=True)
return s
[docs] def prepare_command(self, jobname, host, msj_fname, mae_fname):
"""
Prepare command line
"""
cmd = [os.path.join('$SCHRODINGER', 'utilities', 'multisim')]
cmd += [
'-m',
msj_fname,
'-HOST',
host + ":%d" % self.np,
'-JOBNAME',
jobname,
'-verbose',
'-mode',
"umbrella",
]
cmd += [
mae_fname,
]
return cmd
[docs] def write(self, jobname, host):
'''
writes msj and mae files
'''
msj_fname = jobname + ".msj"
mae_fname = jobname + ".mae"
st = self.prepare_structure()
st.write(mae_fname)
s = self.prepare_msj()
cmd = self.prepare_command(jobname, host, msj_fname, mae_fname)
cmd_line = ' '.join(cmd)
s += '# commandline: ' + cmd_line + '\n'
fh = open(msj_fname, "w")
print(s, file=fh)
fh.close()
return cmd, s
[docs] def run(self, jobname, host):
'''
Run MD REST Job
'''
cmd, s = self.write(jobname, host)
job = jobcontrol.launch_job(cmd)
if not job:
raise RuntimeError("REST MD job submission failed.")
return job