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.remd as remd
import schrodinger.application.desmond.stage as stage
import schrodinger.job.jobcontrol as jobcontrol
import schrodinger.structure as structure
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