wfl's QE calculator with directory handling.

tests/calculators/test_qe.py

@@ -13,9 +13,12 @@
 from ase.build import bulk
 from pytest import approx, fixture, raises, skip
 
-from wfl.calculators.espresso import evaluate_autopara_wrappable, qe_kpoints_and_kwargs
+# from wfl.calculators.espresso import evaluate_autopara_wrappable, qe_kpoints_and_kwargs
+import wfl.calculators.espresso
 from wfl.calculators.dft import evaluate_dft
+from wfl.calculators import generic
 from wfl.configset import ConfigSet, OutputSpec
+from wfl.autoparallelize.autoparainfo import AutoparaInfo
 
 
 @fixture(scope="session")
@@ -57,78 +60,101 @@ def qe_cmd_and_pseudo(tmp_path_factory):
     return cmd, pspot_file
 
 
-def test_qe_kpoints():
+def test_qe_kpoints(tmp_path, qe_cmd_and_pseudo):
+
+    qe_cmd, pspot = qe_cmd_and_pseudo
+
+
+    kw = dict(
+        pseudopotentials=dict(Si=os.path.basename(pspot)),
+        input_data={"SYSTEM": {"ecutwfc": 40, "input_dft": "LDA",}},
+        pseudo_dir=os.path.dirname(pspot),
+        kpts=(2, 3, 4),
+        conv_thr=0.0001,
+        directory=tmp_path
+    ) 
+
     # PBC = TTT
-    kw, prop = qe_kpoints_and_kwargs(
-        Atoms("H", cell=[1, 1, 1], pbc=True), dict(kpts=(2, 3, 4)), ["energy", "stress"]
-    )
-    assert "tstress" in kw
-    assert kw["tstress"]
+    atoms = Atoms("H", cell=[1, 1, 1], pbc=True)
+    properties = ["energy", "stress"] 
+    calc = wfl.calculators.espresso.Espresso(**kw)
+    calc.atoms = atoms.copy()
+    calc.setup_params_for_this_calc(properties)
 
-    assert kw["kpts"] == (2, 3, 4)
+    assert "tstress" in calc.parameters
+    assert calc.parameters['kpts'] == (2, 3, 4)
 
     # PBC = FFF
-    kw, prop = qe_kpoints_and_kwargs(
-        Atoms("H", cell=[1, 1, 1], pbc=False),
-        dict(kpts=(2, 3, 4)),
-        ["energy", "stress", "forces"],
-    )
-    assert "tstress" in kw
-    assert not kw["tstress"]
+    atoms = Atoms("H", cell=[1, 1, 1], pbc=False)
+    properties = ["energy", "stress", "forces"] 
+    calc = wfl.calculators.espresso.Espresso(**kw)
+    calc.atoms = atoms.copy()
+    properties = calc.setup_params_for_this_calc(properties)
 
-    assert "tprnfor" in kw
-    assert kw["tprnfor"]
+    assert "tstress" in calc.parameters
+    assert not calc.parameters["tstress"]
 
-    assert "stress" not in prop
-    assert kw["kpts"] is None
-    assert kw["kspacing"] is None
-    assert kw["koffset"] is False
+    assert "tprnfor" in calc.parameters
+    assert calc.parameters["tprnfor"]
+
+    assert "stress" not in properties
+    assert calc.parameters["kpts"] is None
+    assert calc.parameters["kspacing"] is None
+    assert calc.parameters["koffset"] is False
 
     # PBC mixed -- kpts
-    kw, prop = qe_kpoints_and_kwargs(
-        Atoms("H", cell=[1, 1, 1], pbc=[True, False, False]),
-        dict(kpts=(2, 3, 4), koffset=True),
-        ["energy", "stress", "forces"],
-    )
+    atoms = Atoms("H", cell=[1, 1, 1], pbc=[True, False, False])
+    properties = ["energy", "stress", "forces"] 
+    kw["koffset"] = True
+    calc = wfl.calculators.espresso.Espresso(**kw)
+    calc.atoms = atoms.copy()
+    properties = calc.setup_params_for_this_calc(properties)
+
+    assert "tstress" in calc.parameters
+    assert not calc.parameters["tstress"]
 
