Merge branch 'materialsproject:master' into master

Author: benrich37

pyproject.toml

@@ -297,7 +297,6 @@ namespace_packages = true
 no_implicit_optional = false
 disable_error_code = ["annotation-unchecked", "override", "operator", "attr-defined", "union-attr", "misc", "call-overload", "index"]
 exclude = ['src/pymatgen/analysis', 'src/pymatgen/io/cp2k', 'src/pymatgen/io/lammps']
-plugins = ["numpy.typing.mypy_plugin"]
 
 [[tool.mypy.overrides]]
 module = ["requests.*", "tabulate.*", "monty.*", "matplotlib.*"]

requirements.txt

@@ -85,5 +85,5 @@ tzdata==2024.2
     # via pandas
 uncertainties==3.2.2
     # via pymatgen (pyproject.toml)
-urllib3==2.2.3
+urllib3==2.5.0
     # via requests

src/pymatgen/core/periodic_table.py

@@ -1088,15 +1088,17 @@ def __repr__(self) -> str:
         return f"Species {self}"
 
     def __str__(self) -> str:
-        output = self.name if hasattr(self, "name") else self.symbol
-        if self.oxi_state is not None:
-            abs_charge = formula_double_format(abs(self.oxi_state))
+        name = getattr(self, "name", None)
+        output = name or self.symbol
+        oxi_state = self.oxi_state
+        if oxi_state is not None:
+            abs_charge = formula_double_format(abs(oxi_state))
             if isinstance(abs_charge, float):
                 abs_charge = f"{abs_charge:.2f}"  # type: ignore[assignment]
-            output += f"{abs_charge}{'+' if self.oxi_state >= 0 else '-'}"
+            output += f"{abs_charge}{'+' if oxi_state >= 0 else '-'}"
 
-        if self._spin is not None:
-            spin = self._spin
+        spin = self._spin
+        if spin is not None:
             output += f",{spin=}"
         return output
 
@@ -1484,10 +1486,11 @@ def __repr__(self) -> str:
 
     def __str__(self) -> str:
         output = self.symbol
-        if self.oxi_state is not None:
-            output += f"{formula_double_format(abs(self.oxi_state))}{'+' if self.oxi_state >= 0 else '-'}"
-        if self._spin is not None:
-            spin = self._spin
+        oxi_state = self.oxi_state
+        if oxi_state is not None:
+            output += f"{formula_double_format(abs(oxi_state))}{'+' if oxi_state >= 0 else '-'}"
+        spin = self._spin
+        if spin is not None:
             output += f",{spin=}"
         return output
 

src/pymatgen/ext/matproj.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import gzip
 import itertools
 import json
 import logging
@@ -19,17 +20,21 @@
 from functools import partial
 from typing import TYPE_CHECKING, NamedTuple
 
+import numpy as np
 import orjson
 import requests
 from monty.json import MontyDecoder
 
-from pymatgen.core import SETTINGS
+from pymatgen.core import SETTINGS, Lattice
 from pymatgen.core import __version__ as PMG_VERSION
 from pymatgen.core.composition import Composition
+from pymatgen.electronic_structure.bandstructure import Kpoint
+from pymatgen.phonon import CompletePhononDos, PhononBandStructureSymmLine
 from pymatgen.symmetry.analyzer import SpacegroupAnalyzer
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Sequence
+    from typing import Any
 
     from typing_extensions import Self
 
@@ -474,31 +479,77 @@ def get_entries_in_chemsys(self, elements: str | list[str], *args, **kwargs):
 
         return self.get_entries(criteria, *args, **kwargs)
 
-    def get_phonon_bandstructure_by_material_id(self, material_id: str):
+    def _retrieve_object_from_s3(self, material_id: str, bucket: str, prefix: str, timeout: float = 60) -> Any:
+        """
+        Retrieve data from Amazon S3 OpenData the non-canonical way, using requests.
+
+        This should be transitioned to boto3 if long-term support is desired,
+        or to expand pymatgen support of, e.g., electronic DOS, bandstructure, etc.
+
+        Args:
+            material_id (str): Materials Project material_id
+            bucket (str): the Materials Project bucket, either materialsproject-parsed
+                or materialsproject-build
+            prefix (str) : the prefix of the particular S3 key.
+            timeout (float = 60) : timeout in seconds for the requests command.
+
+        Returns:
+            json loaded object
+        """
+        response = requests.get(
+            f"https://s3.us-east-1.amazonaws.com/{bucket}/{prefix}/{material_id}.json.gz",
+            timeout=timeout,
+        )
+        if response.status_code not in {200, 400}:
+            raise MPRestError(
+                f"Failed to retrieve data from OpenData with status code {response.status_code}:\n{response.reason}"
+            )
+        return orjson.loads(gzip.decompress(response.content))
+
+    def get_phonon_bandstructure_by_material_id(self, material_id: str) -> PhononBandStructureSymmLine:
         """Get phonon bandstructure by material_id.
 
+        Note that this method borrows constructor methods built into
+        in the emmet-core model for this data. Calling the `to_pmg`
+        method of the emmet-core data model handles this.
+
         Args:
             material_id (str): Materials Project material_id
 
         Returns:
             PhononBandStructureSymmLine: A phonon band structure.
         """
