[WIP] First draft of the API of WF generators for atomate v2.0.

atomate/abinit/firetasks/__init__.py

@@ -0,0 +1 @@
+__author__ = 'David Waroquiers <[email protected]>'

atomate/abinit/firetasks/common.py

@@ -0,0 +1,11 @@
+# coding: utf-8
+
+
+"""
+This module defines variables and functions for Abinit tasks.
+"""
+
+
+TMPDIR_NAME = "tmpdata"
+OUTDIR_NAME = "outdata"
+INDIR_NAME = "indata"

atomate/abinit/firetasks/control_tasks.py

@@ -0,0 +1,34 @@
+# coding: utf-8
+
+
+"""
+This module defines tasks that checks the proper execution of Abinit, and possibly takes appropriate measures to
+correct errors or problems.
+"""
+
+
+from fireworks import explicit_serialize, FiretaskBase
+
+from atomate.utils.utils import get_logger
+
+
+__all__ = ("CheckTask", )
+
+
+logger = get_logger(__name__)
+
+__author__ = 'David Waroquiers'
+__email__ = '[email protected]'
+
+
+@explicit_serialize
+class CheckTask(FiretaskBase):
+    """
+    Check an Abinit Run and take appropriate corrections if errors occurred.
+    """
+
+    required_params = []
+    optional_params = []
+
+    def run_task(self, fw_spec):
+        raise NotImplementedError

atomate/abinit/firetasks/glue_tasks.py

@@ -0,0 +1,77 @@
+# coding: utf-8
+
+
+"""
+This module defines tasks that acts as a glue between other Abinit Firetasks to allow communication
+between different Firetasks and Fireworks.
+"""
+
+
+from fireworks import explicit_serialize, FiretaskBase
+
+from atomate.utils.utils import get_logger
+from atomate.common.firetasks.glue_tasks import CopyFiles
+
+
+__all__ = ("CopyAbinitOutputs",
+           "CreateAbinitFolders",
+           "ResolveDeps",)
+
+
+logger = get_logger(__name__)
+
+__author__ = 'David Waroquiers'
+__email__ = '[email protected]'
+
+
+@explicit_serialize
+class CopyAbinitOutputs(CopyFiles):
+    """
+    Copy files from one or multiple previous Abinit run directories to the current directory.
+
+    TODO: define what needs to be copied, what is required as inputs etc ... also think about a more general CopyFiles
+    TODO: is this task needed or do we put everything in a ResolveDeps task or the ResolveDeps task just defines what needs to be copied to a general CopyFiles task ?
+    """
+
+    required_params = []
+    optional_params = []
+
+    def run_task(self, fw_spec):
+        raise NotImplementedError
+
+
+@explicit_serialize
+class CreateAbinitFolders(FiretaskBase):
+    """
+    Firetask to create in, out and tmp folders for abinit.
+    TODO: is it worth to allow changing them ??
+    TODO: see if this should be a subclass of a general CreateFolders Task ? (there is the CreateFolder task that creates just ONE folder and optionally moves to it)
+
+    Optional params:
+        in_dir(str): Directory for abinit input files (e.g. _DEN file from previous calculation).
+        out_dir(str): Directory for abinit output files (e.g. final _DEN file of the current calculation).
+        tmp_dir(str): Directory for temporary abinit files during calculation.
+    """
+
+    required_params = []
+    optional_params = ["in_dir", "out_dir", "tmp_dir"]
+
+    def run_task(self, fw_spec):
+        raise NotImplementedError
+
+
+@explicit_serialize
+class ResolveDeps(FiretaskBase):
+    """
+    Firetask resolve dependencies for abinit.
+
+    Should deal whether its a restart of a non converged calculation or a follow up calculation (e.g. nscf after scf).
+
+    TODO: define required and optional params
+    """
+
+    required_params = []
+    optional_params = []
+
+    def run_task(self, fw_spec):
+        raise NotImplementedError

atomate/abinit/firetasks/run_tasks.py

@@ -0,0 +1,52 @@
+# coding: utf-8
+
+
+"""
+This module defines tasks to run Abinit or other related executables (e.g. anaddb, mrgddb, ...).
+"""
+
+
+from fireworks import explicit_serialize, FiretaskBase
+
+from atomate.utils.utils import get_logger
+
+
+__all__ = ("RunAbinit",
+           "RunAnaDdb",)
+
+
+logger = get_logger(__name__)
+
+__author__ = 'David Waroquiers'
+__email__ = '[email protected]'
+
+
+
+@explicit_serialize
+class RunAbinit(FiretaskBase):
+    """
+    Run Abinit.
+
+    TODO: define if/what are required/optional params
+    """
+
+    required_params = []
+    optional_params = []
+
+    def run_task(self, fw_spec):
+        raise NotImplementedError
+
+
+@explicit_serialize
+class RunAnaDdb(FiretaskBase):
+    """
+    Run anaddb.
+
+    TODO: define if/what are required/optional params
+    """
+
+    required_params = []
+    optional_params = []
+
+    def run_task(self, fw_spec):
+        raise NotImplementedError

