Pull Request: Automatic K-Point Convergence

dmrdjenovich/Convergence.py

@@ -0,0 +1,126 @@
+from computing.executable import Executable
+from crystal import reciprocal
+from crystal.voronoi import Voronoi
+from simulation.qe_spec import QESpec
+from simulation.qe_run import QERun
+from simulation.qe_process import QEProcess
+from simulation.simulation import Simulation
+
+import math
+import os
+import numpy as np
+
+class Convergence(Executable):
+    """
+    Class responsible for running a series of
+    Simulations to do a convergence test for
+    a given material system.
+    """
+
+    def __init__(self, dir, encut, thresh, rsx, input_name="input.txt", meta=None, obtusify=True, homogeneous_k=False):
+        """
+        Sets up a convergence test in the specified
+        directory, using the provided QESpec, and
+        a metadata string for logging purposes.
+        """
+        self.dir = dir
+        self.input_name = input_name
+        self.spec = QESpec.parse_file(os.path.join(dir, input_name))
+        self.spec.set_encut(encut)
+        self.thresh = thresh
+        if obtusify:
+            Convergence._set_obtuse_reciprocal_lattice(self.spec)
+        lattice = self.spec.get_lattice()
+        r_lattice = np.transpose(reciprocal.get_reciprocal(lattice))
+        r_lengths = r_lattice[0]*r_lattice[0]
+        for i in range(1, len(r_lattice)):
+            r_lengths += r_lattice[i]*r_lattice[i]
+        self.r_lengths = list(r_lengths)
+        for i in range(0, len(self.r_lengths)):
+            self.r_lengths[i] = self.r_lengths[i]**0.5
+        self.meta = meta
+        self.homogeneous_k = homogeneous_k
+        self.rsx = rsx
+        self.k = self.spec.get_k_points()
+        self.ks = []
+        self.values = []
+
+    @staticmethod
+    def _set_obtuse_reciprocal_lattice(spec):
+        if spec.get_lattice_type() != 0:
+            return
+        try:
+            start = spec.get_lattice()
+            reci = reciprocal.get_reciprocal(start)
+            reci_obtuse = Voronoi.obtuse_basis(reci)
+            real_reci_obtuse = reciprocal.get_reciprocal(reci_obtuse)
+            unimodular = np.matmul(np.linalg.inv(start), real_reci_obtuse)
+            b_start = spec.get_positions()
+            b_end = list(np.matmul(np.linalg.inv(unimodular), b_start[1]))
+            spec.set_lattice(real_reci_obtuse)
+            spec.set_positions(b_start[0], b_end)
+        except Exception, e:
+            return
+
+    def get_results(self, analysis):
+        """
+        Extract the value used to test convergence using the
+        provided QEAnalysis object.
+        """
+        pass
+
+    def get_resources(self):
+        return self.rsx
+
+    def run(self, envr):
+        sim_start = self.next_executable()
+        if not sim_start.run(envr):
+            print("Error encountered in starting simulation.")
+            return False
+        self.next_k()
+        sim_next = self.next_executable()
+        if not sim_next.run(envr):
+            print("Error encountered in simulation.")
+            return False
+        while abs(self.values[len(self.values) - 1] -
+                  self.values[len(self.values) - 2]) > self.thresh:
+            self.next_k()
+            sim_next = self.next_executable()
+            if not sim_next.run(envr):
+                print("Error encountered in simulation.")
+                return False
+        return True
+
+    def next_executable(self):
+        self.spec.set_k_points(self.k)
+        run = Convergence.CustomRun(self.dir, self.rsx, self.input_name)
+        process = Convergence.CustomProcess(self.dir, self.spec, self)
+        sim = Simulation.construct_override(self.dir, self.spec, self.rsx, None, run, process)
+        return sim
+
+    def next_k(self):
+        last_k = self.k
+        if self.homogeneous_k:
+            self.k = list(last_k)
+            self.k[0] += 1
+            self.k[1] += 1
+            self.k[2] += 1
+        else:
+            lin_density = [last_k[i]/self.r_lengths[i] for i in range(0, len(self.r_lengths))]
+            choice = lin_density.index(min(lin_density))
+            self.k = [x for x in last_k]
+            self.k[choice] += 1
+
+    class CustomRun(QERun):
+        def __init__(self, dir, spec, input_file):
+            super(Convergence.CustomRun, self).__init__(dir, spec, input_file=input_file)
+
+    class CustomProcess(QEProcess):
+        def __init__(self, dir, spec, parent):
+            super(Convergence.CustomProcess, self).__init__(dir, spec)
+            self.parent = parent
+        def get_results(self, analysis):
+            outcome = self.parent.get_results(analysis)
+            self.parent.ks.append(self.parent.k)
+            self.parent.values.append(outcome)
+            return outcome

