From 10457d41e7a71e9d60fcf98be9751b2de04b9b8a Mon Sep 17 00:00:00 2001 From: Max Gallant Date: Mon, 8 Jun 2026 11:06:25 -0700 Subject: [PATCH] linting --- src/rxn_network/jobs/core.py | 2 +- src/rxn_network/jobs/schema.py | 58 +++++++++++++++++----------------- src/rxn_network/utils/ray.py | 2 +- tests/utils/test_ray.py | 4 +-- 4 files changed, 33 insertions(+), 33 deletions(-) diff --git a/src/rxn_network/jobs/core.py b/src/rxn_network/jobs/core.py index aa780f08..093f27aa 100644 --- a/src/rxn_network/jobs/core.py +++ b/src/rxn_network/jobs/core.py @@ -129,7 +129,7 @@ def make(self, chemsys: str): ) else: try: - from mp_api.client import MPRester + from mp_api.client import MPRester # noqa: PLC0415 except ImportError as err: raise ImportError("You may need to install the Materials Project API: pip install -U mp-api") from err diff --git a/src/rxn_network/jobs/schema.py b/src/rxn_network/jobs/schema.py index d9966f24..d0fbca44 100644 --- a/src/rxn_network/jobs/schema.py +++ b/src/rxn_network/jobs/schema.py @@ -1,6 +1,6 @@ """Core definition for various task and synthesis recipe documents.""" -from typing import Any, Optional +from typing import Any from pydantic import BaseModel, Field from pymatgen.core.composition import Element @@ -18,17 +18,17 @@ class EntrySetDocument(BaseModel): """A single entry set object as produced by the GetEntrySet job.""" - task_label: Optional[str] = Field(None, description="The name of the task.") + task_label: str | None = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) entries: GibbsEntrySet = Field(description="The entry set object.") - e_above_hull: Optional[float] = Field(None, description="The e_above_hull cutoff.") + e_above_hull: float | None = Field(None, description="The e_above_hull cutoff.") include_polymorphs: bool = Field( default=False, description="Whether to include metastable polymorphs in the entry set." ) - formulas_to_include: Optional[list[str]] = Field( + formulas_to_include: list[str] | None = Field( None, description="The required formulas to include during construciton." ) @@ -36,20 +36,20 @@ class EntrySetDocument(BaseModel): class EnumeratorTaskDocument(BaseModel): """A single task object from the reaction enumerator workflow.""" - task_label: Optional[str] = Field(None, description="The name of the task.") + task_label: str | None = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) rxns: ReactionSet = Field(description="The reaction set.") - targets: Optional[list[str]] = Field(None, description="The target formulas used in the enumerator(s).") - elements: Optional[list[Element]] = Field(None, description="The elements of the total chemical system") - chemsys: Optional[str] = Field(None, description="The total chemical system string (e.g., Fe-Li-O).") - added_elements: Optional[list[Element]] = Field( + targets: list[str] | None = Field(None, description="The target formulas used in the enumerator(s).") + elements: list[Element] | None = Field(None, description="The elements of the total chemical system") + chemsys: str | None = Field(None, description="The total chemical system string (e.g., Fe-Li-O).") + added_elements: list[Element] | None = Field( None, description="The elements added beyond the elements of the target(s)." ) - added_chemsys: Optional[str] = Field(None, description="The chemical system of the added elements") - enumerators: Optional[list[Enumerator]] = Field( + added_chemsys: str | None = Field(None, description="The chemical system of the added elements") + enumerators: list[Enumerator] | None = Field( None, description="A list of the enumerator objects used to calculate the reactions.", ) @@ -60,7 +60,7 @@ class CompetitionTaskDocument(BaseModel): CalculateCompetitionMaker job. """ - task_label: Optional[str] = Field(None, description="The name of the task.") + task_label: str | None = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", @@ -71,26 +71,26 @@ class CompetitionTaskDocument(BaseModel): " information stored in their data attribute." ) ) - precursor_formula: Optional[str] = Field( + precursor_formula: str | None = Field( None, description="The reduced chemical formula of the precursor material (used in extraction flows)." ) - target_formula: Optional[str] = Field(None, description="The reduced chemical formula of the target material.") - open_elem: Optional[Element] = Field(None, description="The open element (if any).") - chempot: Optional[float] = Field(None, description="The chemical potential of the open element.") - added_elements: Optional[list[Element]] = Field( + target_formula: str | None = Field(None, description="The reduced chemical formula of the target material.") + open_elem: Element | None = Field(None, description="The open element (if any).") + chempot: float | None = Field(None, description="The chemical potential of the open element.") + added_elements: list[Element] | None = Field( None, description="The elements added beyond the elements of the target(s)." ) - added_chemsys: Optional[str] = Field(None, description="The chemical system of the added elements.") - calculate_competition: Optional[bool] = Field(None, description="Whether to calculate competition scores.") - calculate_chempot_distances: Optional[bool] = Field( + added_chemsys: str | None = Field(None, description="The chemical system of the added elements.") + calculate_competition: bool | None = Field(None, description="Whether to calculate competition scores.") + calculate_chempot_distances: bool | None = Field( None, description="Whether to calculate chemical potential distances." ) - temp: Optional[float] = Field( + temp: float | None = Field( None, description=("The temperature in K used to determine the primary competition weightings."), ) - batch_size: Optional[int] = Field(None, description="The batch size for the reaction set") - cpd_kwargs: Optional[dict[str, Any]] = Field(None, description="The kwargs for ChempotDistanceCalculator.") + batch_size: int | None = Field(None, description="The batch size for the reaction set") + cpd_kwargs: dict[str, Any] | None = Field(None, description="The kwargs for ChempotDistanceCalculator.") class NetworkTaskDocument(BaseModel): @@ -99,16 +99,16 @@ class NetworkTaskDocument(BaseModel): Optionally includes unbalanced paths found during pathfinding. """ - task_label: Optional[str] = Field(None, description="The name of the task.") + task_label: str | None = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", ) network: Network = Field(description="The reaction network") - paths: Optional[PathwaySet] = Field(None, description="The (simple/unbalanced) reaction pathways") - k: Optional[int] = Field(None, description="The number of paths solved for, if any.") - precursors: Optional[list[str]] = Field(None, description="The precursor formulas.") - targets: Optional[list[str]] = Field(None, description="The target formulas.") + paths: PathwaySet | None = Field(None, description="The (simple/unbalanced) reaction pathways") + k: int | None = Field(None, description="The number of paths solved for, if any.") + precursors: list[str] | None = Field(None, description="The precursor formulas.") + targets: list[str] | None = Field(None, description="The target formulas.") class PathwaySolverTaskDocument(BaseModel): @@ -116,7 +116,7 @@ class PathwaySolverTaskDocument(BaseModel): solver object and its calculated balanced pathways. """ - task_label: Optional[str] = Field(None, description="The name of the task.") + task_label: str | None = Field(None, description="The name of the task.") last_updated: str = Field( default_factory=datetime_str, description="Timestamp of when the document was last updated.", diff --git a/src/rxn_network/utils/ray.py b/src/rxn_network/utils/ray.py index 2bd2f311..a3fe0ff2 100644 --- a/src/rxn_network/utils/ray.py +++ b/src/rxn_network/utils/ray.py @@ -27,7 +27,7 @@ def initialize_ray(quiet: bool = False): logger.setLevel("INFO") if not ray.is_initialized(): logger.info("Ray is not initialized. Checking for existing cluster...") - if os.environ.get("IP_HEAD") or int(os.environ.get("PBS_NNODES", 0)) > 1: + if os.environ.get("IP_HEAD") or int(os.environ.get("PBS_NNODES", "0")) > 1: ray.init( address="auto", ) diff --git a/tests/utils/test_ray.py b/tests/utils/test_ray.py index 27d0feaf..064e6844 100644 --- a/tests/utils/test_ray.py +++ b/tests/utils/test_ray.py @@ -51,7 +51,7 @@ def test_initialize_ray_with_slurm_cluster(capsys): try: initialize_ray() except Exception as error: # if it does fail, make sure it's the right error - assert "ConnectionError" in str(error.type) # noqa + assert "ConnectionError" in type(error).__name__ if "Connected" in capsys.readouterr().err: pytest.skip("Skipping test_initialize_ray_with_slurm_cluster due to existing cluster") @@ -67,7 +67,7 @@ def test_initialize_ray_with_pbs_cluster(capsys): try: initialize_ray() except Exception as error: # if it does fail, make sure it's the right error - assert "ConnectionError" in str(error.type) # noqa + assert "ConnectionError" in type(error).__name__ if "Connected" in capsys.readouterr().err: pytest.skip("Skipping test_initialize_ray_with_slurm_cluster due to existing cluster")