Refactoring...

src/atomate2/abinit/flows/core.py

@@ -13,6 +13,7 @@
 
 @dataclass
 class LineBandStructureMaker(Maker):
+    # TODO: make this more similar to Vasp
     """
     Maker to generate line abinit band structure.
 

src/atomate2/abinit/jobs/base.py

@@ -10,47 +10,27 @@
 from typing import Any, ClassVar, Optional, Sequence, Union
 
 import jobflow
-from abipy.flowtk.events import AbinitEvent
+
+# from abipy.flowtk.events import AbinitEvent
+from abipy.flowtk.events import as_event_class
 from jobflow import Maker, Response, job
-from monty.inspect import all_subclasses
 from monty.json import jsanitize
 from monty.serialization import dumpfn
-from monty.string import is_string
 from pymatgen.core.structure import Structure
 
 from atomate2.abinit.files import write_abinit_input_set
 from atomate2.abinit.run import run_abinit
 from atomate2.abinit.schemas.core import AbinitTaskDocument, Status
 from atomate2.abinit.sets.base import AbinitInputSetGenerator
-from atomate2.abinit.utils.common import (
-    LOG_FILE_NAME,
-    STDERR_FILE_NAME,
-    InitializationError,
-    UnconvergedError,
-)
+from atomate2.abinit.utils.common import InitializationError, UnconvergedError
 from atomate2.abinit.utils.history import JobHistory
-from atomate2.abinit.utils.settings import AbinitAtomateSettings, get_abipy_manager
+from atomate2.settings import Atomate2Settings
 
 logger = logging.getLogger(__name__)
 
 __all__ = ["BaseAbinitMaker"]
 
 
-def as_event_class(event_string):
-    """Convert event string into a subclass of AbinitEvent.
-
-    The string can be the class name or the YAML tag.
-    """
-    if is_string(event_string):
-        for c in all_subclasses(AbinitEvent):
-            if c.__name__ == event_string or c.yaml_tag == event_string:
-                return c
-        raise ValueError(f"Cannot find event class associated to {event_string}.")
-    raise ValueError(
-        f"Cannot convert event_string of type {type(event_string)}. Should be a string."
-    )
-
-
 ResponseArgs = namedtuple(
     "ResponseArgs",
     ["detour", "addition", "replace", "stop_children", "stop_jobflow", "stored_data"],
@@ -104,8 +84,10 @@ def setup_job(
     # Load the atomate settings for abinit to get configuration parameters
     # TODO: how to allow for tuned parameters on a per-job basis ?
     #  (similar to fw_spec-passed settings)
-    settings = AbinitAtomateSettings()
-    abipy_manager = get_abipy_manager(settings)
+    settings = Atomate2Settings()
+    abipy_manager = None  # Currently disabled as it is needed for autoparal,
+    # which is not yet supported
+    # abipy_manager = get_abipy_manager(settings)
 
     # set walltime, if possible
     # TODO: see in set_walltime, where to put this walltime_command
@@ -211,10 +193,8 @@ def make(
 
         # Run abinit
         run_abinit(
-            abinit_cmd="abinit",
-            mpirun_cmd="mpirun",
-            log_file_path=LOG_FILE_NAME,
-            stderr_file_path=STDERR_FILE_NAME,
+            abinit_cmd=job_config.settings.ABINIT_CMD,
+            mpirun_cmd=job_config.settings.ABINIT_MPIRUN_CMD,
             wall_time=job_config.wall_time,
         )
 
@@ -231,7 +211,7 @@ def make(
         response_args = self.get_response_args(
             task_document=task_doc,
             history=job_config.history,
-            max_restarts=job_config.settings.MAX_RESTARTS,
+            max_restarts=job_config.settings.ABINIT_MAX_RESTARTS,
         )
 
         dumpfn(jsanitize(task_doc.dict()), fn="task_document.json", indent=2)
@@ -246,59 +226,6 @@ def make(
             stored_data=response_args.stored_data,
         )
 
-    # def set_walltime(self):
-    #     """Set the walltime."""
-
-    # def get_abinit_input_set(
-    #     self,
-    #     structure: Optional[Structure] = None,
-    #     prev_outputs=None,
-    #     restart_from=None,
-    # ):
-    #     """Set up AbinitInputSet.
-    #
-    #     Parameters
-    #     ----------
-    #     structure : Structure
-    #         Structure of this job.
-    #     prev_outputs : TBD
-    #         TBD
-    #     restart_from : TBD
-    #         restart from a directory, from a previous job, from a previous uuid,
-    #         from a previous ...
-    #     """
-    #     # gen_kwargs: Dict[str, Any] = {"extra_abivars": self.extra_abivars}
-    #
-    #     if restart_from is not None:
-    #         return self.input_set_generator.get_input_set(
-    #             structure=structure,
-    #             restart_from=restart_from,
-    #             prev_outputs=prev_outputs,
-    #             # **gen_kwargs,
-    #         )
-    #
-    #     if self.input_set_generator is None and restart_from is None:
-    #         raise RuntimeError(
-    #             "Cannot create abinit input set from structure without"
-    #             "input set generator."
-    #         )
-    #
-    #     return self.input_set_generator.get_input_set(
-    #         structure=structure,
-    #         restart_from=None,
-    #         prev_outputs=prev_outputs,
-    #         # **gen_kwargs,
-    #     )
-
-    # def run_abinit(self):
-    #     """Execute abinit."""
-    #     run_abinit(
-    #         abinit_cmd="abinit",
-    #         mpirun_cmd="mpirun",
-    #         log_file_path=LOG_FILE_NAME,
-    #         stderr_file_path=STDERR_FILE_NAME,
-    #     )
-
     def get_response_args(
         self,
         task_document: AbinitTaskDocument,

src/atomate2/abinit/run.py

@@ -24,8 +24,6 @@
 def run_abinit(
     abinit_cmd: str = "abinit",
     mpirun_cmd: str = None,
-    log_file_path: str = LOG_FILE_NAME,
-    stderr_file_path: str = STDERR_FILE_NAME,
     wall_time: int = None,
 ):
     """Run ABINIT."""
@@ -38,7 +36,7 @@ def run_abinit(
         command = [mpirun_cmd, abinit_cmd, INPUT_FILE_NAME]
     else:
         command = [abinit_cmd, INPUT_FILE_NAME]
-    with open(log_file_path, "w") as stdout, open(stderr_file_path, "w") as stderr:
+    with open(LOG_FILE_NAME, "w") as stdout, open(STDERR_FILE_NAME, "w") as stderr:
         process = subprocess.Popen(command, stdout=stdout, stderr=stderr)
 
         if wall_time is not None:

src/atomate2/abinit/sets/__init__.py

@@ -0,0 +1 @@
+"""Module defining ABINIT input sets used in atomate2."""

src/atomate2/abinit/sets/core.py

@@ -3,7 +3,6 @@
 from dataclasses import dataclass, field
 from typing import ClassVar, Optional
 
-import pymatgen.io.abinit.abiobjects as aobj
 from abipy.abio.factories import ebands_from_gsinput, ion_ioncell_relax_input, scf_input
 from abipy.abio.input_tags import MOLECULAR_DYNAMICS, NSCF, RELAX, SCF
 
@@ -226,34 +225,44 @@ def get_abinit_input(
                 "(e.g. relaxation) calculation, use restart_from argument of "
                 "get_input_set method instead."
             )
-
-        try:
-            atom_constraints = kwargs.pop("atoms_constraints")
-        except KeyError:
-            atom_constraints = None
-
-        try:
-            relax_cell = kwargs.pop("relax_cell")
-        except KeyError:
-            relax_cell = self.relax_cell
-
-        if relax_cell:
-            relax_method = aobj.RelaxationMethod.atoms_and_cell(
-                atoms_constraints=atom_constraints
-            )
-        else:
-            relax_method = aobj.RelaxationMethod.atoms_only(
-                atoms_constraints=atom_constraints
-            )
+        if kwargs.get("atoms_constraints", None) is not None:
+            raise NotImplementedError("Atoms constraints not implemented.")
 
         try:
             tolmxf = kwargs.pop("tolmxf")
         except KeyError:
             tolmxf = self.tolmxf
-
-        relax_method.abivars.update(tolmxf=tolmxf)
-
-        relax_input = ion_ioncell_relax_input(structure, pseudos=pseudos, **kwargs)[0]
-        relax_input.set_vars(relax_method.to_abivars())
+        ind = 1 if self.relax_cell else 0
+        relax_input = ion_ioncell_relax_input(structure, pseudos=pseudos, **kwargs)[ind]
+        relax_input["tolmxf"] = tolmxf
+
+        # try:
+        #     atom_constraints = kwargs.pop("atoms_constraints")
+        # except KeyError:
+        #     atom_constraints = None
+        #
+        # try:
+        #     relax_cell = kwargs.pop("relax_cell")
+        # except KeyError:
+        #     relax_cell = self.relax_cell
+        #
+        # if relax_cell:
+        #     relax_method = aobj.RelaxationMethod.atoms_and_cell(
+        #         atoms_constraints=atom_constraints
+        #     )
+        # else:
+        #     relax_method = aobj.RelaxationMethod.atoms_only(
+        #         atoms_constraints=atom_constraints
+        #     )
+        #
+        # try:
+        #     tolmxf = kwargs.pop("tolmxf")
+        # except KeyError:
+        #     tolmxf = self.tolmxf
+        #
+        # relax_method.abivars.update(tolmxf=tolmxf)
+        #
+        # relax_input = ion_ioncell_relax_input(structure, pseudos=pseudos, **kwargs)[0]
+        # relax_input.set_vars(relax_method.to_abivars())
 
         return relax_input

src/atomate2/abinit/utils/settings.py

@@ -1,31 +1,32 @@
 """Settings for abinit flows in atomate2."""
 
 from abipy.flowtk.tasks import TaskManager
-from pydantic import BaseSettings, Field
 
-
-class AbinitAtomateSettings(BaseSettings):
-    """Settings for abinit in atomate2."""
-
-    MPIRUN_CMD: str = Field("mpirun", description="Mpirun command.")
-
-    # ABINIT specific settings
-    ABINIT_CMD: str = Field("abinit", description="Abinit command.")
-    MRGDDB_CMD: str = Field("mrgddb", description="Mrgddb command.")
-    ANADDB_CMD: str = Field("anaddb", description="Anaddb command.")
-    COPY_DEPS: bool = Field(
-        False,
-        description="Copy (True) or link file dependencies between jobs.",
-    )
-    AUTOPARAL: bool = Field(
-        False,
-        description="Use autoparal to determine optimal parallel configuration.",
-    )
-    ABIPY_MANAGER_FILE: str = Field(
-        None,
-        description="Config file for task manager of abipy.",
-    )
-    MAX_RESTARTS: int = Field(5, description="Maximum number of restarts of a job.")
+# from pydantic import BaseSettings, Field
+
+
+# class AbinitAtomateSettings(BaseSettings):
+#     """Settings for abinit in atomate2."""
+#
+#     MPIRUN_CMD: str = Field("mpirun", description="Mpirun command.")
+#
+#     # ABINIT specific settings
+#     ABINIT_CMD: str = Field("abinit", description="Abinit command.")
+#     MRGDDB_CMD: str = Field("mrgddb", description="Mrgddb command.")
+#     ANADDB_CMD: str = Field("anaddb", description="Anaddb command.")
+#     COPY_DEPS: bool = Field(
+#         False,
+#         description="Copy (True) or link file dependencies between jobs.",
+#     )
+#     AUTOPARAL: bool = Field(
+#         False,
+#         description="Use autoparal to determine optimal parallel configuration.",
+#     )
+#     ABIPY_MANAGER_FILE: str = Field(
+#         None,
+#         description="Config file for task manager of abipy.",
+#     )
+#     MAX_RESTARTS: int = Field(5, description="Maximum number of restarts of a job.")
 
 
 def get_abipy_manager(settings):
@@ -38,8 +39,4 @@ def get_abipy_manager(settings):
     """
     if settings.ABIPY_MANAGER_FILE:
         return TaskManager.from_file(settings.ABIPY_MANAGER_FILE)
-    try:
-        return TaskManager.from_user_config()
-    except RuntimeError:
-        # logger.warning("Couldn't load the abipy task manager.")
-        return None
+    return TaskManager.from_user_config()

src/atomate2/settings.py

@@ -92,6 +92,27 @@ class Atomate2Settings(BaseSettings):
         None, description="Additional settings applied to AMSET settings file."
     )
 
+    # ABINIT settings
+    ABINIT_MPIRUN_CMD: str = Field("mpirun", description="Mpirun command.")
+    ABINIT_CMD: str = Field("abinit", description="Abinit command.")
+    ABINIT_MRGDDB_CMD: str = Field("mrgddb", description="Mrgddb command.")
+    ABINIT_ANADDB_CMD: str = Field("anaddb", description="Anaddb command.")
+    ABINIT_COPY_DEPS: bool = Field(
+        False,
+        description="Copy (True) or link file dependencies between jobs.",
+    )
+    ABINIT_AUTOPARAL: bool = Field(
+        False,
+        description="Use autoparal to determine optimal parallel configuration.",
+    )
+    ABINIT_ABIPY_MANAGER_FILE: str = Field(
+        None,
+        description="Config file for task manager of abipy.",
+    )
+    ABINIT_MAX_RESTARTS: int = Field(
+        5, description="Maximum number of restarts of a job."
+    )
+
     class Config:
         """Pydantic config settings."""