-        prop = "phonon_bandstructure"
-        response = self.materials.phonon.search(material_ids=material_id)
-        return response[0][prop]
+        data = self._retrieve_object_from_s3(
+            material_id, bucket="materialsproject-parsed", prefix="ph-bandstructures/dfpt"
+        )
+        rlatt = Lattice(data["reciprocal_lattice"])
+        return PhononBandStructureSymmLine(
+            [Kpoint(q, lattice=rlatt).frac_coords for q in data["qpoints"]],
+            np.array(data["frequencies"]),
+            rlatt,
+            has_nac=data["has_nac"],
+            eigendisplacements=np.array(data["eigendisplacements"]),
+            structure=data["structure"],
+            labels_dict={k: Kpoint(v, lattice=rlatt).frac_coords for k, v in (data["labels_dict"] or {}).items()},
+            coords_are_cartesian=False,
+        )
 
-    def get_phonon_dos_by_material_id(self, material_id: str):
+    def get_phonon_dos_by_material_id(self, material_id: str) -> CompletePhononDos:
         """Get phonon density of states by material_id.
 
+        Note that this method borrows constructor methods built into
+        in the emmet-core model for this data. Calling the `to_pmg`
+        method of the emmet-core data model handles this.
+
         Args:
             material_id (str): Materials Project material_id
 
         Returns:
             CompletePhononDos: A phonon DOS object.
         """
-        prop = "phonon_dos"
-        response = self.request(f"materials/phonon/?material_ids={material_id}&_fields={prop}")
-        return response[0][prop]
+        data = self._retrieve_object_from_s3(material_id, bucket="materialsproject-parsed", prefix="ph-dos/dfpt")
+        data["pdos"] = data.pop("projected_densities", None)
+        return CompletePhononDos.from_dict(data)
 
 
 class MPRestError(Exception):

src/pymatgen/io/cp2k/inputs.py

@@ -44,14 +44,13 @@
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Sequence
-    from pathlib import Path
     from typing import Any, Literal
 
     from typing_extensions import Self
 
     from pymatgen.core.lattice import Lattice
     from pymatgen.core.structure import Molecule, Structure
-    from pymatgen.util.typing import Kpoint
+    from pymatgen.util.typing import Kpoint, PathLike
 
 __author__ = "Nicholas Winner"
 __version__ = "2.0"
@@ -152,7 +151,7 @@ def from_dict(cls, dct: dict) -> Self:
         )
 
     @classmethod
-    def from_str(cls, s: str) -> Self:
+    def from_str(cls, s: str, description: str | None) -> Self:
         """
         Initialize from a string.
 
@@ -163,12 +162,6 @@ def from_str(cls, s: str) -> Self:
         Returns:
             Keyword or None
         """
-        s = s.strip()
-        if "!" in s or "#" in s:
-            s, description = re.split("(?:!|#)", s)
-            description = description.strip()
-        else:
-            description = None
         units = re.findall(r"\[(.*)\]", s) or [None]
         s = re.sub(r"\[(.*)\]", "", s)
         args: list[Any] = s.split()
