From 322d02365953005ae87068ec323b12b42994d83a Mon Sep 17 00:00:00 2001 From: Jonah Marks Date: Fri, 17 Apr 2026 11:05:27 +0200 Subject: [PATCH 1/4] initial ouptput file added --- examples/fsm_example.py | 22 ++- src/mlfsm/cos.py | 26 +++ src/mlfsm/output.py | 412 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 459 insertions(+), 1 deletion(-) create mode 100644 src/mlfsm/output.py diff --git a/examples/fsm_example.py b/examples/fsm_example.py index fb14478..4ba1b44 100644 --- a/examples/fsm_example.py +++ b/examples/fsm_example.py @@ -23,8 +23,10 @@ import numpy as np +from mlfsm import __version__ from mlfsm.cos import FreezingString from mlfsm.opt import CartesianOptimizer, InternalsOptimizer, Optimizer +from mlfsm.output import FSMOutput from mlfsm.utils import load_xyz, load_xyz_fixed HERE = Path(__file__).parent @@ -74,6 +76,10 @@ def run_fsm( outdir.mkdir(parents=True, exist_ok=True) + fsm_output: FSMOutput | None = None + if not interpolate: + fsm_output = FSMOutput(outdir) + # get fixed atom indices def parse_indices(text): if text is None or text.strip() == "": @@ -166,8 +172,18 @@ def parse_indices(text): else: raise ValueError(f"Unknown calculator {calculator}") + if fsm_output is not None: + fsm_output.write_header(__version__) + fsm_output.write_parameters(optcoords, interp, method, maxiter, maxls, dmax, nnodes_min, ninterp, stepsize) + fsm_output.write_system_info(reactant, product, chg, mult, fixed_atoms if len(fixed_atoms) > 0 else None) + fsm_output.write_calculator_info(calc) + fsm_output.write_initial_structures(reactant, product) + # Initialize FSM string - string = FreezingString(reactant, product, nnodes_min, interp, ninterp, stepsize) + string = FreezingString(reactant, product, nnodes_min, interp, ninterp, stepsize, output=fsm_output) + if fsm_output is not None: + fsm_output.write_path_init(string.dist, string.stepsize, string.nnodes_min) + if interpolate: string.interpolate(outdir) return @@ -187,6 +203,10 @@ def parse_indices(text): string.optimize(optimizer) string.write(outdir) + if fsm_output is not None: + fsm_output.write_final_summary(string) + fsm_output.close() + print(f"Gradient calls: {string.ngrad}") diff --git a/src/mlfsm/cos.py b/src/mlfsm/cos.py index 10b32a6..ee58ccb 100644 --- a/src/mlfsm/cos.py +++ b/src/mlfsm/cos.py @@ -11,6 +11,8 @@ if TYPE_CHECKING: from numpy.typing import NDArray + from mlfsm.output import FSMOutput + from mlfsm.coords import Cartesian from mlfsm.geom import ( calculate_arc_length, @@ -80,7 +82,9 @@ def __init__( interp_method: str = "ric", ninterp: int = 100, stepsize: float = 0.0, + output: Optional["FSMOutput"] = None, ) -> None: + self.output = output self.interp: Any self.interp_method = interp_method self.nnodes_min = int(nnodes_min) @@ -177,6 +181,9 @@ def grow(self) -> None: ``self.growing = False`` when the two frontiers are within one step-size of each other. """ + if self.output is not None: + self.output._ensure_iteration_header(self.iteration + 1) + r_atoms = self.r_string[-1] p_atoms = self.p_string[-1] @@ -228,6 +235,8 @@ def grow(self) -> None: self.r_energy += [None] self.r_tangent += [normalize(dxds)] self.r_nnodes = len(self.r_string) + if self.output is not None: + self.output.write_frontier_node("r", r_frontier, self.dist) if self.dist <= 2 * self.stepsize: self.growing = False @@ -261,6 +270,8 @@ def grow(self) -> None: self.p_energy += [None] self.p_tangent += [normalize(dxds)] self.p_nnodes = len(self.p_string) + if self.output is not None: + self.output.write_frontier_node("p", p_frontier, self.dist) else: string = interp() @@ -282,6 +293,8 @@ def grow(self) -> None: self.r_energy += [None] self.r_tangent += [normalize(cs(s[r_idx], 1))] self.r_nnodes = len(self.r_string) + if self.output is not None: + self.output.write_frontier_node("r", r_frontier, self.dist) if self.dist <= 2 * self.stepsize: self.growing = False @@ -295,6 +308,8 @@ def grow(self) -> None: self.p_energy += [None] self.p_tangent += [normalize(cs(s[p_idx], 1))] self.p_nnodes = len(self.p_string) + if self.output is not None: + self.output.write_frontier_node("p", p_frontier, self.dist) def optimize(self, optimizer: Any) -> None: """Relax all unfixed frontier nodes perpendicular to the local tangent. @@ -317,6 +332,8 @@ def optimize(self, optimizer: Any) -> None: if self.r_energy[i] is None and self.r_fix[i]: energy = optimizer.calc.get_potential_energy(self.r_string[i]) self.r_energy[i] = float_check(energy) + if self.output is not None: + self.output.write_optimized_node("r", i, self.r_string[i], self.r_energy[i], 0) elif not self.r_fix[i]: assert self.r_tangent[i] is not None atoms = self.r_string[i] @@ -330,11 +347,15 @@ def optimize(self, optimizer: Any) -> None: ngrad = 0 self.r_fix[i] = True self.ngrad += ngrad + if self.output is not None: + self.output.write_optimized_node("r", i, self.r_string[i], self.r_energy[i], ngrad) for i in range(self.p_nnodes): if self.p_energy[i] is None and self.p_fix[i]: energy = optimizer.calc.get_potential_energy(self.p_string[i]) self.p_energy[i] = float_check(energy) + if self.output is not None: + self.output.write_optimized_node("p", i, self.p_string[i], self.p_energy[i], 0) elif not self.p_fix[i]: assert self.p_tangent[i] is not None atoms = self.p_string[i] @@ -348,6 +369,8 @@ def optimize(self, optimizer: Any) -> None: ngrad = 0 self.p_fix[i] = True self.ngrad += ngrad + if self.output is not None: + self.output.write_optimized_node("p", i, self.p_string[i], self.p_energy[i], ngrad) self.dist = distance(self.r_string[-1].get_positions().flatten(), self.p_string[-1].get_positions().flatten()) @@ -399,6 +422,9 @@ def write(self, outdir: Path | str) -> None: energy_str = np.array2string(energy, precision=1, floatmode="fixed") logging.info(f"ITERATION: {self.iteration} DIST: {self.dist:.2f} ENERGY: {energy_str}") + if self.output is not None: + self.output.write_iteration_summary(self.iteration, self.r_energy, self.p_energy, self.dist) + if not self.growing: with gradfile.open("w") as f: f.write(f"{self.ngrad}\n") diff --git a/src/mlfsm/output.py b/src/mlfsm/output.py new file mode 100644 index 0000000..4574c16 --- /dev/null +++ b/src/mlfsm/output.py @@ -0,0 +1,412 @@ +"""Incremental output file writer for FSM calculations.""" + +from __future__ import annotations + +import datetime +from collections import Counter +from pathlib import Path +from typing import TYPE_CHECKING, Any, Optional, TextIO + +import numpy as np +from ase import Atoms +from numpy.typing import NDArray + +if TYPE_CHECKING: + from mlfsm.cos import FreezingString + +_SEP = "=" * 70 +_SEP_THIN = "-" * 70 + + +def _write_section(f: TextIO, title: str) -> None: + f.write(f"\n {_SEP}\n") + pad = (68 - len(title)) // 2 + f.write(f" {' ' * pad}{title}\n") + f.write(f" {_SEP}\n") + + +def _format_atoms_block(atoms: Atoms, indent: str = " ") -> str: + symbols = atoms.get_chemical_symbols() + positions = atoms.get_positions() + lines = [ + f"{indent}{i + 1:4d} {sym:<3s} {pos[0]:12.6f} {pos[1]:12.6f} {pos[2]:12.6f}" + for i, (sym, pos) in enumerate(zip(symbols, positions, strict=True)) + ] + return "\n".join(lines) + + +def _chemical_formula(atoms: Atoms) -> str: + counts: Counter[str] = Counter(atoms.get_chemical_symbols()) + order = ["C", "H"] + sorted(k for k in counts if k not in ("C", "H")) + parts = [] + for sym in order: + if sym in counts: + parts.append(sym if counts[sym] == 1 else f"{sym}{counts[sym]}") + return "".join(parts) + + +def get_calculator_info(calc: Any) -> dict[str, Any]: + """Extract available information from an ASE calculator without raising.""" + info: dict[str, Any] = {"name": type(calc).__name__} + + for attr in ("label", "method", "basis", "charge", "multiplicity"): + try: + val = getattr(calc, attr, None) + if val is not None: + info[attr] = val + except Exception: + pass + + try: + params = calc.parameters + if isinstance(params, dict): + for k, v in params.items(): + if k not in info: + info[k] = v + except Exception: + pass + + try: + d = calc.todict() + if isinstance(d, dict): + for k, v in d.items(): + if k not in info: + info[k] = v + except Exception: + pass + + # FAIRChem / UMA + for attr in ("task_name",): + try: + val = getattr(calc, attr, None) + if val is not None: + info[attr] = val + except Exception: + pass + try: + ckpt = calc.predictor.checkpoint_path # type: ignore[union-attr] + info["checkpoint"] = str(ckpt) + except Exception: + pass + + return info + + +class FSMOutput: + """Manages incremental writing of a human-readable FSM output file. + + Parameters + ---------- + outdir : path-like + Directory in which to write the output file. + filename : str, optional + Output file name. Default is ``"fsm.out"``. + """ + + def __init__(self, outdir: Path | str, filename: str = "fsm.out") -> None: + self._path = Path(outdir) / filename + self._f: TextIO = self._path.open("w", encoding="utf-8") + self._current_iteration: int = 0 + self._node_lines: list[str] = [] + + def close(self) -> None: + """Flush and close the output file.""" + self._f.flush() + self._f.close() + + # ------------------------------------------------------------------ + # Setup sections (called once before the main loop) + # ------------------------------------------------------------------ + + def write_header(self, version: str) -> None: + """Write the banner and timestamp.""" + f = self._f + f.write(f" {_SEP}\n") + title = "ML-FSM: Machine Learning Freezing String Method" + pad = (68 - len(title)) // 2 + f.write(f" {' ' * pad}{title}\n") + ver_line = f"Version {version}" + pad2 = (68 - len(ver_line)) // 2 + f.write(f" {' ' * pad2}{ver_line}\n") + f.write(f" {_SEP}\n") + now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + f.write(f"\n Date/Time: {now}\n") + f.flush() + + def write_parameters( + self, + optcoords: str, + interp: str, + method: str, + maxiter: int, + maxls: int, + dmax: float, + nnodes_min: int, + ninterp: int, + stepsize: float, + ) -> None: + """Write the input parameter block.""" + f = self._f + _write_section(f, "INPUT PARAMETERS") + f.write(f"\n Optimization coordinates : {optcoords}\n") + f.write(f" Interpolation method : {interp}\n") + f.write(f" Optimizer : {method}\n") + f.write(f" Max optimizer iterations : {maxiter}\n") + f.write(f" Max line search iterations : {maxls}\n") + f.write(f" Max displacement (dmax) : {dmax:.4f} Å\n") + f.write(f" Target node count : {nnodes_min}\n") + f.write(f" Interpolation points : {ninterp}\n") + if stepsize > 0.0: + f.write(f" Step size (explicit) : {stepsize:.4f} Å\n") + else: + f.write(f" Step size (explicit) : derived from target node count\n") + f.flush() + + def write_system_info( + self, + reactant: Atoms, + product: Atoms, + chg: int, + mult: int, + fixed_atoms: Optional[NDArray[np.integer[Any]]], + ) -> None: + """Write molecular system information.""" + f = self._f + _write_section(f, "MOLECULAR SYSTEM") + formula = _chemical_formula(reactant) + natoms = len(reactant) + f.write(f"\n Formula : {formula}\n") + f.write(f" Atoms : {natoms}\n") + f.write(f" Charge : {chg}\n") + f.write(f" Multiplicity : {mult}\n") + if fixed_atoms is None or len(fixed_atoms) == 0: + f.write(" Fixed atoms : None\n") + else: + idx_str = ", ".join(str(i + 1) for i in fixed_atoms) + f.write(f" Fixed atoms : {idx_str} (1-indexed)\n") + f.flush() + + def write_calculator_info(self, calc: Any) -> None: + """Write calculator name and available parameters.""" + f = self._f + info = get_calculator_info(calc) + _write_section(f, "CALCULATOR") + f.write(f"\n Calculator : {info.pop('name')}\n") + skip = {"kwargs", "restart", "ignore_bad_restart_file", "directory"} + for k, v in info.items(): + if k in skip: + continue + label = k.replace("_", " ").capitalize() + f.write(f" {label:<20s}: {v}\n") + f.flush() + + def write_initial_structures(self, reactant: Atoms, product: Atoms) -> None: + """Write reactant and product coordinate blocks.""" + f = self._f + _write_section(f, "INITIAL STRUCTURES") + f.write("\n Standard Orientation — Reactant (Angstroms)\n") + f.write(f" {_SEP_THIN}\n") + f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") + f.write(f" {_SEP_THIN}\n") + f.write(_format_atoms_block(reactant)) + f.write("\n") + + f.write(f"\n Standard Orientation — Product (Angstroms)\n") + f.write(f" {_SEP_THIN}\n") + f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") + f.write(f" {_SEP_THIN}\n") + f.write(_format_atoms_block(product)) + f.write("\n") + f.flush() + + def write_path_init(self, dist: float, stepsize: float, nnodes_min: int) -> None: + """Write path initialization summary.""" + f = self._f + _write_section(f, "PATH INITIALIZATION") + f.write(f"\n Total path distance : {dist:.4f} Å\n") + f.write(f" Step size : {stepsize:.4f} Å\n") + f.write(f" Target node count : {nnodes_min}\n") + f.flush() + + # ------------------------------------------------------------------ + # Per-iteration sections (called from cos.py hooks) + # ------------------------------------------------------------------ + + def _ensure_iteration_header(self, iteration: int) -> None: + if iteration != self._current_iteration: + self._current_iteration = iteration + self._node_lines = [] + _write_section(self._f, f"ITERATION {iteration}") + self._f.write("\n") + + def write_frontier_node(self, side: str, atoms: Atoms, dist: float) -> None: + """Write the newly selected frontier node structure. + + Parameters + ---------- + side : {"r", "p"} + Which end of the string. + atoms : Atoms + The frontier node geometry. + dist : float + Current distance between frontier nodes. + """ + f = self._f + label = "Reactant-side frontier" if side == "r" else "Product-side frontier" + f.write(f" {_SEP_THIN}\n") + f.write(f" {label} (frontier distance = {dist:.4f} Å)\n") + f.write(f" {_SEP_THIN}\n") + f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") + f.write(_format_atoms_block(atoms)) + f.write("\n") + f.flush() + + def write_optimized_node( + self, + side: str, + idx: int, + atoms: Atoms, + energy: Optional[float], + ngrad: int, + ) -> None: + """Record an optimized (or endpoint-evaluated) node. + + Called from within ``FreezingString.optimize()``. Results are + buffered and flushed together with the iteration summary. + + Parameters + ---------- + side : {"r", "p"} + Which string the node belongs to. + idx : int + Position in the string list (0 = endpoint). + atoms : Atoms + Final geometry after optimization. + energy : float or None + Energy in eV. + ngrad : int + Number of gradient calls used (0 for endpoint-only evaluation). + """ + tag = f"{side}[{idx}]" + kind = "endpoint " if ngrad == 0 else "optimized" + if energy is not None: + e_str = f"{energy:+.6f} eV" + else: + e_str = "N/A" + grad_str = "" if ngrad == 0 else f" ngrad = {ngrad}" + self._node_lines.append(f" {tag:<8s} {kind} : energy = {e_str}{grad_str}") + + f = self._f + f.write(f" {_SEP_THIN}\n") + f.write(f" Optimized node {tag} ({kind})\n") + f.write(f" {_SEP_THIN}\n") + f.write(f" Energy: {e_str}{grad_str}\n") + f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") + f.write(_format_atoms_block(atoms)) + f.write("\n") + f.flush() + + def write_iteration_summary( + self, + iteration: int, + r_energies: list[Optional[float]], + p_energies: list[Optional[float]], + dist: float, + ) -> None: + """Write per-iteration energy table and distance. + + Called from ``FreezingString.write()`` after the XYZ file is written. + """ + f = self._f + all_energies = r_energies + p_energies[::-1] + valid = [e for e in all_energies if e is not None] + if not valid: + return + e_min = min(valid) + + f.write(f"\n {_SEP_THIN}\n") + f.write(f" Iteration {iteration} summary (frontier distance = {dist:.4f} Å)\n") + f.write(f" {_SEP_THIN}\n") + f.write(f" {'Node':<8s} {'Side':<8s} {'Energy (eV)':>14s} {'Rel. Energy (eV)':>18s}\n") + f.write(f" {_SEP_THIN}\n") + + nr = len(r_energies) + for i, e in enumerate(r_energies): + tag = "R" if i == 0 else "" + e_str = f"{e:+.6f}" if e is not None else " N/A " + rel_str = f"{e - e_min:+.4f}" if e is not None else " N/A " + f.write(f" {i + 1:<8d} {'r' + tag:<8s} {e_str:>14s} {rel_str:>18s}\n") + + for j, e in enumerate(p_energies[::-1]): + node_idx = nr + j + 1 + tag = "P" if j == len(p_energies) - 1 else "" + e_str = f"{e:+.6f}" if e is not None else " N/A " + rel_str = f"{e - e_min:+.4f}" if e is not None else " N/A " + f.write(f" {node_idx:<8d} {'p' + tag:<8s} {e_str:>14s} {rel_str:>18s}\n") + + f.write(f" {_SEP_THIN}\n\n") + f.flush() + + # ------------------------------------------------------------------ + # Final summary (called once after the loop) + # ------------------------------------------------------------------ + + def write_final_summary(self, string: "FreezingString") -> None: + """Write TS guess identification and full string energy profile.""" + f = self._f + _write_section(f, "CALCULATION COMPLETE") + + all_energies = string.r_energy + string.p_string[::-1] # type: ignore[operator] + all_energies = string.r_energy + string.p_energy[::-1] + valid_pairs = [(i, e) for i, e in enumerate(all_energies) if e is not None] + + f.write(f"\n Total iterations : {string.iteration}\n") + f.write(f" Total gradient calls : {string.ngrad}\n\n") + + if not valid_pairs: + f.write(" No energies available.\n") + f.flush() + return + + e_values = np.array([e for _, e in valid_pairs]) + e_min = float(e_values.min()) + ts_local = int(np.argmax(e_values)) + ts_global_idx = valid_pairs[ts_local][0] + ts_energy = valid_pairs[ts_local][1] + assert ts_energy is not None + + nr = len(string.r_string) + if ts_global_idx < nr: + ts_label = f"r[{ts_global_idx}]" + else: + p_idx = len(all_energies) - 1 - ts_global_idx + ts_label = f"p[{p_idx}]" + + f.write(f" {_SEP_THIN}\n") + f.write(f" TS Guess: node {ts_label} (highest-energy node)\n") + f.write(f" Absolute energy : {ts_energy:+.6f} eV\n") + f.write(f" Relative energy : {ts_energy - e_min:+.4f} eV (above string minimum)\n") + f.write(f" {_SEP_THIN}\n\n") + + f.write(f" Full String Energies (relative to minimum, eV)\n") + f.write(f" {_SEP_THIN}\n") + f.write(f" {'Node':<8s} {'Side':<8s} {'Energy (eV)':>14s} {'Rel. Energy (eV)':>18s}\n") + f.write(f" {_SEP_THIN}\n") + + for local_i, (global_i, e) in enumerate(valid_pairs): + if global_i < nr: + side = "R" if global_i == 0 else "r" + else: + p_pos = len(all_energies) - 1 - global_i + side = "P" if p_pos == 0 else "p" + ts_marker = " <-- TS guess" if local_i == ts_local else "" + e_str = f"{e:+.6f}" + rel_str = f"{e - e_min:+.4f}" + f.write( + f" {local_i + 1:<8d} {side:<8s} {e_str:>14s} {rel_str:>18s}{ts_marker}\n" + ) + + f.write(f" {_SEP_THIN}\n") + f.write(f"\n Full string written to: vfile_{string.iteration:02d}.xyz\n") + f.write(f"\n {_SEP}\n") + f.flush() From 68e26379d74e0b48314e7c1cedfe7030305a317c Mon Sep 17 00:00:00 2001 From: Jonah Marks Date: Sat, 18 Apr 2026 21:57:26 +0200 Subject: [PATCH 2/4] update outfile formatting, changes to optimzation returns to give accurate info --- examples/fsm_example.py | 2 +- src/mlfsm/__init__.py | 2 +- src/mlfsm/cos.py | 62 +++++++--- src/mlfsm/opt.py | 12 +- src/mlfsm/output.py | 258 +++++++++++++++++++++++++++------------- 5 files changed, 228 insertions(+), 108 deletions(-) diff --git a/examples/fsm_example.py b/examples/fsm_example.py index 4ba1b44..71b966e 100644 --- a/examples/fsm_example.py +++ b/examples/fsm_example.py @@ -182,7 +182,7 @@ def parse_indices(text): # Initialize FSM string string = FreezingString(reactant, product, nnodes_min, interp, ninterp, stepsize, output=fsm_output) if fsm_output is not None: - fsm_output.write_path_init(string.dist, string.stepsize, string.nnodes_min) + fsm_output.write_path_init(string.dist, string.stepsize, string.nnodes_min, string.init_coordsobj) if interpolate: string.interpolate(outdir) diff --git a/src/mlfsm/__init__.py b/src/mlfsm/__init__.py index fc2257f..64cb838 100644 --- a/src/mlfsm/__init__.py +++ b/src/mlfsm/__init__.py @@ -1,3 +1,3 @@ """mlfsm package.""" -__version__ = "1.0.0" +__version__ = "1.0.1" diff --git a/src/mlfsm/cos.py b/src/mlfsm/cos.py index ee58ccb..08e2033 100644 --- a/src/mlfsm/cos.py +++ b/src/mlfsm/cos.py @@ -104,17 +104,26 @@ def __init__( self.natoms = len(self.atoms.numbers) if not self.use_cartesian_distance: - interp = self.interp(reactant, product, ninterp=self.ninterp) - s = calculate_arc_length(interp()) + _interp_init = self.interp(reactant, product, ninterp=self.ninterp) + s = calculate_arc_length(_interp_init()) self.dist = s[-1] self.stepsize = self.dist / self.nnodes_min else: - interp = Linear(reactant, product, ninterp=self.ninterp) - s = calculate_arc_length(interp()) + _interp_init = Linear(reactant, product, ninterp=self.ninterp) + s = calculate_arc_length(_interp_init()) self.dist = s[-1] self.stepsize = float(stepsize) self.nnodes_min = int(self.dist / self.stepsize) + if interp_method == "ric": + self.init_coordsobj = ( + _interp_init.coords # type: ignore[union-attr] + if isinstance(_interp_init, RIC) + else RIC(reactant, product, ninterp=2).coords + ) + else: + self.init_coordsobj = None + logger.info(f"NNODES_MIN: {self.nnodes_min}") logger.info(f"DIST: {self.dist:.3f} STEPSIZE: {self.stepsize:.3f}") @@ -182,7 +191,7 @@ def grow(self) -> None: step-size of each other. """ if self.output is not None: - self.output._ensure_iteration_header(self.iteration + 1) + self.output._ensure_iteration_header(self.iteration + 1, self.dist) r_atoms = self.r_string[-1] p_atoms = self.p_string[-1] @@ -207,8 +216,12 @@ def grow(self) -> None: self.growing = False return + if self.output is not None: + self.output.write_current_frontier_node("r", r_atoms) + r_prev = r_xyz.copy().reshape(-1, 3) r_idx = 1 + r_s = 0.0 for qtarget in string[1:-1]: r_next = interp.coords.x(r_prev, qtarget) _, r_next = project_trans_rot(r_xyz.reshape(-1, 3), r_next) @@ -236,14 +249,18 @@ def grow(self) -> None: self.r_tangent += [normalize(dxds)] self.r_nnodes = len(self.r_string) if self.output is not None: - self.output.write_frontier_node("r", r_frontier, self.dist) + self.output.write_frontier_node("r", r_frontier, r_s) if self.dist <= 2 * self.stepsize: self.growing = False return + if self.output is not None: + self.output.write_current_frontier_node("p", p_atoms) + p_prev = p_xyz.copy().reshape(-1, 3) p_idx = 1 + p_s = 0.0 for qtarget in string[1:-1][::-1]: p_next = interp.coords.x(p_prev, qtarget) _, p_next = project_trans_rot(p_xyz.reshape(-1, 3), p_next) @@ -271,7 +288,7 @@ def grow(self) -> None: self.p_tangent += [normalize(dxds)] self.p_nnodes = len(self.p_string) if self.output is not None: - self.output.write_frontier_node("p", p_frontier, self.dist) + self.output.write_frontier_node("p", p_frontier, p_s) else: string = interp() @@ -285,6 +302,10 @@ def grow(self) -> None: r_idx = np.abs(s - self.stepsize).argmin() p_idx = np.abs(s - (s[-1] - self.stepsize)).argmin() + + if self.output is not None: + self.output.write_current_frontier_node("r", r_atoms) + r_frontier = self.atoms.copy() r_frontier.set_positions(string[r_idx].reshape(-1, 3)) @@ -294,12 +315,15 @@ def grow(self) -> None: self.r_tangent += [normalize(cs(s[r_idx], 1))] self.r_nnodes = len(self.r_string) if self.output is not None: - self.output.write_frontier_node("r", r_frontier, self.dist) + self.output.write_frontier_node("r", r_frontier, float(s[r_idx])) if self.dist <= 2 * self.stepsize: self.growing = False return + if self.output is not None: + self.output.write_current_frontier_node("p", p_atoms) + p_frontier = self.atoms.copy() p_frontier.set_positions(string[p_idx].reshape(-1, 3)) @@ -309,7 +333,7 @@ def grow(self) -> None: self.p_tangent += [normalize(cs(s[p_idx], 1))] self.p_nnodes = len(self.p_string) if self.output is not None: - self.output.write_frontier_node("p", p_frontier, self.dist) + self.output.write_frontier_node("p", p_frontier, float(s[-1] - s[p_idx])) def optimize(self, optimizer: Any) -> None: """Relax all unfixed frontier nodes perpendicular to the local tangent. @@ -333,44 +357,44 @@ def optimize(self, optimizer: Any) -> None: energy = optimizer.calc.get_potential_energy(self.r_string[i]) self.r_energy[i] = float_check(energy) if self.output is not None: - self.output.write_optimized_node("r", i, self.r_string[i], self.r_energy[i], 0) + self.output.write_optimized_node("r", i, self.r_string[i], self.r_energy[i], 0, 0) elif not self.r_fix[i]: assert self.r_tangent[i] is not None atoms = self.r_string[i] try: - atoms, energy, ngrad = optimizer.optimize(atoms, self.r_tangent[i]) + atoms, energy, nfev, nit = optimizer.optimize(atoms, self.r_tangent[i]) self.r_string[i] = atoms self.r_energy[i] = float_check(energy) except Exception: energy = optimizer.calc.get_potential_energy(atoms) self.r_energy[i] = float_check(energy) - ngrad = 0 + nfev, nit = 0, 0 self.r_fix[i] = True - self.ngrad += ngrad + self.ngrad += nfev if self.output is not None: - self.output.write_optimized_node("r", i, self.r_string[i], self.r_energy[i], ngrad) + self.output.write_optimized_node("r", i, self.r_string[i], self.r_energy[i], nfev, nit) for i in range(self.p_nnodes): if self.p_energy[i] is None and self.p_fix[i]: energy = optimizer.calc.get_potential_energy(self.p_string[i]) self.p_energy[i] = float_check(energy) if self.output is not None: - self.output.write_optimized_node("p", i, self.p_string[i], self.p_energy[i], 0) + self.output.write_optimized_node("p", i, self.p_string[i], self.p_energy[i], 0, 0) elif not self.p_fix[i]: assert self.p_tangent[i] is not None atoms = self.p_string[i] try: - atoms, energy, ngrad = optimizer.optimize(atoms, self.p_tangent[i]) + atoms, energy, nfev, nit = optimizer.optimize(atoms, self.p_tangent[i]) self.p_string[i] = atoms self.p_energy[i] = float_check(energy) except Exception: energy = optimizer.calc.get_potential_energy(atoms) self.p_energy[i] = float_check(energy) - ngrad = 0 + nfev, nit = 0, 0 self.p_fix[i] = True - self.ngrad += ngrad + self.ngrad += nfev if self.output is not None: - self.output.write_optimized_node("p", i, self.p_string[i], self.p_energy[i], ngrad) + self.output.write_optimized_node("p", i, self.p_string[i], self.p_energy[i], nfev, nit) self.dist = distance(self.r_string[-1].get_positions().flatten(), self.p_string[-1].get_positions().flatten()) diff --git a/src/mlfsm/opt.py b/src/mlfsm/opt.py index 99d434a..b0c3951 100644 --- a/src/mlfsm/opt.py +++ b/src/mlfsm/opt.py @@ -100,8 +100,8 @@ def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, i Returns ------- - tuple[ASE.Atoms,float,int]: ASE.Atoms with final positions, energy of final structure, and number - of gradient calculations used by optimization. + tuple[ASE.Atoms,float,int,int]: ASE.Atoms with final positions, energy, total function + evaluations (nfev), and number of optimizer iterations (nit). """ xyz = atoms.get_positions().flatten() config = { @@ -119,7 +119,7 @@ def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, i res = minimize(**config) atomsf = atoms.copy() atomsf.set_positions(res.x.reshape(-1, 3)) - return atomsf, res.fun, res.njev + return atomsf, res.fun, res.nfev, res.nit @dataclass @@ -220,8 +220,8 @@ def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, i Returns ------- - tuple[ASE.Atoms,float,int]: ASE.Atoms with final positions, energy of final structure, and number - of gradient calculations used by optimization. + tuple[ASE.Atoms,float,int,int]: ASE.Atoms with final positions, energy, total function + evaluations (nfev), and number of optimizer iterations (nit). """ assert self.coordsobj is not None, "Coordsobj must be initialized" @@ -241,4 +241,4 @@ def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, i atomsf = atoms.copy() atomsf.set_positions(xf) - return atomsf, res.fun, res.njev + return atomsf, res.fun, res.nfev, res.nit diff --git a/src/mlfsm/output.py b/src/mlfsm/output.py index 4574c16..feb55bd 100644 --- a/src/mlfsm/output.py +++ b/src/mlfsm/output.py @@ -17,6 +17,29 @@ _SEP = "=" * 70 _SEP_THIN = "-" * 70 +_CREDIT = """\ + Developed and maintained by: + The Gomes Research Group + University of Iowa, Department of Chemical and Biochemical Engineering + + Contributors: Jonah Marks, Jonathon Vandezande, Joe Gomes\ +""" + +_CITATION = """\ + If you use ML-FSM in your research, please cite: + Marks, Jonah, and Joseph Gomes. "Incorporation of Internal Coordinates + Interpolation into the Freezing String Method." Journal of Chemical + Theory and Computation 21.23 (2025): 12110-12120. + + Additionally, please consider citing: + Marks, Jonah, Jonathon Vandezande, and Joseph Gomes. + "Reliable and Efficient Automated Transition-State Searches with + Machine-Learned Interatomic Potentials." + arXiv preprint arXiv:2604.00405 (2026).\ +""" + +_ATOMS_HEADER = f" {'Sym':<4s} {'X':>12s} {'Y':>12s} {'Z':>12s}" + def _write_section(f: TextIO, title: str) -> None: f.write(f"\n {_SEP}\n") @@ -29,12 +52,19 @@ def _format_atoms_block(atoms: Atoms, indent: str = " ") -> str: symbols = atoms.get_chemical_symbols() positions = atoms.get_positions() lines = [ - f"{indent}{i + 1:4d} {sym:<3s} {pos[0]:12.6f} {pos[1]:12.6f} {pos[2]:12.6f}" - for i, (sym, pos) in enumerate(zip(symbols, positions, strict=True)) + f"{indent}{sym:<4s} {pos[0]:12.6f} {pos[1]:12.6f} {pos[2]:12.6f}" + for sym, pos in zip(symbols, positions, strict=True) ] return "\n".join(lines) +def _write_atoms(f: TextIO, atoms: Atoms) -> None: + """Write a coordinate block with header to f.""" + f.write(f"\n{_ATOMS_HEADER}\n") + f.write(_format_atoms_block(atoms)) + f.write("\n") + + def _chemical_formula(atoms: Atoms) -> str: counts: Counter[str] = Counter(atoms.get_chemical_symbols()) order = ["C", "H"] + sorted(k for k in counts if k not in ("C", "H")) @@ -107,7 +137,7 @@ def __init__(self, outdir: Path | str, filename: str = "fsm.out") -> None: self._path = Path(outdir) / filename self._f: TextIO = self._path.open("w", encoding="utf-8") self._current_iteration: int = 0 - self._node_lines: list[str] = [] + self._optimizing_written: bool = False def close(self) -> None: """Flush and close the output file.""" @@ -119,7 +149,7 @@ def close(self) -> None: # ------------------------------------------------------------------ def write_header(self, version: str) -> None: - """Write the banner and timestamp.""" + """Write the banner, credit block, and timestamp.""" f = self._f f.write(f" {_SEP}\n") title = "ML-FSM: Machine Learning Freezing String Method" @@ -128,7 +158,8 @@ def write_header(self, version: str) -> None: ver_line = f"Version {version}" pad2 = (68 - len(ver_line)) // 2 f.write(f" {' ' * pad2}{ver_line}\n") - f.write(f" {_SEP}\n") + f.write(f" {_SEP}\n\n") + f.write(f"{_CREDIT}\n") now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") f.write(f"\n Date/Time: {now}\n") f.flush() @@ -159,7 +190,7 @@ def write_parameters( if stepsize > 0.0: f.write(f" Step size (explicit) : {stepsize:.4f} Å\n") else: - f.write(f" Step size (explicit) : derived from target node count\n") + f.write(f" Step size : derived from target node count\n") f.flush() def write_system_info( @@ -204,61 +235,99 @@ def write_initial_structures(self, reactant: Atoms, product: Atoms) -> None: """Write reactant and product coordinate blocks.""" f = self._f _write_section(f, "INITIAL STRUCTURES") - f.write("\n Standard Orientation — Reactant (Angstroms)\n") - f.write(f" {_SEP_THIN}\n") - f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") - f.write(f" {_SEP_THIN}\n") - f.write(_format_atoms_block(reactant)) - f.write("\n") - f.write(f"\n Standard Orientation — Product (Angstroms)\n") + f.write("\n Reactant (Angstroms)\n") f.write(f" {_SEP_THIN}\n") - f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") + _write_atoms(f, reactant) + + f.write(f"\n Product (Angstroms)\n") f.write(f" {_SEP_THIN}\n") - f.write(_format_atoms_block(product)) - f.write("\n") + _write_atoms(f, product) f.flush() - def write_path_init(self, dist: float, stepsize: float, nnodes_min: int) -> None: - """Write path initialization summary.""" + def write_path_init( + self, + dist: float, + stepsize: float, + nnodes_min: int, + coordsobj: Optional[Any] = None, + ) -> None: + """Write path initialization summary and optional internal coordinate set.""" f = self._f _write_section(f, "PATH INITIALIZATION") f.write(f"\n Total path distance : {dist:.4f} Å\n") f.write(f" Step size : {stepsize:.4f} Å\n") f.write(f" Target node count : {nnodes_min}\n") + if coordsobj is not None: + self._write_coords_section(f, coordsobj) f.flush() + def _write_coords_section(self, f: TextIO, coordsobj: Any) -> None: + keys: list[str] = coordsobj.keys + type_counts: Counter[str] = Counter() + for k in keys: + if "linearbnd" in k: + type_counts["Linear bends"] += 1 + elif "bond" in k or "stre" in k: + type_counts["Stretches"] += 1 + elif "bend" in k: + type_counts["Bends"] += 1 + elif "tors" in k: + type_counts["Torsions"] += 1 + elif "oop" in k: + type_counts["Out-of-plane bends"] += 1 + else: + type_counts["Other"] += 1 + f.write(f"\n Internal coordinate set : {len(keys)} coordinates\n") + for name, count in type_counts.items(): + f.write(f" {name:<22s}: {count}\n") + # ------------------------------------------------------------------ # Per-iteration sections (called from cos.py hooks) # ------------------------------------------------------------------ - def _ensure_iteration_header(self, iteration: int) -> None: + def _ensure_iteration_header(self, iteration: int, frontier_dist: float) -> None: if iteration != self._current_iteration: self._current_iteration = iteration - self._node_lines = [] + self._optimizing_written = False _write_section(self._f, f"ITERATION {iteration}") - self._f.write("\n") + self._f.write(f"\n Frontier distance : {frontier_dist:.4f} Å\n") + self._f.flush() - def write_frontier_node(self, side: str, atoms: Atoms, dist: float) -> None: - """Write the newly selected frontier node structure. + def write_current_frontier_node(self, side: str, atoms: Atoms) -> None: + """Write the current frontier node geometry before interpolation.""" + f = self._f + label = "Reactant" if side == "r" else "Product" + f.write(f"\n Current {label} Frontier Node:\n") + _write_atoms(f, atoms) + f.flush() + + def write_frontier_node( + self, + side: str, + atoms: Atoms, + actual_dist: float, + ) -> None: + """Write the interpolated frontier node structure. Parameters ---------- side : {"r", "p"} Which end of the string. atoms : Atoms - The frontier node geometry. - dist : float - Current distance between frontier nodes. + The interpolated frontier node geometry. + actual_dist : float + Actual Cartesian step distance from the frontier node to the + selected interpolated structure (may differ from target stepsize). """ f = self._f - label = "Reactant-side frontier" if side == "r" else "Product-side frontier" - f.write(f" {_SEP_THIN}\n") - f.write(f" {label} (frontier distance = {dist:.4f} Å)\n") - f.write(f" {_SEP_THIN}\n") - f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") - f.write(_format_atoms_block(atoms)) - f.write("\n") + label = "Reactant-side" if side == "r" else "Product-side" + f.write(f"\n Interpolating...\n") + f.write( + f"\n {label} interpolated structure" + f" (actual step: {actual_dist:.4f} Å from frontier node):\n" + ) + _write_atoms(f, atoms) f.flush() def write_optimized_node( @@ -267,12 +336,10 @@ def write_optimized_node( idx: int, atoms: Atoms, energy: Optional[float], - ngrad: int, + nfev: int, + nit: int, ) -> None: - """Record an optimized (or endpoint-evaluated) node. - - Called from within ``FreezingString.optimize()``. Results are - buffered and flushed together with the iteration summary. + """Write an optimized (or endpoint-evaluated) node with its structure. Parameters ---------- @@ -284,26 +351,26 @@ def write_optimized_node( Final geometry after optimization. energy : float or None Energy in eV. - ngrad : int - Number of gradient calls used (0 for endpoint-only evaluation). + nfev : int + Total function evaluations (true gradient call count); 0 for endpoint. + nit : int + Number of optimizer iterations; 0 for endpoint. """ + f = self._f + if not self._optimizing_written: + f.write("\n Optimizing...\n") + self._optimizing_written = True + tag = f"{side}[{idx}]" - kind = "endpoint " if ngrad == 0 else "optimized" - if energy is not None: - e_str = f"{energy:+.6f} eV" + kind = "endpoint" if nfev == 0 else "optimized" + e_str = f"{energy:+.6f} eV" if energy is not None else "N/A" + if nfev > 0: + nls = max(0, nfev - nit) + grad_str = f" nfev = {nfev} (nit = {nit}, nls = {nls})" else: - e_str = "N/A" - grad_str = "" if ngrad == 0 else f" ngrad = {ngrad}" - self._node_lines.append(f" {tag:<8s} {kind} : energy = {e_str}{grad_str}") - - f = self._f - f.write(f" {_SEP_THIN}\n") - f.write(f" Optimized node {tag} ({kind})\n") - f.write(f" {_SEP_THIN}\n") - f.write(f" Energy: {e_str}{grad_str}\n") - f.write(f" {'Idx':>4s} {'Sym':<3s} {'X':>12s} {'Y':>12s} {'Z':>12s}\n") - f.write(_format_atoms_block(atoms)) - f.write("\n") + grad_str = "" + f.write(f"\n {tag} ({kind}): energy = {e_str}{grad_str}\n") + _write_atoms(f, atoms) f.flush() def write_iteration_summary( @@ -313,7 +380,7 @@ def write_iteration_summary( p_energies: list[Optional[float]], dist: float, ) -> None: - """Write per-iteration energy table and distance. + """Write per-iteration energy table. Called from ``FreezingString.write()`` after the XYZ file is written. """ @@ -325,7 +392,7 @@ def write_iteration_summary( e_min = min(valid) f.write(f"\n {_SEP_THIN}\n") - f.write(f" Iteration {iteration} summary (frontier distance = {dist:.4f} Å)\n") + f.write(f" Iteration {iteration} energy summary (frontier distance = {dist:.4f} Å)\n") f.write(f" {_SEP_THIN}\n") f.write(f" {'Node':<8s} {'Side':<8s} {'Energy (eV)':>14s} {'Rel. Energy (eV)':>18s}\n") f.write(f" {_SEP_THIN}\n") @@ -352,19 +419,22 @@ def write_iteration_summary( # ------------------------------------------------------------------ def write_final_summary(self, string: "FreezingString") -> None: - """Write TS guess identification and full string energy profile.""" + """Write the full optimized string, TS guess, and citation block.""" f = self._f - _write_section(f, "CALCULATION COMPLETE") - - all_energies = string.r_energy + string.p_string[::-1] # type: ignore[operator] all_energies = string.r_energy + string.p_energy[::-1] + all_atoms = string.r_string + string.p_string[::-1] valid_pairs = [(i, e) for i, e in enumerate(all_energies) if e is not None] + # ------------------------------------------------------------------ + # Stats + # ------------------------------------------------------------------ + _write_section(f, "CALCULATION COMPLETE") f.write(f"\n Total iterations : {string.iteration}\n") - f.write(f" Total gradient calls : {string.ngrad}\n\n") + f.write(f" Total gradient calls : {string.ngrad}\n") if not valid_pairs: - f.write(" No energies available.\n") + f.write("\n No energies available.\n") + self._write_citation() f.flush() return @@ -382,31 +452,57 @@ def write_final_summary(self, string: "FreezingString") -> None: p_idx = len(all_energies) - 1 - ts_global_idx ts_label = f"p[{p_idx}]" - f.write(f" {_SEP_THIN}\n") - f.write(f" TS Guess: node {ts_label} (highest-energy node)\n") + f.write(f"\n TS Guess: node {ts_label} (highest-energy node)\n") f.write(f" Absolute energy : {ts_energy:+.6f} eV\n") f.write(f" Relative energy : {ts_energy - e_min:+.4f} eV (above string minimum)\n") - f.write(f" {_SEP_THIN}\n\n") - f.write(f" Full String Energies (relative to minimum, eV)\n") - f.write(f" {_SEP_THIN}\n") - f.write(f" {'Node':<8s} {'Side':<8s} {'Energy (eV)':>14s} {'Rel. Energy (eV)':>18s}\n") - f.write(f" {_SEP_THIN}\n") + # ------------------------------------------------------------------ + # Full string — every node with coordinates and energy + # ------------------------------------------------------------------ + _write_section(f, "FULL OPTIMIZED STRING") + f.write( + "\n The complete optimized string is shown below.\n" + " Each node is listed with its energy and atomic coordinates (Angstroms).\n" + ) for local_i, (global_i, e) in enumerate(valid_pairs): if global_i < nr: - side = "R" if global_i == 0 else "r" + if global_i == 0: + role = " — Reactant" + else: + role = "" else: p_pos = len(all_energies) - 1 - global_i - side = "P" if p_pos == 0 else "p" - ts_marker = " <-- TS guess" if local_i == ts_local else "" - e_str = f"{e:+.6f}" - rel_str = f"{e - e_min:+.4f}" - f.write( - f" {local_i + 1:<8d} {side:<8s} {e_str:>14s} {rel_str:>18s}{ts_marker}\n" - ) + role = " — Product" if p_pos == 0 else "" + + is_ts = local_i == ts_local + ts_tag = " *** TS Guess ***" if is_ts else "" + + f.write(f"\n Node {local_i + 1}{role}{ts_tag}\n") + f.write(f" {'Energy (abs)':<18s}: {e:+.6f} eV\n") + f.write(f" {'Energy (rel)':<18s}: {e - e_min:+.4f} eV\n") + _write_atoms(f, all_atoms[global_i]) + + # ------------------------------------------------------------------ + # TS guess standalone section + # ------------------------------------------------------------------ + _write_section(f, "TRANSITION STATE GUESS") + f.write( + f"\n Node {ts_local + 1} ({ts_label}) is identified as the TS guess\n" + f" based on being the highest-energy node along the optimized string.\n" + ) + f.write(f"\n Energy (absolute) : {ts_energy:+.6f} eV\n") + f.write(f" Energy (relative) : {ts_energy - e_min:+.4f} eV (above string minimum)\n") + _write_atoms(f, all_atoms[ts_global_idx]) + + # ------------------------------------------------------------------ + # Citation + # ------------------------------------------------------------------ + self._write_citation() + f.flush() - f.write(f" {_SEP_THIN}\n") - f.write(f"\n Full string written to: vfile_{string.iteration:02d}.xyz\n") + def _write_citation(self) -> None: + f = self._f + _write_section(f, "CITATION") + f.write(f"\n{_CITATION}\n") f.write(f"\n {_SEP}\n") - f.flush() From cbab68aa583753439e6817429ec506e571fad2f2 Mon Sep 17 00:00:00 2001 From: Jonah Marks Date: Sun, 26 Apr 2026 22:52:03 +0200 Subject: [PATCH 3/4] linting --- src/mlfsm/output.py | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/src/mlfsm/output.py b/src/mlfsm/output.py index feb55bd..5e58473 100644 --- a/src/mlfsm/output.py +++ b/src/mlfsm/output.py @@ -8,10 +8,11 @@ from typing import TYPE_CHECKING, Any, Optional, TextIO import numpy as np -from ase import Atoms -from numpy.typing import NDArray if TYPE_CHECKING: + from ase import Atoms + from numpy.typing import NDArray + from mlfsm.cos import FreezingString _SEP = "=" * 70 @@ -67,7 +68,7 @@ def _write_atoms(f: TextIO, atoms: Atoms) -> None: def _chemical_formula(atoms: Atoms) -> str: counts: Counter[str] = Counter(atoms.get_chemical_symbols()) - order = ["C", "H"] + sorted(k for k in counts if k not in ("C", "H")) + order = ["C", "H", *sorted(k for k in counts if k not in ("C", "H"))] parts = [] for sym in order: if sym in counts: @@ -190,7 +191,7 @@ def write_parameters( if stepsize > 0.0: f.write(f" Step size (explicit) : {stepsize:.4f} Å\n") else: - f.write(f" Step size : derived from target node count\n") + f.write(" Step size : derived from target node count\n") f.flush() def write_system_info( @@ -240,7 +241,7 @@ def write_initial_structures(self, reactant: Atoms, product: Atoms) -> None: f.write(f" {_SEP_THIN}\n") _write_atoms(f, reactant) - f.write(f"\n Product (Angstroms)\n") + f.write("\n Product (Angstroms)\n") f.write(f" {_SEP_THIN}\n") _write_atoms(f, product) f.flush() @@ -322,11 +323,8 @@ def write_frontier_node( """ f = self._f label = "Reactant-side" if side == "r" else "Product-side" - f.write(f"\n Interpolating...\n") - f.write( - f"\n {label} interpolated structure" - f" (actual step: {actual_dist:.4f} Å from frontier node):\n" - ) + f.write("\n Interpolating...\n") + f.write(f"\n {label} interpolated structure (actual step: {actual_dist:.4f} Å from frontier node):\n") _write_atoms(f, atoms) f.flush() From 56a2e37216013fbe7fb855316943d5bfd07ffd20 Mon Sep 17 00:00:00 2001 From: Jonah Marks Date: Sun, 26 Apr 2026 23:06:04 +0200 Subject: [PATCH 4/4] fix mypy errors introduced by output file additions --- src/mlfsm/cos.py | 7 +++---- src/mlfsm/opt.py | 4 ++-- src/mlfsm/output.py | 2 +- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/src/mlfsm/cos.py b/src/mlfsm/cos.py index 08e2033..7031f56 100644 --- a/src/mlfsm/cos.py +++ b/src/mlfsm/cos.py @@ -13,7 +13,7 @@ from mlfsm.output import FSMOutput -from mlfsm.coords import Cartesian +from mlfsm.coords import Cartesian, Redundant from mlfsm.geom import ( calculate_arc_length, distance, @@ -115,11 +115,10 @@ def __init__( self.stepsize = float(stepsize) self.nnodes_min = int(self.dist / self.stepsize) + self.init_coordsobj: Optional[Redundant] = None if interp_method == "ric": self.init_coordsobj = ( - _interp_init.coords # type: ignore[union-attr] - if isinstance(_interp_init, RIC) - else RIC(reactant, product, ninterp=2).coords + _interp_init.coords if isinstance(_interp_init, RIC) else RIC(reactant, product, ninterp=2).coords ) else: self.init_coordsobj = None diff --git a/src/mlfsm/opt.py b/src/mlfsm/opt.py index b0c3951..fa28327 100644 --- a/src/mlfsm/opt.py +++ b/src/mlfsm/opt.py @@ -91,7 +91,7 @@ def obj(self, xyz: NDArray[Any], tangent: NDArray[Any], atoms: Atoms) -> tuple[f pgrads = proj @ grads return energy, pgrads - def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, int]: + def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, int, int]: """Run optimization in Cartesian coordinates using user specified method. Args: @@ -211,7 +211,7 @@ def obj( return energy, pgrads - def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, int]: + def optimize(self, atoms: Atoms, tangent: NDArray[Any]) -> tuple[Atoms, float, int, int]: """Run optimization in internal coordinates using user specified method. Args: diff --git a/src/mlfsm/output.py b/src/mlfsm/output.py index 5e58473..6592c1f 100644 --- a/src/mlfsm/output.py +++ b/src/mlfsm/output.py @@ -115,7 +115,7 @@ def get_calculator_info(calc: Any) -> dict[str, Any]: except Exception: pass try: - ckpt = calc.predictor.checkpoint_path # type: ignore[union-attr] + ckpt = calc.predictor.checkpoint_path info["checkpoint"] = str(ckpt) except Exception: pass