precommit

src/atomate2/common/schemas/neb.py

@@ -15,15 +15,19 @@ class NebResult(BaseModel, extra="allow"):  # type: ignore[call-arg]
     """Container class to store high-level NEB calculation info."""
 
     images: Optional[list[Structure | Molecule]] = Field(
-        None, description="Relaxed structures/molecules along the reaction pathway, including endpoints."
+        None,
+        description=(
+            "Relaxed structures/molecules along the reaction pathway, "
+            "including endpoints."
+        ),
     )
 
-    initial_endpoints : Optional[dict[str, Structure | Molecule]] = Field(
-        None, description = "Initial endpoint structures"
+    initial_endpoints: Optional[dict[str, Structure | Molecule]] = Field(
+        None, description="Initial endpoint structures"
     )
 
-    relaxed_endpoints : Optional[dict[str, Structure | Molecule]] = Field(
-        None, description = "Relaxed endpoint structures"
+    relaxed_endpoints: Optional[dict[str, Structure | Molecule]] = Field(
+        None, description="Relaxed endpoint structures"
     )
 
     initial_images: Optional[list[Structure | Molecule]] = Field(
@@ -60,7 +64,9 @@ class NebResult(BaseModel, extra="allow"):  # type: ignore[call-arg]
         ),
     )
 
-    state: Optional[TaskState] = Field(None, description="Whether the NEB calculation succeeded.")
+    state: Optional[TaskState] = Field(
+        None, description="Whether the NEB calculation succeeded."
+    )
 
     method: Optional[NebMethod] = Field(
         None, description="Variety of NEB used in this calculation."
@@ -79,7 +85,8 @@ def set_barriers(self) -> Self:
         """Perform analysis on barrier if needed."""
         if (
             (not self.forward_barrier or not self.reverse_barrier)
-            and isinstance(self.energies,list) and len(self.energies) > 0
+            and isinstance(self.energies, list)
+            and len(self.energies) > 0
         ):
             self.barrier_analysis = neb_barrier_spline_fit(self.energies)
             for k in ("forward", "reverse"):

src/atomate2/vasp/flows/approx_neb.py

@@ -68,7 +68,7 @@ def make(
         inserted_coords_dict: dict | list,
         inserted_coords_combo: list,
         n_images: int = 5,
-        min_images_per_hop : int | None = 3,
+        min_images_per_hop: int | None = 3,
         prev_dir: str | Path | None = None,
     ) -> Flow:
         """
@@ -142,7 +142,7 @@ def make(
             working_ion,
             ep_relax_jobs.output,
             image_relax_jobs.output,
-            min_images_per_hop = min_images_per_hop
+            min_images_per_hop=min_images_per_hop,
         )
 
         # to permit the flow to succeed even when prior jobs fail

src/atomate2/vasp/jobs/approx_neb.py

@@ -33,12 +33,15 @@
     from atomate2.vasp.jobs.base import BaseVaspMaker
     from atomate2.vasp.sets.base import VaspInputGenerator
 
+
 class HopFailureReason(Enum):
-
+    """Define failure modes for ApproxNEB calculations."""
+
     ENDPOINT = "Endpoint structure relaxation failed"
     MIN_DIST = "Linear distance traversed by working ion is below threshold."
     MIN_IMAGE = "Too few image calculations succeeded"
 
+
 @dataclass
 class ApproxNebHostRelaxMaker(DoubleRelaxMaker):
     """Maker to perform a double relaxation on an ApproxNEB host structure."""
@@ -233,7 +236,7 @@ def get_images_and_relax(
             # At least one endpoint calculation failed
             skip_reasons.append(HopFailureReason.ENDPOINT)
         if (
-            isinstance(min_hop_distance, float)  
+            isinstance(min_hop_distance, float)
             and get_hop_distance_from_endpoints(
                 [ep_structures[ini_ind], ep_structures[fin_ind]], working_ion
             )
@@ -477,7 +480,7 @@ def collate_results(
     working_ion: CompositionLike,
     endpoint_calc_output: dict,
     image_calc_output: dict[str, list],
-    min_images_per_hop : int | None = None,
+    min_images_per_hop: int | None = None,
 ) -> NebPathwayResult:
     """Collect output from an ApproxNEB workflow.
 
@@ -492,7 +495,7 @@ def collate_results(
     image_calc_output : dict[str,list]
         Output of get_images_and_relax
     min_images_per_hop : int or None (default)
-        If an integer, the minimum number of succesful image calculations
+        If an integer, the minimum number of successful image calculations
         to mark a calculation as successful
 
     Returns
@@ -502,36 +505,33 @@ def collate_results(
     hop_dict = {}
     hop_dist = {}
 
-    endpoint_idxs = set()
+    endpoint_idxs = []
     for combo_name in image_calc_output:
-        endpoint_idxs.update(combo_name.split("+"))
-    endpoint_idxs = sorted(endpoint_idxs)
-
-    for combo_name, images in image_calc_output.items():
+        endpoint_idxs.extend(combo_name.split("+"))
+    endpoint_idxs = sorted(set(endpoint_idxs))
 
+    for combo_name, entry in image_calc_output.items():
         metadata = {}
         task_state = TaskState.SUCCESS
-        if all(isinstance(image,str) for image in images):
+        images = entry
+        if all(isinstance(v, str) for v in entry):
             # hop calculation failed
-            metadata = {"failure_reasons": images}
+            metadata = {"failure_reasons": entry}
             task_state = TaskState.FAILED
-            if HopFailureReason.ENDPOINT.value in images:
+            if HopFailureReason.ENDPOINT.value in entry:
                 # Cannot populate any NEB fields, skip entirely
-                hop_dict[combo_name] = NebResult(
-                    state=task_state,
-                    metadata = metadata
-                )
+                hop_dict[combo_name] = NebResult(state=task_state, metadata=metadata)
                 continue
-
-            # set images to empty list, no hop calculations performed
             images = []
 
         endpoint_calcs = [endpoint_calc_output[idx] for idx in combo_name.split("+")]
         hop = [endpoint_calcs[0], *images, endpoint_calcs[1]]
 
         if min_images_per_hop is not None and task_state == TaskState.SUCCESS:
-            num_success_calcs = len([calc for calc in hop if calc["structure"] is not None])
-            if num_success_calcs < min_images_per_hop: 
+            num_success_calcs = len(
+                [calc for calc in hop if calc["structure"] is not None]
+            )
+            if num_success_calcs < min_images_per_hop:
                 task_state = TaskState.FAILED
                 if "failure_reasons" not in metadata:
                     metadata["failure_reasons"] = []
@@ -545,16 +545,17 @@ def collate_results(
                 if calc["initial_structure"] is not None
             ],
             energies=[calc["energy"] for calc in hop if calc["energy"] is not None],
-            initial_endpoints = {
-                idx : endpoint_calc_output[idx]["initial_structure"] for idx in endpoint_idxs
+            initial_endpoints={
+                idx: endpoint_calc_output[idx]["initial_structure"]
+                for idx in endpoint_idxs
             },
-            relaxed_endpoints = {
+            relaxed_endpoints={
                 idx: endpoint_calc_output[idx]["structure"] for idx in endpoint_idxs
             },
             ionic_steps=None,
             method=NebMethod.APPROX,
-            state = task_state,
-            metadata = metadata if len(metadata) > 0 else None,
+            state=task_state,
+            metadata=metadata if len(metadata) > 0 else None,
         )
 
         hop_dist[combo_name] = get_hop_distance_from_endpoints(