"""
calphy: a Python library and command line interface for automated free
energy calculations.
Copyright 2021-2026 (c) Sarath Menon, Yury Lysogorskiy, Ralf Drautz
Interdisciplinary Centre for Advanced Materials Simulation (ICAMS),
Ruhr University Bochum, 44801 Bochum, Germany
calphy is published and distributed under the Academic Software Licence v1.0 (ASL).
calphy is distributed in the hope that it will be useful for non-commercial academic
research, but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the LICENSE file for details.
The ASL permits academic non-commercial use only. Contact
sarath.menon@ruhr-uni-bochum.de to enquire about commercial use rights.
More information about the program can be found in:
Menon, Sarath, Yury Lysogorskiy, Jutta Rogal, and Ralf Drautz.
"Automated Free Energy Calculation from Atomistic Simulations." Physical Review Materials 5(10), 2021
DOI: 10.1103/PhysRevMaterials.5.103801
For more information contact:
sarath.menon@ruhr-uni-bochum.de
"""
import subprocess as sub
import os
import stat
[docs]class Local:
"""
Local submission script
"""
[docs] def __init__(self, options, cores=1, directory=os.getcwd()):
self.queueoptions = {
"scheduler": "local",
"jobname": "tis",
"queuename": None,
"memory": None,
"cores": cores,
"hint": None,
"directory": directory,
"options": [],
"commands": [],
"header": "#!/bin/bash",
}
for key, val in options.items():
if key in self.queueoptions.keys():
if val is not None:
self.queueoptions[key] = val
self.maincommand = ""
[docs] def write_script(self, outfile):
"""
Write the script file
"""
jobout = os.path.join(self.queueoptions["directory"], "local.out")
joberr = os.path.join(self.queueoptions["directory"], "local.err")
with open(outfile, "w") as fout:
fout.write(self.queueoptions["header"])
fout.write("\n")
# now finally commands
for command in self.queueoptions["commands"]:
fout.write("%s\n" % command)
fout.write("%s > %s 2> %s\n" % (self.maincommand, jobout, joberr))
self.script = outfile
[docs] def submit(self):
"""
Submit the job
"""
st = os.stat(self.script)
os.chmod(self.script, st.st_mode | stat.S_IEXEC)
cmd = [self.script]
proc = sub.Popen(cmd, stdin=sub.PIPE, stdout=sub.PIPE, stderr=sub.PIPE)
return proc
[docs]class SLURM:
"""
Slurm class for writing submission script
"""
[docs] def __init__(self, options, cores=1, directory=os.getcwd()):
"""
Create class
"""
self.queueoptions = {
"scheduler": "slurm",
"jobname": "tis",
"queuename": None,
"walltime": "23:59:00",
"memory": None,
"cores": cores,
"cpus_per_task": None,
"gres": None,
"account": None,
"hint": "nomultithread",
"directory": directory,
"options": {},
"commands": [
"uss=$(whoami)",
"find /dev/shm/ -user $uss -type f -mmin +30 -delete",
],
"header": "#!/bin/bash",
}
for key, val in options.items():
if key in self.queueoptions.keys():
self.queueoptions[key] = val
self.maincommand = ""
[docs] def write_script(self, outfile):
"""
Write the script file
"""
jobout = os.path.join(self.queueoptions["directory"], f"{outfile}.slurm.out")
joberr = os.path.join(self.queueoptions["directory"], f"{outfile}.slurm.err")
with open(outfile, "w") as fout:
fout.write(self.queueoptions["header"])
fout.write("\n")
# write the main header options
fout.write("#SBATCH --job-name=%s\n" % self.queueoptions["jobname"])
fout.write("#SBATCH --time=%s\n" % self.queueoptions["walltime"])
if self.queueoptions["queuename"] is not None:
fout.write("#SBATCH --partition=%s\n" % self.queueoptions["queuename"])
fout.write("#SBATCH --ntasks=%s\n" % str(self.queueoptions["cores"]))
if self.queueoptions["cpus_per_task"] is not None:
fout.write(
"#SBATCH --cpus-per-task=%s\n"
% str(self.queueoptions["cpus_per_task"])
)
if self.queueoptions["memory"] is not None:
fout.write("#SBATCH --mem=%s\n" % self.queueoptions["memory"])
if self.queueoptions["hint"] is not None:
fout.write("#SBATCH --hint=%s\n" % self.queueoptions["hint"])
fout.write("#SBATCH --chdir=%s\n" % self.queueoptions["directory"])
# now write extra options
for key, value in self.queueoptions["options"].items():
fout.write(f"#SBATCH --{key}={value}\n")
# now finally commands
for command in self.queueoptions["commands"]:
fout.write("%s\n" % command)
fout.write("%s > %s 2> %s\n" % (self.maincommand, jobout, joberr))
self.script = outfile
[docs] def submit(self):
"""
Submit the job
"""
cmd = ["sbatch", self.script]
proc = sub.Popen(cmd, stdin=sub.PIPE, stdout=sub.PIPE, stderr=sub.PIPE)
print(f'submitting {self.queueoptions["jobname"]}')
stdout, stderr = proc.communicate()
if proc.returncode != 0:
print("Failed to submit!")
print("= standard output")
print(stdout.decode("utf8"))
print("= standard error")
print(stderr.decode("utf8"))
return proc
[docs]class SGE:
"""
Slurm class for writing submission script
"""
[docs] def __init__(self, options, cores=1, directory=os.getcwd()):
"""
Create class
"""
self.queueoptions = {
"scheduler": "sge",
"jobname": "tis",
"walltime": "23:59:00",
"queuename": None,
"memory": "3GB",
"system": "smp",
"commands": [],
"options": [
"-j y",
"-R y",
"-P ams.p",
],
"cores": cores,
"hint": None,
"directory": directory,
"header": "#!/bin/bash",
}
for key, val in options.items():
if key in self.queueoptions.keys():
if val is not None:
self.queueoptions[key] = val
self.maincommand = ""
[docs] def write_script(self, outfile):
"""
Write the script file
"""
jobout = os.path.join(self.queueoptions["directory"], "sge.out")
joberr = os.path.join(self.queueoptions["directory"], "sge.err")
with open(outfile, "w") as fout:
fout.write(self.queueoptions["header"])
fout.write("\n")
# write the main header options
fout.write("#$ -N %s\n" % self.queueoptions["jobname"])
fout.write("#$ -l h_rt=%s\n" % self.queueoptions["walltime"])
fout.write("#$ -l qname=%s\n" % self.queueoptions["queuename"])
fout.write(
"#$ -pe %s %s\n"
% (self.queueoptions["system"], str(self.queueoptions["cores"]))
)
fout.write("#$ -l h_vmem=%s\n" % self.queueoptions["memory"])
fout.write("#$ -cwd %s\n" % self.queueoptions["directory"])
# now write extra options
for option in self.queueoptions["options"]:
fout.write("#$ %s\n" % option)
# now finally commands
for command in self.queueoptions["commands"]:
fout.write("%s\n" % command)
fout.write("%s > %s 2> %s\n" % (self.maincommand, jobout, joberr))
self.script = outfile
[docs] def submit(self):
"""
Submit the job
"""
cmd = ["qsub", self.script]
proc = sub.Popen(cmd, stdin=sub.PIPE, stdout=sub.PIPE, stderr=sub.PIPE)
return proc