@@ -660,7 +653,7 @@ class Cp2kInput(Section):
     title and by implementing the file i/o.
     """
 
-    def __init__(self, name: str = "CP2K_INPUT", subsections: dict | None = None, **kwargs):
+    def __init__(self, name: str = "CP2K_INPUT", subsections: dict | None = None, **kwargs) -> None:
         """Initialize Cp2kInput by calling the super."""
         self.name = name
         self.subsections = subsections or {}
@@ -675,12 +668,12 @@ def __init__(self, name: str = "CP2K_INPUT", subsections: dict | None = None, **
             **kwargs,
         )
 
-    def get_str(self):
+    def get_str(self) -> str:
         """Get string representation of the Cp2kInput."""
         return "".join(v.get_str() for v in self.subsections.values())
 
     @classmethod
-    def _from_dict(cls, dct: dict):
+    def _from_dict(cls, dct: dict) -> Self:
         """Initialize from a dictionary."""
         constructor = getattr(
             __import__(dct["@module"], globals(), locals(), dct["@class"], 0),
@@ -690,7 +683,7 @@ def _from_dict(cls, dct: dict):
         return Cp2kInput("CP2K_INPUT", subsections=constructor.from_dict(dct).subsections)
 
     @classmethod
-    def from_file(cls, filename: str | Path) -> Self:
+    def from_file(cls, filename: PathLike) -> Self:
         """Initialize from a file."""
         with zopen(filename, mode="rt", encoding="utf-8") as file:
             txt = preprocessor(file.read(), os.path.dirname(file.name))
@@ -699,33 +692,38 @@ def from_file(cls, filename: str | Path) -> Self:
     @classmethod
     def from_str(cls, s: str) -> Self:
         """Initialize from a string."""
-        lines = s.splitlines()
-        lines = [line.replace("\t", "") for line in lines]
-        lines = [line.strip() for line in lines]
-        lines = [line for line in lines if line]
+        lines: list[str] = [cleaned for line in s.splitlines() if (cleaned := line.replace("\t", "").strip())]
         return cls.from_lines(lines)
 
     @classmethod
-    def from_lines(cls, lines: list | tuple) -> Self:
+    def from_lines(cls, lines: Sequence[str]) -> Self:
         """Helper method to read lines of file."""
         cp2k_input = Cp2kInput("CP2K_INPUT", subsections={})
         Cp2kInput._from_lines(cp2k_input, lines)
         return cp2k_input
 
-    def _from_lines(self, lines):
+    def _from_lines(self, lines: Sequence[str]) -> None:
         """Helper method, reads lines of text to get a Cp2kInput."""
-        current = self.name
-        description = ""
+        current: str = self.name
+        description: str = ""
+
         for line in lines:
-            if line.startswith(("!", "#")):
-                description += line[1:].strip()
-            elif line.upper().startswith("&END"):
+            line, *comment = re.split(r"[!#]", line, maxsplit=1)
+
+            if comment:
+                description += comment[0].strip()
+
+            if not (line := line.strip()):
+                continue
+
+            if line.upper().startswith("&END"):
                 current = "/".join(current.split("/")[:-1])
+
             elif line.startswith("&"):
                 name, subsection_params = line.split()[0][1:], line.split()[1:]
                 subsection_params = (
                     []
-                    if len(subsection_params) == 1 and subsection_params[0].upper() in ("T", "TRUE", "F", "FALSE", "ON")
+                    if len(subsection_params) == 1 and subsection_params[0].upper() in {"T", "TRUE", "F", "FALSE", "ON"}
                     else subsection_params
                 )
                 alias = f"{name} {' '.join(subsection_params)}" if subsection_params else None
@@ -745,8 +743,10 @@ def _from_lines(self, lines):
                 else:
                     self.by_path(current).insert(sec)
                 current = f"{current}/{alias or name}"
+
             else:
-                kwd = Keyword.from_str(line)
+                kwd = Keyword.from_str(line, (description or None))
+
                 if tmp := self.by_path(current).get(kwd.name):
                     if isinstance(tmp, KeywordList):
                         self.by_path(current).get(kwd.name).append(kwd)
@@ -761,15 +761,15 @@ def _from_lines(self, lines):
 
     def write_file(
         self,
-        input_filename: str = "cp2k.inp",
-        output_dir: str = ".",
+        input_filename: PathLike = "cp2k.inp",
+        output_dir: PathLike = ".",
         make_dir_if_not_present: bool = True,
-    ):
+    ) -> None:
         """Write input to a file.
 
         Args:
-            input_filename (str, optional): Defaults to "cp2k.inp".
-            output_dir (str, optional): Defaults to ".".
+            input_filename (PathLike, optional): Defaults to "cp2k.inp".
+            output_dir (PathLike, optional): Defaults to ".".
             make_dir_if_not_present (bool, optional): Defaults to True.
         """
         if not os.path.isdir(output_dir) and make_dir_if_not_present:

tests/ext/test_matproj.py

@@ -104,13 +104,10 @@ def test_get_entry_by_material_id(mprester):
         mprester.get_entry_by_material_id("mp-2022")  # "mp-2022" does not exist
 
 
-@pytest.mark.skip(reason="MP staff broke the API: Jun 2025")
 def test_get_phonon_data_by_material_id(mprester):
     bs = mprester.get_phonon_bandstructure_by_material_id("mp-661")
-    print(type(bs))
     assert isinstance(bs, PhononBandStructureSymmLine)
     dos = mprester.get_phonon_dos_by_material_id("mp-661")
-    print(type(dos))
     assert isinstance(dos, CompletePhononDos)
 
 

tests/files/io/cp2k/cp2k.inp

@@ -4,7 +4,7 @@
 &FORCE_EVAL 
 	METHOD QS
 	STRESS_TENSOR ANALYTICAL
-	&SUBSYS 
+	&SUBSYS ! "/" used to break input file parser in Section start line
 		! Input parameters needed to set up the CELL.
 		&CELL 
 			A 0.0 2.734364 2.734364

tests/io/cp2k/test_inputs.py

@@ -217,10 +217,14 @@ def test_ci_file(self):
         assert isinstance(self.ci["FORCE_EVAL"]["DFT"]["QS"]["EPS_DEFAULT"].values[0], float)
 
         # description retrieval
+        assert (
+            self.ci["FORCE_EVAL"]["SUBSYS"].description == '"/" used to break input file parser in Section start line'
+        )
         assert self.ci["FORCE_EVAL"]["SUBSYS"]["CELL"].description == "Input parameters needed to set up the CELL."
         assert (
             self.ci["FORCE_EVAL"]["DFT"]["MGRID"]["CUTOFF"].description == "Cutoff in [Ry] for finest level of the MG."
         )
+        assert self.ci["FORCE_EVAL"]["METHOD"].description is None
 
     def test_odd_file(self):
         scramble = ""