#!/usr/bin/env python3
# Copyright (c) 2026, PyRETIS Development Team.
# Distributed under the LGPLv2.1+ License. See LICENSE for more info.
"""pyretisrun - An application for running PyRETIS simulations.
This script is a part of the PyRETIS library and can be used for
running simulations from an input script.
usage: pyretisrun.py [-h] -i INPUT [-V] [-f LOG_FILE] [-l LOG_LEVEL] [-p]
PyRETIS
optional arguments:
-h, --help show this help message and exit
-i INPUT, --input INPUT
Location of PyRETIS input file
-V, --version show program's version number and exit
-f LOG_FILE, --log_file LOG_FILE
Specify log file to write
-l LOG_LEVEL, --log_level LOG_LEVEL
Specify log level for log file
-p, --progress Display a progress meter instead of text output for
the simulation
More information about running PyRETIS can be found at: www.pyretis.org
"""
# pylint: disable=invalid-name
import argparse
import datetime
import logging
import os
import pathlib
import signal
import sys
import traceback
import warnings
# Surface PyRETIS-owned DeprecationWarnings (e.g. the rst input
# deprecation) to CLI users. Python hides DeprecationWarning by
# default unless it comes from __main__; this filter scopes the
# override to pyretis modules only, so warnings from other libraries
# stay at the user's default behaviour.
warnings.filterwarnings('default', category=DeprecationWarning,
module=r'pyretis(\..*)?')
# Other libraries:
import tqdm # noqa: E402 For a progress bar
import colorama # noqa: E402 For coloring text
# PyRETIS library imports:
from pyretis import __version__ as VERSION # noqa: E402
from pyretis.info import PROGRAM_NAME, URL, CITE, LOGO # noqa: E402
from pyretis.core.pathensemble import generate_ensemble_name # noqa: E402
from pyretis.setup import create_simulation # noqa: E402
from pyretis.inout.screen import REFERENCE # noqa: E402
from pyretis.inout.common import ( # noqa: E402
check_python_version,
create_backup,
)
from pyretis.inout.formats.formatter import ( # noqa: E402
get_log_formatter,
setup_console_logging,
)
from pyretis.inout.settings import ( # noqa: E402
dump_toml,
parse_settings_file,
write_settings_file
)
_DATE_FMT = '%d.%m.%Y %H:%M:%S'
logger = setup_console_logging()
[docs]
def use_tqdm(progress):
"""Return a progress bar if we want one.
Parameters
----------
progress : boolean
If True, we should use a progress bar, otherwise not.
Returns
-------
out : object like :py:class:`tqdm.tqdm`
The progress bar, if requested. Otherwise, just a dummy
iterator.
"""
if progress:
pbar = tqdm.tqdm
else:
def empty_tqdm(*args, **kwargs):
"""Return an iterator to replace tqdm."""
if args:
return args[0]
return kwargs.get('iterable', None)
pbar = empty_tqdm
return pbar
[docs]
def hello_world(infile, rundir, logfile):
"""Print out a politically correct greeting for PyRETIS.
Parameters
----------
infile : string
String showing the location of the input file.
rundir : string
String showing the location we are running in.
logfile : string
The output log file
"""
timestart = datetime.datetime.now().strftime(_DATE_FMT)
pyversion = sys.version.split()[0]
logger.banner('\n'.join([LOGO]))
logger.banner('%s version: %s', PROGRAM_NAME, VERSION)
logger.banner('Start of execution: %s', timestart)
logger.banner('Python version: %s', pyversion)
logger.progress('\nRunning in directory: %s', rundir)
logger.progress('Input file: %s', infile)
logger.progress('Log file: %s', logfile)
[docs]
def bye_bye_world():
"""Print out the goodbye message for PyRETIS."""
timeend = datetime.datetime.now().strftime(_DATE_FMT)
msgtxt = f'End of {PROGRAM_NAME} execution: {timeend}'
logger.progress(msgtxt)
# display some references:
references = [f'{PROGRAM_NAME} references:']
references.append(('-')*len(references[0]))
for line in CITE.split('\n'):
if line:
references.append(line)
reftxt = '\n'.join(references)
logger.log(REFERENCE, '\n%s', reftxt)
urltxt = str(URL)
logger.log(REFERENCE, urltxt)
[docs]
def run_md_flux_simulation(sim, sim_settings, progress=False):
"""Run a MD-FLUX simulation.
Parameters
----------
sim : object like :py:class:`.Simulation`
This is the simulation to run.
sim_settings : dict
The simulation settings.
progress : boolean, optional
If True, we will display a progress bar, otherwise, we print
results to the screen.
"""
logger.progress('Starting MD-Flux simulation')
tqd = use_tqdm(progress)
sim.engine.exe_dir = sim_settings['simulation']['exe_path']
sim.set_up_output(sim_settings, progress=progress)
for _ in tqd(sim.run(), initial=sim.cycle['startcycle'],
total=sim.cycle['endcycle'], desc='MD-flux'):
pass
# Write final restart file:
sim.write_restart(now=True)
return True
[docs]
def run_md_simulation(sim, sim_settings, progress=False):
"""Run a MD simulation.
Parameters
----------
sim : object like :py:class:`.Simulation`
This is the simulation to run.
sim_settings : dict
The simulation settings.
progress : boolean, optional
If True, we will display a progress bar, otherwise, we print
results to the screen.
"""
logger.progress('Starting MD simulation')
tqd = use_tqdm(progress)
sim.engine.exe_dir = sim_settings['simulation']['exe_path']
sim.set_up_output(sim_settings, progress=progress)
for _ in tqd(sim.run(), initial=sim.cycle['startcycle'],
total=sim.cycle['endcycle'], desc='MD step'):
pass
# Write final restart file:
sim.write_restart(now=True)
return True
[docs]
def make_tis_files(_, settings, progress=False):
"""Create TIS simulations input files PyRETIS.
It just writes out input files for single TIS simulations and
exit without running a simulation.
Parameters
----------
settings : list of dicts or Simulation objects
The settings for the simulations.
"""
_ = progress
logtxt = 'Input settings requests: TIS for multiple path ensembles.'
logger.progress(logtxt)
logtxt = 'Will create input files for the TIS simulations and exit'
logger.progress(logtxt)
i_ens = 0
for i, ens_settings in enumerate(settings['ensemble']):
i_ens += 1
if i == 0 and not settings['simulation']['zero_ensemble']:
i_ens += 1
ens_settings['simulation']['zero_ensemble'] = False
ens_settings['simulation']['task'] = 'tis'
# Drop the per-ensemble scalar ``interface`` that create_ensembles
# leaves at the top level: it is redundant with
# ``[simulation] interfaces`` (the [left, middle, right] triple) and,
# as a stray top-level float, it breaks the TOML settings finalisers
# (``len()`` / mapping access on a scalar). Harmless on the legacy
# .rst round trip; must go for the .toml the scheduler route reads.
ens_settings.pop('interface', None)
ensf = generate_ensemble_name(i_ens)
logtxt = f'Creating input for TIS ensemble: {i_ens} '
logger.progress(logtxt)
# Emit per-ensemble TIS configs as TOML: the legacy .rst format is
# deprecated AND a .rst path-sampling input is not routed to the
# infinite-swapping scheduler, so it would hit the retired native
# in-process loop. A .toml routes through the scheduler like any
# other native path-sampling config.
infile = f'tis-{ensf}.toml'
logtxt = f'Create file: "{infile}"'
logger.progress(logtxt)
exe_dir_file = os.path.join(ens_settings['engine']['exe_path'], infile)
# Ensure generated TIS input files include a human-friendly
# orderparameter name without mutating the caller's settings.
op_section = ens_settings.get('orderparameter', {})
need_restore = False
had_name_key = 'name' in op_section
original_name = op_section.get('name') if had_name_key else None
if not had_name_key or op_section.get('name') is None:
ens_settings.setdefault('orderparameter', {})
ens_settings['orderparameter']['name'] = 'Order Parameter'
need_restore = True
try:
write_settings_file(ens_settings, exe_dir_file, backup=False)
finally:
if need_restore:
if had_name_key:
ens_settings['orderparameter']['name'] = original_name
else:
# remove the temporary key we added
ens_settings['orderparameter'].pop('name', None)
logtxt = 'Command for executing:'
logger.progress(logtxt)
logtxt = f'pyretis run -i {infile} -p -f {ensf}.log'
logger.progress(logtxt)
return True
[docs]
def run_generic_simulation(sim, sim_settings, progress=False):
"""Run a generic PyRETIS simulation.
These are simulations that are just going to complete a given
number of steps. Other simulation may consist of several
simulations tied together and these are NOT handled here.
Parameters
----------
sim : object like :py:class:`.Simulation`
This is the simulation to run.
sim_settings : dict
The simulation settings.
progress : boolean, optional
If True, we will display a progress bar, otherwise, we print
results to the screen.
"""
logtxt = 'Running simulation'
logger.progress(logtxt)
tqd = use_tqdm(progress)
sim.set_up_output(sim_settings, progress=progress)
for _ in tqd(sim.run(), desc='Step'):
pass
# Write final restart file:
sim.write_restart(now=True)
return True
_RUNNERS = {'md-flux': run_md_flux_simulation,
'md-nve': run_md_simulation,
'md': run_md_simulation,
'make-tis-files': make_tis_files}
# The input-TOML task values that select the infinite-swapping
# (replica-exchange) sampler instead of the in-process simulation flow.
_INFINITE_SWAPPING_TASKS = frozenset(
{'infinite_swapping', 'infswap', 'infretis'}
)
# Native path-sampling tasks that are run through the infinite-swapping
# scheduler at n_workers=1 (their config is translated by
# native_to_infswap_config). ALL of retis / tis / explore / pptis / repptis
# route through the scheduler -- the native in-process loop has been retired
# (PathSimulation.run/step raise). A single-ensemble ``tis`` config (one
# [i^+] with no [0^-]) is handled by native_to_infswap_config (``single_tis``);
# the contract is sampling validity, not byte-identity (see
# is_native_scheduler_config). A native path-sampling config the scheduler
# cannot yet run (a port gap, see native_scheduler_port_gap) has NO execution
# path now and is rejected with a clear error at routing -- it is never
# dispatched to the retired loop.
_NATIVE_SCHEDULER_TASKS = frozenset({'retis', 'tis', 'explore', 'pptis',
'repptis'})
[docs]
def is_infinite_swapping_config(inputfile):
"""Return True if the input selects the infinite-swapping sampler.
The infinite-swapping (replica-exchange) sampler is selected
explicitly via ``[simulation] task = "infinite_swapping"`` (or
``infswap``/``infretis``). A recognised native path-sampling task
(``_NATIVE_SCHEDULER_TASKS``: retis/tis/explore/pptis/repptis) is
NEVER treated as infinite-swapping, even if its ``[runner]`` section
is present -- ``[runner]`` (worker count / multi-engine pools) is
legal native syntax too, translated by
:func:`pyretis.inout.config_adapter.native_to_infswap_config` just
like every other native section. Only an input with NO recognised
``task`` at all falls back to the ``[runner]``-presence heuristic
(the still-supported legacy infretis-flavour dialect, which never
sets ``task``). Only ``.toml`` inputs are considered; the in-process
simulation flow handles everything else.
Parameters
----------
inputfile : string
Path to the input file.
Returns
-------
boolean
True if the infinite-swapping scheduler should run this input.
"""
if os.path.splitext(inputfile)[1].lower() != '.toml':
return False
if not os.path.isfile(inputfile):
# main() calls this before set_up_simulation, and the open() below
# would raise a bare FileNotFoundError on a missing .toml. We
# cannot inspect a file that is not there anyway, so return False:
# main() then routes to set_up_simulation, whose descriptive
# "Input file NOT found!" ValueError is raised inside the
# error-logging try/except -- the same path a missing .rst takes.
return False
try:
import tomllib as _toml
except ImportError:
import tomli as _toml
with open(inputfile, 'rb') as infile:
config = _toml.load(infile)
task = str(config.get('simulation', {}).get('task', '')).lower()
if task in _INFINITE_SWAPPING_TASKS:
return True
if task in _NATIVE_SCHEDULER_TASKS:
return False
return 'runner' in config
[docs]
def is_infretis_flavour_config(inputfile):
"""Return True if ``inputfile`` needs the infretis-flavour pipeline.
The structural marker is ``[simulation.tis_set]`` -- the
infretis-flavour dialect's own placement of its path-sampling knobs
(vs. native's ``[tis]``) -- checked directly rather than inferred
from ``task``: real configs in the validation suite carry an
explicit ``task = "infinite_swapping"`` for unambiguous routing
while still using ``[simulation.tis_set]`` throughout, so ``task``
presence alone does not distinguish the two dialects here (see
MERGE_TODO.md S5.6).
A restart continuation is excluded: it is always invoked by
pointing ``-i`` directly at the scheduler's own ``restart.toml``
(never the original config), which is already in the coordinator's
own shape (its ``[current]`` section is state, not input) and must
not be re-normalized. Both the conventional filename and
``[current]``'s presence are checked, since the restart-continuation
convention in :func:`pyretis.simulation.setup.setup_config` already
depends on the exact ``restart.toml`` name, not just this function.
Parameters
----------
inputfile : string
Path to the input file.
Returns
-------
boolean
True if this is a fresh infretis-flavour config that should be
routed through :func:`run_infretis_flavour_config`.
"""
if os.path.splitext(inputfile)[1].lower() != '.toml':
return False
if not os.path.isfile(inputfile):
return False
if os.path.basename(inputfile) == 'restart.toml':
return False
try:
import tomllib as _toml
except ImportError:
import tomli as _toml
with open(inputfile, 'rb') as infile:
config = _toml.load(infile)
if 'current' in config:
return False
return 'tis_set' in config.get('simulation', {})
[docs]
def scheduler_supported_features(config):
"""Return ``(supported, reason)`` for routing a native retis config.
The infinite-swapping scheduler at ``n_workers = 1`` faithfully
reproduces the native RETIS loop for the kick- (or restart-)
initialised internal-engine RETIS family with the ``sh``/``wt``/``wf``/
``ss`` shooting moves (``wf``/``ss`` via the WHAM ``Cxy/HA`` unweighting
on the native-output route). Several native features are still
scheduler PORT GAPS and must keep the in-process loop until they are
ported, so this helper detects them and reports the first one found:
- shooting moves other than ``sh``/``wt``/``wf``/``ss``.
- the permeability ``mirror`` (``mirror_freq``) and ``target`` swap
(``target_freq``) moves route through the scheduler but only at
``n_workers = 1`` (both persist a global order-function mutation on
accept).
- an ``[initial-path] method`` of ``load`` for non-explore tasks.
- a ``[simulation] restart`` continuation.
Parameters
----------
config : dict
The parsed native TOML configuration.
Returns
-------
(boolean, string)
``(True, '')`` when the scheduler faithfully covers the config;
``(False, reason)`` naming the first unsupported feature.
"""
tis = config.get('tis', {})
moves = tis.get('shooting_moves')
if moves is not None:
# A move the translator + scheduler accept routes faithfully; use
# config_adapter._INF_MOVE_CODES as the single source of truth so the
# router never rejects a config the translator would happily run
# (it previously hard-coded {sh,wt,wf,ss} and so falsely flagged
# tr / bias / mwf as port gaps).
from pyretis.inout.config_adapter import _INF_MOVE_CODES
unsupported = sorted(
{str(move).lower() for move in moves} - set(_INF_MOVE_CODES))
if unsupported:
return False, ('unsupported shooting move(s): '
+ ', '.join(unsupported))
if tis.get('mirror_freq') or tis.get('target_freq'):
# Mirror and target swap are ported (scheduler ``mr``/``ts``
# moves). The native in-process loop they used to run on is now
# retired, so the permeability family routes through the scheduler
# by default; the committed references were regenerated from the
# scheduler's deterministic output (a determinism lock -- the
# ``ts``/``mr`` statistical unbiasedness is tracked separately in
# ``targetswap-validation/RESULTS.md``).
#
# Both moves PERSIST a global order-function mutation on accept
# (the mirror flag / the target index), so they are only
# consistent at a single worker. Resolve the worker count exactly
# as config_adapter does (env override, then a [runner] section,
# then the default of 1) and keep the native loop if it would
# be >1.
resolved_workers = int(os.environ.get(
'PYRETIS_NATIVE_WORKERS',
config.get('runner', {}).get('workers', 1)))
if resolved_workers > 1:
return False, ('mirror/target swap moves require n_workers=1 '
f'(got {resolved_workers})')
init_method = str(config.get('initial-path', {}).get('method',
'kick')).lower()
# ``restart`` resumes the scheduler from its own output.toml.
# ``load`` (a pre-existing load folder) is re-written into the
# coordinator's load_dir format by generate_load_dir.
if init_method not in ('kick', '', 'restart', 'load'):
return False, 'initial-path method = ' + init_method
# A ``[simulation] restart`` continuation is honoured by the scheduler
# route: ``run_pyretis_path_sampling`` resumes from its own
# ``output.toml`` when the initial-path method is ``restart`` (the
# native restart filename is not used). A ``[simulation] restart`` WITHOUT
# ``method = "restart"`` would route to the in-process path that reads
# the native restart file, which the scheduler never writes -- keep that
# as the genuine port gap.
wants_restart = config.get('simulation', {}).get('restart')
if wants_restart and init_method != 'restart':
return False, 'simulation restart continuation without method=restart'
return True, ''
[docs]
def native_scheduler_port_gap(inputfile):
"""Return the scheduler port-gap reason for a native path-sampling TOML.
A native path-sampling task (tis/retis/explore/pptis/repptis) whose
config the scheduler cannot yet run (see
:func:`scheduler_supported_features`) has NO execution path -- the native
in-process loop is retired -- so the caller must reject it with this
reason rather than dispatch the retired loop. Returns ``None`` for inputs
that are not native path-sampling ``.toml`` tasks (``md`` / ``md-flux`` /
a ``.rst`` input / an infinite-swapping config), which run their normal
route, and ``None`` for a native path-sampling config the scheduler does
cover.
Parameters
----------
inputfile : string
Path to the input file.
Returns
-------
string or None
The first unsupported-feature reason, or ``None``.
"""
if os.path.splitext(inputfile)[1].lower() != '.toml':
return None
if not os.path.isfile(inputfile):
return None
if is_infinite_swapping_config(inputfile):
return None
try:
import tomllib as _toml
except ImportError:
import tomli as _toml
with open(inputfile, 'rb') as infile:
config = _toml.load(infile)
task = str(config.get('simulation', {}).get('task', '')).lower()
if task not in _NATIVE_SCHEDULER_TASKS:
return None
supported, reason = scheduler_supported_features(config)
return None if supported else reason
[docs]
def is_native_scheduler_config(inputfile):
"""Return True if a native path-sampling TOML routes to the scheduler.
As of the Stage C collapse, native ``task = "retis"`` ``.toml``
inputs are run through the infinite-swapping scheduler at
``n_workers = 1`` (their config is translated by
:func:`pyretis.inout.config_adapter.native_to_infswap_config`). The
native ``tis``, ``explore``, ``pptis``, and ``repptis`` tasks all route
through the scheduler too (validated by VALIDITY, not byte-identity --
the coordinator RNG differs from the native loop's). Inputs that already
select the infinite-swapping sampler are excluded. A native
path-sampling config that hits a scheduler PORT GAP (see
:func:`native_scheduler_port_gap`) returns ``False`` here and is rejected
with a clear error by the caller -- it is NOT dispatched to the retired
in-process loop. Only ``.toml`` inputs are considered (a ``.rst`` input
is not a scheduler config).
Parameters
----------
inputfile : string
Path to the input file.
Returns
-------
boolean
True for a native path-sampling ``.toml`` the scheduler can run.
"""
if os.path.splitext(inputfile)[1].lower() != '.toml':
return False
if not os.path.isfile(inputfile):
return False
if is_infinite_swapping_config(inputfile):
return False
try:
import tomllib as _toml
except ImportError:
import tomli as _toml
with open(inputfile, 'rb') as infile:
config = _toml.load(infile)
task = str(config.get('simulation', {}).get('task', '')).lower()
if task not in _NATIVE_SCHEDULER_TASKS:
return False
return native_scheduler_port_gap(inputfile) is None
[docs]
def run_infinite_swapping(inputfile):
"""Run an infinite-swapping (replica-exchange) input via its scheduler.
This is the programmatic entry for the infinite-swapping sampler (the
scheduler + config loader), run from the current working directory.
``pyretisrun`` calls it for inputs that select infinite swapping, so
it is the single way to drive that sampler.
Parameters
----------
inputfile : string
Path to the infinite-swapping input TOML.
"""
from pyretis.simulation.scheduler import scheduler
from pyretis.simulation.setup import setup_config
config = setup_config(inputfile)
# setup_config returns None when there is nothing to run (e.g. a
# restart whose cstep has already reached the target, or a missing
# active trajectory). Exit cleanly rather than crashing scheduler().
if config is None:
logger.progress('Nothing to run (restart already complete or no '
'active path); exiting.')
return
scheduler(config)
[docs]
def _apply_restart_steps(restart_file, steps):
"""Set the step target in a scheduler run file in place.
On a continuation the native config carries the new (usually higher)
total step target. The scheduler stops once ``cstep`` reaches
``[simulation] steps``, so a resumed restart must adopt the new target
or it would have nothing left to run. This reads, updates and rewrites
the run TOML (``output.toml``, or a legacy ``restart.toml``), leaving
every other key (the persisted ``current`` state -- RNG, frac, active
paths -- and all settings) untouched.
Parameters
----------
restart_file : string
Path to the scheduler run file (``output.toml`` or a legacy
``restart.toml``).
steps : int
The new total step target to write into ``[simulation] steps``.
"""
import tomli
with open(restart_file, 'rb') as handle:
restart = tomli.load(handle)
restart.setdefault('simulation', {})['steps'] = int(steps)
with open(restart_file, 'wb') as handle:
dump_toml(restart, handle)
[docs]
def run_pyretis_path_sampling(inputfile, runpath):
"""Run a native RETIS config through the infinite-swapping coordinator.
This is the opt-in compatibility route. It translates the native
configuration to the coordinator's config dictionary
(:func:`pyretis.inout.config_adapter.native_to_infswap_config`) and
generates the coordinator's ``load_dir`` when requested. By default
(``method = "load"``, or ``"kick"``) this goes through the proven,
sequential, single-process native initiation
(:func:`pyretis.inout.config_adapter.generate_load_dir`), unchanged.
Opting into ``[initial-path] kick-parallel = true`` (with ``method =
"kick"`` and ``kick-from = "initial"``, the default) instead routes
through the parallel, engine-agnostic kick phase
(:func:`pyretis.simulation.setup.run_kick_phase`, one job per
ensemble across a worker pool) -- proven so far only for engines that
drive their kick search through ``propagate()``/``_propagate_from()``
(e.g. TurtleMD); internal engines' own ``kick_across_middle`` is not
yet streaming-aware, so ``kick-parallel`` stays opt-in rather than the
default until that gap is closed. It then writes the resolved config
to the single ``output.toml`` and hands it to the unchanged scheduler
via :func:`run_infinite_swapping`.
Parameters
----------
inputfile : string
Path to the native RETIS input TOML.
runpath : string
The directory the simulation runs from (where ``load`` and the
translated ``output.toml`` are written).
"""
from pyretis.inout.config_adapter import (
native_to_infswap_config,
generate_load_dir,
)
from pyretis.simulation.setup import apply_config_defaults, run_kick_phase
native_config = parse_settings_file(inputfile)
config = native_to_infswap_config(native_config)
# Applied early (idempotent, setdefault-based): run_kick_phase needs
# ensemble_engines/load_dir resolved before it can dispatch kick jobs.
apply_config_defaults(config)
method = str(
native_config.get('initial-path', {}).get('method', 'kick')
).lower()
kick_from = str(
native_config.get('initial-path', {}).get('kick-from', 'initial')
).lower()
# Opt-in only (see the docstring): the parallel kick phase is proven
# so far only for engines whose kick search drives propagate()/
# _propagate_from() (e.g. TurtleMD); internal engines' own
# kick_across_middle is not yet streaming-aware. Default False keeps
# every existing config on the proven sequential route, unchanged.
kick_parallel = bool(
native_config.get('initial-path', {}).get('kick-parallel', False)
)
# --- Continuation / fresh-start handling --------------------------
# The scheduler persists its full state (cstep, frac, per-worker RNG)
# in ``./output.toml`` (the single run file: resolved config + the
# running [current]). A native config requesting a restart continuation
# therefore RESUMES from that file rather than re-translating a fresh
# config. setup_config reads ``output.toml`` as both the input and the
# restart file, so ``inp == re_inp`` and the guard is inert; exe_dir ==
# cwd (see main), so the bare name resolves in the run dir. A run
# started before the output.toml consolidation persists its state in a
# legacy ``restart.toml``; prefer output.toml but fall back to it so
# existing on-disk runs still resume.
output_file = os.path.join(runpath, 'output.toml')
legacy_restart = os.path.join(runpath, 'restart.toml')
resume_file = None
if os.path.isfile(output_file):
resume_file = output_file
elif os.path.isfile(legacy_restart):
resume_file = legacy_restart
if method == 'restart' and resume_file is not None:
# Carry the native step target into the run file so a continuation
# to a higher cycle count actually advances (the scheduler stops
# once cstep reaches steps).
_apply_restart_steps(resume_file, config['simulation']['steps'])
logger.progress('Resuming the scheduler from %s.', resume_file)
run_infinite_swapping(os.path.basename(resume_file))
return
if method == 'restart':
logger.progress('initial-path method=restart but no %s found; '
'starting fresh via kick.', output_file)
method = 'kick'
# Fresh start: a fresh output.toml (written below WITHOUT a [current])
# truncates any prior scheduler state. Also clear the atomic-write
# siblings and any legacy restart.toml so a stale state file cannot be
# mistaken for a resume on a later run.
for stale in (output_file + '.prev', output_file + '.tmp',
legacy_restart, legacy_restart + '.prev',
legacy_restart + '.tmp'):
if os.path.isfile(stale):
logger.progress('Discarding stale %s for a fresh scheduler run.',
stale)
os.remove(stale)
if method == 'load':
logger.progress('Generating initial paths via native load '
'initiation for the coordinator load directory.')
generate_load_dir(native_config, runpath)
elif method == 'kick' and kick_from == 'initial' and kick_parallel:
# The opt-in parallel route: one kick job per ensemble, dispatched
# across a worker pool (see run_kick_phase). Only "kick-from =
# initial" is supported here -- "previous" seeds each ensemble
# from the closest phase point of the PREVIOUS ensemble's
# just-kicked path, which is inherently sequential.
logger.progress('Generating initial paths via the parallel '
'kick phase (one job per ensemble) for the '
'coordinator load directory.')
run_kick_phase(config, runpath)
elif method == 'kick':
logger.progress('Generating initial paths via native kick '
'initiation for the coordinator load directory.')
generate_load_dir(native_config, runpath)
# The config (with defaults already applied above) is written WITHOUT
# a [current]: setup_config then builds a fresh cstep=0 state, which
# the scheduler persists back into output.toml.
with open(output_file, 'wb') as handle:
dump_toml(config, handle)
logger.progress('Translated native config written to: %s', output_file)
# Hand the scheduler the BARE name so it equals setup_config's
# ``re_inp`` default (the run file is also the state file now): exe_dir
# == cwd == runpath (the per-cycle write_toml emits ``./output.toml``),
# so the bare name resolves in the run dir and the inp/re_inp guard
# stays inert. Passing the full path would make inp != re_inp and the
# freshly-written output.toml would be mis-flagged as a foreign state
# file.
run_infinite_swapping(os.path.basename(output_file))
[docs]
def _warn_infretis_dialect_deprecated(inputfile):
"""Emit a single :py:class:`DeprecationWarning` for infretis-flavour input.
Mirrors :func:`pyretis.inout.settings._warn_rst_deprecated`'s rst
precedent: the legacy dialect keeps running (via the same
normalize/translate pipeline every native config already goes
through, see :func:`run_infretis_flavour_config`), this only warns.
"""
warnings.warn(
f"The infretis-flavour TOML dialect ({inputfile}: a [runner] "
"section, [simulation.tis_set], no [simulation] task) is "
"deprecated and will be removed in a future release. It still "
"runs, so existing scripts keep working. Migrate to the "
"pyretis-native dialect (task = \"retis\"/\"tis\"/..., [tis] "
"instead of [simulation.tis_set]) with the automated converter: "
f"'python -m pyretis.tools.convert_infretis_dialect {inputfile}' "
"(see MERGE_TODO.md S5.6 / docs/examples/migration.rst for the "
"full mapping).",
DeprecationWarning,
stacklevel=3,
)
[docs]
def run_infretis_flavour_config(inputfile, runpath):
"""Run a legacy infretis-flavour config via the native-schema pipeline.
Reshapes the raw infretis-flavour dict
(:func:`pyretis.inout.config_adapter.normalize_infretis_dialect`),
parses the result through the same strict native-schema validator
every hand-written native TOML goes through
(:func:`pyretis.inout.settings.parse_settings_toml`), translates it
(:func:`pyretis.inout.config_adapter.native_to_infswap_config`), and
hands the result to the unchanged scheduler via
:func:`run_infinite_swapping` -- the same three-stage pipeline
:func:`run_pyretis_path_sampling` uses for native input, so both
dialects now share one code path from this point on.
A restart continuation is excluded from this: it is always invoked
by pointing ``-i`` directly at the scheduler's own ``restart.toml``
(never the original config -- ``setup_config`` itself refuses a
fresh input file alongside an existing ``restart.toml``), and
``restart.toml`` is already in the coordinator's own shape, not a
user-authored dialect needing translation.
Parameters
----------
inputfile : string
Path to the infretis-flavour input TOML.
runpath : string
The directory the simulation runs from (where the intermediate
native-shaped TOML and the final translated ``infswap.toml`` are
written).
"""
_warn_infretis_dialect_deprecated(inputfile)
import tomli_w
try:
import tomllib as _toml
except ImportError:
import tomli as _toml
from pyretis.inout.config_adapter import (
native_to_infswap_config,
normalize_infretis_dialect,
)
with open(inputfile, 'rb') as handle:
raw = _toml.load(handle)
normalized = normalize_infretis_dialect(raw)
# The normalized dict is native-shaped but not yet a user-facing
# artifact (unlike the final translated infswap.toml below, which
# mirrors run_pyretis_path_sampling's own convention) -- write it to
# a throwaway temp file purely so parse_settings_toml (which reads
# from a path, not a dict) can validate it.
import tempfile
with tempfile.NamedTemporaryFile(
mode='wb', suffix='.toml', delete=False
) as handle:
tomli_w.dump(normalized, handle)
normalized_path = handle.name
try:
native_config = parse_settings_file(normalized_path)
finally:
os.remove(normalized_path)
config = native_to_infswap_config(native_config)
translated = os.path.join(runpath, 'infswap.toml')
with open(translated, 'wb') as handle:
tomli_w.dump(config, handle)
logger.progress('Translated infretis-flavour config written to: %s',
translated)
run_infinite_swapping(translated)
[docs]
def set_up_simulation(inputfile, runpath):
"""Run all the needed generic set-up.
Parameters
----------
inputfile : string
The input file which defines the simulation.
runpath : string
The base path we are running the simulation from.
Returns
-------
runner : method
A method which can be used to execute the simulation.
sim : object like :py:class:`.Simulation`
The simulation defined by the input file.
syst : object like :py:class:`.System`
The system created.
sim_settings : dict
The input settings read from the input file.
"""
if not os.path.isfile(inputfile):
raise ValueError(f'Input file "{inputfile}" NOT found!')
logger.progress('Reading input settings from: %s', inputfile)
logger.progress('Setting up simulation')
sim_settings = parse_settings_file(inputfile)
# NB this is not transmitted to the ensembles
sim_settings['simulation']['exe_path'] = runpath
sim_settings['engine']['exe_path'] = runpath
for ens in sim_settings.get('ensemble', []):
ens['simulation']['exe_path'] = runpath
ens['engine']['exe_path'] = runpath
logtxt = 'Set up and create simulation.'
logger.info(logtxt)
sim = create_simulation(sim_settings)
task = sim_settings['simulation']['task'].lower()
logger.success('Setup for simulation "%s" is done.', task)
runner = _RUNNERS.get(task, run_generic_simulation)
return runner, sim, sim_settings
[docs]
def store_simulation_settings(settings, indir, backup, ext='.rst'):
"""Store the parsed input settings.
Parameters
----------
settings : dict
The simulation settings.
indir : string
The directory which contains the input script.
backup : boolean
If True, an existing settings file will be backed up.
ext : string
Extension for the regenerated settings dump. Matches the
input file extension, so a ``.toml`` run writes ``out.toml``
and a ``.rst`` run writes ``out.rst``. Defaults to ``.rst``
for backwards compatibility.
"""
out_file = os.path.join(indir, f'out{ext}')
logger.info('Full simulation settings written to: %s', out_file)
write_settings_file(settings, out_file, backup=backup)
[docs]
def remove_exit_file(exit_file):
"""Remove the EXIT file after a completed soft exit."""
try:
os.remove(exit_file)
except FileNotFoundError:
return
except OSError as error:
logger.warning('Could not remove EXIT file "%s": %s', exit_file,
error)
else:
logger.info('Removed EXIT file: %s', exit_file)
[docs]
def soft_exit_ignore(turn_keyboard_interruption_off=True, exe_dir=None):
"""Manage the KeyboardInterrupt exception.
Parameters
----------
turn_keyboard_interruption_off : boolean
If True, instead of regular exiting from the program,
the file 'EXIT' is created to stop the PyRETIS.
exe_dir : string, optional
The path where EXIT file is expected.
"""
def soft_exit_handler(signum, frame): # pragma: no cover
"""Handle with a keyboard interruption signal."""
# pylint: disable=unused-argument
logger.progress('Attempting soft exit - terminating soon...')
pathlib.Path(os.path.join(exe_dir, 'EXIT')).touch(exist_ok=True)
if turn_keyboard_interruption_off:
return signal.signal(signal.SIGINT, soft_exit_handler)
return signal.signal(signal.SIGINT, signal.default_int_handler)
[docs]
def _report_execution_error(error, log_level):
"""Log a stopped execution and write its traceback to the log only.
Shared by both CLI flows (the native in-process simulation and the
infinite-swapping scheduler) so a failure is reported the same way
regardless of which path ran. The friendly one-line message goes to
the screen; the full traceback goes to the log file only.
Parameters
----------
error : Exception
The exception currently being handled.
log_level : integer
The active log level. At ``DEBUG`` or below the caller should
re-raise so the traceback also reaches the screen.
Returns
-------
reraise : boolean
``True`` when the caller should re-raise (debug mode), so the
error is never silently swallowed.
"""
logger.error('"%s: %s".', error.__class__.__name__, error.args)
logger.error('ERROR - execution stopped.')
logger.error(
'Please see the LOG for the error message and traceback.'
)
# Print the traceback to the log-file, but not to the screen.
screen = logger.handlers[0]
lvl = screen.level
screen.setLevel(logging.CRITICAL + 1)
logger.error(traceback.format_exc())
screen.setLevel(lvl)
return log_level <= logging.DEBUG
[docs]
def main(infile, indir, exe_dir, progress, log_level):
"""Execute PyRETIS.
Parameters
----------
infile : string
The input file to open with settings for PyRETIS.
indir : string
The folder containing the settings file.
exe_dir : string
The directory we are working from.
progress : boolean
Determines if we should use a progress bar or not.
log_level : integer
Determines if we should display the error traceback or not.
"""
simulation = None
settings = {}
exit_status = 0
def _dispatch_to_scheduler(run_callable):
"""Run a scheduler dispatch with a one-shot SIGTERM clean-unwind.
HPC schedulers (SLURM at walltime), ``kill`` and ``scancel`` send
SIGTERM, whose default action terminates the interpreter without
running atexit / finally handlers -- orphaning the worker pool (and
any live engine subprocesses). Convert SIGTERM into the same clean
unwind as Ctrl-C so the pool is released; ``output.toml`` is written
atomically (``os.replace``), so interrupting cannot corrupt the
restart state. Returns the process exit code.
"""
def _sigterm_handler(signum, frame): # pragma: no cover
# pylint: disable=unused-argument
# One-shot: ignore any further SIGTERM during the unwind below so
# a second signal (``timeout`` can SIGTERM a routed run again
# while it logs "stopped cleanly" / tears the pool down) cannot
# re-enter and raise a second KeyboardInterrupt that escapes the
# handler. The finally clause restores old_sigterm.
signal.signal(signal.SIGTERM, signal.SIG_IGN)
raise KeyboardInterrupt
old_sigterm = signal.signal(signal.SIGTERM, _sigterm_handler)
try:
run_callable()
except KeyboardInterrupt:
logger.progress('Termination signal received; '
'stopped the scheduler cleanly.')
bye_bye_world()
return 130
except Exception as error: # pylint: disable=broad-exception-caught
if _report_execution_error(error, log_level):
raise
bye_bye_world()
return 1
finally:
signal.signal(signal.SIGTERM, old_sigterm)
logger.success('\nSimulation done\n')
bye_bye_world()
return 0
# Stage C: all native path-sampling tasks (tis/retis/explore/pptis/
# repptis) route through the infinite-swapping scheduler at n_workers=1.
# The native in-process loop is retired, so a config the scheduler cannot
# run is rejected with a clear error below (native_scheduler_port_gap),
# never dispatched to the retired loop.
if is_native_scheduler_config(infile):
logger.progress('Routing the native TIS/RETIS config through the '
'infinite-swapping scheduler at n_workers=1.')
return _dispatch_to_scheduler(
lambda: run_pyretis_path_sampling(infile, exe_dir))
# Single-entry-point dispatch: an infinite-swapping input is run by
# its own scheduler (replica-exchange worker pool), not the
# in-process simulation flow below. A fresh legacy infretis-flavour
# config (still supported, now deprecated -- MERGE_TODO.md S5.6)
# goes through the same normalize/translate pipeline every native
# config already does; a restart continuation (already in the
# coordinator's own shape) is excluded and runs unchanged.
if is_infinite_swapping_config(infile):
if is_infretis_flavour_config(infile):
logger.progress('Input uses the infretis-flavour dialect; '
'running the replica-exchange scheduler.')
return _dispatch_to_scheduler(
lambda: run_infretis_flavour_config(infile, exe_dir))
logger.progress('Input selects the infinite-swapping sampler; '
'running the replica-exchange scheduler.')
return _dispatch_to_scheduler(lambda: run_infinite_swapping(infile))
exit_file = os.path.join(exe_dir, 'EXIT')
if os.path.isfile(exit_file):
logger.progress(
'Exit file found - Remove it before executing PyRETIS.')
logger.error('* %s file found *', exit_file)
logger.error('Remove the file to execute PyRETIS')
bye_bye_world()
return 1
in_ext = os.path.splitext(infile)[1].lower() or '.rst'
try:
# A native path-sampling task the scheduler cannot run has no
# execution path (the in-process loop is retired): fail with the
# specific reason instead of crashing deep in PathSimulation.run().
port_gap = native_scheduler_port_gap(infile)
if port_gap:
raise ValueError(
'This native path-sampling configuration cannot be run: '
f'{port_gap}. The native in-process loop has been retired; '
'adjust the configuration (e.g. workers = 1, a supported '
'shooting move, or initial-path method = "restart") so it '
'routes through the infinite-swapping scheduler.')
run, simulation, settings = set_up_simulation(infile, exe_dir)
store_simulation_settings(settings, indir, True, ext=in_ext)
# Run the simulation:
soft_exit_ignore(turn_keyboard_interruption_off=True, exe_dir=exe_dir)
run(simulation, settings, progress=progress)
soft_exit_ignore(turn_keyboard_interruption_off=False,
exe_dir=exe_dir)
except Exception as error: # pylint: disable=broad-exception-caught
exit_status = 1
if _report_execution_error(error, log_level):
raise
finally:
# Write out the simulation settings as they were parsed and
# add some additional info:
if simulation is not None:
end = getattr(simulation, 'cycle', {'step': None})['step']
if end is not None:
settings['simulation']['endcycle'] = end
logtxt = f'Execution ended at step {end}'
logger.progress(logtxt)
logger.success('\nSimulation done\n')
store_simulation_settings(settings, indir, False, ext=in_ext)
if exit_status == 0:
remove_exit_file(exit_file)
bye_bye_world()
return exit_status
[docs]
def entry_point(): # pragma: no cover
"""entry_point - The entry point for the pip install of pyretisrun."""
colorama.init(autoreset=True)
parser = argparse.ArgumentParser(description=PROGRAM_NAME)
parser.add_argument('-i', '--input',
help=f'Location of {PROGRAM_NAME} input file',
required=True)
parser.add_argument('-V', '--version', action='version',
version=f'{PROGRAM_NAME} {VERSION}')
parser.add_argument('-f', '--log_file',
help='Specify log file to write',
required=False,
default=f'{PROGRAM_NAME.lower()}.log')
parser.add_argument('-l', '--log_level',
help='Specify log level for log file',
required=False,
default='INFO')
parser.add_argument('-p', '--progress', action='store_true',
help=('Display a progress meter instead of text '
'output for the simulation'))
args_dict = vars(parser.parse_args())
input_file = args_dict['input']
# Store directories:
cwd_dir = os.getcwd()
input_dir = os.path.dirname(input_file)
if not os.path.isdir(input_dir):
input_dir = os.getcwd()
# Define a file logger:
create_backup(args_dict['log_file'])
fileh = logging.FileHandler(args_dict['log_file'], mode='a')
log_levl = getattr(logging, args_dict['log_level'].upper(),
logging.INFO)
fileh.setLevel(log_levl)
fileh.setFormatter(get_log_formatter(log_levl))
logger.addHandler(fileh)
# Here, we just check the python version. PyRETIS should anyway
# fail before this for python2.
check_python_version()
hello_world(input_file, cwd_dir, args_dict['log_file'])
sys.exit(main(input_file, input_dir, cwd_dir,
args_dict['progress'], log_levl))
[docs]
def entry_point_deprecated(): # pragma: no cover
"""Legacy ``pyretisrun`` entry point: warn, then run ``pyretis run``.
The standalone ``pyretisrun`` command is kept working for now but is
deprecated in favour of ``pyretis run``; from PyRETIS 5 only the
unified command will be supported. This wrapper emits that warning
and then delegates to :func:`entry_point` unchanged.
"""
from pyretis.bin._cli_common import warn_deprecated_executable
warn_deprecated_executable('pyretisrun', 'run', logger=logger)
entry_point()
if __name__ == '__main__': # pragma: no cover
entry_point()