Espresso Bands job and flow

docs/user/recipes/recipes_list.md

@@ -202,6 +202,10 @@ The list of available quacc recipes is shown below. The "Req'd Extras" column sp
 | Espresso DOS Flow        | `#!Python @flow` | [quacc.recipes.espresso.dos.dos_flow][]             |              |
 | Espresso Projwfc         | `#!Python @job`  | [quacc.recipes.espresso.dos.projwfc_job][]          |              |
 | Espresso Projwfc Flow    | `#!Python @flow` | [quacc.recipes.espresso.dos.projwfc_flow][]         |              |
+| Espresso Bands Flow      | `#!Python @flow` | [quacc.recipes.espresso.bands.bands_flow][]         |              |
+| Espresso Bands PW        | `#!Python @job`  | [quacc.recipes.espresso.bands.bands_pw_job][]       |              |
+| Espresso Bands PP        | `#!Python @job`  | [quacc.recipes.espresso.bands.bands_pp_job][]       |              |
+| Espresso Fermi Surface   | `#!Python @job`  | [quacc.recipes.espresso.bands.fermi_surface_job][]  |              |
 
 </center>
 

src/quacc/calculators/espresso/espresso.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 import os
 import re
 from pathlib import Path
@@ -19,6 +20,7 @@
     write_espresso_ph,
     write_fortran_namelist,
 )
+from ase.io.espresso_namelist.keys import ALL_KEYS
 
 from quacc import SETTINGS
 from quacc.calculators.espresso.utils import get_pseudopotential_info, sanity_checks
@@ -28,6 +30,8 @@
 if TYPE_CHECKING:
     from typing import Any
 
