Source code for pyretis.core.engine_time

# Copyright (c) 2026, PyRETIS Development Team.
# Distributed under the LGPLv2.1+ License. See LICENSE for more info.
"""Resolve the engine time step for the flux and rate analysis.

The simulation time per PyRETIS step (``timestep * subcycles``) is the
single factor that turns path lengths, counted in integration steps,
into times when computing fluxes and rates. The time step belongs to the
engine, so this module resolves it the same way for every engine: from
the ``[engine]`` settings when present, otherwise from the engine's own
input file -- as PyRETIS does when it builds the engine.

Important method defined here
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

engine_time_per_step (:py:func:`.engine_time_per_step`)
    Return ``timestep * subcycles`` from a settings/run configuration.
"""
import math
import os


__all__ = ['engine_time_per_step', 'validate_engine_timestep']


def validate_engine_timestep(given, actual, engine_name):
    """Check a supplied time step against the engine's own time step.

    External engines own their time step in their own input file or
    integrator, so an ``[engine] timestep`` is optional for them. When
    one is supplied it must agree with the value the engine actually
    uses -- otherwise the simulation and the analysis would silently run
    with different time steps. This fails loud on a mismatch and does
    nothing when no time step is supplied.

    Parameters
    ----------
    given : float or None
        The time step from the ``[engine]`` settings, or ``None`` when it
        was not supplied.
    actual : float
        The time step the engine itself uses (read from its input file or
        integrator).
    engine_name : string
        Name of the engine, used in the error message.

    Raises
    ------
    ValueError
        If ``given`` is supplied and differs from ``actual``.

    """
    if given is None:
        return
    if not math.isclose(float(given), float(actual),
                        rel_tol=1e-9, abs_tol=1e-12):
        raise ValueError(
            f'The [engine] timestep ({given}) does not match the '
            f'{engine_name} time step ({actual}). For {engine_name} the '
            f'time step is defined by the engine itself; omit it from the '
            f'[engine] settings or make the two values agree.'
        )


def engine_time_per_step(settings):
    """Return the simulation time per PyRETIS step from settings.

    This is the engine ``timestep`` multiplied by its ``subcycles``
    (defaulting to 1) -- the physical time between two recorded path
    points, in the engine's own time unit. It is the single factor used
    to turn path lengths (counted in steps) into times when computing
    fluxes and rates, so every analysis routine derives it the same way.

    The time step belongs to the engine. When it is not present in the
    ``[engine]`` settings (e.g. LAMMPS, which keeps it in the LAMMPS
    input file), it is recovered from the engine's own input file -- as
    PyRETIS does when it builds the engine -- before giving up.

    Parameters
    ----------
    settings : dict
        The simulation settings, or a loaded run configuration. The
        engine information is read from the ``engine`` section.

    Returns
    -------
    float or None
        ``timestep * subcycles``, or ``None`` if the time step cannot be
        determined from the settings or the engine input (so each caller
        can decide whether that is an error or a reason to fall back to a
        per-step quantity).

    """
    engine = settings.get('engine', {})
    timestep = engine.get('timestep')
    if timestep is None:
        timestep = _engine_timestep_from_input(settings)
    if timestep is None:
        return None
    return float(timestep) * float(engine.get('subcycles', 1))


def _engine_timestep_from_input(settings):
    """Read the engine time step from the external program's input file.

    This mirrors how the engines read their own time step when they are
    built, so the analysis can recover it for engines that own the time
    step in their input file rather than in the ``[engine]`` settings.

    Only engines whose input file can be read for a time step are handled
    (currently LAMMPS). ``None`` is returned otherwise -- for example for
    OpenMM, whose time step lives in the integrator object and is not
    available from a file, and for engines (internal, GROMACS, CP2K, ...)
    that take their time step from the ``[engine]`` settings.

    Parameters
    ----------
    settings : dict
        The simulation settings, or a loaded run configuration.

    Returns
    -------
    float or None
        The engine time step read from the input file, or ``None``.

    """
    engine = settings.get('engine', {})
    klass = str(engine.get('class', '')).lower()
    if 'lammps' in klass:
        # Local import: the engines depend on this package, so importing
        # them at module level would create a circular import.
        # pylint: disable=import-outside-toplevel
        from pyretis.engines.lammps import read_lammps_input
        input_path = engine.get('input_path', '.')
        exe_path = settings.get('simulation', {}).get(
            'exe_path', engine.get('exe_path')
        )
        if exe_path is None:
            exe_path = os.path.abspath('.')
        template = os.path.join(exe_path, input_path, 'lammps.in')
        if os.path.isfile(template):
            for key, value in read_lammps_input(template):
                if key == 'timestep':
                    return float(value)
    return None