"""Load infinite-swapping paths from disk into :class:`.Path` objects.
These loaders were historically part of the inf-flavour ``Path`` module
(``pyretis.core._path_inf``). After the A3.1b Path collapse there is a
single :class:`pyretis.core.path.Path`, so the loaders live here, free of
the now-deleted ``_path_inf`` module. They build native ``Path`` objects
whose phasepoints are file-backed snapshot :class:`.System` objects (the
external-engine representation), exactly as the infinite-swapping
scheduler expects.
"""
from __future__ import annotations
import logging
import os
from typing import Any, Dict, List
from pyretis.core.path import Path
from pyretis.core.system_core import System
from pyretis.inout.formats.energy import EnergyPathFile
from pyretis.inout.formats.order import OrderPathFile
from pyretis.inout.formats.path import PathExtFile
logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())
[docs]
def load_path(pdir: str) -> Path:
"""Load a path from the given directory."""
trajtxt = os.path.join(pdir, "traj.txt")
ordertxt = os.path.join(pdir, "order.txt")
if not os.path.isfile(trajtxt):
raise FileNotFoundError(trajtxt)
if not os.path.isfile(ordertxt):
raise FileNotFoundError(ordertxt)
# load trajtxt
with PathExtFile(trajtxt, "r") as trajfile:
# Just get the first trajectory:
traj = next(trajfile.load())
# Update trajectory to use full path names. Resolve to an ABSOLUTE
# path: external engines re-read the trajectory file at propagation
# time from the per-worker exe_dir (e.g. "worker0/"), so a path
# relative to the load dir ("load/1/accepted/...") would not resolve
# there. Internal engines read at load time and are unaffected.
for i, snapshot in enumerate(traj["data"]):
traj["data"][i][1] = os.path.abspath(
os.path.join(pdir, "accepted", snapshot[1])
)
traj["data"][i][2] = int(snapshot[2])
traj["data"][i][3] = int(snapshot[3]) == -1
for config in set(frame[1] for frame in traj["data"]):
if not os.path.isfile(config):
raise FileNotFoundError(config)
# load ordertxt
with OrderPathFile(ordertxt, "r") as orderfile:
orderblock = next(orderfile.load())
orderdata = orderblock["data"][:, 1:]
path = Path()
# Recover the move that generated this path from the order.txt comment
# header (``# Cycle: N, status: S, move: M``). The path formatters
# persist the full ``generated`` tuple there, so a path reloaded from
# disk keeps the move (sh/wf/ss/tr/...) that produced it instead of a
# synthetic load/restart tag. ``load_paths_from_disk`` decides when to
# honour this (a restart must report the original move; a fresh load
# keeps its own load tag).
path.generated = _parse_move_comment(orderblock["comment"])
for snapshot, order in zip(traj["data"], orderdata):
frame = System()
frame.order = order
frame.config = (snapshot[1], snapshot[2])
frame.vel_rev = snapshot[3]
path.phasepoints.append(frame)
_load_energies_for_path(path, pdir)
# Known limitation: the loaded path is NOT run through the validity
# check that the native load route applies via
# ``pyretis.initiation.initiate_load._check_path`` (start/end interface
# and middle-crossing test -> accept, status). That check needs a
# :class:`.PathEnsemble` (its ``interfaces``, ``start_condition`` and
# ``ensemble_name``), which is not available here: ``load_path`` is
# given only a directory, and the ensembles are built later in
# ``simulation.setup`` (after this function returns). It is currently
# safe to skip because the infinite-swapping scheduler does not gate a
# loaded path on a per-ensemble accept/reject status; instead
# ``InfSwapState.load_paths`` assigns the path its per-ensemble
# occupancy weights (``calc_cv_vector`` / ``compute_weight``), so a path
# that does not cross a given interface simply receives zero weight
# there. A real validation pass would have to thread a PathEnsemble (or
# the bare interface tuple + start condition) into this loader and
# decide what to do with a rejected initial path on the inf route.
return path
[docs]
def _parse_move_field(token: str) -> Any:
"""Convert one ``repr``-formatted move-tuple token to its scalar value.
Parameters
----------
token : str
A single, already-stripped element of the move tuple, e.g.
``"'sh'"``, ``"-0.2478"``, ``"2"`` or ``"nan"``.
Returns
-------
object
The string (quotes removed), float or int the token represents.
Integers keep their ``int`` type so the recovered tuple's ``repr``
matches the persisted one exactly.
"""
if token.startswith("'") and token.endswith("'"):
return token[1:-1]
if token.startswith('"') and token.endswith('"'):
return token[1:-1]
if token == "nan":
return float("nan")
if token.lstrip("+-").isdigit():
return int(token)
return float(token)
[docs]
def _load_energies_for_path(path: Path, dirname: str) -> None:
"""Load energy data for a path.
Parameters
----------
path : Path
The path we are to set up/fill.
dirname : str
The path to the directory with the input files.
"""
energy_file_name = os.path.join(dirname, "energy.txt")
try:
with EnergyPathFile(energy_file_name, "r") as energyfile:
energy = next(energyfile.load())
path.update_energies(
energy["data"]["ekin"],
energy["data"]["vpot"],
energy["data"].get("etot", []),
energy["data"].get("temp", []),
)
except FileNotFoundError:
pass
[docs]
def load_paths_from_disk(config: Dict[str, Any]) -> List[Path]:
"""Load paths from disk."""
load_dir = config["simulation"]["load_dir"]
# The authoritative per-path move table a resume restores from (P7.9):
# ``[current] generated`` in output.toml, written by
# ``InfSwapState.write_toml`` from the in-memory ``native_rows``. The
# order.txt-header recovery below it is the fallback for runs restarted
# from an output.toml that predates this table: it is correct for paths
# the scheduler itself archived mid-run, but a still-live INITIAL
# path's load-dir header either lacks the move entirely
# (internal-engine load dirs) or holds the initiation-time tag rather
# than the ``"ld"`` a fresh run assigns in memory only -- exactly the
# gap that made a restarted run's Mc/No.-shoot diverge from the
# continuous run's.
persisted_moves = config["current"].get("generated", {})
paths = []
for pnumber in config["current"]["active"]:
new_path = load_path(os.path.join(load_dir, str(pnumber)))
# ``load_path`` recovers the exact ``generated`` move recorded when
# the path was archived, read back from its order.txt comment header
# (``None`` when the archived move was itself ``None``). On restart
# we keep that move verbatim: the per-ensemble native output applies
# the same ``None -> ('sh', 0.0, 0, 0)`` fallback as an uninterrupted
# run, so the reported move, the per-ensemble shoot counters and the
# O-shoot column all reproduce the continuous run. A fresh load was
# not produced by an MC move in this run, so it keeps an "ld" tag.
if "restarted_from" not in config["current"]:
new_path.generated = ("ld", float("nan"), 0, 0)
else:
persisted = persisted_moves.get(str(pnumber))
if persisted is not None:
new_path.generated = tuple(persisted)
new_path.maxlen = config["simulation"]["tis_set"]["maxlength"]
paths.append(new_path)
# assign pnumber
paths[-1].path_number = pnumber
return paths