dmrdjenovich/README.md

@@ -0,0 +1,37 @@
+# K-point convergence tracker (Materials)
+
+> Ideal candidate: scientists skilled in Density Functional Theory and proficient in python.
+
+# Overview
+
+The aim of this task is to create a python package that implements automatic convergence tracking mechanism for a materials simulations engine. The convergence is tracked with respect to the k-point sampling inside a reciprocal cell of a crystalline compound.
+
+# Requirements
+
+1. automatically find the dimensions of a k-point mesh that satisfy a certain criteria for total energy (eg. total energy is converged within dE = 0.01meV)
+1. the code shall be written in a way that can facilitate easy addition of convergence wrt other characteristics extracted from simulations (forces, pressures, phonon frequencies etc)
+1. the code shall support VASP or Quantum ESPRESSO
+
+# Expectations
+
+- correctly find k-point mesh that satisfies total energy convergence parameters for a set of 10 materials, starting from Si2, as simplest, to a 10-20-atom supercell of your choice
+- modular and object-oriented implementation
+- commit early and often - at least once per 24 hours
+
+# Timeline
+
+We leave exact timing to the candidate. Must fit Within 5 days total.
+
+# User story
+
+As a user of this software I can start it passing:
+
+- path to input data (eg. pw.in / POSCAR, INCAR, KPOINTS) and
+- kinetic energy cutoff
+
+as parameters and get the k-point dimensions (eg. 5 5 5).
+
+# Notes
+
+- create an account at exabyte.io and use it for the calculation purposes
+- suggested modeling engine: Quantum ESPRESSO

dmrdjenovich/computing/Executable.py

@@ -0,0 +1,23 @@
+
+class Executable(object):
+    """
+    Class representing a chunk of computational work.
+
+    Has time and node / cpu requirements, specified
+    by a Resources object.
+    """
+
+    def get_resources(self):
+        """
+        Returns the required resources for this
+        computational task.
+        """
+        pass
+
+    def run(self, envr):
+        """
+        Given a Resource object representing the
+        available computational resources, runs
+        this computational task.
+        """
+        pass

dmrdjenovich/computing/Resources.py

@@ -0,0 +1,16 @@
+
+class Resources(object):
+    """
+    Class representing a measure of computational
+    work: has time and node / cpu requirements.
+    """
+
+    def __init__(self, nodes, time):
+        self.nodes = nodes
+        self.time = time
+
+    def get_nodes(self):
+        return self.nodes
+
+    def get_time(self):
+        return self.time

dmrdjenovich/computing/b_executable_p2.py