atomate/abinit/firetasks/write_input_tasks.py

@@ -0,0 +1,37 @@
+# coding: utf-8
+
+
+"""
+This module defines tasks to write inputs for Abinit or other related executables.
+"""
+
+
+from fireworks import explicit_serialize, FiretaskBase
+
+from atomate.utils.utils import get_logger
+
+
+__all__ = ("WriteInput",)
+
+
+logger = get_logger(__name__)
+
+__author__ = 'David Waroquiers'
+__email__ = '[email protected]'
+
+
+
+@explicit_serialize
+class WriteInput(FiretaskBase):
+    """
+    Write input for Abinit.
+
+    TODO: define if/what are required/optional params
+    """
+
+    required_params = []
+    optional_params = []
+
+    def run_task(self, fw_spec):
+        raise NotImplementedError
+

atomate/abinit/fireworks/__init__.py

@@ -0,0 +1 @@
+__author__ = 'David Waroquiers <[email protected]>'

atomate/abinit/fireworks/core.py

@@ -0,0 +1,44 @@
+# coding: utf-8
+
+"""
+This module defines the function to generate Fireworks for abinit.
+"""
+
+
+from fireworks import Firework
+from atomate.abinit.firetasks.glue_tasks import CreateAbinitFolders
+from atomate.abinit.firetasks.glue_tasks import ResolveDeps
+from atomate.abinit.firetasks.glue_tasks import CopyAbinitOutputs
+from atomate.abinit.firetasks.write_input_tasks import WriteInput
+from atomate.abinit.firetasks.run_tasks import RunAbinit
+from atomate.abinit.firetasks.control_tasks import CheckTask
+from atomate.common.firetasks.glue_tasks import PassCalcLocs
+
+
+__all__ = ("scf_fw", )
+
+
+def scf_fw(structure):
+    """Generate a Firework for a Self-Consistent Field (SCF) calculation.
+
+    Args:
+        structure: Pymatgen Structure object.
+        TODO: Add arguments needed here
+
+    Returns:
+        Firework: Firework object with all Firetasks needed for an SCF calculation.
+    """
+
+    spec = {'structure': structure}
+    #TODO: only a very general structure is done here :) goal is to foster discussion on the overall architecture
+    tasks = []
+    tasks.append(CreateAbinitFolders())
+    tasks.append(ResolveDeps())  # Does it link/move/copy the files directly or is it just to define which files have to be copied where and then the next task does the work ?
+    tasks.append(CopyAbinitOutputs())  # Or a general CopyFiles task, what about moving or symbolic links ?
+    tasks.append(WriteInput())
+    tasks.append(RunAbinit())
+    tasks.append(CheckTask())
+    tasks.append(PassCalcLocs())
+
+    raise NotImplementedError
+    # return Firework(tasks=tasks, spec=spec)

atomate/abinit/workflows/__init__.py

@@ -0,0 +1 @@
+__author__ = 'David Waroquiers <[email protected]>'

atomate/abinit/workflows/core.py

@@ -0,0 +1,52 @@
+# coding: utf-8
+
+"""
+This module defines Workflow generators for abinit.
+"""
+
+
+from atomate.common.workflows.base import BaseStructureWFGenerator
+from atomate.abinit.fireworks.core import scf_fw
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Union
+
+from pymatgen.core.structure import Structure
+from fireworks import Workflow
+
+
+__all__ = ("ScfWFGenerator", )
+
+
+class ScfWFGenerator(BaseStructureWFGenerator):
+    """Workflow generator for Self-Consistent Field (SCF) calculations."""
+
+    def __init__(self):
+        """Constructor for the SCF Workflow generator."""
+
+        raise NotImplementedError
+
+    def get_wf(self, structure: Optional[Structure] = None,
+               name: str = 'Structure Workflow',
+               prev_calc_locs: Optional[Union[List[str], str, bool]] = None,
+               metadata: Optional[Union[Dict["str", Any], bool]] = None,
+               name_tag: Optional[str] = None):
+
+        """
+        Generate and return a SCF Workflow.
+
+        Args:
+            structure: Pymatgen Structure object to be used as an input for the Workflow.
+            name: Name of this workflow.
+            prev_calc_locs:
+            metadata:
+            name_tag:
+
+        Returns:
+            Fireworks' Workflow object
+        """
+
+        fw = scf_fw(structure=structure)
+        return Workflow([fw])

atomate/common/workflows/__init__.py

@@ -0,0 +1 @@
+__author__ = 'David Waroquiers <[email protected]>'