Print invalid kpath_scheme, use_symmetrized_structure and list valid ones in PhononMaker error messages

.pre-commit-config.yaml

@@ -2,7 +2,7 @@ default_language_version:
   python: python3
 repos:
 - repo: https://github.com/charliermarsh/ruff-pre-commit
-  rev: v0.2.1
+  rev: v0.2.2
   hooks:
   - id: ruff
     args: [--fix]

src/atomate2/common/files.py

@@ -347,7 +347,7 @@ def get_zfile(
         found, then ``None`` will be returned.
     """
     for file in directory_listing:
-        if file.name in [base_name, f"{base_name}.gz", f"{base_name}.GZ"]:
+        if file.name in (base_name, f"{base_name}.gz", f"{base_name}.GZ"):
             return file
 
     if allow_missing:

src/atomate2/common/schemas/cclib.py

@@ -291,20 +291,16 @@ def cclib_calculate(
     )
 
     method = method.lower()
-    cube_methods = ["bader", "ddec6", "hirshfeld"]
+    cube_methods = ("bader", "ddec6", "hirshfeld")
 
     if method in cube_methods and not cube_file:
-        raise FileNotFoundError(
-            f"A cube file must be provided for {method}. Returning None."
-        )
-    if method in ["ddec6", "hirshfeld"] and not proatom_dir:
+        raise FileNotFoundError(f"A cube file must be provided for {method}.")
+    if method in ("ddec6", "hirshfeld") and not proatom_dir:
         if os.getenv("PROATOM_DIR") is None:
-            raise OSError("PROATOM_DIR environment variable not set. Returning None.")
+            raise OSError("PROATOM_DIR environment variable not set.")
         proatom_dir = os.path.expandvars(os.environ["PROATOM_DIR"])
-    if proatom_dir and not os.path.exists(proatom_dir):
-        raise FileNotFoundError(
-            f"Protatom directory {proatom_dir} does not exist. Returning None."
-        )
+    if proatom_dir and not os.path.isdir(proatom_dir):
+        raise FileNotFoundError(f"{proatom_dir=} does not exist.")
 
     if cube_file and method in cube_methods:
         vol = volume.read_from_cube(str(cube_file))

src/atomate2/cp2k/schemas/calc_types/utils.py

@@ -73,12 +73,12 @@ def task_type(inputs: dict) -> TaskType:
     cp2k_run_type = inputs.get("cp2k_global", {}).get("Run_type", "")
     ci = Cp2kInput.from_dict(inputs["cp2k_input"])
 
-    if cp2k_run_type.upper() in [
+    if cp2k_run_type.upper() in (
         "ENERGY",
         "ENERGY_FORCE",
         "WAVEFUNCTION_OPTIMIZATION",
         "WFN_OPT",
-    ]:
+    ):
         if ci.check("FORCE_EVAL/DFT/SCF"):
             tmp = ci["force_eval"]["dft"]["scf"].get("MAX_SCF", Keyword("", 50))
             if tmp.values[0] == 1:
@@ -101,34 +101,34 @@ def task_type(inputs: dict) -> TaskType:
         else:
             calc_type.append("Static")
 
-    elif cp2k_run_type.upper() in ["GEO_OPT", "GEOMETRY_OPTIMIZATION", "CELL_OPT"]:
+    elif cp2k_run_type.upper() in ("GEO_OPT", "GEOMETRY_OPTIMIZATION", "CELL_OPT"):
         calc_type.append("Structure Optimization")
 
     elif cp2k_run_type.upper() == "BAND":
         calc_type.append("Band")
 
-    elif cp2k_run_type.upper() in ["MOLECULAR_DYNAMICS", "MD"]:
+    elif cp2k_run_type.upper() in ("MOLECULAR_DYNAMICS", "MD"):
         calc_type.append("Molecular Dynamics")
 
-    elif cp2k_run_type.upper() in ["MONTE_CARLO", "MC", "TMC", "TAMC"]:
+    elif cp2k_run_type.upper() in ("MONTE_CARLO", "MC", "TMC", "TAMC"):
         calc_type.append("Monte Carlo")
 
-    elif cp2k_run_type.upper() in ["LINEAR_RESPONSE", "LR"]:
+    elif cp2k_run_type.upper() in ("LINEAR_RESPONSE", "LR"):
         calc_type.append("Linear Response")
 
-    elif cp2k_run_type.upper() in ["VIBRATIONAL_ANALYSIS", "NORMAL_MODES"]:
+    elif cp2k_run_type.upper() in ("VIBRATIONAL_ANALYSIS", "NORMAL_MODES"):
         calc_type.append("Vibrational Analysis")
 
-    elif cp2k_run_type.upper() in ["ELECTRONIC_SPECTRA", "SPECTRA"]:
+    elif cp2k_run_type.upper() in ("ELECTRONIC_SPECTRA", "SPECTRA"):
         calc_type.append("Electronic Spectra")
 
     elif cp2k_run_type.upper() == "NEGF":
         calc_type.append("Non-equilibrium Green's Function")
 
-    elif cp2k_run_type.upper() in ["PINT", "DRIVER"]:
+    elif cp2k_run_type.upper() in ("PINT", "DRIVER"):
         calc_type.append("Path Integral")
 
-    elif cp2k_run_type.upper() in ["RT_PROPAGATION", "EHRENFEST_DYN"]:
+    elif cp2k_run_type.upper() in ("RT_PROPAGATION", "EHRENFEST_DYN"):
         calc_type.append("Real-time propagation")
 
     elif cp2k_run_type.upper() == "BSSE":

src/atomate2/cp2k/schemas/calculation.py

@@ -394,9 +394,9 @@ def from_cp2k_files(
             # TODO vasp version calls bader_analysis_from_path but cp2k
             # cube files don't support that yet, do it manually
             bader = {
-                "min_dist": [d["min_dist"] for d in ba.data],
-                "charge": [d["charge"] for d in ba.data],
-                "atomic_volume": [d["atomic_vol"] for d in ba.data],
+                "min_dist": [dct["min_dist"] for dct in ba.data],
+                "charge": [dct["charge"] for dct in ba.data],
+                "atomic_volume": [dct["atomic_vol"] for dct in ba.data],
                 "vacuum_charge": ba.vacuum_charge,
                 "vacuum_volume": ba.vacuum_volume,
                 "reference_used": bool(ba.chgref_filename),
@@ -479,9 +479,7 @@ def _get_basis_and_potential_files(dir_name: Path) -> dict[Cp2kObject, DataFile]
     """
     data: dict[Cp2kObject, DataFile] = {}
     if Path.exists(dir_name / "BASIS"):
-        data[Cp2kObject.BASIS] = BasisFile.from_file(  # type: ignore[index]
-            str(dir_name / "BASIS")
-        )
+        data[Cp2kObject.BASIS] = BasisFile.from_file(str(dir_name / "BASIS"))  # type: ignore[index]
     if Path.exists(dir_name / "POTENTIAL"):
         data[Cp2kObject.POTENTIAL] = PotentialFile.from_file(  # type: ignore[index]
             str(dir_name / "POTENTIAL")

src/atomate2/cp2k/sets/base.py

@@ -216,7 +216,7 @@ def get_input_set(
         Cp2kInput
             A Cp2k input set.
         """
-        structure, prev_input, cp2k_output = self._get_previous(structure, prev_dir)
+        structure, prev_input, _cp2k_output = self._get_previous(structure, prev_dir)
 
         input_updates = self.get_input_updates(
             structure,

src/atomate2/forcefields/flows/phonons.py

@@ -180,27 +180,25 @@ def make(
             instead of min_length, also a supercell_matrix can
             be given, e.g. [[1.0,0.0,0.0],[0.0,1.0,0.0],[0.0,0.0,1.0]
         """
-        if self.use_symmetrized_structure not in [None, "primitive", "conventional"]:
+        use_symmetrized_structure = self.use_symmetrized_structure
+        kpath_scheme = self.kpath_scheme
+        valid_structs = (None, "primitive", "conventional")
+        if use_symmetrized_structure not in valid_structs:
             raise ValueError(
-                "use_symmetrized_structure can only be primitive, conventional, None"
+                f"Invalid {use_symmetrized_structure=}, use one of {valid_structs}"
             )
 
-        if (
-            self.use_symmetrized_structure != "primitive"
-            and self.kpath_scheme != "seekpath"
-        ):
+        if use_symmetrized_structure != "primitive" and kpath_scheme != "seekpath":
             raise ValueError(
-                "You can only use other kpath schemes with the primitive standard "
-                "structure"
+                f"You can't use {kpath_scheme=} with the primitive standard "
+                "structure, please use seekpath"
             )
 
-        if self.kpath_scheme not in [
-            "seekpath",
-            "hinuma",
-            "setyawan_curtarolo",
-            "latimer_munro",
-        ]:
-            raise ValueError("kpath scheme is not implemented")
+        valid_schemes = ("seekpath", "hinuma", "setyawan_curtarolo", "latimer_munro")
+        if kpath_scheme not in valid_schemes:
+            raise ValueError(
+                f"{kpath_scheme=} is not implemented, use one of {valid_schemes}"
+            )
 
         jobs = []
 

src/atomate2/lobster/schemas.py

@@ -806,10 +806,8 @@ def from_directory(
             "dos_comparison": True,
             "n_bins": 256,
             "bva_comp": True,
+            **(calc_quality_kwargs or {}),
         }
-        if calc_quality_kwargs:
-            for args, values in calc_quality_kwargs.items():
-                calc_quality_kwargs_default[args] = values
 
         calc_quality_summary = CalcQualitySummary.from_directory(
             dir_name,
@@ -1319,16 +1317,12 @@ def read_saved_json(
         Returns a dictionary with lobster task json data corresponding to query.
     """
     with gzip.open(filename, "rb") as file:
-        lobster_data = {}
-        objects = ijson.items(file, "item", use_float=True)
-        for obj in objects:
-            if query is None:
-                for field, data in obj.items():
-                    lobster_data[field] = data
-            elif query in obj:
-                for field, data in obj.items():
-                    lobster_data[field] = data
-                break
+        lobster_data = {
+            field: data
+            for obj in ijson.items(file, "item", use_float=True)
+            for field, data in obj.items()
+            if query is None or query in obj
+        }
         if not lobster_data:
             raise ValueError(
                 "Please recheck the query argument. "

src/atomate2/utils/file_client.py

@@ -465,7 +465,7 @@ def gunzip(
             path.unlink()
         else:
             ssh = self.get_ssh(host)
-            _, stdout, _ = ssh.exec_command(f"gunzip -f {path!s}")
+            _stdin, _stdout, _stderr = ssh.exec_command(f"gunzip -f {path!s}")
 
     def close(self) -> None:
         """Close all connections."""

src/atomate2/vasp/files.py

@@ -77,7 +77,7 @@ def copy_vasp_outputs(
     # find optional files; do not fail if KPOINTS is missing, this might be KSPACING
     # note: POTCAR files never have the relax extension, whereas KPOINTS files should
     optional_files = []
-    for file in ["POTCAR", "POTCAR.spec", "KPOINTS" + relax_ext]:
+    for file in ("POTCAR", "POTCAR.spec", "KPOINTS" + relax_ext):
         found_file = get_zfile(directory_listing, file, allow_missing=True)
         if found_file is not None:
             optional_files.append(found_file)

src/atomate2/vasp/flows/phonons.py

@@ -189,25 +189,25 @@ def make(
             Instead of min_length, also a supercell_matrix can be given, e.g.
             [[1.0,0.0,0.0],[0.0,1.0,0.0],[0.0,0.0,1.0]
         """
-        if self.use_symmetrized_structure not in [None, "primitive", "conventional"]:
+        use_symmetrized_structure = self.use_symmetrized_structure
+        kpath_scheme = self.kpath_scheme
+        valid_structs = (None, "primitive", "conventional")
+        if use_symmetrized_structure not in valid_structs:
             raise ValueError(
-                "use_symmetrized_structure can only be primitive, conventional, None"
+                f"Invalid {use_symmetrized_structure=}, use one of {valid_structs}"
             )
 
-        if (
-            self.use_symmetrized_structure != "primitive"
-            and self.kpath_scheme != "seekpath"
-        ):
+        if use_symmetrized_structure != "primitive" and kpath_scheme != "seekpath":
             raise ValueError(
-                "You can only use other kpath schemes with the primitive standard "
-                "structure"
+                f"You can't use {kpath_scheme=} with the primitive standard "
+                "structure, please use seekpath"
             )
 
-        if (
-            self.kpath_scheme
-            not in "seekpath hinuma setyawan_curtarolo latimer_munro".split()
-        ):
-            raise ValueError("kpath scheme is not implemented")
+        valid_schemes = ("seekpath", "hinuma", "setyawan_curtarolo", "latimer_munro")
+        if kpath_scheme not in valid_schemes:
+            raise ValueError(
+                f"{kpath_scheme=} is not implemented, use one of {valid_schemes}"
+            )
 
         jobs = []
 

src/atomate2/vasp/sets/base.py

@@ -168,7 +168,7 @@ def is_valid(self) -> bool:
             )
 
         if (
-            all(k.is_metal for k in self.poscar.structure.composition)
+            all(elem.is_metal for elem in self.poscar.structure.composition)
             and self.incar.get("NSW", 0) > 0
             and self.incar.get("ISMEAR", 1) < 1
         ):
@@ -179,11 +179,11 @@ def is_valid(self) -> bool:
                 stacklevel=1,
             )
 
-        if self.incar.get("LHFCALC") and self.incar.get("ALGO", "Normal") not in [
+        if self.incar.get("LHFCALC") and self.incar.get("ALGO", "Normal") not in (
             "Normal",
             "All",
             "Damped",
-        ]:
+        ):
             warnings.warn(
                 "Hybrid functionals only support Algo = All, Damped, or Normal.",
                 BadInputSetWarning,
@@ -266,7 +266,7 @@ class VaspInputGenerator(InputGenerator):
     auto_kspacing
         If true, automatically use the VASP recommended KSPACING based on bandgap,
         i.e. higher kpoint spacing for insulators than metals. Can be boolean or float.
-        If float, then the value will interpreted as the bandgap in eV to use for the
+        If a float, the value will be interpreted as the bandgap in eV to use for the
         KSPACING calculation.
     constrain_total_magmom
         Whether to constrain the total magmom (NUPDOWN in INCAR) to be the sum of the

tests/common/schemas/test_defect.py

@@ -22,20 +22,23 @@ def abs_d(s1, s2):
         return np.linalg.norm(np.array(s1) - np.array(s2))
 
     points_on_line_2d = [(1, 1), (-2, -2), (0, 0), (2, 2), (-1, -1)]
-    r, d = sort_pos_dist(points_on_line_2d, s1=(0, 0), s2=(1.5, 1.5), dist=abs_d)
-    assert r == [(-2, -2), (-1, -1), (0, 0), (1, 1), (2, 2)]
+    sorted_pts, dists = sort_pos_dist(
+        points_on_line_2d, s1=(0, 0), s2=(1.5, 1.5), dist=abs_d
+    )
+    assert sorted_pts == [(-2, -2), (-1, -1), (0, 0), (1, 1), (2, 2)]
+    assert dists == pytest.approx([-2.8284271, -1.4142135, 0, 1.41421356, 2.82842712])
 
-    r, d = sort_pos_dist(points_on_line_2d, s1=(0, 0), s2=(-2.5, -2.5), dist=abs_d)
-    assert r == [(2, 2), (1, 1), (0, 0), (-1, -1), (-2, -2)]
+    sorted_pts, dists = sort_pos_dist(
+        points_on_line_2d, s1=(0, 0), s2=(-2.5, -2.5), dist=abs_d
+    )
+    assert sorted_pts == [(2, 2), (1, 1), (0, 0), (-1, -1), (-2, -2)]
+    assert dists == pytest.approx([-2.8284271, -1.4142135, 0, 1.41421356, 2.82842712])
 
 
 # schemas where all fields have default values
 @pytest.mark.parametrize(
     "model_cls",
-    [
-        FormationEnergyDiagramDocument,
-        CCDDocument,
-    ],
+    [FormationEnergyDiagramDocument, CCDDocument],
 )
 def test_model_validate(model_cls):
     model_cls.model_validate_json(json.dumps(model_cls(), cls=MontyEncoder))

tests/common/test_files.py

@@ -34,7 +34,7 @@ def test_gunzip_force_overwrites(tmp_path):
 
 
 def test_zip_outputs(tmp_dir):
-    for file_name in ["a", "b"]:
+    for file_name in ("a", "b"):
         (Path.cwd() / file_name).touch()
 
     gzip_output_folder(directory=Path.cwd(), setting=False, files_list=["a"])

tests/cp2k/sets/test_core.py

@@ -13,29 +13,29 @@ def test_input_generators(si_structure, basis_and_potential):
         StaticSetGenerator,
     )
 
-    for gen in [
+    for gen in (
         StaticSetGenerator(user_input_settings=basis_and_potential),
         HybridStaticSetGenerator(user_input_settings=basis_and_potential),
-    ]:
+    ):
         input_set = gen.get_input_set(si_structure)
         assert input_set.cp2k_input["GLOBAL"]["RUN_TYPE"].values[0] == "ENERGY_FORCE"
         assert input_set.cp2k_input.check("FORCE_EVAL/DFT/KPOINTS")
 
-    for gen in [
+    for gen in (
         RelaxSetGenerator(user_input_settings=basis_and_potential),
         HybridRelaxSetGenerator(user_input_settings=basis_and_potential),
-    ]:
+    ):
         input_set = gen.get_input_set(si_structure)
         assert input_set.cp2k_input["GLOBAL"]["RUN_TYPE"].values[0] == "GEO_OPT"
         assert input_set.cp2k_input.get("MOTION")
         assert input_set.cp2k_input["MOTION"]["GEO_OPT"]["BFGS"]["TRUST_RADIUS"].values[
             0
         ] == pytest.approx(0.1)
 
-    for gen in [
+    for gen in (
         CellOptSetGenerator(user_input_settings=basis_and_potential),
         HybridCellOptSetGenerator(user_input_settings=basis_and_potential),
-    ]:
+    ):
         input_set = gen.get_input_set(si_structure)
         assert input_set.cp2k_input["GLOBAL"]["RUN_TYPE"].values[0] == "CELL_OPT"