@@ -0,0 +1,94 @@
+from executable import Executable
+
+import os
+import subprocess
+
+class BExecutableP2(Executable):
+    """
+    Class automatically handling the details of scheduling a
+    SHELL task using python 2.
+    """
+
+    def get_shell_string(self):
+        """
+        Returns the command that will be interpreted by the SHELL.
+        """
+        pass
+
+    def get_shell_name(self):
+        """
+        Returns the name of the SHELL that will be invoked to run
+        the command.
+        """
+        pass
+
+    def get_std_out(self):
+        """
+        Returns the absolute canonical filepath for the redirected
+        standard out stream. If null, no redirect will occur.
+        """
+        pass
+
+    def get_std_err(self):
+        """
+        Returns the absolute canonical filepath for the redirected
+        standard error stream. If null, no redirect will occur.
+        """
+        pass
+
+    def get_working_dir(self):
+        """
+        Returns the absolute canonical filepath for the working
+        directory in which the command will be run.
+        """
+        pass
+
+    def receive_environment(self, envr):
+        """
+        Given the current execution envrionment resources,
+        sets up the executable to run appropriately.
+        """
+        pass
+
+    def run(self, envr):
+        """
+        Runs the specified shell string computational task.
+        """
+        cmd = self.get_shell_string()
+        wd_str = self.get_working_dir()
+        if not os.path.isdir(wd_str):
+            raise RuntimeException("Unable to locate working directory.")
+        std_out = None
+        std_err = None
+
+        try:
+            std_out_call = self.get_std_out()
+            if not std_out_call is None:
+                std_out = open(std_out_call, "w")
+            std_err_call = self.get_std_err()
+            if not std_err_call is None:
+                std_err = open(std_err_call, "w")
+            result = subprocess.call([self.get_shell_name(), "-c", cmd],
+                                    stdout=std_out, stderr=std_err,
+                                    cwd=wd_str)
+            return result == 0
+
+        except IOError as e:
+            print(e)
+            return False
+
+        finally:
+            if not std_out is None:
+                try:
+                    std_out.close()
+                except IOError as e2:
+                    pass
+            if not std_err is None:
+                try:
+                    std_err.close()
+                except IOError as e2:
+                    pass
+
+
+
+

dmrdjenovich/computing/b_executable_p3.py

@@ -0,0 +1,94 @@
+from executable import Executable
+
+import os
+import subprocess
+
+class BExecutableP3(Executable):
+    """
+    Class automatically handling the details of scheduling a
+    SHELL task using python 3.
+    """
+
+    def get_shell_string(self):
+        """
+        Returns the command that will be interpreted by the SHELL.
+        """
+        pass
+
+    def get_shell_name(self):
+        """
+        Returns the name of the SHELL that will be invoked to run
+        the command.
+        """
+        pass
+
+    def get_std_out(self):
+        """
+        Returns the absolute canonical filepath for the redirected
+        standard out stream. If null, no redirect will occur.
+        """
+        pass
+
+    def get_std_err(self):
+        """
+        Returns the absolute canonical filepath for the redirected
+        standard error stream. If null, no redirect will occur.
+        """
+        pass
+
+    def get_working_dir(self):
+        """
+        Returns the absolute canonical filepath for the working
+        directory in which the command will be run.
+        """
+        pass
+
+    def receive_environment(self, envr):
+        """
+        Given the current execution envrionment resources,
+        sets up the executable to run appropriately.
+        """
+        pass
+
+    def run(self, envr):
+        """
+        Runs the specified shell string computational task.
+        """
+        cmd = self.get_shell_string()
+        wd_str = self.get_working_dir()
+        if not os.path.isdir(wd_str):
+            raise RuntimeException("Unable to locate working directory.")
+        std_out = None
+        std_err = None
+
+        try:
+            std_out_call = self.get_std_out()
+            if not std_out_call is None:
+                std_out = open(std_out_call, "w")
+            std_err_call = self.get_std_err()
+            if not std_err_call is None:
+                std_err = open(std_err_call, "w")
+            result = subprocess.run([self.get_shell_name(), "-c", cmd],
+                                    stdout=std_out, stderr=std_err,
+                                    cwd=wd_str)
+            return True
+
+        except IOError as e:
+            print(e)
+            return False
+
+        finally:
+            if not std_out is None:
+                try:
+                    std_out.close()
+                except IOError as e2:
+                    pass
+            if not std_err is None:
+                try:
+                    std_err.close()
+                except IOError as e2:
+                    pass
+
+
+
+