added pdos job and flow

docs/user/recipes/recipes_list.md

@@ -199,6 +199,8 @@ The list of available quacc recipes is shown below. The "Req'd Extras" column sp
 | Espresso Non-SCF         | `#!Python @job`  | [quacc.recipes.espresso.core.non_scf_job][]         |              |
 | Espresso DOS             | `#!Python @job`  | [quacc.recipes.espresso.dos.dos_job][]              |              |
 | 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][]         |              |
 
 </center>
 

src/quacc/calculators/espresso/espresso.py

@@ -66,7 +66,7 @@ def __init__(
             "wfcdir": os.environ.get("ESPRESSO_TMPDIR", "."),
         }
 
-        self.outfiles = {"fildos": "pwscf.dos"}
+        self.outfiles = {"fildos": "pwscf.dos", "filpdos": "pwscf.pdos_tot"}
 
         self.test_run = test_run
 
@@ -208,9 +208,17 @@ def read_results(self, directory: Path | str) -> dict[str, Any]:
             fildos = self.outfiles["fildos"]
             with Path(fildos).open("r") as fd:
                 lines = fd.readlines()
-                fermi = float(re.search(r"-?\d+\.?\d*", lines[0])[0])
+                fermi = float(re.search(r"-?\d+\.?\d*", lines[0]).group(0))
                 dos = np.loadtxt(lines[1:])
             results = {fildos.name: {"dos": dos, "fermi": fermi}}
+        elif self.binary == "projwfc":
+            filpdos = self.outfiles["filpdos"]
+            with Path(filpdos).open("r") as fd:
+                lines = np.loadtxt(fd.readlines())
+                energy = lines[1:, 0]
+                dos = lines[1:, 1]
+                pdos = lines[1:, 2]
+            results = {filpdos.name: {"energy": energy, "dos": dos, "pdos": pdos}}
         else:
             results = {}
 

src/quacc/recipes/espresso/dos.py

@@ -30,6 +30,11 @@ class DosSchema(TypedDict):
         non_scf_job: RunSchema
         dos_job: RunSchema
 
