begin refactor to isolate pmg / emmet dependences

Author: esoteric-ephemera

config.yaml.gz

@@ -1,26 +1,24 @@
 """Check common issues with VASP calculations."""
 
 from __future__ import annotations
-from dataclasses import dataclass, field
+from pydantic import Field, field_validator
 import numpy as np
-
-from typing import TYPE_CHECKING
+from functools import cached_property
+from typing import Any
 
 from emmet.core.vasp.calc_types.enums import TaskType
 from pymatgen.core import Structure
 
 from pymatgen.io.validation.common import BaseValidator
 
-if TYPE_CHECKING:
-    from emmet.core.tasks import TaskDoc
-    from emmet.core.vasp.calc_types.enums import RunType
-    from emmet.core.vasp.task_valid import TaskDocument
-    from pymatgen.io.vasp.inputs import Incar
-    from typing import Sequence
-    from numpy.typing import ArrayLike
+from emmet.core.tasks import TaskDoc
+from emmet.core.vasp.calc_types.enums import RunType
+from emmet.core.vasp.task_valid import TaskDocument
+from pymatgen.io.vasp.inputs import Incar
+from collections.abc import Sequence
+from numpy.typing import ArrayLike
 
 
-@dataclass
 class CheckCommonErrors(BaseValidator):
     """
     Check for common calculation errors.
@@ -59,22 +57,33 @@ class CheckCommonErrors(BaseValidator):
 
     reasons: list[str]
     warnings: list[str]
-    task_doc: TaskDoc | TaskDocument = None
+    task_doc: dict = None
     parameters: dict = None
     structure: Structure = None
     run_type: RunType = None
     name: str = "Check common errors"
     fast: bool = False
     defaults: dict | None = None
     # TODO: make this also work for elements Gd and Eu, which have magmoms >5 in at least one of their pure structures
-    valid_max_magmoms: dict[str, float] = field(default_factory=lambda: {"Gd": 10.0, "Eu": 10.0})
-    exclude_elements: set[str] = field(default_factory=lambda: {"Am", "Po"})
+    valid_max_magmoms: dict[str, float] = Field(default_factory=lambda: {"Gd": 10.0, "Eu": 10.0})
+    exclude_elements: set[str] = Field(default_factory=lambda: {"Am", "Po"})
     valid_max_allowed_scf_gradient: float | None = None
     num_ionic_steps_to_avg_drift_over: int | None = None
 
-    def __post_init__(self):
-        self.incar = self.task_doc["calcs_reversed"][0]["input"]["incar"]
-        self.ionic_steps = self.task_doc["calcs_reversed"][0]["output"]["ionic_steps"]
+    @field_validator("task_doc",mode="before")
+    @classmethod
+    def deserialize_task_doc(cls,val : Any) -> dict:
+        if hasattr(val,"model_dump"):
+            return val.model_dump()
+        return val
+
+    @cached_property
+    def incar(self) -> dict:
+        return self.task_doc["calcs_reversed"][0]["input"]["incar"]
+
+    @cached_property
+    def ionic_steps(self) -> list:
+        return self.task_doc["calcs_reversed"][0]["output"]["ionic_steps"]
 
     def _check_run_type(self) -> None:
         if f"{self.run_type}".upper() not in {"GGA", "GGA+U", "PBE", "PBE+U", "R2SCAN"}:
@@ -214,8 +223,6 @@ def _check_unused_elements(self) -> None:
                 "which are not currently being accepted."
             )
 
-
-@dataclass
 class CheckVaspVersion(BaseValidator):
     """
     Check for common errors related to the version of VASP used.
@@ -280,27 +287,33 @@ def _check_vasp_version(self) -> None:
                 "but we only allow versions 5.4.4 and >=6.0.0 (as of July 2023)."
             )
 
-
-@dataclass
 class CheckStructureProperties(BaseValidator):
     """Check structure for options that are not suitable for thermodynamic calculations."""
 
     reasons: list[str]
     warnings: list[str]
-    structures: list[dict | Structure | None] = None
+    structures: list[Structure]
     task_type: TaskType = None
     name: str = "VASP POSCAR properties validator"
     site_properties_to_check: tuple[str, ...] = ("selective_dynamics", "velocities")
 
-    def __post_init__(self) -> None:
+    @field_validator("structures",mode="before")
+    @classmethod
+    def serialize_structures(cls, val : list[Structure | dict | None]) -> list[Structure]:
         """Extract required structure site properties."""
