diff --git a/dpgen2/exploration/task/lmp_template_task_group.py b/dpgen2/exploration/task/lmp_template_task_group.py index d1f8e9fc..73e081c9 100644 --- a/dpgen2/exploration/task/lmp_template_task_group.py +++ b/dpgen2/exploration/task/lmp_template_task_group.py @@ -1,11 +1,14 @@ import itertools import random +import re +import warnings from pathlib import ( Path, ) from typing import ( List, Optional, + Set, ) from dpgen2.constants import ( @@ -101,6 +104,34 @@ def make_task( if self.plm_set: templates.append(self.plm_template) conts = self.make_cont(templates, self.revisions) + # Validate: check for unreplaced V_* variables in substituted templates + if self.revisions: + template_raw = "\n".join(self.lmp_template) + if self.plm_set: + template_raw += "\n" + "\n".join(self.plm_template) + # Flatten all template variants (LAMMPS + PLUMED) for validation + all_conts = [c for c_list in conts for c in c_list] + check_revisions_completeness( + all_conts, + list(self.revisions.keys()), + template_raw=template_raw, + ) + else: + # Warn (but don't error) if template has V_* but no revisions provided. + # This can be legitimate for customized-lmp-template workflows or tests. + combined = "\n".join(self.lmp_template) + if self.plm_set: + combined += "\n" + "\n".join(self.plm_template) + unreplaced = find_unreplaced_variables(combined) + if unreplaced: + warnings.warn( + f"LAMMPS template contains revision variable(s) {sorted(unreplaced)} " + f"but no 'revisions' dict was provided. " + f"These variables will NOT be substituted. " + f"If this is unintentional, add them to 'revisions' in your " + f"exploration config.", + stacklevel=2, + ) nconts = len(conts[0]) for cc, ii in itertools.product(confs, range(nconts)): # type: ignore if not self.plm_set: @@ -218,3 +249,101 @@ def revise_by_keys(lmp_lines, keys, values): for ii in range(len(lmp_lines)): lmp_lines[ii] = lmp_lines[ii].replace(kk, str(vv)) return lmp_lines + + +# Regex pattern for dpgen-style revision placeholders: V_ followed by uppercase letters/digits/underscores. +# This matches the universal convention in dpgen v1/v2 (all tests, docs, and examples use V_XXX). +_REVISION_VARIABLE_PATTERN = re.compile( + r"(? str: + """Remove LAMMPS-style comments (# to end of line) to avoid false positives. + + This prevents V_* patterns in comments (e.g., "# set V_PRESS later") + from being flagged as unreplaced variables. + """ + lines = content.split("\n") + stripped = [] + for line in lines: + # LAMMPS comments start with # (not inside quotes for our purposes) + idx = line.find("#") + if idx >= 0: + stripped.append(line[:idx]) + else: + stripped.append(line) + return "\n".join(stripped) + + +def find_unreplaced_variables(content: str) -> Set[str]: + """Scan text for remaining V_* revision placeholders that were not substituted. + + Strips LAMMPS comments before scanning to avoid false positives from + commented-out variable references. + + Parameters + ---------- + content : str + The LAMMPS input content after revision substitution. + + Returns + ------- + Set[str] + Set of variable names (e.g. {"V_PRESS", "V_UNDEFINED"}) still present. + """ + stripped = _strip_lammps_comments(content) + return set(_REVISION_VARIABLE_PATTERN.findall(stripped)) + + +def check_revisions_completeness( + templates_content: List[str], + revision_keys: List[str], + template_raw: str = "", +) -> None: + """Validate that all V_* placeholders in the template have been substituted. + + This function performs two checks: + 1. **Post-substitution residual check**: After applying revisions, scan the output + for any remaining V_* variables that were not replaced. This catches typos in + template variables or missing keys in revisions. + 2. **Unused key warning**: If a revision key is defined but never appears in the + raw template, emit a warning (possible typo in the key name). + + Parameters + ---------- + templates_content : List[str] + List of template strings after revision substitution (one per revision combo). + revision_keys : List[str] + The keys defined in the revisions dict. + template_raw : str + The raw template content before substitution (for unused key detection). + + Raises + ------ + ValueError + If unreplaced V_* variables are found in substituted templates. + """ + # Check 1: Residual unreplaced variables + all_unreplaced: Set[str] = set() + for content in templates_content: + all_unreplaced.update(find_unreplaced_variables(content)) + + if all_unreplaced: + raise ValueError( + f"LAMMPS template contains undefined revision variable(s): " + f"{sorted(all_unreplaced)}. " + f"Defined revisions: {sorted(revision_keys)}. " + f"Please add missing variables to 'revisions' in your exploration config, " + f"or remove them from the template." + ) + + # Check 2: Unused revision keys (warning only) + if template_raw and revision_keys: + for key in revision_keys: + if key not in template_raw: + warnings.warn( + f"Revision key '{key}' is defined but does not appear in the " + f"LAMMPS/PLUMED template. Possible typo?", + stacklevel=3, + ) diff --git a/tests/exploration/test_lmp_templ_task_group.py b/tests/exploration/test_lmp_templ_task_group.py index 6211137d..d899caf9 100644 --- a/tests/exploration/test_lmp_templ_task_group.py +++ b/tests/exploration/test_lmp_templ_task_group.py @@ -388,6 +388,7 @@ def test_lmp_plm(self): idx += 1 def test_lmp_empty(self): + """Empty revisions with template containing V_* should warn but succeed.""" task_group = LmpTemplateTaskGroup() task_group.set_conf(self.confs) task_group.set_lmp( @@ -396,24 +397,16 @@ def test_lmp_empty(self): revisions=self.rev_empty, traj_freq=self.traj_freq, ) - task_group.make_task() + import warnings as _warnings + + with _warnings.catch_warnings(record=True) as w: + _warnings.simplefilter("always") + task_group.make_task() + var_warnings = [x for x in w if "V_NSTEPS" in str(x.message)] + self.assertGreater(len(var_warnings), 0) + # Should still produce tasks ngroup = len(task_group) - self.assertEqual( - ngroup, - len(self.confs), - ) - idx = 0 - for cc in range(len(self.confs)): - ee = expected_lmp_template.split("\n") - self.assertEqual( - task_group[idx].files()[lmp_conf_name], - self.confs[cc], - ) - self.assertEqual( - task_group[idx].files()[lmp_input_name].split("\n"), - ee, - ) - idx += 1 + self.assertEqual(ngroup, len(self.confs)) def test_lmp_pimd(self): task_group = LmpTemplateTaskGroup() @@ -436,3 +429,229 @@ def test_lmp_pimd(self): task_group[0].files()[lmp_input_name].split("\n"), ee, ) + + +class TestRevisionVariablePrecheck(unittest.TestCase): + """Test PR6: validation of revision variables in LAMMPS templates.""" + + def setUp(self): + self.lmp_template_fname = Path("lmp_precheck.template") + self.numb_models = 4 + self.confs = ["foo"] + self.traj_freq = 10 + + def tearDown(self): + if self.lmp_template_fname.exists(): + os.remove(self.lmp_template_fname) + + def _write_template(self, content): + self.lmp_template_fname.write_text(content) + + def test_undefined_variable_raises(self): + """Template has V_PRESS but revisions only define V_NSTEPS and V_TEMP.""" + template = textwrap.dedent( + """\ + variable NSTEPS equal V_NSTEPS + variable TEMP equal V_TEMP + variable PRESS equal V_PRESS + + pair_style deepmd + pair_coeff * * + dump dpgen_dump + run ${NSTEPS} + """ + ) + self._write_template(template) + task_group = LmpTemplateTaskGroup() + task_group.set_conf(self.confs) + task_group.set_lmp( + self.numb_models, + self.lmp_template_fname, + revisions={"V_NSTEPS": [1000], "V_TEMP": [300]}, + traj_freq=self.traj_freq, + ) + with self.assertRaises(ValueError) as ctx: + task_group.make_task() + self.assertIn("V_PRESS", str(ctx.exception)) + self.assertIn("undefined revision variable", str(ctx.exception).lower()) + + def test_no_revisions_but_template_has_variables(self): + """Template has V_* variables but no revisions — should warn, not error.""" + template = textwrap.dedent( + """\ + variable NSTEPS equal V_NSTEPS + variable TEMP equal V_TEMP + + pair_style deepmd + pair_coeff * * + dump dpgen_dump + run ${NSTEPS} + """ + ) + self._write_template(template) + task_group = LmpTemplateTaskGroup() + task_group.set_conf(self.confs) + task_group.set_lmp( + self.numb_models, + self.lmp_template_fname, + revisions={}, + traj_freq=self.traj_freq, + ) + import warnings as _warnings + + with _warnings.catch_warnings(record=True) as w: + _warnings.simplefilter("always") + task_group.make_task() + var_warnings = [x for x in w if "V_NSTEPS" in str(x.message)] + self.assertGreater(len(var_warnings), 0) + # Should still succeed + self.assertEqual(len(task_group), 1) + + def test_all_variables_defined_no_error(self): + """All V_* variables are covered by revisions — should succeed.""" + template = textwrap.dedent( + """\ + variable NSTEPS equal V_NSTEPS + variable TEMP equal V_TEMP + + pair_style deepmd + pair_coeff * * + dump dpgen_dump + run ${NSTEPS} + """ + ) + self._write_template(template) + task_group = LmpTemplateTaskGroup() + task_group.set_conf(self.confs) + task_group.set_lmp( + self.numb_models, + self.lmp_template_fname, + revisions={"V_NSTEPS": [1000], "V_TEMP": [300, 600]}, + traj_freq=self.traj_freq, + ) + # Should not raise + task_group.make_task() + self.assertEqual(len(task_group), 2) # 1 conf * 2 V_TEMP values + + def test_unused_revision_key_warns(self): + """Revision defines V_TYPO that doesn't appear in template — should warn.""" + template = textwrap.dedent( + """\ + variable NSTEPS equal V_NSTEPS + + pair_style deepmd + pair_coeff * * + dump dpgen_dump + run ${NSTEPS} + """ + ) + self._write_template(template) + task_group = LmpTemplateTaskGroup() + task_group.set_conf(self.confs) + task_group.set_lmp( + self.numb_models, + self.lmp_template_fname, + revisions={"V_NSTEPS": [1000], "V_TYPO": [42]}, + traj_freq=self.traj_freq, + ) + import warnings as _warnings + + with _warnings.catch_warnings(record=True) as w: + _warnings.simplefilter("always") + task_group.make_task() + # Should have at least one warning about V_TYPO + typo_warnings = [x for x in w if "V_TYPO" in str(x.message)] + self.assertGreater(len(typo_warnings), 0) + + def test_lammps_internal_variables_not_flagged(self): + """${NSTEPS} and similar LAMMPS internal refs should NOT be flagged.""" + template = textwrap.dedent( + """\ + variable NSTEPS equal V_NSTEPS + + pair_style deepmd + pair_coeff * * + dump dpgen_dump + velocity all create ${TEMP} 12345 + run ${NSTEPS} + """ + ) + self._write_template(template) + task_group = LmpTemplateTaskGroup() + task_group.set_conf(self.confs) + task_group.set_lmp( + self.numb_models, + self.lmp_template_fname, + revisions={"V_NSTEPS": [1000]}, + traj_freq=self.traj_freq, + ) + # ${TEMP} is LAMMPS syntax, not a dpgen revision variable — should not raise + task_group.make_task() + self.assertEqual(len(task_group), 1) + + def test_plumed_template_undefined_variable_raises(self): + """V_* in PLUMED template but not in revisions should also be caught.""" + lmp_template = textwrap.dedent( + """\ + variable NSTEPS equal V_NSTEPS + variable TEMP equal V_TEMP + + pair_style deepmd + pair_coeff * * + dump dpgen_dump + fix dpgen_plm + run ${NSTEPS} + """ + ) + plm_template = textwrap.dedent( + """\ + DISTANCE ATOMS=3,5 LABEL=d1 + RESTRAINT ARG=d1 AT=V_DIST0 KAPPA=150.0 LABEL=restraint + """ + ) + self._write_template(lmp_template) + plm_fname = Path("plm_precheck.template") + plm_fname.write_text(plm_template) + try: + task_group = LmpTemplateTaskGroup() + task_group.set_conf(self.confs) + task_group.set_lmp( + self.numb_models, + self.lmp_template_fname, + plm_template_fname=str(plm_fname), + # V_DIST0 is used in PLUMED template but NOT defined here + revisions={"V_NSTEPS": [1000], "V_TEMP": [300]}, + traj_freq=self.traj_freq, + ) + with self.assertRaises(ValueError) as ctx: + task_group.make_task() + self.assertIn("V_DIST0", str(ctx.exception)) + finally: + plm_fname.unlink(missing_ok=True) + + def test_commented_variables_not_flagged(self): + """V_* in LAMMPS comments should NOT trigger errors.""" + template = textwrap.dedent( + """\ + variable NSTEPS equal V_NSTEPS + # TODO: add V_PRESS support later + # variable PRESS equal V_PRESS + + pair_style deepmd + pair_coeff * * + dump dpgen_dump + run ${NSTEPS} + """ + ) + self._write_template(template) + task_group = LmpTemplateTaskGroup() + task_group.set_conf(self.confs) + task_group.set_lmp( + self.numb_models, + self.lmp_template_fname, + revisions={"V_NSTEPS": [1000]}, + traj_freq=self.traj_freq, + ) + # V_PRESS is only in comments — should NOT raise + task_group.make_task() + self.assertEqual(len(task_group), 1)