# Copyright (c) 2026, PyRETIS Development Team.
# Distributed under the LGPLv2.1+ License. See LICENSE for more info.
"""This module defines common methods for the settings handling.
Important methods defined here
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
create_external (:py:func:`.create_external`)
Method to create objects from settings.
check_settings (:py:func:`.check_settings`)
Check that required simulation settings are actually given.
create_engine (:py:func:`.create_engine`)
Method to create an engine from settings.
create_orderparameter (:py:func:`.create_orderparameter`)
Method to create order parameters from settings.
create_potential (:py:func:`.create_potential`)
Method to create a potential from settings.
"""
import logging
import os
from pyretis.core.common import initiate_instance, import_from
from pyretis.engines import engine_factory
from pyretis.orderparameter import order_factory
from pyretis.orderparameter.orderparameter import (
CompositeOrderParameter,
normalize_order_output,
wrap_orderparameter,
)
from pyretis.forcefield.factory import potential_factory
logger = logging.getLogger(__name__) # pylint: disable=invalid-name
logger.addHandler(logging.NullHandler())
__all__ = ['create_external', 'check_settings', 'create_orderparameter',
'create_orderparameters', 'check_mirror_position',
'create_engine', 'create_potential']
[docs]def check_settings(settings, required):
"""Check that required simulation settings are actually given.
This method will look for required settings in the given
`settings`. If one or more keys from the given `required` list of
strings are not found, this method will return False. Otherwise,
it will return True. Typically, an exception should be raised if
False is returned, this is handled outside the method in case
someone wants to add some magic handling of missing settings.
Parameters
----------
settings : dict
This dict contains the given settings
required : list of strings
This list contains the settings that are required and which
we will check the presence of.
Returns
-------
result : boolean
True if all required settings are present, False otherwise.
not_found : list of strings
There are the required settings we did not find.
"""
result = True
not_found = []
for setting in required:
if setting not in settings:
result = False
not_found.append(setting)
return result, not_found
[docs]def create_external(settings, key, factory, required_methods,
key_settings=None):
"""Create external objects from settings.
This method will handle the creation of objects from settings. The
requested objects can be PyRETIS internals or defined in external
modules.
Parameters
----------
settings : dict
This dictionary contains the settings for the simulation.
key : string
The setting we are creating for.
factory : callable
A method to call that can handle the creation of internal
objects for us.
required_methods : list of strings
The methods we need to have if creating an object from external
files.
key_settings : dict, optional
This dictionary contains the settings for the specific key we
are processing. If this is not given, we will try to obtain
these settings by `settings[key]`. The reason why we make it
possible to pass these as settings is in case we are processing
a key which does not give a simple setting, but a list of settings.
It that case `settings[key]` will give a list to process. That list
is iterated somewhere else and `key_settings` can then be used to
process these elements.
Returns
-------
out : object
This object represents the class we are requesting here.
"""
if key_settings is None:
try:
key_settings = settings[key]
except KeyError:
logger.debug('No "%s" setting found. Skipping set-up', key)
return None
module = key_settings.get('module', None)
klass = None
try:
klass = key_settings['class']
except KeyError:
logger.debug('No "class" setting for "%s" specified. Skipping set-up',
key)
return None
if module is None:
return factory(key_settings)
# Here we assume we are to load from a file. Before we import
# we need to check that the path is ok or if we should include
# the 'exe_path' from settings.
# 1) Check if we can find the module:
if os.path.isfile(module):
obj = import_from(module, klass)
else:
if 'exe_path' in settings['simulation']:
module = os.path.join(settings['simulation']['exe_path'],
module)
obj = import_from(module, klass)
else:
msg = f'Could not find module "{module}" for {key}!'
raise ValueError(msg)
# run some checks:
for function in required_methods:
objfunc = getattr(obj, function, None)
if not objfunc:
msg = f'Could not find method {klass}.{function}'
logger.critical(msg)
raise ValueError(msg)
if not callable(objfunc):
msg = f'Method {klass}.{function} is not callable!'
logger.critical(msg)
raise ValueError(msg)
instance = initiate_instance(obj, key_settings)
if key in ('orderparameter', 'collective-variable'):
return wrap_orderparameter(instance)
return instance
[docs]def create_orderparameter(settings):
"""Create order parameters from settings.
Parameters
----------
settings : dict
This dictionary contains the settings for the simulation.
Returns
-------
out : object like :py:class:`.OrderParameter`
This object represents the order parameter.
"""
main_order = create_external(
settings,
'orderparameter',
order_factory,
['calculate'],
)
if main_order is None:
logger.info('No order parameter created')
return None
main_order = wrap_orderparameter(main_order)
logger.progress('Created main order parameter:\n%s', main_order)
extra_cv = []
order_settings = settings.get('collective-variable', [])
for order_setting in order_settings:
order = create_external(
settings,
'collective-variable',
order_factory,
['calculate'],
key_settings=order_setting
)
order = wrap_orderparameter(order)
logger.progress('Created additional collective variable:\n%s', order)
extra_cv.append(order)
if not extra_cv:
return main_order
all_order = [main_order] + extra_cv
order = CompositeOrderParameter(order_parameters=all_order)
order = wrap_orderparameter(order)
logger.progress('Composite order parameter:\n%s', order)
return order
[docs]def create_orderparameters(engines, settings):
"""Assign an order parameter to every engine in the pool.
This is the plural counterpart of
:py:func:`.create_orderparameter`. It is used by the
infinite-swapping scheduler, which keeps a per-worker pool of
engines and needs each engine to carry its own order-parameter
instance (the order parameter is intentionally created once per
engine rather than shared, so the engines never cross the worker
process boundary together with a shared order parameter).
Parameters
----------
engines : dict of lists
A mapping from an engine key to a list of engine objects. Each
engine receives a freshly created order parameter assigned to
its ``order_function`` attribute.
settings : dict
This dictionary contains the settings for the simulation. The
``"orderparameter"`` section is used to build the order
parameter (see :py:func:`.create_orderparameter`).
"""
for engine_key in engines:
for engine in engines[engine_key]:
engine.order_function = create_orderparameter(settings)
check_mirror_position(engine.order_function, settings)
[docs]def check_mirror_position(order_function, settings):
"""Fail fast on a mis-placed permeability mirror plane.
Ports the native ``create_ensemble`` check
(:mod:`pyretis.setup.createsimulation`) onto the infinite-swapping
scheduler: in a permeability simulation the order function's mirror
plane must sit halfway between the outer interfaces of the ``[0^-]``
ensemble (``lambda_minus_one`` on the left, the first interface on
the right), or every mirror move would silently sample a different
distribution. A no-op outside permeability or when the order
function has no ``mirror_pos``.
Parameters
----------
order_function : object like :py:class:`.OrderParameter`
The order parameter to validate.
settings : dict
The coordinator configuration; ``simulation.tis_set`` provides
``permeability`` and ``lambda_minus_one``, and
``simulation.interfaces`` the first interface.
Raises
------
ValueError
If the mirror plane is not at the required position.
"""
tis_set = settings.get('simulation', {}).get('tis_set', {})
if not tis_set.get('permeability', False):
return
if not hasattr(order_function, 'mirror_pos'):
return
zero_left = tis_set.get('lambda_minus_one', False)
if zero_left is False:
return
offset = getattr(order_function, 'offset', 0)
first_interface = settings['simulation']['interfaces'][0]
correct_mirror = (zero_left + first_interface) / 2. - offset
if abs(order_function.mirror_pos - correct_mirror) > 1E-5:
msg = "Order function should have a mirror at "
msg += f"{correct_mirror}, found one at "
msg += f"{order_function.mirror_pos} instead."
raise ValueError(msg)
[docs]def create_engine(settings):
"""Create an engine from settings.
Parameters
----------
settings : dict
This dictionary contains the settings for the simulation.
Returns
-------
out : object like :py:class:`.EngineBase`
This object represents the engine.
"""
engine = create_external(settings, 'engine', engine_factory,
['integration_step'])
if not engine:
raise ValueError('Could not create engine from settings!')
# The time step belongs to the engine. Record the value it resolved
# (from the [engine] setting or its own input/integrator) back into
# the engine settings when it is not already given there, so the
# persisted configuration is self-describing for the flux/rate
# analysis -- notably for engines such as OpenMM that keep the time
# step in the integrator and not in a readable input file.
engine_settings = settings.get('engine')
if (engine_settings is not None
and engine_settings.get('timestep') is None
and getattr(engine, 'timestep', None) is not None):
engine_settings['timestep'] = engine.timestep
engine_settings.setdefault(
'subcycles', getattr(engine, 'subcycles', 1)
)
return engine
[docs]def create_potential(settings, key_settings):
"""Create a potential from settings.
Parameters
----------
settings : dict
This dictionary contains the settings for the simulation.
key_settings : dict
Settings for the potential we are creating.
Returns
-------
out : object like :py:class:`.PotentialFunction`
The object representing the potential function.
"""
return create_external(settings, 'potential', potential_factory,
['force', 'potential', 'potential_and_force'],
key_settings=key_settings)