-    assert "tstress" in kw
-    assert not kw["tstress"]
+    assert "tprnfor" in calc.parameters
+    assert calc.parameters["tprnfor"]
 
-    assert "tprnfor" in kw
-    assert kw["tprnfor"]
+    assert "stress" not in properties
+
+    assert calc.parameters["kpts"] == (2, 1, 1)
+    assert calc.parameters["kspacing"] is None
+    assert calc.parameters["koffset"] == (1, 0, 0)
 
-    assert "stress" not in prop
-    assert kw["kpts"] == (2, 1, 1)
-    assert kw["kspacing"] is None
-    assert kw["koffset"] == (1, 0, 0)
 
     # koffset in mixed PBC
-    kw, prop = qe_kpoints_and_kwargs(
-        Atoms("H", cell=[1, 1, 1], pbc=[True, False, False]),
-        dict(kpts=(2, 3, 4), koffset=False),
-        ["energy", "forces"],
-    )
-    assert kw["koffset"] is False
+    atoms = Atoms("H", cell=[1, 1, 1], pbc=[True, False, False])
+    properties = ["energy", "forces"] 
+    kw["koffset"] = False 
+    calc = wfl.calculators.espresso.Espresso(**kw)
+    calc.atoms = atoms.copy()
+    properties = calc.setup_params_for_this_calc(properties)
+
+    assert calc.parameters["koffset"] is False
+
 
     # PBC mixed -- kspacing
-    kw, prop = qe_kpoints_and_kwargs(
-        Atoms("H", cell=[1, 1, 1], pbc=[True, False, False]),
-        dict(kspacing=0.1, koffset=(0, 1, 0)),
-        ["energy", "stress", "forces"],
-    )
-    assert "tstress" in kw
-    assert not kw["tstress"]
+    atoms = Atoms("H", cell=[1, 1, 1], pbc=[True, False, False])
+    properties = ["energy", "stress", "forces"] 
+    kw["kspacing"] = 0.1
+    kw["koffset"] = (0, 1, 0)
+    calc = wfl.calculators.espresso.Espresso(**kw)
+    calc.atoms = atoms.copy()
+    properties = calc.setup_params_for_this_calc(properties)
+
+    assert "tstress" in calc.parameters
+    assert not calc.parameters["tstress"]
 
-    assert "tprnfor" in kw
-    assert kw["tprnfor"]
+    assert "tprnfor" in calc.parameters
+    assert calc.parameters["tprnfor"]
 
-    assert "stress" not in prop
-    assert kw["kpts"] == (63, 1, 1)
-    assert kw["kspacing"] is None
-    assert kw["koffset"] == (0, 0, 0)
+    assert "stress" not in properties
+    assert calc.parameters["kpts"] == (63, 1, 1)
+    assert calc.parameters["kspacing"] is None
+    assert calc.parameters["koffset"] == (0, 0, 0)
 
 
-@pytest.mark.xfail(reason="PP file changes. Even before that hard-wired values are wrong, also calculation does not converge with default conv_thr")
 def test_qe_calculation(tmp_path, qe_cmd_and_pseudo):
     # command and pspot
     qe_cmd, pspot = qe_cmd_and_pseudo
@@ -143,8 +169,12 @@ def test_qe_calculation(tmp_path, qe_cmd_and_pseudo):
         input_data={"SYSTEM": {"ecutwfc": 40, "input_dft": "LDA",}},
         pseudo_dir=os.path.dirname(pspot),
         kpts=(2, 2, 2),