-
-        for idx, struct in enumerate(self.structures):
-            if isinstance(struct, dict):
-                self.structures[idx] = Structure.from_dict(struct)
-
-        self._site_props = {
-            k: [struct.site_properties.get(k) for struct in self.structures if struct]  # type: ignore[union-attr]
+        
+        out_val = []
+        for struct in val:
+            if struct:
+                if isinstance(struct, dict):
+                    struct = Structure.from_dict(struct)
+                out_val.append(struct)
+        return out_val
+    
+    @cached_property
+    def site_properties(self) -> dict[str, Any]:
+        return {
+            k: [struct.site_properties.get(k) for struct in self.structures]
             for k in self.site_properties_to_check
         }
 
@@ -314,7 +327,7 @@ def _has_frozen_degrees_of_freedom(selective_dynamics_array: ArrayLike[bool] | N
     def _check_selective_dynamics(self) -> None:
         """Check structure for inappropriate site properties."""
 
-        if (selec_dyn := self._site_props.get("selective_dynamics")) is not None and self.task_type in {
+        if (selec_dyn := self.site_properties.get("selective_dynamics")) is not None and self.task_type in {
             TaskType.Structure_Optimization,
             TaskType.Deformation,
         }:
@@ -334,7 +347,7 @@ def _has_nonzero_velocities(velocities: ArrayLike | None, tol: float = 1.0e-8) -
     def _check_velocities(self) -> None:
         """Check structure for non-zero velocities."""
 
-        if (velos := self._site_props.get("velocities")) is not None and self.task_type != TaskType.Molecular_Dynamics:
+        if (velos := self.site_properties.get("velocities")) is not None and self.task_type != TaskType.Molecular_Dynamics:
             if any(self._has_nonzero_velocities(velo) for velo in velos):
                 self.warnings.append(
                     "At least one of the structures had non-zero velocities. "

pymatgen/io/validation/check_common_errors.py

@@ -1,25 +1,21 @@
 """Validate VASP INCAR files."""
 
 from __future__ import annotations
+from collections.abc import Sequence
 import copy
-from dataclasses import dataclass
 import numpy as np
 from emmet.core.vasp.calc_types.enums import TaskType
 
-from pymatgen.io.validation.common import BaseValidator, BasicValidator
-from pymatgen.io.validation.vasp_defaults import InputCategory, VaspParam
+from pymatgen.core import Structure
+from pymatgen.io.vasp.sets import VaspInputSet
 
-from typing import TYPE_CHECKING
+from pymatgen.io.validation.common import BaseValidator
+from pymatgen.io.validation.vasp_defaults import InputCategory, VaspParam
 
-if TYPE_CHECKING:
-    from typing import Any, Sequence
-    from pymatgen.core import Structure
-    from pymatgen.io.vasp.sets import VaspInputSet
+from typing import Any
 
 # TODO: fix ISIF getting overwritten by MP input set.
 
-
-@dataclass
 class CheckIncar(BaseValidator):
     """
     Check calculation parameters related to INCAR input tags.
@@ -107,25 +103,16 @@ def check(self) -> None:
         working_params.update_parameters_and_defaults()
 
         # Validate each parameter in the set of working parameters
-        simple_validator = BasicValidator()
-        for key in working_params.defaults:
+        for key, vasp_param in working_params.defaults.items():
             if self.fast and len(self.reasons) > 0:
                 # fast check: stop checking whenever a single check fails
                 break
 
-            simple_validator.check_parameter(
-                reasons=self.reasons,
-                warnings=self.warnings,
-                input_tag=working_params.defaults[key]["alias"],
-                current_values=working_params.parameters[key],
-                reference_values=working_params.valid_values[key],
-                operations=working_params.defaults[key]["operation"],
-                tolerance=working_params.defaults[key]["tolerance"],
-                append_comments=working_params.defaults[key]["comment"],
-                severity=working_params.defaults[key]["severity"],
+            vasp_param.check(
+                working_params.parameters[key],
+                working_params.valid_values[key]
             )
-
-
+            
 class UpdateParameterValues:
     """
     Update a set of parameters according to supplied rules and defaults.

pymatgen/io/validation/check_incar.py

@@ -1,20 +1,14 @@
 """Validate VASP KPOINTS files or the KSPACING/KGAMMA INCAR settings."""
 
 from __future__ import annotations
-from dataclasses import dataclass
 import numpy as np
 from pymatgen.io.vasp import Kpoints
 
-from pymatgen.io.validation.common import BaseValidator
-
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from pymatgen.core import Structure
-    from pymatgen.io.vasp.sets import VaspInputSet
+from pymatgen.core import Structure
+from pymatgen.io.vasp.sets import VaspInputSet
 
+from pymatgen.io.validation.common import BaseValidator
 
-@dataclass
 class CheckKpointsKspacing(BaseValidator):
     """
     Check that k-point density is sufficiently high and is compatible with lattice symmetry.
@@ -54,7 +48,7 @@ class CheckKpointsKspacing(BaseValidator):
     warnings: list[str]
     name: str = "Check k-point density"
     valid_input_set: VaspInputSet = None
-    kpoints: Kpoints | dict = None
+    kpoints: Kpoints = None
     structure: Structure = None
     defaults: dict | None = None
     kpts_tolerance: float | None = None
@@ -85,13 +79,13 @@ def _get_valid_num_kpts(self) -> int:
 
     def _check_user_shifted_mesh(self) -> None:
         # Check for user shifts
-        if (not self.allow_kpoint_shifts) and any(shift_val != 0 for shift_val in self.kpoints["usershift"]):
+        if (not self.allow_kpoint_shifts) and any(shift_val != 0 for shift_val in self.kpoints.kpts_shift):
             self.reasons.append("INPUT SETTINGS --> KPOINTS: shifting the kpoint mesh is not currently allowed.")
 
     def _check_explicit_mesh_permitted(self) -> None:
         # Check for explicit kpoint meshes
 
-        if (not self.allow_explicit_kpoint_mesh) and len(self.kpoints["kpoints"]) > 1:
+        if (not self.allow_explicit_kpoint_mesh) and len(self.kpoints.kpts) > 1:
             self.reasons.append(
                 "INPUT SETTINGS --> KPOINTS: explicitly defining "
                 "the k-point mesh is not currently allowed. "
@@ -106,13 +100,10 @@ def _check_kpoint_density(self) -> None:
         # Check number of kpoints used
         valid_num_kpts = self._get_valid_num_kpts()
 
-        if isinstance(self.kpoints, Kpoints):
-            self.kpoints = self.kpoints.as_dict()
-
         cur_num_kpts = max(
-            self.kpoints.get("nkpoints", 0),
-            np.prod(self.kpoints.get("kpoints")),
-            len(self.kpoints.get("kpoints")),
+            self.kpoints.num_kpts,
+            np.prod(self.kpoints.kpts),
+            len(self.kpoints.kpts),
         )
         if cur_num_kpts < valid_num_kpts:
             self.reasons.append(
@@ -123,17 +114,17 @@ def _check_kpoint_density(self) -> None:
     def _check_kpoint_mesh_symmetry(self) -> None:
         # check for valid kpoint mesh (which depends on symmetry of the structure)
 
-        cur_kpoint_style = self.kpoints.get("generation_style").lower()
+        cur_kpoint_style = self.kpoints.style.name.lower()
         is_hexagonal = self.structure.lattice.is_hexagonal()
         is_face_centered = self.structure.get_space_group_info()[0][0] == "F"
         monkhorst_mesh_is_invalid = is_hexagonal or is_face_centered
         if (
             cur_kpoint_style == "monkhorst"
             and monkhorst_mesh_is_invalid
-            and any(x % 2 == 0 for x in self.kpoints.get("kpoints")[0])
+            and any(x % 2 == 0 for x in self.kpoints.kpts[0])
         ):
             # only allow Monkhorst with all odd number of subdivisions per axis.
-            kx, ky, kz = self.kpoints.get("kpoints")[0]
+            kx, ky, kz = self.kpoints.kpts[0]
             self.reasons.append(
                 f"INPUT SETTINGS --> KPOINTS or KGAMMA: ({kx}x{ky}x{kz}) "
                 "Monkhorst-Pack kpoint mesh was used."

pymatgen/io/validation/check_kpoints_kspacing.py

@@ -1,23 +1,18 @@
 """Check POTCAR against known POTCARs in pymatgen, without setting up psp_resources."""
 
 from __future__ import annotations
-from dataclasses import dataclass, field
+from pydantic import Field
 from importlib.resources import files as import_resource_files
 from monty.serialization import loadfn
 import numpy as np
 
-from pymatgen.io.validation.common import BaseValidator
-
-from typing import TYPE_CHECKING
+from pymatgen.core import Structure
+from pymatgen.io.vasp.sets import VaspInputSet
 
-if TYPE_CHECKING:
-    from pymatgen.core import Structure
-    from pymatgen.io.vasp.sets import VaspInputSet
+from pymatgen.io.validation.common import BaseValidator
 
 _potcar_summary_stats = loadfn(import_resource_files("pymatgen.io.vasp") / "potcar-summary-stats.json.bz2")
 
-
-@dataclass
 class CheckPotcar(BaseValidator):
     """
     Check POTCAR against library of known valid POTCARs.
@@ -52,9 +47,9 @@ class CheckPotcar(BaseValidator):
     warnings: list[str]
     valid_input_set: VaspInputSet = None
     structure: Structure = None
-    potcars: dict = None
+    potcars: list[dict] = None
     name: str = "Check POTCARs"
-    potcar_summary_stats: dict = field(default_factory=lambda: _potcar_summary_stats)
+    potcar_summary_stats: dict = Field(default_factory=lambda: _potcar_summary_stats)
     data_match_tol: float = 1.0e-6
     fast: bool = False
 

pymatgen/io/validation/check_potcar.py

@@ -1,10 +1,11 @@
 """Common class constructor for validation checks."""
 
 from __future__ import annotations
-from dataclasses import dataclass
 from math import isclose
 from typing import TYPE_CHECKING, Literal
 
+from pydantic import BaseModel
+
 if TYPE_CHECKING:
     from typing import Any
 
@@ -214,8 +215,7 @@ def check_parameter(
             )
 
 
-@dataclass
-class BaseValidator:
+class BaseValidator(BaseModel):
     """
     Template for validation classes.
 
@@ -261,5 +261,5 @@ def check(self) -> None:
             if self.fast and len(self.reasons) > 0:
                 # fast check: stop checking whenever a single check fails
                 break
-
+            
             getattr(self, attr)()