# Copyright (c) 2026, PyRETIS Development Team.
# Distributed under the LGPLv2.1+ License. See LICENSE for more info.
"""Convert an infretis-flavour TOML config to pyretis-native syntax.
Usage::
python -m pyretis.tools.convert_infretis_dialect path/to/infswap.toml
python -m pyretis.tools.convert_infretis_dialect path/to/infswap.toml \
out.toml
If the output path is omitted, the file is converted in place (native
syntax is still ``.toml``, so there is no extension change the way
``pyretis.tools.convert_settings`` has for ``.rst`` -> ``.toml``).
Mirrors that converter's validated round-trip pattern
(:func:`pyretis.tools.convert_settings.convert`): the conversion is
verified before being trusted, here by comparing the coordinator
config :func:`pyretis.simulation.setup.setup_config` builds from the
ORIGINAL file against the one it builds from the CONVERTED file
(both go through the identical shared second stage,
MERGE_TODO.md S5.6), not by re-running a full simulation -- any
mismatch aborts with an error rather than silently writing a
translation nobody checked.
"""
import argparse
import json
import os
import sys
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,
)
from pyretis.inout.settings import parse_settings_toml
from pyretis.simulation.setup import setup_config
#: Coordinator-config keys the converter's own tis_set comparison
#: leaves out. These are keys native_to_infswap_config carries
#: unconditionally (post-defaulting) whose value equals the
#: coordinator's own internal fallback default regardless -- see
#: normalize_infretis_dialect's docstring for the full reasoning
#: (sigma_v, rescale_energy, mirror_freq, target_freq all confirmed
#: against their pyretis.core.moves.py / repex.py .get(key, default)
#: reads). Comparing only the keys the OLD route's dict actually has
#: is what the rest of this module does instead of hardcoding this
#: list a second time.
_TIS_SET_HARMLESS_EXTRAS = frozenset({
'sigma_v', 'rescale_energy', 'mirror_freq', 'target_freq',
'target_indices', 'high_accept',
})
[docs]
def _norm(value):
"""Stable string form for equality comparison and diagnostics."""
return json.dumps(value, sort_keys=True, default=str)
[docs]
def _strip_incidental(engine):
"""Drop keys that legitimately differ by run location, not content.
``input_files`` (GROMACS/CP2K, ``pyretis/inout/formats/{gromacs,
cp2k}.py``) is a run-time cache of ``input_path`` resolved to
absolute paths on whatever machine ``setup_config`` happens to run
on -- derived, not user-specified, so it is exactly as
run-location-incidental as ``exe_path``.
``cp2k_format`` is likewise derived, not user content: the engine
factory auto-detects it from ``input_path/initial.*`` (else ``"xyz"``)
deterministically and idempotently
(``pyretis.engines.factory``'s cp2k-format checker). The two dialects
only differ in WHEN that detection runs -- the native route resolves
it at config-build time, the infretis-flavour route defers it to
engine construction -- so the resolved value is identical and its
presence on only one side is not a translation difference.
"""
return {k: v for k, v in engine.items()
if k not in ('exe_path', 'rgen', 'type', 'input_files',
'cp2k_format')}
[docs]
def _compare_configs(old_config, new_config):
"""Compare two coordinator config dicts; return a list of mismatches.
Only the fields a conversion could plausibly perturb are checked
(mirroring the manual validation this module formalizes, see
MERGE_TODO.md S5.6 Phase 3/4): interfaces, shooting_moves, the
topology flags, zeroswap, ensemble_engines, runner (minus the
infinit-only ``files`` key, which normalize_infretis_dialect drops
deliberately), engine/engine0 (minus run-location-specific keys),
output's delete_old family, and every tis_set key the OLD route
actually sets (extra keys the NEW route carries are the documented,
harmless post-defaulting additions -- see _TIS_SET_HARMLESS_EXTRAS).
"""
mismatches = []
def check(name, old_value, new_value):
if _norm(old_value) != _norm(new_value):
mismatches.append(
f'{name}: OLD={old_value!r} NEW={new_value!r}'
)
old_sim, new_sim = old_config['simulation'], new_config['simulation']
check('simulation.interfaces', old_sim['interfaces'],
new_sim['interfaces'])
# shooting_moves only needs one entry per ensemble (len(interfaces));
# normalize_infretis_dialect truncates to that length (see its own
# docstring), matching what the coordinator itself only ever indexes
# into -- entries beyond that are unused on both routes, not a
# translation difference, so truncate the OLD side the same way
# before comparing (a longer, untruncated infretis-flavour list, e.g.
# copied verbatim from a larger example, would otherwise read as a
# false-positive mismatch).
n_ens = len(old_sim['interfaces'])
check('simulation.shooting_moves',
old_sim['shooting_moves'][:n_ens], new_sim['shooting_moves'])
for flag in ('noswap', 'repptis', 'single_tis', 'explore'):
check(f'simulation.{flag}', old_sim.get(flag, False),
new_sim.get(flag, False))
check('simulation.zeroswap', old_sim.get('zeroswap'),
new_sim.get('zeroswap'))
check('simulation.ensemble_engines', old_sim.get('ensemble_engines'),
new_sim.get('ensemble_engines'))
old_runner = {k: v for k, v in old_config['runner'].items()
if k != 'files'}
check('runner', old_runner, new_config['runner'])
check('engine', _strip_incidental(old_config['engine']),
_strip_incidental(new_config['engine']))
check('engine0', _strip_incidental(old_config.get('engine0', {})),
_strip_incidental(new_config.get('engine0', {})))
old_output, new_output = old_config['output'], new_config['output']
for key in ('delete_old', 'delete_old_all', 'keep_maxop_trajs'):
check(f'output.{key}', old_output[key], new_output[key])
old_order_energy = (
old_output.get('order-file', 1), old_output.get('energy-file', 1)
)
new_order_energy = (
new_output['order_file'], new_output['energy_file']
)
check('output.order_file/energy_file', old_order_energy,
new_order_energy)
old_tis_set = old_sim['tis_set']
new_tis_set = new_sim['tis_set']
for key, old_value in old_tis_set.items():
if key in _TIS_SET_HARMLESS_EXTRAS and key not in new_tis_set:
continue
check(f'tis_set.{key}', old_value, new_tis_set.get(key, '<MISSING>'))
return mismatches
[docs]
def _validate_conversion(infretis_path, native_path):
"""Compare the OLD and NEW routes' coordinator configs; raise on mismatch.
Both files must be read relative to the SAME working directory
(module files, ``initial.xyz``, ``load/`` all resolve relative to
cwd, not the input path), so the caller is expected to already be
chdir'd there -- see :func:`convert`.
"""
old_config = setup_config(infretis_path)
native_config = parse_settings_toml(native_path)
translated = native_to_infswap_config(native_config)
translated_path = native_path + '.converter-check.toml'
try:
with open(translated_path, 'wb') as handle:
tomli_w.dump(translated, handle)
new_config = setup_config(translated_path)
finally:
os.remove(translated_path)
mismatches = _compare_configs(old_config, new_config)
if mismatches:
raise RuntimeError(
f'Conversion validation failed for {infretis_path}:\n '
+ '\n '.join(mismatches)
)
[docs]
def convert(infretis_path, native_path=None, *, force=False, validate=True):
"""Convert one infretis-flavour TOML file to native syntax in place.
Returns the path of the written TOML file.
Parameters
----------
infretis_path : str
Path to the infretis-flavour input TOML.
native_path : str, optional
Output path. Defaults to overwriting ``infretis_path`` (native
syntax is still ``.toml``, unlike the rst -> toml converter).
force : bool, optional
Overwrite ``native_path`` if it already exists and differs from
``infretis_path``.
validate : bool, optional
Verify the conversion (see :func:`_validate_conversion`) before
writing. Runs with the working directory changed to
``infretis_path``'s directory, since the config's own relative
references (engine modules, ``load/``, ...) require it; restored
afterwards regardless of outcome.
"""
if not os.path.isfile(infretis_path):
raise FileNotFoundError(infretis_path)
if native_path is None:
native_path = infretis_path
in_place = native_path == infretis_path
if os.path.exists(native_path) and not in_place and not force:
raise FileExistsError(
f'{native_path} already exists (pass --force to overwrite)'
)
with open(infretis_path, 'rb') as handle:
raw = _toml.load(handle)
normalized = normalize_infretis_dialect(raw)
directory = os.path.dirname(os.path.abspath(infretis_path)) or '.'
infretis_name = os.path.basename(infretis_path)
native_name = (
infretis_name if in_place else os.path.basename(native_path)
)
# Write the converted content to a scratch name first, in the SAME
# directory (so validation's relative references -- engine modules,
# load/, ... -- resolve identically to the real thing), so the
# original is never even transiently overwritten: setup_config
# reading infretis_path for the "OLD route" comparison must see the
# untouched original, not a half-finished conversion.
scratch_name = native_name + '.converting'
scratch_path = os.path.join(directory, scratch_name)
with open(scratch_path, 'wb') as handle:
tomli_w.dump(normalized, handle)
if validate:
cwd = os.getcwd()
try:
os.chdir(directory)
_validate_conversion(infretis_name, scratch_name)
except Exception:
os.remove(scratch_path)
raise
finally:
os.chdir(cwd)
os.replace(scratch_path, native_path)
return native_path
[docs]
def main(argv=None):
"""Run the infretis-flavour -> native converter command line interface."""
parser = argparse.ArgumentParser(
prog='pyretis.tools.convert_infretis_dialect',
description=(
'Convert an infretis-flavour TOML config ([runner], '
'[simulation.tis_set], no task) to pyretis-native syntax.'
),
)
parser.add_argument('infretis', help='input infretis-flavour .toml file')
parser.add_argument(
'native', nargs='?', default=None,
help='output .toml file (default: convert in place)',
)
parser.add_argument(
'--force', '-f', action='store_true',
help='overwrite an existing output file',
)
parser.add_argument(
'--no-validate', dest='validate', action='store_false',
help='skip the old-vs-new coordinator config equality check',
)
args = parser.parse_args(argv)
try:
out = convert(
args.infretis, args.native,
force=args.force, validate=args.validate,
)
except (FileNotFoundError, FileExistsError, RuntimeError) as exc:
print(f'error: {exc}', file=sys.stderr)
return 1
print(f'wrote {out}')
return 0
if __name__ == '__main__':
sys.exit(main())