"""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 classic ``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.archive_paths import resolve_path_dir
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 validate_path_for_ensemble(path: Path, ensemble: Dict[str, Any]) -> None:
"""Raise when a loaded path does not satisfy an ensemble window.
The start must satisfy the ensemble's ``start_cond`` and the path
must end at an interface. The middle-interface crossing is only
required where it is an OCCUPANT invariant, i.e. for the
single-sided positive windows (where a non-crossing path can only
mean mis-staged input) -- there it catches a path staged into the
wrong ensemble that the start/end checks alone would accept. It is
deliberately NOT required for:
* a window whose paths may start at either end (``start_cond``
covering both ``L`` and ``R``: the REPPTIS/PPTIS windows and the
permeability ``[0^-]`` window on the standard route) -- the
sampler's own ``shoot`` skips the crossing check there too
("paths that can start everywhere"), so LL/RR occupants are
legitimate;
* a right-starting zero window with a genuine midpoint middle (the
REPPTIS-route ``[0^-]`` permeability window) -- kick initiation
legitimately accepts a non-crossing seed there and sampling burns
it in; the historical loader tolerated exactly this.
"""
interfaces = tuple(ensemble["interfaces"])
start, end, _, cross = path.check_interfaces(interfaces)
start_condition = ensemble["start_cond"]
allowed_starts = (
(start_condition,)
if isinstance(start_condition, str)
else tuple(start_condition)
)
middle_required = not (
set(allowed_starts) == {"L", "R"}
or (set(allowed_starts) == {"R"}
and interfaces[1] != interfaces[2])
)
problems = []
if start is None or start not in allowed_starts:
problems.append(
f"starts at {start!r}, expected one of {allowed_starts}"
)
if end not in ("L", "R"):
problems.append(f"ends at {end!r}, expected 'L' or 'R'")
if middle_required and (
cross is None or len(cross) < 2 or not cross[1]):
problems.append("does not cross the middle interface")
if problems:
path_number = getattr(path, "path_number", None)
name = ensemble.get("ens_name", "unknown")
raise ValueError(
f"Loaded path {path_number!r} is invalid for ensemble {name}: "
+ "; ".join(problems)
+ f" (interfaces: {interfaces}, path order"
f" min/max: {path.ordermin[0]:.6g}/{path.ordermax[0]:.6g})"
)
[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-ensemble exe_dir (``NNN/generate/``), so a path
# relative to the load dir would not resolve there. Internal
# engines read at load time and are unaffected. The frames live
# flat in the path directory (``<pdir>/<frame>``); a user-staged
# load dir (the upstream infretis staging contract) keeps them in a
# legacy ``<pdir>/accepted/<frame>`` subdir, so fall back to that
# when the flat file is absent.
for i, snapshot in enumerate(traj["data"]):
flat = os.path.abspath(os.path.join(pdir, snapshot[1]))
legacy = os.path.abspath(
os.path.join(pdir, "accepted", snapshot[1])
)
traj["data"][i][1] = (
legacy if (not os.path.isfile(flat)
and os.path.isfile(legacy)) else flat
)
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:]
trajdata = traj["data"]
if len(trajdata) != len(orderdata):
raise ValueError(
f"Trajectory/order frame count differs in '{pdir}': "
f"{len(trajdata)} != {len(orderdata)}"
)
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(trajdata, orderdata, strict=True):
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)
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")
# Starting from a path that carries no energies is legitimate: a
# sparse or external load supplies coordinates only. The simulation
# continues (the energies are filled in as the path is regenerated),
# but say so explicitly -- the state is worth knowing about. The
# pre-check replaces the FileIO machinery's low-level "Could not
# open" + "I/O error" pair, which reported this expected condition as
# a filesystem failure. A file that exists but cannot be read is a
# real error and still raises, rather than leaving the path with
# silently missing energies.
if not os.path.isfile(energy_file_name):
logger.warning(
'No energy data for the path loaded from "%s"; '
'continuing with undefined energies for these frames.',
dirname,
)
return
with EnergyPathFile(energy_file_name, "r") as energyfile:
# An existing file with no block is a truncated or corrupted
# reference, not the "loaded without energies" case handled
# above, so report it as such instead of letting the bare
# StopIteration from next() escape.
energy = next(energyfile.load(), None)
if energy is None:
raise ValueError(
f'No energy data block in "{energy_file_name}"; the file '
'exists but carries no energies.'
)
path.update_energies(
energy["data"]["ekin"],
energy["data"]["vpot"],
energy["data"].get("etot", []),
energy["data"].get("temp", []),
)
[docs]
def load_paths_from_disk(config: Dict[str, Any]) -> List[Path]:
"""Load paths from disk."""
# 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 ``pathensemble_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", {})
# Birth ensemble per path number, for locating its nested archive dir
# (<ens_save_idx>/<subdir>/<pn>). resolve_path_dir falls back to the
# legacy flat <load_dir>/<pn> when the nested directory is absent: a
# user-staged flat load directory read by a fresh run, or a run whose
# paths were archived before the per-ensemble nesting (with or without
# a stamped ens_save_idx -- a resume records the map without moving
# the files).
ens_map = config["current"].get("ens_save_idx", {})
paths = []
for pnumber in config["current"]["active"]:
new_path = load_path(
resolve_path_dir(config, pnumber, ens_map.get(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 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