fix(cp2k): say what is missing from an AIMD output directory

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 <PROJECT_NAME>-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:

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()