-        conv_thr=0.0001
-    )
+        conv_thr=0.0001,
+        calculator_command=qe_cmd,
+        directory=tmp_path
+    ) 
+
+    calc = (wfl.calculators.espresso.Espresso, [], kw)
 
     # output container
     c_out = OutputSpec(
@@ -154,16 +184,14 @@ def test_qe_calculation(tmp_path, qe_cmd_and_pseudo):
         all_or_none=True,
     )
 
-    results = evaluate_dft(
-        calculator_name="QE",
+    results = generic.run(
         inputs=[at0, at],
-        outputs=c_out,
-        workdir_root=tmp_path,
-        calculator_command=qe_cmd,
-        calculator_kwargs=kw,
-        output_prefix="QE_",
+        outputs=c_out, 
+        calculator=calc, 
+        output_prefix='QE_',
     )
 
+
     # unpack the configset
     si_single, si2 = list(results)
 
@@ -174,72 +202,78 @@ def test_qe_calculation(tmp_path, qe_cmd_and_pseudo):
     # single atoms tests
     assert "QE_stress" not in si_single.info
     assert "QE_energy" in si_single.info
-    assert si_single.info["QE_energy"] == approx(-601.8757092817176)
+    assert si_single.info["QE_energy"] == approx(expected=-101.20487969465684, abs=1e-3)
     assert si_single.get_volume() == approx(6.0 ** 3)
 
     # bulk Si tests
     assert "QE_energy" in si2.info
-    assert si2.info["QE_energy"] == approx(-1214.4189734323988)
+    assert si2.info["QE_energy"] == approx(expected=-213.10730256386654, abs=1e-3)
     assert "QE_stress" in si2.info
+    print(si2.info["QE_stress"])
     assert si2.info["QE_stress"] == approx(
-        abs=1e-5,
-        expected=np.array([-0.040234, -0.040265, -0.040265, -0.002620, 0.0, 0.0]),
+        abs=1e-3,
+        expected = np.array([-0.03510667, -0.03507546, -0.03507546, -0.00256625, -0., -0.,]),
     )
     assert "QE_forces" in si2.arrays
-    assert si2.arrays["QE_forces"][0, 0] == approx(-0.17277428)
+    assert si2.arrays["QE_forces"][0, 0] == approx(-0.17099353)
     assert si2.arrays["QE_forces"][:, 1:] == approx(0.0)
     assert si2.arrays["QE_forces"][0] == approx(-1 * si2.arrays["QE_forces"][1])
 
 
-def test_qe_errors():
-    with raises(
-        ValueError, match="QE will not perform a calculation without settings given!"
-    ):
-        evaluate_autopara_wrappable(Atoms(), calculator_kwargs=None)
-
+def test_wfl_Espresso_calc(tmp_path, qe_cmd_and_pseudo):
+    # command and pspot
+    qe_cmd, pspot = qe_cmd_and_pseudo
 
-def test_qe_no_calculation(tmp_path, qe_cmd_and_pseudo):
-    # call just to skip if pw.x is missing
-    _, _ = qe_cmd_and_pseudo
+    atoms = Atoms("Si", cell=(2, 2, 2), pbc=[True] * 3)
+    kw = dict(
+        pseudopotentials=dict(Si=os.path.basename(pspot)),
+        input_data={"SYSTEM": {"ecutwfc": 40, "input_dft": "LDA",}},
+        pseudo_dir=os.path.dirname(pspot),
+        kpts=(2, 2, 2),
+        conv_thr=0.0001
+    ) 
 
-    results = evaluate_autopara_wrappable(bulk("Si"), calculator_kwargs=dict(), output_prefix="dummy_", workdir_root=tmp_path)
+    calc = wfl.calculators.espresso.Espresso(
+        directory=tmp_path,
+        **kw)
+    atoms.calc = calc
 
-    assert isinstance(results, Atoms)
-    assert "dummy_energy" not in results.info
-    assert "dummy_stress" not in results.info
-    assert "dummy_forces" not in results.arrays
+    atoms.get_potential_energy()
+    atoms.get_forces()
+    atoms.get_stress()
 
 
-def test_qe_to_spc(tmp_path, qe_cmd_and_pseudo):
-    # command and pspot
-    _, pspot = qe_cmd_and_pseudo
+def test_wfl_Espresso_calc_via_generic(tmp_path, qe_cmd_and_pseudo):
 
-    # mainly a copy of VASP test
-    ase.io.write(
-        tmp_path / "qe_in.xyz",
-        Atoms("Si", cell=(2, 2, 2), pbc=[True] * 3),
-        format="extxyz",
-    )
+    qe_cmd, pspot = qe_cmd_and_pseudo
 
+    atoms = Atoms("Si", cell=(2, 2, 2), pbc=[True] * 3)
     kw = dict(
         pseudopotentials=dict(Si=os.path.basename(pspot)),
         input_data={"SYSTEM": {"ecutwfc": 40, "input_dft": "LDA",}},
         pseudo_dir=os.path.dirname(pspot),
         kpts=(2, 2, 2),
-        conv_thr=0.0001
+        conv_thr=0.0001,
+        directory=tmp_path
+    ) 
+
+    calc = (wfl.calculators.espresso.Espresso, [], kw)
+
+    cfgs = [atoms]*3 + [Atoms("Cu", cell=(2, 2, 2), pbc=[True]*3)]
+    ci = ConfigSet(input_configs=cfgs)
+    co = OutputSpec()
+    autoparainfo = AutoparaInfo(
+        num_python_subprocesses=0
     )
 
-    configs_eval = evaluate_dft(
-        inputs=ConfigSet(input_files=tmp_path /  "qe_in.xyz"),
-        outputs=OutputSpec(file_root=tmp_path, output_files="qe_out.to_SPC.xyz"),
-        calculator_name="QE",
-        workdir_root=tmp_path,
-        calculator_kwargs=kw,
-        output_prefix=None,
+    ci = generic.run(
+        inputs=ci,
+        outputs=co, 
+        calculator=calc, 
+        output_prefix='qe_',
+        autopara_info=autoparainfo
     )
 
-    ats = list(configs_eval)
-    assert "energy" in ats[0].calc.results
-    assert "stress" in ats[0].calc.results
-    assert "forces" in ats[0].calc.results
-    # ase.io.write(sys.stdout, list(configs_eval), format='extxyz')
+    assert "qe_calculation_failed" in list(ci)[-1].info
+
+

wfl/calculators/dft.py

@@ -7,7 +7,7 @@
 - Quantum Espresso
 """
 from wfl.autoparallelize import _autoparallelize_ll
-from wfl.calculators import castep, vasp, espresso
+from wfl.calculators import castep, vasp 
 
 
 def evaluate_dft(
@@ -31,8 +31,8 @@ def evaluate_dft(
         input atomic configs, needs to be iterable
     outputs: list(Atoms) / Configset_out
         output atomic configs
-    calculator_name: str {"CASTEP", "VASP", "QE"}
-        name of Plane Wave DFT calculator, options are: "CASTEP", "VASP", "QE"
+    calculator_name: str {"CASTEP", "VASP"}
+        name of Plane Wave DFT calculator, options are: "CASTEP", "VASP"
     workdir_root: path-like, default os.getcwd()
         directory to put calculation directories into
     dir_prefix: str, default 'DFT\_'
@@ -73,7 +73,7 @@ def evaluate_dft(
     elif calculator_name == "VASP":
         op = vasp.evaluate_autopara_wrappable
     elif calculator_name == "QE":
-        op = espresso.evaluate_autopara_wrappable
+        raise ValueError(f"wfl.calculators.espresso.Espresso is compatible with wfl.calculators.generic.run.")
     else:
         raise ValueError(f"Calculator name `{calculator_name}` not understood")