+    class ProjwfcSchema(TypedDict):
+        static_job: RunSchema
+        non_scf_job: RunSchema
+        projwfc_job: RunSchema
+
 
 @job
 def dos_job(
@@ -75,6 +80,50 @@ def dos_job(
     )
 
 
+@job
+def projwfc_job(
+    prev_dir: str | Path,
+    parallel_info: dict[str] | None = None,
+    test_run: bool = False,
+    **calc_kwargs,
+) -> RunSchema:
+    """
+    Function to carry out a basic projwfc.x calculation.
+    It is mainly used to extract the charge density and wavefunction from a previous pw.x calculation.
+    It can generate partial dos, local dos, spilling paramenter and more. Fore more details please see
+    https://www.quantum-espresso.org/Doc/INPUT_PROJWFC.html
+
+    Parameters
+    ----------
+    prev_dir
+        Outdir of the previously ran pw.x calculation. This is used to copy
+        the entire tree structure of that directory to the working directory
+        of this calculation.
+    parallel_info
+        Dictionary containing information about the parallelization of the
+        calculation. See the ASE documentation for more information.
+    **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
+        `ase.io.espresso.write_fortran_namelist` 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(
+        template=EspressoTemplate("projwfc", test_run=test_run),
+        calc_defaults={},
+        calc_swaps=calc_kwargs,
+        parallel_info=parallel_info,
+        additional_fields={"name": "projwfc.x Projects-wavefunctions"},
+        copy_files=prev_dir,
+    )
+
+
 @flow
 def dos_flow(
     atoms: Atoms,
@@ -167,3 +216,97 @@ def dos_flow(
         "non_scf_job": non_scf_results,
         "dos_job": dos_results,
     }
+
+
+@flow
+def projwfc_flow(
+    atoms: Atoms,
+    job_decorators: dict[str, Callable | None] | None = None,
+    job_params: dict[str, Any] | None = None,
+) -> ProjwfcSchema:
+    """
+    This function performs a projwfc calculation.
+
+    Consists of following jobs that can be modified:
+
+    1. pw.x static
+        - name: "static_job"
+        - job: [quacc.recipes.espresso.core.static_job][]
+
+    2. pw.x non self-consistent
+        - name: "non_scf_job"
+        - job: [quacc.recipes.espresso.core.non_scf_job][]
+
+    3. projwfc.x job
+        - name: "projwfc_job"
+        - job: [quacc.recipes.espresso.dos.projwfc_job][]
+
+    Parameters
+    ----------
+    atoms
+        Atoms object
+    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
+    -------
+    ProjwfcSchema
+        Dictionary of results from [quacc.schemas.ase.summarize_run][].
+        See the type-hint for the data structure.
+    """
+
+    static_job_defaults = {
+        "kspacing": 0.2,
+        "input_data": {"system": {"occupations": "tetrahedra"}},
+    }
+    non_scf_job_defaults = recursive_dict_merge(
+        job_params.get("static_job", {}),
+        {
+            "kspacing": 0.01,
+            "input_data": {
+                "control": {"calculation": "nscf", "verbosity": "high"},
+                "system": {"occupations": "tetrahedra"},
+            },
+        },
+    )
+    projwfc_job_defaults = {}
+
+    calc_defaults = {
+        "static_job": static_job_defaults,
+        "non_scf_job": non_scf_job_defaults,
+        "projwfc_job": projwfc_job_defaults,
+    }
+    job_params = recursive_dict_merge(calc_defaults, job_params)
+
+    static_job_, non_scf_job_, projwfc_job_ = customize_funcs(
+        ["static_job", "non_scf_job", "projwfc_job"],
+        [static_job, non_scf_job, projwfc_job],
+        parameters=job_params,
+        decorators=job_decorators,
+    )
+
+    static_results = static_job_(atoms)
+    file_to_copy = pw_copy_files(
+        job_params["static_job"].get("input_data"),
+        static_results["dir_name"],
+        include_wfc=False,
+    )
+
+    non_scf_results = non_scf_job_(atoms, prev_dir=file_to_copy)
+    file_to_copy = pw_copy_files(
+        job_params["non_scf_job"].get("input_data"),
+        non_scf_results["dir_name"],
+        include_wfc=True,
+    )
+
+    projwfc_results = projwfc_job_(prev_dir=file_to_copy)
+
+    return {
+        "static_job": static_results,
+        "non_scf_job": non_scf_results,
+        "projwfc_job": projwfc_results,
+    }

tests/core/recipes/espresso_recipes/test_dos.py

@@ -1,12 +1,14 @@
+from __future__ import annotations
+
 from pathlib import Path
 from shutil import which
 
 import pytest
 from ase.build import bulk
 from numpy.testing import assert_allclose
 
-from quacc.recipes.espresso.dos import dos_flow
-from quacc.utils.files import copy_decompress_files
+from quacc.recipes.espresso.dos import dos_flow, projwfc_flow, projwfc_job
+from quacc.utils.files import copy_decompress_files, copy_decompress_tree
 
 pytestmark = pytest.mark.skipif(
     which("pw.x") is None or which("dos.x") is None, reason="QE not installed"
@@ -24,6 +26,16 @@
 #    assert output["results"]["pwscf.dos"]["fermi"] == pytest.approx(7.199)
 
 
+def test_projwfc_job(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    copy_decompress_tree({DATA_DIR / "dos_test/": "pwscf.save/*.gz"}, tmp_path)
+    copy_decompress_files([DATA_DIR / "Si.upf.gz"], tmp_path)
+    output = projwfc_job(tmp_path)
+    print(output)
+    assert output["name"] == "projwfc.x Projects-wavefunctions"
+    assert output["parameters"]["input_data"]["projwfc"] == {}
+
+
 def test_dos_flow(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)
 
@@ -76,3 +88,55 @@ def test_dos_flow(tmp_path, monkeypatch):
     assert output["non_scf_job"]["results"]["nspins"] == 1
 
     assert output["dos_job"]["results"]["pwscf.dos"]["fermi"] == pytest.approx(6.772)
+
+
+def test_projwfc_flow(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+
+    copy_decompress_files([DATA_DIR / "Si.upf.gz"], tmp_path)
+    atoms = bulk("Si")
+    input_data = {
+        "control": {"calculation": "scf", "pseudo_dir": tmp_path},
+        "electrons": {"mixing_mode": "TF", "mixing_beta": 0.7, "conv_thr": 1.0e-6},
+    }
+
+    pseudopotentials = {"Si": "Si.upf"}
+
+    job_params = {
+        "static_job": {"input_data": input_data, "pseudopotentials": pseudopotentials},
+        "non_scf_job": {"kspacing": 0.05},
+    }
+
+    output = projwfc_flow(atoms, job_params=job_params)
+    assert_allclose(
+        output["static_job"]["atoms"].get_positions(),
+        atoms.get_positions(),
+        atol=1.0e-4,
+    )
+    assert (
+        output["static_job"]["parameters"]["input_data"]["control"]["calculation"]
+        == "scf"
+    )
+    assert (
+        output["static_job"]["parameters"]["input_data"]["electrons"]["mixing_mode"]
+        == "TF"
+    )
+
+    assert output["static_job"]["results"]["nbands"] == 8
+    assert output["static_job"]["results"]["nspins"] == 1
+
+    assert_allclose(
+        output["non_scf_job"]["atoms"].get_positions(),
+        atoms.get_positions(),
+        atol=1.0e-4,
+    )
+    assert (
+        output["non_scf_job"]["parameters"]["input_data"]["control"]["calculation"]
+        == "nscf"
+    )
+    assert (
+        output["non_scf_job"]["parameters"]["input_data"]["electrons"]["mixing_mode"]
+        == "TF"
+    )
+    assert output["non_scf_job"]["results"]["nbands"] == 8
+    assert output["non_scf_job"]["results"]["nspins"] == 1