From 57318f2267eb477e919a828e2610f5b20ac73ba3 Mon Sep 17 00:00:00 2001 From: Enrico Marazzi Date: Thu, 3 Apr 2025 16:16:18 +0200 Subject: [PATCH 1/5] Phonon workflow for ABINIT first pr --- src/atomate2/abinit/files.py | 31 ++ src/atomate2/abinit/flows/dfpt.py | 191 ++++++- src/atomate2/abinit/jobs/anaddb.py | 72 ++- src/atomate2/abinit/jobs/core.py | 67 +++ src/atomate2/abinit/jobs/mrgdv.py | 143 +++++ src/atomate2/abinit/jobs/response.py | 92 +++- src/atomate2/abinit/run.py | 49 +- src/atomate2/abinit/schemas/anaddb.py | 368 ++++++++++++- src/atomate2/abinit/schemas/calculation.py | 2 + src/atomate2/abinit/schemas/mrgdvdb.py | 499 ++++++++++++++++++ src/atomate2/abinit/schemas/outfiles.py | 2 +- src/atomate2/abinit/sets/anaddb.py | 40 ++ src/atomate2/abinit/sets/mrgddb.py | 4 +- src/atomate2/abinit/sets/mrgdvdb.py | 219 ++++++++ src/atomate2/abinit/sets/response.py | 16 + src/atomate2/abinit/utils/common.py | 26 + src/atomate2/settings.py | 1 + tests/abinit/conftest.py | 72 ++- tests/abinit/flows/test_dfpt.py | 8 + tests/abinit/test_dfpt.py | 0 .../silicon_carbide_shg/ref_paths.json.gz | Bin 234 -> 236 bytes 21 files changed, 1868 insertions(+), 34 deletions(-) create mode 100644 src/atomate2/abinit/jobs/mrgdv.py create mode 100644 src/atomate2/abinit/schemas/mrgdvdb.py create mode 100644 src/atomate2/abinit/sets/mrgdvdb.py create mode 100644 tests/abinit/test_dfpt.py diff --git a/src/atomate2/abinit/files.py b/src/atomate2/abinit/files.py index 4377356977..be79e5bc57 100644 --- a/src/atomate2/abinit/files.py +++ b/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, diff --git a/src/atomate2/abinit/flows/dfpt.py b/src/atomate2/abinit/flows/dfpt.py index 219d02ae1f..82ccaed8c8 100644 --- a/src/atomate2/abinit/flows/dfpt.py +++ b/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,27 @@ class DfptFlowMaker(Maker): The maker to use for the DDE calculations. dte_maker : .BaseAbinitMaker The maker to use for the DTE calculations. + 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. + 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 +90,18 @@ 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 = field( + default_factory=PhononResponseMaker + ) # | mrgddb_maker: Maker | None = field(default_factory=MrgddbMaker) # | + mrgdv_maker: Maker | None = field(default_factory=MrgdvMaker) # | anaddb_maker: Maker | None = field(default_factory=AnaddbMaker) # | use_dde_sym: bool = True dte_skip_permutations: bool | None = False + qpt_list: list | None = None + ngqpt: list | None = None + qptopt: int = None def __post_init__(self) -> None: """Process post-init configuration.""" @@ -105,6 +133,7 @@ def make( self, structure: Structure | None = None, restart_from: str | Path | None = None, + **anaddb_kwargs, ) -> Flow: """ Create a DFPT flow. @@ -121,9 +150,23 @@ def make( 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 + ): + 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,12 +228,37 @@ def make( ) jobs.append(dte_calcs) + if self.phonon_maker: + # generate the perturbations for the phonon calculations + prev_outputs = [static_job.output.dir_name] + if self.dde_maker: + prev_outputs.append(dde_calcs.output["dirs"]) + + 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) + + phonon_calcs = run_rf( + perturbations=phonon_perts_qpt_list.output["perts"], + rf_maker=self.phonon_maker, + prev_outputs=prev_outputs, + with_dde=bool(self.dde_maker), + ) + jobs.append(phonon_calcs) + if self.mrgddb_maker: # merge the DDE and DTE DDB. - - prev_outputs = [dde_calcs.output["dirs"]] + 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, @@ -198,12 +266,34 @@ def make( jobs.append(mrgddb_job) + if self.mrgdv_maker: + 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: + 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) + 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) @@ -233,6 +323,8 @@ class ShgFlowMaker(DfptFlowMaker): name: str = "DFPT Chi2 SHG" anaddb_maker: Maker | None = field(default_factory=AnaddbDfptDteMaker) + phonon_maker: Maker | None = None + mrgdv_maker: Maker | None = None use_dde_sym: bool = False static_maker: BaseAbinitMaker = field( default_factory=lambda: StaticMaker(input_set_generator=ShgStaticSetGenerator()) @@ -243,6 +335,7 @@ def make( self, structure: Structure | None = None, restart_from: str | Path | None = None, + **anaddb_kwargs, ) -> Flow: """ Create a DFPT flow. @@ -259,7 +352,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 +364,83 @@ 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. + """ + + name: str = "Phbands-PhDOS 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) + ) + ) + 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 band structure and DOS 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 band structure and DOS flow. + """ + return super().make( + structure=structure, + restart_from=restart_from, + **anaddb_kwargs, + ) diff --git a/src/atomate2/abinit/jobs/anaddb.py b/src/atomate2/abinit/jobs/anaddb.py index 27db25ee6d..0b59421ba0 100644 --- a/src/atomate2/abinit/jobs/anaddb.py +++ b/src/atomate2/abinit/jobs/anaddb.py @@ -14,12 +14,16 @@ from atomate2.abinit.jobs.base import setup_job from atomate2.abinit.run import run_anaddb from atomate2.abinit.schemas.anaddb import AnaddbTaskDoc +from atomate2.abinit.schemas.outfiles import AbinitStoredFile from atomate2.abinit.sets.anaddb import ( AnaddbDfptDteInputGenerator, AnaddbInputGenerator, + AnaddbPhbandsDOSInputGenerator, ) if TYPE_CHECKING: + from collections.abc import Callable + from pymatgen.core.structure import Structure from atomate2.abinit.utils.history import JobHistory @@ -29,6 +33,46 @@ __all__ = ["AnaddbDfptDteMaker", "AnaddbMaker"] +_DATA_OBJECTS = [ + AbinitStoredFile, +] + + +def anaddb_job(method: Callable) -> job: + """ + Decorate the ``make`` method of ANADDB job makers. + + This is a thin wrapper around :obj:`~jobflow.core.job.job` that configures common + settings for all anaddb jobs. For example, it ensures that large data objects + (band structures, density of states, DDB, etc) are all stored in the + atomate2 data store. It also configures the output schema to be an Abinit + :obj:`.TaskDocument`. + + Any makers that return Anaddb jobs (not flows) should decorate the ``make`` method + with @anaddb_job. For example: + + .. code-block:: python + + class MyAnaddbMaker(AnaddbMaker): + @anaddb_job + def make(structure): + # code to run abinit job. + pass + + Parameters + ---------- + method : callable + A AnaddbMaker.make method. This should not be specified directly and is + implied by the decorator. + + Returns + ------- + callable + A decorated version of the make function that will generate Anaddb jobs. + """ + return job(method, data=_DATA_OBJECTS, output_schema=AnaddbTaskDoc) + + @dataclass class AnaddbMaker(Maker): """Maker to create a job to analyze a DDB file with the utility anaddb. @@ -44,18 +88,21 @@ class AnaddbMaker(Maker): default_factory=AnaddbInputGenerator ) wall_time: int | None = None + factory_kwargs: dict = field(default_factory=dict) + task_document_kwargs: dict = field(default_factory=dict) @property def calc_type(self) -> str: """Get the type of calculation for this maker.""" return self.input_set_generator.calc_type - @job(output_schema=AnaddbTaskDoc) + @anaddb_job def make( self, structure: Structure, prev_outputs: str | list[str] | None = None, history: JobHistory | None = None, + **anaddb_factory_kwargs, ) -> jobflow.Response: """ Return an AnaDDB jobflow.Job. @@ -66,6 +113,8 @@ def make( history : JobHistory A JobHistory object containing the history of this job. """ + self.input_set_generator.factory_kwargs = anaddb_factory_kwargs + # Setup job and get general job configuration config = setup_job( structure=None, @@ -92,7 +141,7 @@ def make( # parse Anaddb DDB output task_doc = AnaddbTaskDoc.from_directory( Path.cwd(), - # **self.task_document_kwargs, + **self.task_document_kwargs, ) task_doc.task_label = self.name @@ -113,3 +162,22 @@ class AnaddbDfptDteMaker(AnaddbMaker): input_set_generator: AnaddbInputGenerator = field( default_factory=AnaddbDfptDteInputGenerator ) + + +@dataclass +class AnaddbPhBandsDOSMaker(AnaddbMaker): + """Maker to compute phonon bands and DOS from a merged DDB file. + + Parameters + ---------- + name : str + The job name. + """ + + name: str = "Anaddb PhbandsDOS" + input_set_generator: AnaddbInputGenerator = field( + default_factory=AnaddbPhbandsDOSInputGenerator + ) + task_document_kwargs: dict = field( + default_factory=lambda: {"files_to_store": ["PHBST", "PHDOS"]} + ) diff --git a/src/atomate2/abinit/jobs/core.py b/src/atomate2/abinit/jobs/core.py index bfbd2b763c..b508a74a26 100644 --- a/src/atomate2/abinit/jobs/core.py +++ b/src/atomate2/abinit/jobs/core.py @@ -3,16 +3,20 @@ from __future__ import annotations import logging +import tempfile from dataclasses import dataclass, field from typing import TYPE_CHECKING, ClassVar +import numpy as np from abipy.flowtk.events import ( AbinitCriticalWarning, NscfConvergenceWarning, RelaxConvergenceWarning, ScfConvergenceWarning, ) +from jobflow import Response, job +from atomate2.abinit.files import load_abinit_input, write_abinit_input_set from atomate2.abinit.jobs.base import BaseAbinitMaker, abinit_job from atomate2.abinit.sets.core import ( LineNonSCFSetGenerator, @@ -22,9 +26,11 @@ StaticSetGenerator, UniformNonSCFSetGenerator, ) +from atomate2.abinit.utils.history import JobHistory if TYPE_CHECKING: from collections.abc import Sequence + from pathlib import Path from jobflow import Job from pymatgen.core.structure import Structure @@ -56,6 +62,67 @@ class StaticMaker(BaseAbinitMaker): ) +@dataclass +class StaticMakerforPhonons(StaticMaker): + """ + Maker to create ABINIT scf jobs for phonons. + + The make method is modified in order to stop a gs calculation + if the q-points are not commensurate with the k-points. + + Parameters + ---------- + name : str + The job name. + """ + + @job + def validate_grids( + self, + structure: Structure | None, + ngqpt: list[int] | None = None, + prev_outputs: str | Path | list[str] | None = None, + restart_from: str | Path | list[str] | None = None, + history: JobHistory | None = None, + ) -> dict[str, dict]: + """ + Validate the q-point grid for the phonon calculation. + + If the q- and k-grids are commensurate the method returns + a static job maker, otherwise it raises a ValueError. + + Parameters + ---------- + structure : Structure + A pymatgen structure object. + ngqpt : list + q-point grid for phonon calculation. + shiftq : list + Shifts for the q-point grid. + qptopt : int + Option for the q-point grid. + qpt_list : list + List of q-points for the phonon calculation. + """ + with tempfile.TemporaryDirectory(dir=".") as temp_dir: + write_abinit_input_set( + structure=structure, + input_set_generator=self.input_set_generator, + directory=temp_dir, + ) + ai = load_abinit_input(temp_dir) + if np.any(np.array(ai["ngkpt"]) % np.array(ngqpt) != np.array([0, 0, 0])): + raise ValueError("q-points are not commensurate with k-points") + self.name = "SCF Calculation for Phonons" + static_job = self.make( + structure=structure, + prev_outputs=prev_outputs, + restart_from=restart_from, + history=history, + ) + return Response(replace=static_job, output=static_job.output) + + @dataclass class LineNonSCFMaker(BaseAbinitMaker): """Maker to create a job with a non-scf ABINIT calculation along a line. diff --git a/src/atomate2/abinit/jobs/mrgdv.py b/src/atomate2/abinit/jobs/mrgdv.py new file mode 100644 index 0000000000..ef9aa7911f --- /dev/null +++ b/src/atomate2/abinit/jobs/mrgdv.py @@ -0,0 +1,143 @@ +"""Merge DV jobs for merging POT files from ABINIT calculations.""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING + +import jobflow +import numpy as np +from jobflow import Maker, Response, job + +from atomate2.abinit.files import write_mrgdv_input_set +from atomate2.abinit.jobs.base import setup_job +from atomate2.abinit.run import run_mrgdv +from atomate2.abinit.schemas.mrgdvdb import MrgdvdbTaskDoc +from atomate2.abinit.schemas.outfiles import AbinitStoredFile +from atomate2.abinit.sets.mrgdvdb import MrgdvInputGenerator + +if TYPE_CHECKING: + from collections.abc import Callable + + from atomate2.abinit.utils.history import JobHistory + +logger = logging.getLogger(__name__) + +__all__ = [ + "MrgdvMaker", +] + +_MRGDV_DATA_OBJECTS = [ + AbinitStoredFile, +] + + +def mrgdv_job(method: Callable) -> job: + """ + Decorate the ``make`` method of mrgdv job makers. + + This is a thin wrapper around :obj:`~jobflow.core.job.job` that configures common + settings for all mrgdv jobs. For example, it ensures that the mrgdvdb file is + stored in the atomate2 data store. It also configures the output schema to be a + mrgdvdb + :obj:`.TaskDocument`. + + Any makers that return mrgdv jobs (not flows) should decorate the ``make`` method + with @mrgdv_job. For example: + + .. code-block:: python + + class MyMrgdvMaker(MrgdvMaker): + @mrgdv_job + def make(structure): + # code to run mrgdv job. + pass + + Parameters + ---------- + method : callable + A BaseMrgdvMaker.make method. This should not be specified directly and is + implied by the decorator. + + Returns + ------- + callable + A decorated version of the make function that will generate mrgdv jobs. + """ + return job(method, data=_MRGDV_DATA_OBJECTS, output_schema=MrgdvdbTaskDoc) + + +@dataclass +class MrgdvMaker(Maker): + """Maker to create a job with a merge of POT files from ABINIT. + + Parameters + ---------- + name : str + The job name. + """ + + name: str = "Merge DVDB" + input_set_generator: MrgdvInputGenerator = field( + default_factory=MrgdvInputGenerator + ) + wall_time: int | None = None + + @property + def calc_type(self) -> str: + """Get the type of calculation for this maker.""" + return self.input_set_generator.calc_type + + @mrgdv_job + def make( + self, + prev_outputs: list[str] | None = None, + history: JobHistory | None = None, + ) -> jobflow.Response: + """ + Return a MRGdv jobflow.Job. + + Parameters + ---------- + prev_outputs : TODO: add description from sets.base + history : JobHistory + A JobHistory object containing the history of this job. + """ + # Flatten the list of previous outputs dir + # prev_outputs = [item for sublist in prev_outputs for item in sublist] + prev_outputs = list(np.hstack(prev_outputs)) + + # Setup job and get general job configuration + config = setup_job( + structure=None, + prev_outputs=prev_outputs, + restart_from=None, + history=history, + wall_time=self.wall_time, + ) + + # Write mrgdv input set + write_mrgdv_input_set( + input_set_generator=self.input_set_generator, + prev_outputs=prev_outputs, + directory=config.workdir, + ) + + # Run mrgdv + run_mrgdv( + wall_time=config.wall_time, + start_time=config.start_time, + ) + + # parse Mrgdv dv output + task_doc = MrgdvdbTaskDoc.from_directory( + Path.cwd(), + # **self.task_document_kwargs, + ) + task_doc.task_label = self.name + + return Response( + output=task_doc, + ) diff --git a/src/atomate2/abinit/jobs/response.py b/src/atomate2/abinit/jobs/response.py index 8d5f389dd4..42c3161e63 100644 --- a/src/atomate2/abinit/jobs/response.py +++ b/src/atomate2/abinit/jobs/response.py @@ -1,4 +1,4 @@ -"""Jobs for running ABINIT response to perturbations.""" +"""Response function jobs for running ABINIT calculations.""" from __future__ import annotations @@ -22,6 +22,7 @@ DdeSetGenerator, DdkSetGenerator, DteSetGenerator, + PhononSetGenerator, ) if TYPE_CHECKING: @@ -39,9 +40,11 @@ "DdeMaker", "DdkMaker", "DteMaker", + "PhononResponseMaker", "ResponseMaker", "generate_dde_perts", "generate_dte_perts", + "generate_phonon_perts", "run_rf", ] @@ -162,6 +165,27 @@ class DteMaker(ResponseMaker): ) +@dataclass +class PhononResponseMaker(ResponseMaker): + """Maker to create a job with a Phonon ABINIT calculation. + + Parameters + ---------- + name : str + The job name. + """ + + calc_type: str = "Phonon" + name: str = "Phonon calculation" + input_set_generator: AbinitInputGenerator = field( + default_factory=PhononSetGenerator + ) + + CRITICAL_EVENTS: ClassVar[Sequence[AbinitCriticalWarning]] = ( + ScfConvergenceWarning, + ) + + @job def generate_dde_perts( gsinput: AbinitInput, @@ -240,11 +264,58 @@ def generate_dte_perts( return outputs +@job +def generate_phonon_perts( + gsinput: AbinitInput, + ngqpt: list | tuple | None = None, + qptopt: int | None = 1, + qpt_list: list | None = None, + with_wfq: bool = False, +) -> dict[str, list[Any] | tuple[Any, ...] | Any]: + """ + Generate the perturbations for the phonon calculations. + + Parameters + ---------- + gsinput : an |AbinitInput| representing a ground state calculation, + likely the SCF performed to get the WFK. + qpt: list or tuple + q-point for the phonon calculations. + """ + gsinput = gsinput.deepcopy() + gsinput.pop_vars(["autoparal"]) + outputs = {} + if qpt_list is None: + qpt_list = gsinput.abiget_ibz( + ngkpt=ngqpt, shiftk=[0, 0, 0], kptopt=qptopt + ).points + outputs["ngqpt"] = ngqpt if ngqpt else gsinput["ngkpt"] + + perturbations = list() + outdirs = list() + for q in qpt_list: + perts = gsinput.abiget_irred_phperts(qpt=q) + perturbations.append(perts) + outdirs.append(Path(os.getcwd())) # to make the dir accessible + # when a wfq_maker will be available something like ... can be added here + # if q not in kpt_list and with_wfq: + # wfq_job = wfq_maker.make(q=q, prev_outputs=prev_outputs) + # outputs["dirs"].append(wfq_job.output.dir_name) + # and the last if removed + outputs["perts"] = list(np.hstack(perturbations)) + outputs["dir_name"] = list(np.hstack(outdirs)) + if np.any(np.array(gsinput["ngkpt"]) % np.array(outputs["ngqpt"])) and not with_wfq: + raise ValueError("q-points are not commensurate with k-points.") + + return outputs + + @job def run_rf( perturbations: list[dict], rf_maker: ResponseMaker, prev_outputs: str | list[str] | None = None, + with_dde: bool = False, ) -> Flow: """ Run the RF calculations. @@ -257,19 +328,32 @@ def run_rf( prev_outputs : a list of previous output directories """ rf_jobs = [] + is_phonon = isinstance(rf_maker, PhononResponseMaker) outputs: dict[str, Any] = {"dirs": []} - if isinstance(rf_maker, DdeMaker | DteMaker): + if isinstance(rf_maker, DdeMaker | DteMaker | PhononResponseMaker): # Flatten the list of previous outputs dir # prev_outputs = [item for sublist in prev_outputs for item in sublist] prev_outputs = list(np.hstack(prev_outputs)) for ipert, pert in enumerate(perturbations): + is_gamma = np.allclose(pert["qpt"], [0.0, 0.0, 0.0]) if "qpt" in pert else True + + prev_out = ( + [prev_outputs[0]] + if is_phonon and with_dde and not is_gamma + else prev_outputs + ) rf_job = rf_maker.make( perturbation=pert, - prev_outputs=prev_outputs, + prev_outputs=prev_out, ) - rf_job.append_name(f"{ipert+1}/{len(perturbations)}") + + if is_phonon: + qpt_str = f"{pert['qpt'][0]:.2f},{pert['qpt'][1]:.2f},{pert['qpt'][2]:.2f}" + rf_job.append_name(f", q = {qpt_str} ({ipert+1}/{len(perturbations)})") + else: + rf_job.append_name(f"{ipert+1}/{len(perturbations)}") rf_jobs.append(rf_job) outputs["dirs"].append(rf_job.output.dir_name) # TODO: determine outputs diff --git a/src/atomate2/abinit/run.py b/src/atomate2/abinit/run.py index 9b60ba490b..7f2f134628 100644 --- a/src/atomate2/abinit/run.py +++ b/src/atomate2/abinit/run.py @@ -13,10 +13,11 @@ INPUT_FILE_NAME, LOG_FILE_NAME, MRGDDB_INPUT_FILE_NAME, + MRGDV_INPUT_FILE_NAME, STDERR_FILE_NAME, ) -__all__ = ["run_abinit", "run_anaddb", "run_mrgddb"] +__all__ = ["run_abinit", "run_anaddb", "run_mrgddb", "run_mrgdv"] SLEEP_TIME_STEP = 30 @@ -114,6 +115,52 @@ def run_mrgddb( process.wait() +def run_mrgdv( + mrgdv_cmd: str = None, + mpirun_cmd: str = None, + wall_time: int = None, + start_time: float = None, +) -> None: + """Run mrgdv.""" + mrgdv_cmd = mrgdv_cmd or SETTINGS.ABINIT_MRGDV_CMD + mpirun_cmd = mpirun_cmd or SETTINGS.ABINIT_MPIRUN_CMD + command = [] + if mpirun_cmd: + if not any(opt in mpirun_cmd.split() for opt in ("-c", "-n", "--n", "-np")): + mpirun_cmd += " -n 1" + command.extend(mpirun_cmd.split()) + command.append(mrgdv_cmd) + start_time = start_time or time.time() + + max_end_time = 0.0 + if wall_time is not None: + mrgdv_timelimit = wall_time + if mrgdv_timelimit > 480: + # TODO: allow tuning this timelimit buffer for mrgddb, + # e.g. using a config variable or possibly per job + mrgdv_timelimit -= 240 + max_end_time = start_time + wall_time + + with ( + open(MRGDV_INPUT_FILE_NAME) as stdin, + open(LOG_FILE_NAME, "w") as stdout, + open(STDERR_FILE_NAME, "w") as stderr, + ): + process = subprocess.Popen(command, stdin=stdin, stdout=stdout, stderr=stderr) + + if wall_time is not None: + while True: + time.sleep(SLEEP_TIME_STEP) + if process.poll() is not None: + break + current_time = time.time() + remaining_time = max_end_time - current_time + if remaining_time < 5 * SLEEP_TIME_STEP: + process.terminate() + + process.wait() + + def run_anaddb( anaddb_cmd: str = None, mpirun_cmd: str = None, diff --git a/src/atomate2/abinit/schemas/anaddb.py b/src/atomate2/abinit/schemas/anaddb.py index 6ea5299ed6..68bb858d4d 100644 --- a/src/atomate2/abinit/schemas/anaddb.py +++ b/src/atomate2/abinit/schemas/anaddb.py @@ -13,14 +13,28 @@ import abipy.core.abinit_units as abu import numpy as np from abipy.dfpt.anaddbnc import AnaddbNcFile +from abipy.dfpt.converters import abinit_to_phonopy +from abipy.dfpt.phonons import PhononBands, PhononDos from abipy.flowtk import events from abipy.flowtk.utils import File +from emmet.core.math import Matrix3D from emmet.core.structure import StructureMetadata +from monty.serialization import loadfn from pydantic import BaseModel, Field from pymatgen.core.structure import Structure +from pymatgen.phonon.bandstructure import PhononBandStructureSymmLine +from pymatgen.phonon.dos import PhononDos as pmgPhononDos from atomate2.abinit.schemas.calculation import AbinitObject, TaskState +from atomate2.abinit.schemas.outfiles import AbinitStoredFile from atomate2.abinit.utils.common import get_event_report +from atomate2.common.schemas.phonons import ( + ForceConstants, + PhononComputationalSettings, + PhononJobDirs, + PhononUUIDs, + ThermalDisplacementData, +) from atomate2.utils.path import get_uri logger = logging.getLogger(__name__) @@ -48,11 +62,110 @@ class CalculationOutput(BaseModel): epsinf: Optional[list] = Field( None, description="Electronic contribution to the dielectric tensor" ) + phonon_bandstructure: Optional[PhononBandStructureSymmLine] = Field( + None, + description="Phonon band structure object.", + ) + + phonon_dos: Optional[pmgPhononDos] = Field( + None, + description="Phonon density of states object.", + ) + + free_energies: Optional[list[float]] = Field( + None, + description="vibrational part of the free energies in J/mol per " + "formula unit for temperatures in temperature_list", + ) + + heat_capacities: Optional[list[float]] = Field( + None, + description="heat capacities in J/K/mol per " + "formula unit for temperatures in temperature_list", + ) + + internal_energies: Optional[list[float]] = Field( + None, + description="internal energies in J/mol per " + "formula unit for temperatures in temperature_list", + ) + entropies: Optional[list[float]] = Field( + None, + description="entropies in J/(K*mol) per formula unit" + "for temperatures in temperature_list ", + ) + + temperatures: Optional[list[int]] = Field( + None, + description="temperatures at which the vibrational" + " part of the free energies" + " and other properties have been computed", + ) + + total_dft_energy: Optional[float] = Field( + None, description="total DFT energy per formula unit in eV" + ) + + volume_per_formula_unit: Optional[float] = Field( + None, description="volume per formula unit in Angstrom**3" + ) + + formula_units: Optional[int] = Field(None, description="Formula units per cell") + + has_imaginary_modes: Optional[bool] = Field( + None, description="if true, structure has imaginary modes" + ) + + # needed, e.g. to compute Grueneisen parameter etc + force_constants: Optional[ForceConstants] = Field( + None, description="Force constants between every pair of atoms in the structure" + ) + + born: Optional[list[Matrix3D]] = Field( + None, + description="Born charges as computed from phonopy. Only for symmetrically " + "different atoms", + ) + + epsilon_static: Optional[Matrix3D] = Field( + None, description="The high-frequency dielectric constant" + ) + + supercell_matrix: Optional[Matrix3D] = Field( + None, description="matrix describing the supercell" + ) + primitive_matrix: Optional[Matrix3D] = Field( + None, description="matrix describing relationship to primitive cell" + ) + + code: Optional[str] = Field( + None, description="String describing the code for the computation" + ) + + phonopy_settings: Optional[PhononComputationalSettings] = Field( + None, description="Field including settings for Phonopy" + ) + + thermal_displacement_data: Optional[ThermalDisplacementData] = Field( + None, + description="Includes all data of the computation of the thermal displacements", + ) + + jobdirs: Optional[PhononJobDirs] = Field( + None, description="Field including all relevant job directories" + ) + + uuids: Optional[PhononUUIDs] = Field( + None, description="Field including all relevant uuids" + ) @classmethod def from_abinit_anaddb( cls, + dir_name: Path | str, output: AnaddbNcFile, + output_phbands: PhononBands = None, + output_phdos: PhononDos = None, ) -> CalculationOutput: """ Create an Anaddb output document from an AnaddbNcFile. @@ -67,13 +180,106 @@ def from_abinit_anaddb( The Anaddb calculation output document. """ structure = output.structure - dijk = list( - output.dchide * 16 * np.pi**2 * abu.Bohr_Ang**2 * 1e-8 * abu.eps0 / abu.e_Cb - ) # for pm/V units (SI) - epsinf = list(output.epsinf) + if output_phbands: + phonon_bandstructure = output_phbands.to_pymatgen() + phonon_bandstructure.labels_dict = { + k.strip("$"): v for k, v in phonon_bandstructure.labels_dict.items() + } + else: + phonon_bandstructure = None + phonon_dos = output_phdos.to_pymatgen() if output_phdos else None + try: + phonopy = abinit_to_phonopy( + anaddbnc=output, + supercell_matrix=loadfn(f"{dir_name}/anaddb_input.json")["ngqpt"], + ) + except (AttributeError, KeyError): + phonopy = None + if phonon_dos: + temperatures = [int(t) for t in output_phdos.get_free_energy().mesh] + free_energies = [ + phonon_dos.helmholtz_free_energy(temp, structure=structure) + for temp in temperatures + ] + heat_capacities = [ + phonon_dos.cv(temp=temp, structure=structure) for temp in temperatures + ] + internal_energies = [ + phonon_dos.internal_energy(temp, structure=structure) + for temp in temperatures + ] + entropies = [ + phonon_dos.entropy(temp, structure=structure) for temp in temperatures + ] + else: + temperatures = None + free_energies = None + heat_capacities = None + internal_energies = None + entropies = None + total_dft_energy = None # TODO: get the total energy from the scf gs I guess ? + formula_units = ( + structure.composition.num_atoms + / structure.composition.reduced_composition.num_atoms + ) + volume_per_formula_unit = structure.volume / formula_units + + has_imaginary_modes = ( + phonon_bandstructure.has_imaginary_freq() if phonon_bandstructure else None + ) + + if phonopy: + force_constants = ForceConstants(phonopy.force_constants) + born = phonopy.nac_params["born"].tolist() + supercell_matrix = phonopy.supercell_matrix.tolist() + primitive_matrix = phonopy.primitive_matrix.tolist() + else: + force_constants = None + born = None + supercell_matrix = None + primitive_matrix = None + epsilon_static = None # ??? + code = "abinit" + + # for pm/V units (SI) + dijk = ( + list( + output.dchide + * 16 + * np.pi**2 + * abu.Bohr_Ang**2 + * 1e-8 + * abu.eps0 + / abu.e_Cb + ) + if output.dchide is not None and output.dchide.any() + else None + ) + epsinf = ( + list(output.epsinf) + if output.epsinf is not None and output.epsinf.any() + else None + ) return cls( structure=structure, + phonon_bandstructure=phonon_bandstructure, + phonon_dos=phonon_dos, + free_energies=free_energies, + heat_capacities=heat_capacities, + internal_energies=internal_energies, + entropies=entropies, + temperatures=temperatures, + total_dft_energy=total_dft_energy, + volume_per_formula_unit=volume_per_formula_unit, + formula_units=formula_units, + has_imaginary_modes=has_imaginary_modes, + force_constants=force_constants, + born=born, + epsilon_static=epsilon_static, + supercell_matrix=supercell_matrix, + primitive_matrix=primitive_matrix, + code=code, dijk=dijk, epsinf=epsinf, ) @@ -123,6 +329,9 @@ def from_abinit_files( task_name: str, abinit_anaddb_file: Path | str = "out_anaddb.nc", abinit_analog_file: Path | str = "run.log", + abinit_phbst_file: Path | str = "out_PHBST.nc", + abinit_phdos_file: Path | str = "out_PHDOS.nc", + files_to_store: list | None = None, ) -> tuple[Calculation, dict[AbinitObject, dict]]: """ Create a anaddb calculation document from a directory and file paths. @@ -137,6 +346,10 @@ def from_abinit_files( Path to the merged DDB file, relative to dir_name. abinit_analog_file: Path or str Path to the main log of anaddb job, relative to dir_name. + abinit_phbst_file: Path or str + Path to the PHBST file, relative to dir_name + abinit_phdos_file: Path or str + Path to the PHDOS file, relative to dir_name Returns ------- @@ -145,18 +358,38 @@ def from_abinit_files( """ dir_name = Path(dir_name) abinit_anaddb_file = dir_name / abinit_anaddb_file + abinit_phbst_file = dir_name / abinit_phbst_file + abinit_phdos_file = dir_name / abinit_phdos_file output_doc = None if abinit_anaddb_file.exists(): abinit_anaddb = AnaddbNcFile.from_file(abinit_anaddb_file) - output_doc = CalculationOutput.from_abinit_anaddb(abinit_anaddb) + if abinit_phbst_file.exists(): + abinit_phbst = PhononBands.from_file(abinit_phbst_file) + if abinit_phdos_file.exists(): + abinit_phdos = PhononDos.as_phdos(str(abinit_phdos_file)) + if abinit_anaddb_file.exists(): + output_doc = CalculationOutput.from_abinit_anaddb( + dir_name=dir_name, + output=abinit_anaddb, + output_phbands=abinit_phbst if abinit_phbst else None, + output_phdos=abinit_phdos if abinit_phdos else None, + ) completed_at = str( datetime.fromtimestamp( os.stat(abinit_anaddb_file).st_mtime, tz=timezone.utc ) ) - + abinit_objects: dict[AbinitObject, Any] = {} + if abinit_phbst_file.exists() and "PHBST" in files_to_store: + abinit_objects[AbinitObject.PHBSTFILE] = AbinitStoredFile.from_file( # type: ignore[index] + filepath=abinit_phbst_file, data_type=bytes + ) + if abinit_phdos_file.exists() and "PHDOS" in files_to_store: + abinit_objects[AbinitObject.PHDOSFILE] = AbinitStoredFile.from_file( # type: ignore[index] + filepath=abinit_phdos_file, data_type=bytes + ) report = None has_anaddb_completed = TaskState.FAILED try: @@ -178,7 +411,7 @@ def from_abinit_files( output=output_doc, event_report=report, ), - None, # abinit_objects, + abinit_objects, ) @@ -195,13 +428,111 @@ class OutputDoc(BaseModel): The electronic contribution to the dielectric tensor """ - structure: Union[Structure] = Field(None, description="The output structure object") + structure: Union[Structure] = Field( + None, description="The final structure from the calculation" + ) dijk: Optional[list] = Field( None, description="Conventional SHG tensor in pm/V (Chi^(2)/2)" ) epsinf: Optional[list] = Field( None, description="Electronic contribution to the dielectric tensor" ) + phonon_bandstructure: Optional[PhononBandStructureSymmLine] = Field( + None, + description="Phonon band structure object.", + ) + + phonon_dos: Optional[pmgPhononDos] = Field( + None, + description="Phonon density of states object.", + ) + + free_energies: Optional[list[float]] = Field( + None, + description="vibrational part of the free energies in J/mol per " + "formula unit for temperatures in temperature_list", + ) + + heat_capacities: Optional[list[float]] = Field( + None, + description="heat capacities in J/K/mol per " + "formula unit for temperatures in temperature_list", + ) + + internal_energies: Optional[list[float]] = Field( + None, + description="internal energies in J/mol per " + "formula unit for temperatures in temperature_list", + ) + entropies: Optional[list[float]] = Field( + None, + description="entropies in J/(K*mol) per formula unit" + "for temperatures in temperature_list ", + ) + + temperatures: Optional[list[int]] = Field( + None, + description="temperatures at which the vibrational" + " part of the free energies" + " and other properties have been computed", + ) + + total_dft_energy: Optional[float] = Field( + None, description="total DFT energy per formula unit in eV" + ) + + volume_per_formula_unit: Optional[float] = Field( + None, description="volume per formula unit in Angstrom**3" + ) + + formula_units: Optional[int] = Field(None, description="Formula units per cell") + + has_imaginary_modes: Optional[bool] = Field( + None, description="if true, structure has imaginary modes" + ) + + # needed, e.g. to compute Grueneisen parameter etc + force_constants: Optional[ForceConstants] = Field( + None, description="Force constants between every pair of atoms in the structure" + ) + + born: Optional[list[Matrix3D]] = Field( + None, + description="Born charges as computed from phonopy. Only for symmetrically " + "different atoms", + ) + + epsilon_static: Optional[Matrix3D] = Field( + None, description="The high-frequency dielectric constant" + ) + + supercell_matrix: Optional[Matrix3D] = Field( + None, description="matrix describing the supercell" + ) + primitive_matrix: Optional[Matrix3D] = Field( + None, description="matrix describing relationship to primitive cell" + ) + + code: Optional[str] = Field( + None, description="String describing the code for the computation" + ) + + phonopy_settings: Optional[PhononComputationalSettings] = Field( + None, description="Field including settings for Phonopy" + ) + + thermal_displacement_data: Optional[ThermalDisplacementData] = Field( + None, + description="Includes all data of the computation of the thermal displacements", + ) + + jobdirs: Optional[PhononJobDirs] = Field( + None, description="Field including all relevant job directories" + ) + + uuids: Optional[PhononUUIDs] = Field( + None, description="Field including all relevant uuids" + ) @classmethod def from_abinit_calc_doc(cls, calc_doc: Calculation) -> OutputDoc: @@ -219,6 +550,23 @@ def from_abinit_calc_doc(cls, calc_doc: Calculation) -> OutputDoc: """ return cls( structure=calc_doc.output.structure, + phonon_bandstructure=calc_doc.output.phonon_bandstructure, + phonon_dos=calc_doc.output.phonon_dos, + free_energies=calc_doc.output.free_energies, + heat_capacities=calc_doc.output.heat_capacities, + internal_energies=calc_doc.output.internal_energies, + entropies=calc_doc.output.entropies, + temperatures=calc_doc.output.temperatures, + total_dft_energy=calc_doc.output.total_dft_energy, + volume_per_formula_unit=calc_doc.output.volume_per_formula_unit, + formula_units=calc_doc.output.formula_units, + has_imaginary_modes=calc_doc.output.has_imaginary_modes, + force_constants=calc_doc.output.force_constants, + born=calc_doc.output.born, + epsilon_static=calc_doc.output.epsilon_static, + supercell_matrix=calc_doc.output.supercell_matrix, + primitive_matrix=calc_doc.output.primitive_matrix, + code=calc_doc.output.code, dijk=calc_doc.output.dijk, epsinf=calc_doc.output.epsinf, ) @@ -378,6 +726,10 @@ def _get_task_files(files: list[Path], suffix: str = "") -> dict: abinit_files["abinit_anaddb_file"] = Path(file).relative_to(path) elif file.match(f"*run.log{suffix}*"): abinit_files["abinit_analog_file"] = Path(file).relative_to(path) + if file.match(f"*outdata/out_PHBST.nc{suffix}*"): + abinit_files["abinit_phbst_file"] = Path(file).relative_to(path) + if file.match(f"*outdata/out_PHDOS.nc{suffix}*"): + abinit_files["abinit_phdos_file"] = Path(file).relative_to(path) return abinit_files diff --git a/src/atomate2/abinit/schemas/calculation.py b/src/atomate2/abinit/schemas/calculation.py index 868e52b79d..bebffa4942 100644 --- a/src/atomate2/abinit/schemas/calculation.py +++ b/src/atomate2/abinit/schemas/calculation.py @@ -47,6 +47,8 @@ class AbinitObject(ValueEnum): TRAJECTORY = "trajectory" DDBFILE = "ddb" # DDB file as string POTENTIAL = "potential" # POT file as b-string + PHBSTFILE = "phbst" # Phonon Bandstructure file + PHDOSFILE = "phdos" # Phonon DOS file class CalculationOutput(BaseModel): diff --git a/src/atomate2/abinit/schemas/mrgdvdb.py b/src/atomate2/abinit/schemas/mrgdvdb.py new file mode 100644 index 0000000000..a2c2621ab7 --- /dev/null +++ b/src/atomate2/abinit/schemas/mrgdvdb.py @@ -0,0 +1,499 @@ +"""Core definitions of mrgdvdb calculations documents.""" + +from __future__ import annotations + +import logging +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from emmet.core.structure import StructureMetadata +from jobflow.utils import ValueEnum +from pydantic import BaseModel, Field + +from atomate2.abinit.schemas.outfiles import AbinitStoredFile +from atomate2.abinit.utils.common import get_mrgdv_report +from atomate2.utils.path import get_uri + +logger = logging.getLogger(__name__) + + +class TaskState(ValueEnum): + """Mrgdv calculation state.""" + + SUCCESS = "successful" + FAILED = "failed" + UNCONVERGED = "unconverged" + + +# # need to inherit from MSONable to be stored in the data store +# # I tried to combine it with @dataclass, but didn't work... +# class dvFileStr(MSONable): +# """Object storing the raw string of a dv file.""" + +# def __init__(self, dvfilepath: str | Path, dv_as_str: str) -> None: +# self.dvfilepath: str | Path = dvfilepath +# self.dv_as_str: str = dv_as_str + +# @classmethod +# def from_dvfile(cls, dvfile: dvFile) -> Self: +# """Create a dvFileStr object from the native dvFile abipy object.""" +# with open(dvfile.filepath) as f: +# dv_as_str = f.read() +# return cls(dvfilepath=dvfile.filepath, dv_as_str=dv_as_str) + + +class MrgdvdbObject(ValueEnum): + """Types of Mrgdvdb data objects.""" + + DVDBFILE = "out_DVDB" # DVDB file as string + + +class CalculationOutput(BaseModel): + """Document defining Mrgdv calculation outputs. + + Parameters + ---------- + dvdb_version: int + The DVDB file version + v1scf_number: int + Number of V1SCF potential + qpt_number: int + Number of q-points + have_dielectric: bool + Whether the dielectric tensor is present + have_bec: bool + Whether the Born effective charges are present + have_quadrupoles: bool + Whether the quadrupoles are present + have_efield: bool + Whether the derivative wrt the electric field is present + add_lr: int + Treatment of long-range part of V1scf + qdam: float + Damping factor for Gaussian Filter + """ + + dvdb_version: int = Field(None, description="The DVDB file version") + v1scf_number: int = Field(None, description="Number of V1SCF potential") + qpt_number: int = Field(None, description="Number of q-points") + have_dielectric: bool = Field( + None, description="Whether the dielectric tensor is present" + ) + have_bec: bool = Field( + None, description="Whether the Born effective charges are present" + ) + have_quadrupoles: bool = Field( + None, description="Whether the quadrupoles are present" + ) + have_efield: bool = Field( + None, description="Whether the derivative wrt the electric field is present" + ) + add_lr: int = Field(None, description="Treatment of long-range part of V1scf") + qdam: float = Field(None, description="Damping factor for Gaussian Filter") + + @classmethod + def from_abinit_logdvdb( + cls, + logfile: str | Path, + ) -> CalculationOutput: + """ + Create an Mrgdv output document from the merged Abinit out_dv file. + + Parameters + ---------- + logfile: str or Path + The path to the log of the mrgdv calculation. + + Returns + ------- + The Mrgdv calculation output document. + """ + with open(logfile) as f: + for line in f: + if "DVDB version" in line: + dvdb_version = int(line.split()[2]) + elif "Number of v1scf potentials" in line: + v1scf_number = int(line.split()[4]) + elif "Number of q-points in DVDB" in line: + qpt_number = int(line.split()[5]) + elif "Have dielectric tensor" in line: + have_dielectric = "yes" in line + elif "Have Born effective charges" in line: + have_bec = "yes" in line + elif "Have quadrupoles" in line: + have_quadrupoles = "yes" in line + elif "Have electric field" in line: + have_efield = "yes" in line + elif "Treatment of long-range part in V1scf" in line: + add_lr = int(line.split()[7]) + elif "Damping factor for Gaussian filter" in line: + qdam = float(line.split()[6]) + + return cls( + dvdb_version=dvdb_version, + v1scf_number=v1scf_number, + qpt_number=qpt_number, + have_dielectric=have_dielectric, + have_bec=have_bec, + have_quadrupoles=have_quadrupoles, + have_efield=have_efield, + add_lr=add_lr, + qdam=qdam, + ) + + +class Calculation(BaseModel): + """Full Mrgdv calculation (inputs) and outputs. + + Parameters + ---------- + dir_name: str + The directory for this Mrgdv calculation + has_mrgdv_completed: .TaskState + Whether Mrgdv completed the merge successfully + output: .CalculationOutput + The Mrgdv calculation output + completed_at: str + Timestamp for when the merge was completed + output_file_paths: Dict[str, str] + Paths (relative to dir_name) of the Mrgdv output files + associated with this calculation + """ + + dir_name: str = Field(None, description="The directory for this Abinit calculation") + has_mrgdv_completed: TaskState = Field( + None, description="Whether Abinit completed the calculation successfully" + ) + output: Optional[CalculationOutput] = Field( + None, description="The Abinit calculation output" + ) + completed_at: str = Field( + None, description="Timestamp for when the calculation was completed" + ) + output_file_paths: Optional[dict[str, str]] = Field( + None, + description="Paths (relative to dir_name) of the Abinit output files " + "associated with this calculation", + ) + + @classmethod + def from_abinit_files( + cls, + dir_name: Path | str, + task_name: str, + abinit_outdvdb_file: Path | str = "out_dvdb", + abinit_mrglog_file: Path | str = "run.log", + ) -> tuple[Calculation, dict[MrgdvdbObject, dict]]: + """ + Create a Mrgdv calculation document from a directory and file paths. + + Parameters + ---------- + dir_name: Path or str + The directory containing the calculation outputs. + task_name: str + The task name. + abinit_outdvdb_file: Path or str + Path to the merged dv file, relative to dir_name. + abinit_mrglog_file: Path or str + Path to the main log of mrgdv job, relative to dir_name. + + Returns + ------- + .Calculation + A Mrgdv calculation document. + """ + dir_name = Path(dir_name) + abinit_outdvdb_file = dir_name / Path(abinit_outdvdb_file) + # abinit_mrglog_file = dir_name.parent / abinit_mrglog_file + + output_doc = None + mrgdv_objects: dict[MrgdvdbObject, Any] = {} + if abinit_outdvdb_file.exists() and Path(abinit_mrglog_file).exists(): + mrgdv_objects[MrgdvdbObject.DVDBFILE] = AbinitStoredFile.from_file( # type: ignore[index] + filepath=abinit_outdvdb_file, data_type=bytes + ) + output_doc = CalculationOutput.from_abinit_logdvdb(abinit_mrglog_file) + + completed_at = str( + datetime.fromtimestamp( + os.stat(abinit_outdvdb_file).st_mtime, tz=timezone.utc + ) + ) + + report = None + has_mrgdv_completed = TaskState.FAILED + try: + report = get_mrgdv_report(logfile=abinit_mrglog_file) + + if report["run_completed"] and abinit_outdvdb_file.exists(): + has_mrgdv_completed = TaskState.SUCCESS + + # except (dvError, Exception) as exc: + except Exception as exc: + msg = f"{cls} exception while parsing mrgdv event_report:\n{exc}" + logger.critical(msg) + logging.exception(msg) + + return ( + cls( + dir_name=str(dir_name), + task_name=task_name, + has_mrgdv_completed=has_mrgdv_completed, + completed_at=completed_at, + output=output_doc, + ), + mrgdv_objects, + ) + + +class OutputDoc(BaseModel): + """Summary of the outputs for a Mrgdv calculation. + + Parameters + ---------- + dvdb_version: int + The DVDB file version + v1scf_number: int + Number of V1SCF potential + qpt_number: int + Number of q-points + have_dielectric: bool + Whether the dielectric tensor is present + have_bec: bool + Whether the Born effective charges are present + have_quadrupoles: bool + Whether the quadrupoles are present + have_efield: bool + Whether the derivative wrt the electric field is present + add_lr: int + Treatment of long-range part of V1scf + qdam: float + Damping factor for Gaussian Filter + """ + + dvdb_version: Optional[int] = Field(None, description="The DVDB file version") + v1scf_number: Optional[int] = Field(None, description="Number of V1SCF potential") + qpt_number: Optional[int] = Field(None, description="Number of q-points") + have_dielectric: Optional[bool] = Field( + None, description="Whether the dielectric tensor is present" + ) + have_bec: Optional[bool] = Field( + None, description="Whether the Born effective charges are present" + ) + have_quadrupoles: Optional[bool] = Field( + None, description="Whether the quadrupoles are present" + ) + have_efield: Optional[bool] = Field( + None, description="Whether the derivative wrt the electric field is present" + ) + add_lr: Optional[int] = Field( + None, description="Treatment of long-range part of V1scf" + ) + qdam: Optional[float] = Field( + None, description="Damping factor for Gaussian Filter" + ) + + @classmethod + def from_abinit_calc_doc(cls, calc_doc: Calculation) -> OutputDoc: + """Create a summary from an abinit CalculationDocument. + + Parameters + ---------- + calc_doc: .Calculation + A Mrgdv calculation document. + + Returns + ------- + .OutputDoc + The calculation output summary. + """ + return cls( + dvdb_version=calc_doc.output.dvdb_version, + v1scf_number=calc_doc.output.v1scf_number, + qpt_number=calc_doc.output.qpt_number, + have_dielectric=calc_doc.output.have_dielectric, + have_bec=calc_doc.output.have_bec, + have_quadrupoles=calc_doc.output.have_quadrupoles, + have_efield=calc_doc.output.have_efield, + add_lr=calc_doc.output.add_lr, + qdam=calc_doc.output.qdam, + ) + + +class MrgdvdbTaskDoc(StructureMetadata): + """Definition of task document about an Mrgdv Job. + + Parameters + ---------- + dir_name: str + The directory for this Abinit task + completed_at: str + Timestamp for when this task was completed + output: .OutputDoc + The output of the final calculation + state: .TaskState + State of this task + included_objects: List[.MrgdvdbObject] + List of Abinit objects included with this task document + abinit_objects: Dict[.MrgdvdbObject, Any] + Abinit objects associated with this task + task_label: str + A description of the task + tags: List[str] + Metadata tags for this task document + """ + + dir_name: Optional[str] = Field( + None, description="The directory for this Abinit task" + ) + completed_at: Optional[str] = Field( + None, description="Timestamp for when this task was completed" + ) + output: Optional[OutputDoc] = Field( + None, description="The output of the final calculation" + ) + state: Optional[TaskState] = Field(None, description="State of this task") + included_objects: Optional[list[MrgdvdbObject]] = Field( + None, description="List of Mrgdv objects included with this task document" + ) + mrgdv_objects: Optional[dict[MrgdvdbObject, Any]] = Field( + None, description="Mrgdv objects associated with this task" + ) + task_label: Optional[str] = Field(None, description="A description of the task") + dvdb_version: Optional[int] = Field(None, description="The DVDB file version") + v1scf_number: Optional[int] = Field(None, description="Number of V1SCF potential") + qpt_number: Optional[int] = Field(None, description="Number of q-points") + have_dielectric: Optional[bool] = Field( + None, description="Whether the dielectric tensor is present" + ) + have_bec: Optional[bool] = Field( + None, description="Whether the Born effective charges are present" + ) + have_quadrupoles: Optional[bool] = Field( + None, description="Whether the quadrupoles are present" + ) + have_efield: Optional[bool] = Field( + None, description="Whether the derivative wrt the electric field is present" + ) + add_lr: Optional[int] = Field( + None, description="Treatment of long-range part of V1scf" + ) + qdam: Optional[float] = Field( + None, description="Damping factor for Gaussian Filter" + ) + tags: Optional[list[str]] = Field( + None, description="Metadata tags for this task document" + ) + + @classmethod + def from_directory( + cls, + dir_name: Path | str, + additional_fields: dict[str, Any] = None, + **abinit_calculation_kwargs, + ) -> MrgdvdbTaskDoc: + """Create a task document from a directory containing Abinit/Mrgdv files. + + Parameters + ---------- + dir_name: Path or str + The path to the folder containing the calculation outputs. + additional_fields: Dict[str, Any] + Dictionary of additional fields to add to output document. + **abinit_calculation_kwargs + Additional parsing options that will be passed to the + :obj:`.Calculation.from_abinit_files` function. + + Returns + ------- + .MrgdvdbTaskDoc + A task document for the calculation. + """ + logger.info(f"Getting task doc in: {dir_name}") + + if additional_fields is None: + additional_fields = {} + + dir_name = Path(dir_name) + task_files = _find_abinit_files(dir_name) + + if len(task_files) == 0: + raise FileNotFoundError("No Abinit files found!") + + calcs_reversed = [] + all_mrgdv_objects = [] + for task_name, files in task_files.items(): + calc_doc, mrgdv_objects = Calculation.from_abinit_files( + dir_name, task_name, **files, **abinit_calculation_kwargs + ) + calcs_reversed.append(calc_doc) + all_mrgdv_objects.append(mrgdv_objects) + + tags = additional_fields.get("tags") + + dir_name = get_uri(dir_name) # convert to full uri path + + # only store objects from last calculation + # TODO: make this an option + mrgdv_objects = all_mrgdv_objects[-1] + included_objects = None + if mrgdv_objects: + included_objects = list(mrgdv_objects) + + # rewrite the original structure save! + + data = { + "calcs_reversed": calcs_reversed, + "completed_at": calcs_reversed[-1].completed_at, + "dir_name": dir_name, + "included_objects": included_objects, + "mrgdv_objects": mrgdv_objects, + # "meta_structure": calcs_reversed[-1].output.structure, + "state": calcs_reversed[-1].has_mrgdv_completed, + # "structure": calcs_reversed[-1].output.structure, + "dvdb_version": calcs_reversed[-1].output.dvdb_version, + "v1scf_number": calcs_reversed[-1].output.v1scf_number, + "qpt_number": calcs_reversed[-1].output.qpt_number, + "have_dielectric": calcs_reversed[-1].output.have_dielectric, + "have_bec": calcs_reversed[-1].output.have_bec, + "have_quadrupoles": calcs_reversed[-1].output.have_quadrupoles, + "have_efield": calcs_reversed[-1].output.have_efield, + "add_lr": calcs_reversed[-1].output.add_lr, + "qdam": calcs_reversed[-1].output.qdam, + "tags": tags, + } + + doc = cls(**data) + # doc = doc.model_copy(update=data) + return doc.model_copy(update=additional_fields, deep=True) + + +def _find_abinit_files( + path: Path | str, +) -> dict[str, Any]: + """Find Abinit files.""" + path = Path(path) + task_files = {} + + def _get_task_files(files: list[Path], suffix: str = "") -> dict: + abinit_files = {} + for file in files: + # Here we make assumptions about the output file naming + if file.match(f"*outdata/out_dv{suffix}*"): + abinit_files["abinit_outdvdb_file"] = Path(file).relative_to(path) + elif file.match(f"*run.log{suffix}*"): + abinit_files["abinit_mrglog_file"] = Path(file).relative_to(path) + + return abinit_files + + # get any matching file from the root folder + standard_files = _get_task_files( + list(path.glob("*")) + list(path.glob("outdata/*")) + ) + if len(standard_files) > 0: + task_files["standard"] = standard_files + + return task_files diff --git a/src/atomate2/abinit/schemas/outfiles.py b/src/atomate2/abinit/schemas/outfiles.py index f7b60235f8..6f70607e0b 100644 --- a/src/atomate2/abinit/schemas/outfiles.py +++ b/src/atomate2/abinit/schemas/outfiles.py @@ -45,7 +45,7 @@ def from_dict(cls, d: dict) -> Self: def from_file(cls, filepath: Union[str, Path], data_type: Union[str, type]) -> Self: """Create an AbinitStoredFile from the original file.""" source_filepath = os.path.abspath(filepath) - if data_type in {"bytes", bytes}: + if data_type in {"bytes", bytes} or ".nc" in str(filepath): read_type = "rb" elif data_type in {"str", str}: read_type = "r" diff --git a/src/atomate2/abinit/sets/anaddb.py b/src/atomate2/abinit/sets/anaddb.py index 1647669ae1..4a1b8ce136 100644 --- a/src/atomate2/abinit/sets/anaddb.py +++ b/src/atomate2/abinit/sets/anaddb.py @@ -350,3 +350,43 @@ class AnaddbDfptDteInputGenerator(AnaddbInputGenerator): factory: Callable = anaddbinp_dfpt_dte # partial does not a __name__ so cannot jsanitize... # factory: Callable = partial(AnaddbInput.dfpt, dte=True) + + +def anaddbinp_phbands_dos( + structure: Structure, + **kwargs, +) -> AnaddbInput: + """ + Generate the AnaddbInput for phonon bands and DOS. + + Parameters + ---------- + structure + A structure. + + Returns + ------- + An AnaddbInput + + """ + return AnaddbInput.phbands_and_dos(structure=structure, **kwargs) + + +@dataclass +class AnaddbPhbandsDOSInputGenerator(AnaddbInputGenerator): + """ + A class to generate the AnaddbInput for phonon bands and DOS. + + Parameters + ---------- + factory + A callable to generate the AnaddbInput for phonon bands and DOS. + + Returns + ------- + An AnaddbInput + + """ + + factory: Callable = anaddbinp_phbands_dos + factory_kwargs: dict = field(default_factory=dict) diff --git a/src/atomate2/abinit/sets/mrgddb.py b/src/atomate2/abinit/sets/mrgddb.py index 20a93c7a8d..508ee83e93 100644 --- a/src/atomate2/abinit/sets/mrgddb.py +++ b/src/atomate2/abinit/sets/mrgddb.py @@ -9,7 +9,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING -from abipy.abio.input_tags import DDE, DTE +from abipy.abio.input_tags import DDE, DTE, PH_Q_PERT from abipy.flowtk.utils import Directory from pymatgen.io.core import InputSet @@ -135,7 +135,7 @@ class MrgddbInputGenerator(AbinitMixinInputGenerator): """ calc_type: str = "mrgddb" - prev_outputs_deps: tuple = (f"{DDE}:DDB", f"{DTE}:DDB") + prev_outputs_deps: tuple = (f"{DDE}:DDB", f"{DTE}:DDB", f"{PH_Q_PERT}:DDB") def get_input_set( self, diff --git a/src/atomate2/abinit/sets/mrgdvdb.py b/src/atomate2/abinit/sets/mrgdvdb.py new file mode 100644 index 0000000000..e1c94bfb40 --- /dev/null +++ b/src/atomate2/abinit/sets/mrgdvdb.py @@ -0,0 +1,219 @@ +"""Module defining base MRGDVDB input set and generator.""" + +from __future__ import annotations + +import copy +import logging +import os +import time +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from abipy.abio.input_tags import DDE, DTE, PH_Q_PERT +from abipy.flowtk.utils import Directory +from pymatgen.io.core import InputSet + +from atomate2.abinit.sets.base import AbinitMixinInputGenerator +from atomate2.abinit.utils.common import ( + INDIR_NAME, + MRGDV_INPUT_FILE_NAME, + OUTDIR_NAME, + TMPDIR_NAME, +) + +if TYPE_CHECKING: + from collections.abc import Iterable + from pathlib import Path + +__all__ = ["MrgdvInputGenerator", "MrgdvInputSet"] + +logger = logging.getLogger(__name__) + + +class MrgdvInputSet(InputSet): + """ + A class to represent a set of Mrgdv inputs. + + Parameters + ---------- + mrgdv_input + An input (str) to mrgdv. + input_files + A list of input files needed for the calculation. + """ + + def __init__( + self, + mrgdv_input: str = None, + input_files: Iterable[tuple[str, str]] | None = None, + ) -> None: + self.input_files = input_files + super().__init__( + inputs={ + MRGDV_INPUT_FILE_NAME: mrgdv_input, + } + ) + + def write_input( + self, + directory: str | Path, + make_dir: bool = True, + overwrite: bool = True, + zip_inputs: bool = False, + ) -> None: + """Write Mrgdv input file to a directory.""" + # TODO: do we allow zipping ? not sure if it really makes sense for abinit as + # the abinit input set also sets up links to previous files, sets up the + # indir, outdir and tmpdir, ... + super().write_input( + directory=directory, + make_dir=make_dir, + overwrite=overwrite, + zip_inputs=zip_inputs, + ) + indir, outdir, tmpdir = self.set_workdir(workdir=directory) + + def validate(self) -> bool: + """Validate the input set. + + Check that all input files exist and are dv files. + """ + if not self.input_files: + return False + for _out_filepath, in_file in self.input_files: + if not os.path.isfile(_out_filepath) or not in_file.startswith("in_POT"): + return False + return True + + @property + def mrgdv_input(self) -> str: + """Get the Mrgdv input (str).""" + return self[MRGDV_INPUT_FILE_NAME] + + @staticmethod + def set_workdir(workdir: Path | str) -> tuple[Directory, Directory, Directory]: + """Set up the working directory. + + This also sets up and creates standard input, output and temporary directories. + """ + workdir = os.path.abspath(workdir) + + # Directories with input|output|temporary data. + indir = Directory(os.path.join(workdir, INDIR_NAME)) + outdir = Directory(os.path.join(workdir, OUTDIR_NAME)) + tmpdir = Directory(os.path.join(workdir, TMPDIR_NAME)) + + # Create dirs for input, output and tmp data. + indir.makedirs() + outdir.makedirs() + tmpdir.makedirs() + + return indir, outdir, tmpdir + + def deepcopy(self) -> MrgdvInputSet: + """Deep copy of the input set.""" + return copy.deepcopy(self) + + +@dataclass +class MrgdvInputGenerator(AbinitMixinInputGenerator): + """ + A class to generate Mrgdv input sets. + + Parameters + ---------- + calc_type + A short description of the calculation type + prev_outputs_deps + Defines the files that needs to be linked from previous calculations and + are required for the execution of the current calculation. + The format is a tuple where each element is a list of "|" separated + runlevels (as defined in the AbinitInput object) followed by a colon and + a list of "|" list of extensions of files that needs to be linked. + The runlevel defines the type of calculations from which the file can + be linked. An example is (f"{NSCF}:WFK",). + """ + + calc_type: str = "mrgdv" + prev_outputs_deps: tuple = (f"{DDE}:POT", f"{DTE}:POT", f"{PH_Q_PERT}:POT") + + def get_input_set( + self, + prev_outputs: str | tuple | list | Path | None = None, + workdir: str | Path | None = ".", + ) -> MrgdvInputSet: + """Generate an MrgdvInputSet object. + + Here we assume that prev_outputs is + a list of directories. + + Parameters + ---------- + prev_outputs : str or Path or list or tuple + Directory (as a str or Path) or list/tuple of directories (as a str + or Path) needed as dependencies for the MrgdvInputSet generated. + """ + prev_outputs = self.check_format_prev_dirs(prev_outputs) + + input_files = [] + if prev_outputs is not None and not self.prev_outputs_deps: + raise RuntimeError( + f"Previous outputs not allowed for {self.__class__.__name__}." + ) + irdvars, files = self.resolve_deps(prev_outputs, self.prev_outputs_deps) + input_files.extend(files) + mrgdv_input = self.get_mrgdv_input( + prev_outputs=prev_outputs, + workdir=workdir, + ) + + return MrgdvInputSet( + mrgdv_input=mrgdv_input, + input_files=input_files, + ) + + def get_mrgdv_input( + self, + prev_outputs: list[str] | None = None, + workdir: str | Path | None = ".", + ) -> str: + """ + Generate the mrgdv input (str) for the input set. + + Parameters + ---------- + prev_outputs + A list of previous output directories. + + Returns + ------- + A string + """ + if not prev_outputs: + raise RuntimeError( + f"No previous_outputs. Required for {self.__class__.__name__}." + ) + + if not self.prev_outputs_deps and prev_outputs: + msg = ( + f"Previous outputs not allowed for {self.__class__.__name__} " + "Consider if get_input_set method " + "can fit your needs instead." + ) + raise RuntimeError(msg) + + irdvars, files = self.resolve_deps(prev_outputs, self.prev_outputs_deps) + + workdir = os.path.abspath(workdir) + outdir = Directory(os.path.join(workdir, OUTDIR_NAME, "out_dvdb")) + + generated_input = str(outdir) + generated_input += "\n" + generated_input += f"dvdbs merged on {time.asctime()}" + generated_input += "\n" + generated_input += f"{len(files)}" + for file_path, _ in files: + generated_input += "\n" + generated_input += f"{file_path}" + + return generated_input diff --git a/src/atomate2/abinit/sets/response.py b/src/atomate2/abinit/sets/response.py index 6bf3187b02..3860886903 100644 --- a/src/atomate2/abinit/sets/response.py +++ b/src/atomate2/abinit/sets/response.py @@ -9,6 +9,7 @@ ddepert_from_gsinput, ddkpert_from_gsinput, dtepert_from_gsinput, + phononpert_from_gsinput, ) from abipy.abio.input_tags import DDE, DDK, DTE, SCF @@ -23,6 +24,7 @@ "DdeSetGenerator", "DdkSetGenerator", "DteSetGenerator", + "PhononSetGenerator", ] @@ -66,3 +68,17 @@ class DteSetGenerator(StaticSetGenerator): factory_prev_inputs_kwargs: dict | None = field( default_factory=lambda: {"gs_input": (SCF,)} ) + + +@dataclass +class PhononSetGenerator(StaticSetGenerator): + """Class to generate Abinit Phonon input sets.""" + + calc_type: str = "Phonon" + factory: Callable = phononpert_from_gsinput + pseudos: str | list[str] | PseudoTable | None = None + restart_from_deps: tuple = (f"{SCF}:WFK",) + prev_outputs_deps: tuple = (f"{SCF}:WFK", f"{DDE}:1WF", f"{DDE}:1DEN") + factory_prev_inputs_kwargs: dict | None = field( + default_factory=lambda: {"gs_input": (SCF,)} + ) diff --git a/src/atomate2/abinit/utils/common.py b/src/atomate2/abinit/utils/common.py index ccc2b126c7..7e6cd9b29d 100644 --- a/src/atomate2/abinit/utils/common.py +++ b/src/atomate2/abinit/utils/common.py @@ -41,6 +41,7 @@ OUTNC_FILE_NAME = "out_OUT.nc" INPUT_FILE_NAME: str = "run.abi" MRGDDB_INPUT_FILE_NAME: str = "mrgddb.in" +MRGDV_INPUT_FILE_NAME: str = "mrgdv.in" ANADDB_INPUT_FILE_NAME: str = "anaddb.in" MPIABORTFILE = "__ABI_MPIABORTFILE__" DUMMY_FILENAME = "__DUMMY__" @@ -462,3 +463,28 @@ def get_mrgddb_report( with open(str(logfile)) as f: last_line = f.readlines()[-1] return {"run_completed": "completed successfully" in last_line} + + +def get_mrgdv_report( + logfile: str | Path, +) -> dict: + """Get report from mrgdv. + + This returns a dict with a "run_completed" key that is True + if "completed successfully" is present in the log. + + Parameters + ---------- + logfile : File + Output file to be parsed. Should be either the log file (stdout). + + Returns + ------- + dict + dict with bool "run_completed" key. + """ + if not Path(logfile).exists: + return {"run_completed": False} + with open(str(logfile)) as f: + last_line = f.readlines()[-1] + return {"run_completed": "Done" in last_line} diff --git a/src/atomate2/settings.py b/src/atomate2/settings.py index e810bb341b..d3e8e420d5 100644 --- a/src/atomate2/settings.py +++ b/src/atomate2/settings.py @@ -198,6 +198,7 @@ class Atomate2Settings(BaseSettings): ABINIT_MPIRUN_CMD: Optional[str] = Field(None, description="Mpirun command.") ABINIT_CMD: str = Field("abinit", description="Abinit command.") ABINIT_MRGDDB_CMD: str = Field("mrgddb", description="Mrgddb command.") + ABINIT_MRGDV_CMD: str = Field("mrgdv", description="Mrgdv command.") ABINIT_ANADDB_CMD: str = Field("anaddb", description="Anaddb command.") ABINIT_COPY_DEPS: bool = Field( default=False, diff --git a/tests/abinit/conftest.py b/tests/abinit/conftest.py index 6f7353884f..d3dabb1073 100644 --- a/tests/abinit/conftest.py +++ b/tests/abinit/conftest.py @@ -4,7 +4,9 @@ from pathlib import Path from typing import TYPE_CHECKING, Literal +import numpy as np import pytest +from abipy.abio.abivars import is_anaddb_var if TYPE_CHECKING: from collections.abc import Sequence @@ -26,6 +28,20 @@ def abinit_test_dir(test_dir): return test_dir / "abinit" +@pytest.fixture(scope="session", autouse=True) +def load_pseudos(abinit_test_dir): + import abipy.flowtk.psrepos + + abipy.flowtk.psrepos.REPOS_ROOT = str(abinit_test_dir / "pseudos") + + +@pytest.fixture(scope="session", autouse=True) +def load_manager(abinit_test_dir): + import abipy.flowtk.tasks + + abipy.flowtk.tasks.TaskManager.USER_CONFIG_DIR = str(abinit_test_dir / "abipy") + + @pytest.fixture(scope="session") def abinit_integration_tests(pytestconfig): return pytestconfig.getoption("abinit_integration") @@ -134,7 +150,7 @@ def check_run_abi(ref_path: str | Path): ref_str = file.read() ref = AbinitInputFile.from_string(ref_str.decode("utf-8")) # Ignore the pseudos as the directory depends on the pseudo root directory - diffs = user.get_differences(ref, ignore_vars=["pseudos"]) + diffs = user.get_differences(ref, ignore_vars=["pseudos", "pp_dirpath"]) # TODO: should we still add some check on the pseudos here ? assert diffs == [], "'run.abi' is different from reference." @@ -145,7 +161,19 @@ def check_abinit_input_json(ref_path: str | Path): user = loadfn("abinit_input.json") assert isinstance(user, AbinitInput) + user_abivars = user.structure.to_abivars() + ref = loadfn(ref_path / "inputs" / "abinit_input.json.gz") + ref_abivars = ref.structure.to_abivars() + + for k, user_v in user_abivars.items(): + assert k in ref_abivars, f"{k = } is not a key of the reference input." + ref_v = ref_abivars[k] + if isinstance(user_v, str): + assert user_v == ref_v, f"{k = }-->{user_v = } versus {ref_v = }" + else: + assert np.allclose(user_v, ref_v), f"{k = }-->{user_v = } versus {ref_v = }" + assert user.runlevel == ref.runlevel, f"{user.runlevel = } versus {ref.runlevel = }" assert user.structure == ref.structure assert user.runlevel == ref.runlevel @@ -339,11 +367,19 @@ def convert_file_to_dict(file_path): with file_opener(file_path, mode) as file: for line in file: - key, value = line.split() - try: - result_dict[key] = int(value) # Assuming values are integers - except ValueError: - result_dict[key] = str(value) # Fall back to string if not an integer + if "#" in line or len(line) == 1: + continue + sl = line.strip().split(" ", 1) + if is_anaddb_var(sl[0]) and len(sl) > 1: + try: + result_dict[sl[0]] = int(sl[1]) + except ValueError: + result_dict[sl[0]] = sl[1] + elif is_anaddb_var(sl[0]) and len(sl) == 1: + current_key = sl[0] + result_dict[current_key] = [] + else: + result_dict[current_key].append([float(t) for t in line.split()]) return result_dict @@ -359,9 +395,29 @@ def check_anaddb_input_json(ref_path: str | Path): user = loadfn("anaddb_input.json") assert isinstance(user, AnaddbInput) + user_abivars = user.structure.to_abivars() + ref = loadfn(ref_path / "inputs" / "anaddb_input.json.gz") - assert user.structure == ref.structure - assert user == ref + ref_abivars = ref.structure.to_abivars() + # Check structure + for k, user_v in user_abivars.items(): + assert k in ref_abivars, f"{k = } is not a key of the reference input." + ref_v = ref_abivars[k] + if isinstance(user_v, str): + assert user_v == ref_v, f"{k = }-->{user_v = } versus {ref_v = }" + else: + assert np.allclose(user_v, ref_v), f"{k = }-->{user_v = } versus {ref_v = }" + + # Check anaddb input + user_args = dict(user.as_dict()["anaddb_args"]) + ref_args = dict(ref.as_dict()["anaddb_args"]) + for k, user_v in user_args.items(): + assert k in ref_args, f"{k = } is not a key of the reference input." + ref_v = ref_args[k] + if isinstance(user_v, str): + assert user_v == ref_v, f"{k = }-->{user_v = } versus {ref_v = }" + else: + assert np.allclose(user_v, ref_v), f"{k = }-->{user_v = } versus {ref_v = }" def copy_abinit_outputs(ref_path: str | Path): diff --git a/tests/abinit/flows/test_dfpt.py b/tests/abinit/flows/test_dfpt.py index 3c35c3990f..2198ad453f 100644 --- a/tests/abinit/flows/test_dfpt.py +++ b/tests/abinit/flows/test_dfpt.py @@ -1,3 +1,11 @@ +from shutil import which + +import pytest + + +@pytest.mark.skipif( + which("abinit") is None, reason="abinit must be installed to run this test." +) def test_run_silicon_carbide_shg( mock_abinit, mock_mrgddb, mock_anaddb, abinit_test_dir, clean_dir ): diff --git a/tests/abinit/test_dfpt.py b/tests/abinit/test_dfpt.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/test_data/abinit/flows/dfpt/ShgFlowMaker/silicon_carbide_shg/ref_paths.json.gz b/tests/test_data/abinit/flows/dfpt/ShgFlowMaker/silicon_carbide_shg/ref_paths.json.gz index 8aed1a0731b8bfb57ea83238f279f049aa7f1819..173038e6d749ee1e88ec4e438ce3bde486bf8770 100644 GIT binary patch literal 236 zcmVWt zP#}Oh00)|g`zGQ*6LH@}9B3l$nW$Jzlm-IDYNB)-h=G9E2T&k@`hWxoNPIvF1f+i; mtB1CZ7(0b^Kj*@p|8mDpR#h3+`rf`n%kBk+rLw&r1pol5;A!jt literal 234 zcmV-mv|8iw!UvOb`Xmc)Vb8l_{tQA-E$-kTmMLCg~VC6M8NANUqj$R?7)$cGWL9(r_)78KhTHD-8DST*+< z7hQX?zg*nv5l`A^IrLnLvC^vMQ`ce};Z6j{Xgx$I%dT!Z#w4t6J8nJZWFJU?fW!w7 zAb>ak1Dc5WCSpJnG2cWCXd>pBsN76M1A%fg5#0w8ARzGp1PCBLAO!+aA3%Ws`UmoQ kZ0jh?&SBlpIk)G(+^H8zRbj2~?K`x-0Nm<503HPZ07DpJm;e9( From a25dd21b8fb3c8c5c7152de4b79efc062791a827 Mon Sep 17 00:00:00 2001 From: Enrico Marazzi Date: Fri, 4 Apr 2025 11:11:04 +0200 Subject: [PATCH 2/5] Phonon workflow for ABINIT first pr --- src/atomate2/abinit/flows/dfpt.py | 35 +++++++++++------ src/atomate2/abinit/jobs/anaddb.py | 2 +- src/atomate2/abinit/jobs/core.py | 19 ++++++---- src/atomate2/abinit/jobs/response.py | 34 ++++++++++------- src/atomate2/abinit/schemas/anaddb.py | 54 ++++++++++++++++++++++++++- src/atomate2/abinit/sets/anaddb.py | 4 ++ src/atomate2/abinit/sets/response.py | 2 +- 7 files changed, 113 insertions(+), 37 deletions(-) diff --git a/src/atomate2/abinit/flows/dfpt.py b/src/atomate2/abinit/flows/dfpt.py index 82ccaed8c8..58de5cf947 100644 --- a/src/atomate2/abinit/flows/dfpt.py +++ b/src/atomate2/abinit/flows/dfpt.py @@ -58,10 +58,15 @@ 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 DDBs. + mrgdv_maker : .Maker + The maker to merge the POT files. anaddb_maker : .Maker The maker to analyze the DDBs. use_dde_sym : bool @@ -99,7 +104,7 @@ class DfptFlowMaker(Maker): anaddb_maker: Maker | None = field(default_factory=AnaddbMaker) # | use_dde_sym: bool = True dte_skip_permutations: bool | None = False - qpt_list: list | None = None + qpt_list: list[list] | None = None ngqpt: list | None = None qptopt: int = None @@ -144,6 +149,8 @@ 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 ------- @@ -156,6 +163,9 @@ def make( 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 ) @@ -229,11 +239,10 @@ def make( jobs.append(dte_calcs) if self.phonon_maker: - # generate the perturbations for the phonon calculations prev_outputs = [static_job.output.dir_name] - if self.dde_maker: - prev_outputs.append(dde_calcs.output["dirs"]) - + # 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, @@ -241,7 +250,7 @@ def make( 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, @@ -251,7 +260,7 @@ def make( jobs.append(phonon_calcs) if self.mrgddb_maker: - # merge the DDE and DTE DDB. + # merge the DDE, DTE and Phonon DDB. prev_outputs = [] if self.dde_maker: prev_outputs.append(dde_calcs.output["dirs"]) @@ -267,6 +276,7 @@ def make( 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"]) @@ -281,6 +291,7 @@ def make( 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"]} @@ -288,6 +299,7 @@ def make( 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( @@ -298,11 +310,6 @@ def make( 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 @@ -379,6 +386,10 @@ class PhononMaker(DfptFlowMaker): 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 = "Phbands-PhDOS Flow" diff --git a/src/atomate2/abinit/jobs/anaddb.py b/src/atomate2/abinit/jobs/anaddb.py index 0b59421ba0..bd31654766 100644 --- a/src/atomate2/abinit/jobs/anaddb.py +++ b/src/atomate2/abinit/jobs/anaddb.py @@ -30,7 +30,7 @@ logger = logging.getLogger(__name__) -__all__ = ["AnaddbDfptDteMaker", "AnaddbMaker"] +__all__ = ["AnaddbDfptDteMaker", "AnaddbMaker", "AnaddbPhBandsDOSMaker", "anaddb_job"] _DATA_OBJECTS = [ diff --git a/src/atomate2/abinit/jobs/core.py b/src/atomate2/abinit/jobs/core.py index b508a74a26..955957f13c 100644 --- a/src/atomate2/abinit/jobs/core.py +++ b/src/atomate2/abinit/jobs/core.py @@ -67,8 +67,12 @@ class StaticMakerforPhonons(StaticMaker): """ Maker to create ABINIT scf jobs for phonons. - The make method is modified in order to stop a gs calculation - if the q-points are not commensurate with the k-points. + Based on the StaticMaker class, the method validate_grids + checks if the q-point grid is commensurate with the k-point grid. + If they are commensurate, it returns a static job maker. + Otherwise, it raises a ValueError. + This is performed to avoid running gs computations when we know the + phonon calculation will fail. Parameters ---------- @@ -97,12 +101,11 @@ def validate_grids( A pymatgen structure object. ngqpt : list q-point grid for phonon calculation. - shiftq : list - Shifts for the q-point grid. - qptopt : int - Option for the q-point grid. - qpt_list : list - List of q-points for the phonon calculation. + + Returns + ------- + job + A job object with the static job maker. """ with tempfile.TemporaryDirectory(dir=".") as temp_dir: write_abinit_input_set( diff --git a/src/atomate2/abinit/jobs/response.py b/src/atomate2/abinit/jobs/response.py index 42c3161e63..210a90deec 100644 --- a/src/atomate2/abinit/jobs/response.py +++ b/src/atomate2/abinit/jobs/response.py @@ -269,7 +269,7 @@ def generate_phonon_perts( gsinput: AbinitInput, ngqpt: list | tuple | None = None, qptopt: int | None = 1, - qpt_list: list | None = None, + qpt_list: list[list] | None = None, with_wfq: bool = False, ) -> dict[str, list[Any] | tuple[Any, ...] | Any]: """ @@ -279,8 +279,23 @@ def generate_phonon_perts( ---------- gsinput : an |AbinitInput| representing a ground state calculation, likely the SCF performed to get the WFK. - qpt: list or tuple + ngqpt : list or tuple + 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 + Option for the q-point generation. + qpt_list: list q-point for the phonon calculations. + wfq_maker: bool + True if a wfq_maker is provided for k+q computations. + Not yet implemented, so default is False. + + Returns + ------- + dict + A dictionary with the perturbations, the ngqpt and the + output directory name. """ gsinput = gsinput.deepcopy() gsinput.pop_vars(["autoparal"]) @@ -290,7 +305,9 @@ def generate_phonon_perts( ngkpt=ngqpt, shiftk=[0, 0, 0], kptopt=qptopt ).points outputs["ngqpt"] = ngqpt if ngqpt else gsinput["ngkpt"] - + else: + outputs["ngqpt"] = [1, 1, 1] + qpt_list = [qpt_list] if isinstance(qpt_list[0], int | float) else qpt_list perturbations = list() outdirs = list() for q in qpt_list: @@ -306,7 +323,6 @@ def generate_phonon_perts( outputs["dir_name"] = list(np.hstack(outdirs)) if np.any(np.array(gsinput["ngkpt"]) % np.array(outputs["ngqpt"])) and not with_wfq: raise ValueError("q-points are not commensurate with k-points.") - return outputs @@ -315,7 +331,6 @@ def run_rf( perturbations: list[dict], rf_maker: ResponseMaker, prev_outputs: str | list[str] | None = None, - with_dde: bool = False, ) -> Flow: """ Run the RF calculations. @@ -337,16 +352,9 @@ def run_rf( prev_outputs = list(np.hstack(prev_outputs)) for ipert, pert in enumerate(perturbations): - is_gamma = np.allclose(pert["qpt"], [0.0, 0.0, 0.0]) if "qpt" in pert else True - - prev_out = ( - [prev_outputs[0]] - if is_phonon and with_dde and not is_gamma - else prev_outputs - ) rf_job = rf_maker.make( perturbation=pert, - prev_outputs=prev_out, + prev_outputs=prev_outputs, ) if is_phonon: diff --git a/src/atomate2/abinit/schemas/anaddb.py b/src/atomate2/abinit/schemas/anaddb.py index 68bb858d4d..e21e4a3fa0 100644 --- a/src/atomate2/abinit/schemas/anaddb.py +++ b/src/atomate2/abinit/schemas/anaddb.py @@ -366,14 +366,18 @@ def from_abinit_files( abinit_anaddb = AnaddbNcFile.from_file(abinit_anaddb_file) if abinit_phbst_file.exists(): abinit_phbst = PhononBands.from_file(abinit_phbst_file) + else: + abinit_phbst = None if abinit_phdos_file.exists(): abinit_phdos = PhononDos.as_phdos(str(abinit_phdos_file)) + else: + abinit_phdos = None if abinit_anaddb_file.exists(): output_doc = CalculationOutput.from_abinit_anaddb( dir_name=dir_name, output=abinit_anaddb, - output_phbands=abinit_phbst if abinit_phbst else None, - output_phdos=abinit_phdos if abinit_phdos else None, + output_phbands=abinit_phbst, + output_phdos=abinit_phdos, ) completed_at = str( @@ -426,6 +430,52 @@ class OutputDoc(BaseModel): The conventional static SHG tensor in pm/V (Chi^(2)/2) epsinf: list (3x3) The electronic contribution to the dielectric tensor + phonon_bandstructure: PhononBandStructureSymmLine + The phonon band structure object. + phonon_dos: PhononDos + The phonon density of states object. + free_energies: list + The vibrational part of the free energies in J/mol per + formula unit for temperatures in temperature_list + heat_capacities: list + The heat capacities in J/K/mol per + formula unit for temperatures in temperature_list + internal_energies: list + The internal energies in J/mol per + formula unit for temperatures in temperature_list + entropies: list + The entropies in J/(K*mol) per formula unit + for temperatures in temperature_list + temperatures: list + The temperatures at which the vibrational + part of the free energies and other properties have been computed + total_dft_energy: float + The total DFT energy per formula unit in eV + volume_per_formula_unit: float + The volume per formula unit in Angstrom**3 + formula_units: int + The number of formula units per cell + has_imaginary_modes: bool + Whether the structure has imaginary modes + force_constants: ForceConstants + The force constants between every pair of atoms in the structure + born: list + The Born charges as computed from phonopy. Only for symmetrically + different atoms + epsilon_static: Matrix3D + The high-frequency dielectric constant + supercell_matrix: Matrix3D + The matrix describing the supercell + primitive_matrix: Matrix3D + The matrix describing the relationship to the primitive cell + code: str + The code for the computation + phonopy_settings: PhononComputationalSettings + The settings for Phonopy + thermal_displacement_data: ThermalDisplacementData + The data of the computation of the thermal displacements + jobdirs: PhononJobDirs + The job directories """ structure: Union[Structure] = Field( diff --git a/src/atomate2/abinit/sets/anaddb.py b/src/atomate2/abinit/sets/anaddb.py index 4a1b8ce136..ba6b40ba7a 100644 --- a/src/atomate2/abinit/sets/anaddb.py +++ b/src/atomate2/abinit/sets/anaddb.py @@ -363,6 +363,8 @@ def anaddbinp_phbands_dos( ---------- structure A structure. + kwargs + A dictionary with anaddb keywords. Returns ------- @@ -381,6 +383,8 @@ class AnaddbPhbandsDOSInputGenerator(AnaddbInputGenerator): ---------- factory A callable to generate the AnaddbInput for phonon bands and DOS. + factory_kwargs + A dictionary with factory keywords. Returns ------- diff --git a/src/atomate2/abinit/sets/response.py b/src/atomate2/abinit/sets/response.py index 3860886903..e94fb65157 100644 --- a/src/atomate2/abinit/sets/response.py +++ b/src/atomate2/abinit/sets/response.py @@ -78,7 +78,7 @@ class PhononSetGenerator(StaticSetGenerator): factory: Callable = phononpert_from_gsinput pseudos: str | list[str] | PseudoTable | None = None restart_from_deps: tuple = (f"{SCF}:WFK",) - prev_outputs_deps: tuple = (f"{SCF}:WFK", f"{DDE}:1WF", f"{DDE}:1DEN") + prev_outputs_deps: tuple = (f"{SCF}:WFK",) factory_prev_inputs_kwargs: dict | None = field( default_factory=lambda: {"gs_input": (SCF,)} ) From 1e3083738f8defcc39cd3d2449eb75ed7b26137c Mon Sep 17 00:00:00 2001 From: Enrico Marazzi Date: Mon, 7 Apr 2025 12:16:26 +0200 Subject: [PATCH 3/5] Phonon workflow for ABINIT first pr --- src/atomate2/abinit/flows/dfpt.py | 16 +++++++--------- src/atomate2/abinit/jobs/core.py | 2 +- src/atomate2/abinit/jobs/response.py | 6 +++--- src/atomate2/abinit/utils/common.py | 2 +- 4 files changed, 12 insertions(+), 14 deletions(-) diff --git a/src/atomate2/abinit/flows/dfpt.py b/src/atomate2/abinit/flows/dfpt.py index 58de5cf947..bda17b14c2 100644 --- a/src/atomate2/abinit/flows/dfpt.py +++ b/src/atomate2/abinit/flows/dfpt.py @@ -96,11 +96,9 @@ class DfptFlowMaker(Maker): 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 = field( - default_factory=PhononResponseMaker - ) # | + phonon_maker: BaseAbinitMaker | None = None # | mrgddb_maker: Maker | None = field(default_factory=MrgddbMaker) # | - mrgdv_maker: Maker | None = field(default_factory=MrgdvMaker) # | + mrgdv_maker: Maker | None = None # | anaddb_maker: Maker | None = field(default_factory=AnaddbMaker) # | use_dde_sym: bool = True dte_skip_permutations: bool | None = False @@ -330,8 +328,6 @@ class ShgFlowMaker(DfptFlowMaker): name: str = "DFPT Chi2 SHG" anaddb_maker: Maker | None = field(default_factory=AnaddbDfptDteMaker) - phonon_maker: Maker | None = None - mrgdv_maker: Maker | None = None use_dde_sym: bool = False static_maker: BaseAbinitMaker = field( default_factory=lambda: StaticMaker(input_set_generator=ShgStaticSetGenerator()) @@ -392,7 +388,7 @@ class PhononMaker(DfptFlowMaker): True if the merge of POT files should be included, False otherwise. """ - name: str = "Phbands-PhDOS Flow" + name: str = "Phonon Flow" with_dde: bool = True run_anaddb: bool = True run_mrgddb: bool = True @@ -402,6 +398,8 @@ class PhononMaker(DfptFlowMaker): 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 @@ -434,7 +432,7 @@ def make( **anaddb_kwargs, ) -> Flow: """ - Create a phonon band structure and DOS flow. + Create a phonon flow. Parameters ---------- @@ -448,7 +446,7 @@ def make( Returns ------- Flow - A phonon band structure and DOS flow. + A phonon flow. """ return super().make( structure=structure, diff --git a/src/atomate2/abinit/jobs/core.py b/src/atomate2/abinit/jobs/core.py index 955957f13c..4369168b96 100644 --- a/src/atomate2/abinit/jobs/core.py +++ b/src/atomate2/abinit/jobs/core.py @@ -114,7 +114,7 @@ def validate_grids( directory=temp_dir, ) ai = load_abinit_input(temp_dir) - if np.any(np.array(ai["ngkpt"]) % np.array(ngqpt) != np.array([0, 0, 0])): + if any(np.array(ai["ngkpt"]) % np.array(ngqpt)): raise ValueError("q-points are not commensurate with k-points") self.name = "SCF Calculation for Phonons" static_job = self.make( diff --git a/src/atomate2/abinit/jobs/response.py b/src/atomate2/abinit/jobs/response.py index 210a90deec..25d38dc04a 100644 --- a/src/atomate2/abinit/jobs/response.py +++ b/src/atomate2/abinit/jobs/response.py @@ -1,4 +1,4 @@ -"""Response function jobs for running ABINIT calculations.""" +"""Jobs for running ABINIT response to perturbations.""" from __future__ import annotations @@ -273,7 +273,7 @@ def generate_phonon_perts( with_wfq: bool = False, ) -> dict[str, list[Any] | tuple[Any, ...] | Any]: """ - Generate the perturbations for the phonon calculations. + Generate the qpt-list and perturbations for the phonon calculations. Parameters ---------- @@ -287,7 +287,7 @@ def generate_phonon_perts( Option for the q-point generation. qpt_list: list q-point for the phonon calculations. - wfq_maker: bool + with_wfq: bool True if a wfq_maker is provided for k+q computations. Not yet implemented, so default is False. diff --git a/src/atomate2/abinit/utils/common.py b/src/atomate2/abinit/utils/common.py index 7e6cd9b29d..342847850d 100644 --- a/src/atomate2/abinit/utils/common.py +++ b/src/atomate2/abinit/utils/common.py @@ -476,7 +476,7 @@ def get_mrgdv_report( Parameters ---------- logfile : File - Output file to be parsed. Should be either the log file (stdout). + Logfile file to be parsed. Returns ------- From ee9344ece83eacca460aa7fa1e53ad613b39ef34 Mon Sep 17 00:00:00 2001 From: Enrico Marazzi Date: Mon, 7 Apr 2025 14:13:07 +0200 Subject: [PATCH 4/5] Phonon workflow for ABINIT first pr --- src/atomate2/abinit/jobs/response.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/atomate2/abinit/jobs/response.py b/src/atomate2/abinit/jobs/response.py index 25d38dc04a..ef74bf0539 100644 --- a/src/atomate2/abinit/jobs/response.py +++ b/src/atomate2/abinit/jobs/response.py @@ -321,7 +321,7 @@ def generate_phonon_perts( # and the last if removed outputs["perts"] = list(np.hstack(perturbations)) outputs["dir_name"] = list(np.hstack(outdirs)) - if np.any(np.array(gsinput["ngkpt"]) % np.array(outputs["ngqpt"])) and not with_wfq: + if any(np.array(gsinput["ngkpt"]) % np.array(outputs["ngqpt"])) and not with_wfq: raise ValueError("q-points are not commensurate with k-points.") return outputs From 8ce9ed39b20c99cb7e51c9342664b3af1603824c Mon Sep 17 00:00:00 2001 From: Enrico Marazzi Date: Thu, 10 Apr 2025 14:16:10 +0200 Subject: [PATCH 5/5] Fixing phonon flow --- src/atomate2/abinit/flows/dfpt.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/atomate2/abinit/flows/dfpt.py b/src/atomate2/abinit/flows/dfpt.py index bda17b14c2..9a0444d4db 100644 --- a/src/atomate2/abinit/flows/dfpt.py +++ b/src/atomate2/abinit/flows/dfpt.py @@ -253,7 +253,6 @@ def make( perturbations=phonon_perts_qpt_list.output["perts"], rf_maker=self.phonon_maker, prev_outputs=prev_outputs, - with_dde=bool(self.dde_maker), ) jobs.append(phonon_calcs)