+LOGGER = logging.getLogger(__name__)
+
 
 class EspressoTemplate(EspressoTemplate_):
     """This is a wrapper around the ASE Espresso template that allows for the use of
@@ -349,7 +353,19 @@ def __init__(
         )
         self._bin_path = str(full_path)
         self._binary = template.binary
-        self._cleanup_params()
+
+        if self._binary in ALL_KEYS:
+            self._cleanup_params()
+        else:
+            LOGGER.warning(
+                f"the binary you requested, `{self._binary}`, is not supported by ASE. This means that presets and usual checks will not be carried out, your `input_data` must be provided in nested format."
+            )
+
+            template.binary = None
+
+            self.kwargs["input_data"] = Namelist(self.kwargs.get("input_data"))
+            self._user_calc_params = self.kwargs
+
         self._pseudo_path = (
             self._user_calc_params.get("input_data", {})
             .get("control", {})

src/quacc/recipes/espresso/_base.py

@@ -6,6 +6,7 @@
 
 from ase import Atoms
 from ase.io.espresso import Namelist
+from ase.io.espresso_namelist.keys import ALL_KEYS
 
 from quacc.calculators.espresso.espresso import (
     Espresso,
@@ -199,8 +200,9 @@ def _prepare_atoms(
 
     binary = template.binary if template else "pw"
 
-    calc_defaults["input_data"].to_nested(binary=binary, **calc_defaults)
-    calc_swaps["input_data"].to_nested(binary=binary, **calc_swaps)
+    if binary in ALL_KEYS:
+        calc_defaults["input_data"].to_nested(binary=binary, **calc_defaults)
+        calc_swaps["input_data"].to_nested(binary=binary, **calc_swaps)
 
     calc_flags = recursive_dict_merge(calc_defaults, calc_swaps)
 

src/quacc/recipes/espresso/bands.py

@@ -0,0 +1,313 @@
+"""
+This module, 'bands.py', contains recipes for performing bands and fermi surface calculations using the
+bands.x and fs.x binaries from Quantum ESPRESSO via the quacc library.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from ase.dft.kpoints import bandpath
+from pymatgen.io.ase import AseAtomsAdaptor
+from pymatgen.symmetry.analyzer import SpacegroupAnalyzer
+
+from quacc import flow, job
+from quacc.calculators.espresso.espresso import EspressoTemplate
+from quacc.recipes.espresso._base import base_fn
+from quacc.utils.kpts import convert_pmg_kpts
+from quacc.wflow_tools.customizers import customize_funcs
+
+if TYPE_CHECKING:
+    from typing import Any, Callable, TypedDict
+
+    from ase.atoms import Atoms
+
+    from quacc.schemas._aliases.ase import RunSchema
+    from quacc.utils.files import Filenames, SourceDirectory
+
+    class BandsSchema(TypedDict, total=False):
+        bands_pw: RunSchema
+        bands_pp: RunSchema
+        fermi_surface: RunSchema
+
+
+@job
+def bands_pw_job(
+    atoms: Atoms,
+    copy_files: SourceDirectory | dict[SourceDirectory, Filenames],
+    make_bandpath: bool = True,
+    line_density: float = 20,
+    force_gamma: bool = True,
+    parallel_info: dict[str] | None = None,
+    test_run: bool = False,
+    **calc_kwargs,
+) -> RunSchema:
+    """
+    Function to carry out a basic bands calculation with pw.x.
+
+    Parameters
+    ----------
+    atoms
+        The Atoms object.
+    copy_files
+        Files to copy (and decompress) from source to the runtime directory.
+    make_bandpath
+        If True, it returns the primitive cell for your structure and generates
+        the high symmetry k-path using Latmer-Munro approach.
+        For more information look at
+        [pymatgen.symmetry.bandstructure.HighSymmKpath][]
+    line_density
+        Density of kpoints along the band path if make_bandpath is True
+        For more information [quacc.utils.kpts.convert_pmg_kpts][]
+    force_gamma
+        Forces gamma-centered k-points when using make_bandpath
+        For more information [quacc.utils.kpts.convert_pmg_kpts][]
+    parallel_info
+        Dictionary containing information about the parallelization of the
+        calculation. See the ASE documentation for more information.
+    test_run
+        If True, a test run is performed to check that the calculation input_data is correct or
+        to generate some files/info if needed.
+    **calc_kwargs
+        Additional keyword arguments to pass to the Espresso calculator. Set a value to
+        `quacc.Remove` to remove a pre-existing key entirely. See the docstring of
+        [quacc.calculators.espresso.espresso.Espresso][] for more information.
+
+    Returns
+    -------
+    RunSchema
+        Dictionary of results from [quacc.schemas.ase.summarize_run][].
+        See the type-hint for the data structure.
+    """
+
+    calc_defaults = {
+        "input_data": {"control": {"calculation": "bands", "verbosity": "high"}}
+    }
+    if make_bandpath:
+        structure = AseAtomsAdaptor.get_structure(atoms)
+        primitive = SpacegroupAnalyzer(structure).get_primitive_standard_structure()
+        atoms = primitive.to_ase_atoms()
+        calc_defaults["kpts"] = bandpath(
+            convert_pmg_kpts(
+                {"line_density": line_density}, atoms, force_gamma=force_gamma
+            )[0],
+            cell=atoms.get_cell(),
+        )
+
+    return base_fn(
+        atoms,
+        template=EspressoTemplate("pw", test_run=test_run),
+        calc_defaults=calc_defaults,
+        calc_swaps=calc_kwargs,
+        parallel_info=parallel_info,
+        additional_fields={"name": "pw.x bands"},
+        copy_files=copy_files,
+    )
+
+
+@job
+def bands_pp_job(
+    atoms: Atoms,
+    copy_files: SourceDirectory | dict[SourceDirectory, Filenames],
+    parallel_info: dict[str] | None = None,
+    test_run: bool = False,
+    **calc_kwargs,
+) -> RunSchema:
+    """
+    Function to re-order bands and computes bands-related properties with bands.x.
+
+    Parameters
+    ----------
+    atoms
+        The Atoms object.
+    copy_files
+        Files to copy (and decompress) from source to the runtime directory.
+    parallel_info
+        Dictionary containing information about the parallelization of the
+        calculation. See the ASE documentation for more information.
+    test_run
+        If True, a test run is performed to check that the calculation input_data is correct or
+        to generate some files/info if needed.
+    **calc_kwargs
+        Additional keyword arguments to pass to the Espresso calculator. Set a value to
+        `quacc.Remove` to remove a pre-existing key entirely. See the docstring of
+        [quacc.calculators.espresso.espresso.Espresso][] for more information.
+
+    Returns
+    -------
+    RunSchema
+        Dictionary of results from [quacc.schemas.ase.summarize_run][].
+        See the type-hint for the data structure.
+    """
+
+    return base_fn(
+        atoms,
+        template=EspressoTemplate("bands", test_run=test_run),
+        calc_defaults={},
+        calc_swaps=calc_kwargs,
+        parallel_info=parallel_info,
+        additional_fields={"name": "bands.x post-processing"},
+        copy_files=copy_files,
+    )
+
+
+@job
+def fermi_surface_job(
+    atoms: Atoms,
+    copy_files: SourceDirectory | dict[SourceDirectory, Filenames],
+    parallel_info: dict[str] | None = None,
+    test_run: bool = False,
+    **calc_kwargs,
+) -> RunSchema:
+    """
+    Function to retrieve the fermi surface with fs.x
+    It requires a previous uniform unshifted k-point grid bands calculation.
+
+    Parameters
+    ----------
+    atoms
+        The Atoms object.
+    copy_files
+        Files to copy (and decompress) from source to the runtime directory.
+    parallel_info
+        Dictionary containing information about the parallelization of the
+        calculation. See the ASE documentation for more information.
+    test_run
+        If True, a test run is performed to check that the calculation input_data is correct or
+        to generate some files/info if needed.
+    **calc_kwargs
+        Additional keyword arguments to pass to the Espresso calculator. Set a value to
+        `quacc.Remove` to remove a pre-existing key entirely. See the docstring of
+        [quacc.calculators.espresso.espresso.Espresso][] for more information.
+
+    Returns
+    -------
+    RunSchema
+        Dictionary of results from [quacc.schemas.ase.summarize_run][].
+        See the type-hint for the data structure.
+    """
+
+    return base_fn(
+        atoms,
+        template=EspressoTemplate("fs", test_run=test_run),
+        calc_defaults={},
+        calc_swaps=calc_kwargs,
+        parallel_info=parallel_info,
+        additional_fields={"name": "fs.x fermi_surface"},
+        copy_files=copy_files,
+    )
+
+
+@flow
+def bands_flow(
+    atoms: Atoms,
+    copy_files: SourceDirectory | dict[SourceDirectory, Filenames],
+    run_bands_pp: bool = True,
+    run_fermi_surface: bool = False,
+    make_bandpath: bool = True,
+    line_density: float = 20,
+    force_gamma: bool = True,
+    parallel_info: dict[str] | None = None,
+    test_run: bool = False,
+    job_params: dict[str, Any] | None = None,
+    job_decorators: dict[str, Callable | None] | None = None,
+) -> BandsSchema:
+    """
+    Function to compute bands structure and fermi surface using pw.x, bands.x and fs.x.
+
+    Consists of the following steps:
+
+    1. A pw.x non-self consistent calculation
+        - name: "bands_pw_job"
+        - job : [quacc.recipes.espresso.bands.bands_pw_job][]
+
+    2. A bands.x post-processing calculation
+        - name: "bands_pp_job"
+        - job : [quacc.recipes.espresso.bands.bands_pp_job][]
+
+    3. A fs.x calculation to obtain the fermi surface
+        - name: "fermi_surface_job"
+        - job : [quacc.recipes.espresso.bands.fermi_surface_job][]
+
+    Parameters
+    ----------
+    atoms
+        The Atoms object.
+    copy_files
+        Files to copy (and decompress) from source to the runtime directory.
+    run_bands_pp
+        If True, a bands.x post-processing calculation will be carried out.
+        This allows to re-order bands and computes band-related properties.
+    run_fermi_surface
+        If True, a fs.x calculation will be carried out.
+        This allows to generate the fermi surface of your structure.
+        It requires a uniform unshifted k-point grid bands calculation.
+    make_bandpath
+        If True, it returns the primitive cell for your structure and generates
+        the high symmetry k-path using Latmer-Munro approach.
+        For more information look at
+        [pymatgen.symmetry.bandstructure.HighSymmKpath][]
+    line_density
+        Density of kpoints along the band path if make_bandpath is True
+        For more information [quacc.utils.kpts.convert_pmg_kpts][]
+    force_gamma
+        Forces gamma-centered k-points when using make_bandpath
+        For more information [quacc.utils.kpts.convert_pmg_kpts][]
+    parallel_info
+        Dictionary containing information about the parallelization of the
+        calculation. See the ASE documentation for more information.
+    test_run
+        If True, a test run is performed to check that the calculation input_data is correct or
+        to generate some files/info if needed.
+    job_params
+        Custom parameters to pass to each Job in the Flow. This is a dictinoary where
+        the keys are the names of the jobs and the values are dictionaries of parameters.
+    job_decorators
+        Custom decorators to apply to each Job in the Flow. This is a dictionary where
+        the keys are the names of the jobs and the values are decorators.
+
+    Returns
+    -------
+    BandsSchema
+        Dictionary of results from [quacc.schemas.ase.summarize_run][].
+        See the type-hint for the data structure.
+    """
+
+    results = {}
+    (bands_pw_job_, bands_pp_job_, fermi_surface_job_) = customize_funcs(
+        ["bands_pw_job", "bands_pp_job", "fermi_surface_job"],
+        [bands_pw_job, bands_pp_job, fermi_surface_job],
+        parameters=job_params,
+        decorators=job_decorators,
+    )
+
+    bands_result = bands_pw_job_(
+        atoms,
+        copy_files,
+        make_bandpath=make_bandpath,
+        line_density=line_density,
+        force_gamma=force_gamma,
+        parallel_info=parallel_info,
+        test_run=test_run,
+    )
+    results["bands_pw"] = bands_result
+
+    if run_bands_pp:
+        bands_pp_results = bands_pp_job_(
+            atoms,
+            bands_result["dir_name"],
+            parallel_info=parallel_info,
+            test_run=test_run,
+        )
+        results["bands_pp"] = bands_pp_results
+
+    if run_fermi_surface:
+        fermi_results = fermi_surface_job_(
+            atoms,
+            bands_result["dir_name"],
+            parallel_info=parallel_info,
+            test_run=test_run,
+        )
+        results["fermi_surface"] = fermi_results
+
+    return results