Abinit phonon workflow

src/atomate2/abinit/files.py

@@ -27,6 +27,7 @@
     from atomate2.abinit.sets.anaddb import AnaddbInputGenerator
     from atomate2.abinit.sets.base import AbinitInputGenerator
     from atomate2.abinit.sets.mrgddb import MrgddbInputGenerator
+    from atomate2.abinit.sets.mrgdvdb import MrgdvInputGenerator
 
 __all__ = [
     "del_gzip_files",
@@ -36,6 +37,7 @@
     "write_abinit_input_set",
     "write_anaddb_input_set",
     "write_mrgddb_input_set",
+    "write_mrgdv_input_set",
 ]
 
 logger = logging.getLogger(__name__)
@@ -190,6 +192,35 @@ def write_mrgddb_input_set(
     mrgis.write_input(directory=directory, make_dir=True, overwrite=False)
 
 
+def write_mrgdv_input_set(
+    input_set_generator: MrgdvInputGenerator,
+    prev_outputs: str | Path | list[str] | None = None,
+    directory: str | Path = ".",
+) -> None:
+    """Write the mrgdv input using a given generator.
+
+    Parameters
+    ----------
+    input_set_generator
+        The input generator used to write the mrgddb inputs.
+    prev_outputs
+        The list of previous directories needed for the calculation.
+    directory
+        Directory in which to write the abinit inputs.
+    """
+    mrgis = input_set_generator.get_input_set(
+        prev_outputs=prev_outputs,
+        workdir=directory,
+    )
+    if not mrgis.validate():
+        raise RuntimeError(
+            "MrgdvInputSet is not valid. Some previous outputs \
+        do not exist."
+        )
+
+    mrgis.write_input(directory=directory, make_dir=True, overwrite=False)
+
+
 def write_anaddb_input_set(
     structure: Structure,
     input_set_generator: AnaddbInputGenerator,

src/atomate2/abinit/flows/dfpt.py

@@ -9,15 +9,22 @@
 from abipy.abio.factories import scf_for_phonons
 from jobflow import Flow, Maker
 
-from atomate2.abinit.jobs.anaddb import AnaddbDfptDteMaker, AnaddbMaker
-from atomate2.abinit.jobs.core import StaticMaker
+from atomate2.abinit.jobs.anaddb import (
+    AnaddbDfptDteMaker,
+    AnaddbMaker,
+    AnaddbPhBandsDOSMaker,
+)
+from atomate2.abinit.jobs.core import StaticMaker, StaticMakerforPhonons
 from atomate2.abinit.jobs.mrgddb import MrgddbMaker
+from atomate2.abinit.jobs.mrgdv import MrgdvMaker
 from atomate2.abinit.jobs.response import (
     DdeMaker,
     DdkMaker,
     DteMaker,
+    PhononResponseMaker,
     generate_dde_perts,
     generate_dte_perts,
+    generate_phonon_perts,
     run_rf,
 )
 from atomate2.abinit.powerups import update_user_abinit_settings
@@ -51,14 +58,32 @@ class DfptFlowMaker(Maker):
         The maker to use for the DDE calculations.
     dte_maker : .BaseAbinitMaker
         The maker to use for the DTE calculations.
+    wfq_maker : .BaseAbinitMaker
+        The maeker to use to compute the k+q WFs.
+        Not implemented yet.
+    phonon_maker : .BaseAbinitMaker
+        The maker to use for the phonon calculations.
     mrgddb_maker : .Maker
-        The maker to merge the DDE and DTE DDB.
+        The maker to merge the DDBs.
+    mrgdv_maker : .Maker
+        The maker to merge the POT files.
+    anaddb_maker : .Maker
+        The maker to analyze the DDBs.
     use_dde_sym : bool
         True if only the irreducible DDE perturbations should be considered,
             False otherwise.
     dte_skip_permutations: Since the current version of abinit always performs
         all the permutations of the perturbations, even if only one is asked,
         if True avoids the creation of inputs that will produce duplicated outputs.
+    qpt_list: list or tuple or None
+        A list of q points to compute the phonon band structure.
+        All the q points must be part of the k-mesh used for electrons.
+    ngqpt: list or tuple or None
+        Monkhorst-Pack divisions for the phonon q-mesh.
+        Default is the same as the one used in the GS calculation.
+        Must be a sub-mesh of the k-mesh used for electrons.
+    qptopt: int or None
+        Option for the generation of the q-points list, default same as kptopt in gs.
     """
 
     name: str = "DFPT"
@@ -70,10 +95,16 @@ class DfptFlowMaker(Maker):
     ddk_maker: BaseAbinitMaker | None = field(default_factory=DdkMaker)  # |
     dde_maker: BaseAbinitMaker | None = field(default_factory=DdeMaker)  # |
     dte_maker: BaseAbinitMaker | None = field(default_factory=DteMaker)  # |
+    wfq_maker: BaseAbinitMaker | None = None  # | not implemented
+    phonon_maker: BaseAbinitMaker | None = None  # |
     mrgddb_maker: Maker | None = field(default_factory=MrgddbMaker)  # |
+    mrgdv_maker: Maker | None = None  # |
     anaddb_maker: Maker | None = field(default_factory=AnaddbMaker)  # |
     use_dde_sym: bool = True
     dte_skip_permutations: bool | None = False
+    qpt_list: list[list] | None = None
+    ngqpt: list | None = None
+    qptopt: int = None
 
     def __post_init__(self) -> None:
         """Process post-init configuration."""
@@ -105,6 +136,7 @@ def make(
         self,
         structure: Structure | None = None,
         restart_from: str | Path | None = None,
+        **anaddb_kwargs,
     ) -> Flow:
         """
         Create a DFPT flow.
@@ -115,15 +147,34 @@ def make(
             A pymatgen structure object.
         restart_from : str or Path or None
             One previous directory to restart from.
+        anaddb_kwargs : dict
+            Additional kwargs for the anaddb maker.
 
         Returns
         -------
         Flow
             A DFPT flow
         """
-        static_job = self.static_maker.make(structure, restart_from=restart_from)
+        jobs = []
+        if (
+            isinstance(self.static_maker, StaticMakerforPhonons)
+            and not self.wfq_maker
+            and self.ngqpt
+        ):
+            """A check on the q-mesh is performed in order to avoid gs computations
+            if q and k grids are not commensurate."""
+
+            static_job = self.static_maker.validate_grids(
+                structure, ngqpt=self.ngqpt, restart_from=restart_from
+            )
+            static_job.name = "SCF with grids validation"
+            jobs.append(static_job)
 
-        jobs = [static_job]
+        else:
+            static_job = self.static_maker.make(
+                structure=structure, restart_from=restart_from
+            )
+            jobs.append(static_job)
 
         if self.ddk_maker:
             # the use of symmetries is not implemented for DDK
@@ -185,34 +236,77 @@ def make(
             )
             jobs.append(dte_calcs)
 
-        if self.mrgddb_maker:
-            # merge the DDE and DTE DDB.
+        if self.phonon_maker:
+            prev_outputs = [static_job.output.dir_name]
+            # if self.dde_maker:
+            #    prev_outputs.append(dde_calcs.output["dirs"])
+            # generation of qpt_list (if needed) and corresponding perturbations
+            phonon_perts_qpt_list = generate_phonon_perts(
+                gsinput=static_job.output.input.abinit_input,
+                ngqpt=self.ngqpt,
+                qptopt=self.qptopt,
+                qpt_list=self.qpt_list,
+            )
+            jobs.append(phonon_perts_qpt_list)
+            # perform the phonon calculations
+            phonon_calcs = run_rf(
+                perturbations=phonon_perts_qpt_list.output["perts"],
+                rf_maker=self.phonon_maker,
+                prev_outputs=prev_outputs,
+            )
+            jobs.append(phonon_calcs)
 
-            prev_outputs = [dde_calcs.output["dirs"]]
+        if self.mrgddb_maker:
+            # merge the DDE, DTE and Phonon DDB.
+            prev_outputs = []
+            if self.dde_maker:
+                prev_outputs.append(dde_calcs.output["dirs"])
             if self.dte_maker:
                 prev_outputs.append(dte_calcs.output["dirs"])
+            if self.phonon_maker:
+                prev_outputs.append(phonon_calcs.output["dirs"])
 
             mrgddb_job = self.mrgddb_maker.make(
                 prev_outputs=prev_outputs,
             )
 
             jobs.append(mrgddb_job)
 
+        if self.mrgdv_maker:
+            # merge the DDE and Phonon POT files.
+            prev_outputs = []
+            if self.dde_maker:
+                prev_outputs.append(dde_calcs.output["dirs"])
+            if self.phonon_maker:
+                prev_outputs.append(phonon_calcs.output["dirs"])
+
+            mrgdv_job = self.mrgdv_maker.make(
+                prev_outputs=prev_outputs,
+            )
+            jobs.append(mrgdv_job)
+
         if self.anaddb_maker:
             # analyze a merged DDB.
+            if self.phonon_maker:
+                # set the required args for the anaddb phbstdos input
+                if anaddb_kwargs:
+                    anaddb_kwargs.update(
+                        {"ngqpt": phonon_perts_qpt_list.output["ngqpt"]}
+                    )
+                else:
+                    anaddb_kwargs = {"ngqpt": phonon_perts_qpt_list.output["ngqpt"]}
+                anaddb_kwargs.setdefault("nqsmall", 10)
+                # ifc needed to create the phonopy like outdoc, user can turn it off
+                anaddb_kwargs.setdefault("with_ifc", True)
 
             anaddb_job = self.anaddb_maker.make(
                 structure=mrgddb_job.output.structure,
                 prev_outputs=mrgddb_job.output.dir_name,
+                **anaddb_kwargs,
             )
 
             jobs.append(anaddb_job)
 
-        # TODO: implement the possibility of other DFPT WFs (phonons,...)
-        # if self.wfq_maker:
-        #     ...
-
-        # return Flow(jobs, output=jobs[-1].output, name=self.name)  # TODO: fix outputs
         return Flow(
             jobs, output=[j.output for j in jobs], name=self.name
         )  # TODO: fix outputs
@@ -243,6 +337,7 @@ def make(
         self,
         structure: Structure | None = None,
         restart_from: str | Path | None = None,
+        **anaddb_kwargs,
     ) -> Flow:
         """
         Create a DFPT flow.
@@ -259,7 +354,9 @@ def make(
         Flow
             A DFPT flow
         """
-        shg_flow = super().make(structure=structure, restart_from=restart_from)
+        shg_flow = super().make(
+            structure=structure, restart_from=restart_from, **anaddb_kwargs
+        )
 
         if self.scissor:
             shg_flow = update_user_abinit_settings(
@@ -269,3 +366,89 @@ def make(
             )
 
         return shg_flow
+
+
+@dataclass
+class PhononMaker(DfptFlowMaker):
+    """
+    Maker to generate a phonon band structure and phonon DOS flow with abinit.
+
+    Parameters
+    ----------
+    name : str
+        Name of the flows produced by this maker.
+    with_dde : bool
+        True if the DDE calculations should be included, False otherwise.
+    run_anaddb : bool
+        True if the anaddb calculations should be included, False otherwise.
+    run_mrgddb : bool
+        True if the merge of DDB files should be included, False otherwise.
+    run_mrgdv : bool
+        True if the merge of POT files should be included, False otherwise.
+    """
+
+    name: str = "Phonon Flow"
+    with_dde: bool = True
+    run_anaddb: bool = True
+    run_mrgddb: bool = True
+    run_mrgdv: bool = True
+    static_maker: BaseAbinitMaker = field(
+        default_factory=lambda: StaticMakerforPhonons(
+            input_set_generator=StaticSetGenerator(factory=scf_for_phonons)
+        )
+    )
+    phonon_maker: BaseAbinitMaker = field(default_factory=PhononResponseMaker)
+    mrgdv_maker: BaseAbinitMaker | None = field(default_factory=MrgdvMaker)
+    anaddb_maker: BaseAbinitMaker | None = field(default_factory=AnaddbPhBandsDOSMaker)
+    dte_maker: BaseAbinitMaker | None = None
+
+    def __post_init__(self) -> None:
+        """Process post-init configuration."""
+        if not self.with_dde:
+            """
+            To turn off the DDE calculations, turn off DDK as well.
+            If a DDK maker is provided, it will be removed
+            """
+            self.ddk_maker = None
+            self.dde_maker = None
+
+        if not self.run_mrgddb:
+            """Turn off the merge of POT files"""
+            self.mrgddb_maker = None
+
+        if not self.run_mrgdv:
+            """Turn off the merge of DDB files"""
+            self.mrgdv_maker = None
+
+        if not self.run_anaddb:
+            """Turn off the anaddb calculations"""
+            self.anaddb_maker = None
+
+    def make(
+        self,
+        structure: Structure | None = None,
+        restart_from: str | Path | None = None,
+        **anaddb_kwargs,
+    ) -> Flow:
+        """
+        Create a phonon flow.
+
+        Parameters
+        ----------
+        structure : Structure
+            A pymatgen structure object.
+        restart_from : str or Path or None
+            One previous directory to restart from.
+        anaddb_kwargs : dict
+            Additional kwargs for the anaddb maker.
+
+        Returns
+        -------
+        Flow
+            A phonon flow.
+        """
+        return super().make(
+            structure=structure,
+            restart_from=restart_from,
+            **anaddb_kwargs,
+        )