diff --git a/dpdata/plugins/cp2k.py b/dpdata/plugins/cp2k.py index 61f2eaf9..d226253d 100644 --- a/dpdata/plugins/cp2k.py +++ b/dpdata/plugins/cp2k.py @@ -1,6 +1,7 @@ from __future__ import annotations import glob +import os import dpdata.formats.cp2k.output from dpdata.format import Format @@ -16,11 +17,50 @@ """ +def _find_single_file(directory, pattern, description, hint): + """Resolve one file inside an AIMD output directory. + + ``cp2k/aimd_output`` is pointed at a directory and locates its inputs by + globbing. Without this, a directory missing one of them fails with a bare + ``IndexError`` from subscripting the empty match list, which says neither + what was looked for nor where. + """ + # Only ``pattern`` is a glob. Escaping the directory keeps valid names + # such as ``run[1]`` from being interpreted as character classes. + matches = sorted( + glob.glob(os.path.join(glob.escape(os.fspath(directory)), pattern)) + ) + if not matches: + if not os.path.exists(directory): + found = f" {directory!r} does not exist." + elif not os.path.isdir(directory): + found = f" {directory!r} is not a directory." + else: + listing = sorted(os.listdir(directory)) + found = f" The directory contains: {', '.join(listing) or '(nothing)'}." + raise FileNotFoundError( + f"cp2k/aimd_output found no {description} matching {pattern!r} in " + f"{directory!r}.{found} {hint}" + ) + return matches[0] + + @Format.register("cp2k/aimd_output") class CP2KAIMDOutputFormat(Format): def from_labeled_system(self, file_name, restart=False, **kwargs): - xyz_file = sorted(glob.glob(f"{file_name}/*pos*.xyz"))[0] - log_file = sorted(glob.glob(f"{file_name}/*.log"))[0] + xyz_file = _find_single_file( + file_name, + "*pos*.xyz", + "trajectory file", + "CP2K writes it as -pos-1.xyz when MOTION/PRINT/TRAJECTORY " + "is enabled; pass the directory holding it, not a single file.", + ) + log_file = _find_single_file( + file_name, + "*.log", + "output log", + "Redirect the CP2K stdout into this directory, e.g. cp2k.popt -i input.inp > cp2k.log.", + ) try: return tuple(Cp2kSystems(log_file, xyz_file, restart)) except (StopIteration, RuntimeError) as e: diff --git a/tests/test_cp2k_aimd_missing_inputs.py b/tests/test_cp2k_aimd_missing_inputs.py new file mode 100644 index 00000000..2e59b37b --- /dev/null +++ b/tests/test_cp2k_aimd_missing_inputs.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import os +import shutil +import tempfile +import unittest + +from context import dpdata + +FIXTURE = os.path.join("cp2k", "aimd") + + +class TestCp2kAimdMissingInputs(unittest.TestCase): + """``cp2k/aimd_output`` locates its inputs by globbing a directory. + + A directory missing one of them used to fail with a bare ``IndexError`` + from subscripting the empty match list. + """ + + def setUp(self): + self.tmp_dir = tempfile.mkdtemp() + self.work = os.path.join(self.tmp_dir, "aimd") + shutil.copytree(FIXTURE, self.work) + + def tearDown(self): + shutil.rmtree(self.tmp_dir, ignore_errors=True) + + def _remove(self, name): + os.remove(os.path.join(self.work, name)) + + def test_complete_directory_still_loads(self): + system = dpdata.LabeledSystem(self.work, fmt="cp2k/aimd_output") + self.assertGreater(system.get_nframes(), 0) + + def test_directory_glob_metacharacters_are_literal(self): + bracketed = os.path.join(self.tmp_dir, "aimd[1]") + shutil.copytree(FIXTURE, bracketed) + system = dpdata.LabeledSystem(bracketed, fmt="cp2k/aimd_output") + self.assertGreater(system.get_nframes(), 0) + + def test_missing_trajectory_names_the_pattern(self): + self._remove("DPGEN-pos-1.xyz") + with self.assertRaises(FileNotFoundError) as caught: + dpdata.LabeledSystem(self.work, fmt="cp2k/aimd_output") + message = str(caught.exception) + self.assertIn("*pos*.xyz", message) + self.assertIn("trajectory file", message) + # The listing tells the user what the parser did see. + self.assertIn("cp2k.log", message) + + def test_missing_log_names_the_pattern(self): + self._remove("cp2k.log") + with self.assertRaises(FileNotFoundError) as caught: + dpdata.LabeledSystem(self.work, fmt="cp2k/aimd_output") + message = str(caught.exception) + self.assertIn("*.log", message) + self.assertIn("output log", message) + + def test_pointing_at_a_file_says_so(self): + # A frequent mistake: passing the log file rather than its directory. + log = os.path.join(self.work, "cp2k.log") + with self.assertRaises(FileNotFoundError) as caught: + dpdata.LabeledSystem(log, fmt="cp2k/aimd_output") + self.assertIn("is not a directory", str(caught.exception)) + + def test_missing_directory_says_so(self): + absent = os.path.join(self.tmp_dir, "absent") + with self.assertRaises(FileNotFoundError) as caught: + dpdata.LabeledSystem(absent, fmt="cp2k/aimd_output") + self.assertIn("does not exist", str(caught.exception)) + + +if __name__ == "__main__": + unittest.main()