Perf: remove redundant checks on data integrity

deepmd/utils/data_system.py

@@ -28,9 +28,6 @@
 from deepmd.utils.out_stat import (
     compute_stats_from_redu,
 )
-from deepmd.utils.path import (
-    DPPath,
-)
 
 log = logging.getLogger(__name__)
 
@@ -103,6 +100,7 @@ def __init__(
         del rcut
         self.system_dirs = systems
         self.nsystems = len(self.system_dirs)
+        assert self.nsystems > 0, "No systems provided"
         self.data_systems = []
         for ii in self.system_dirs:
             self.data_systems.append(
@@ -755,23 +753,7 @@ def process_systems(systems: Union[str, list[str]]) -> list[str]:
         systems = expand_sys_str(systems)
     elif isinstance(systems, list):
         systems = systems.copy()
-    help_msg = "Please check your setting for data systems"
-    # check length of systems
-    if len(systems) == 0:
-        msg = "cannot find valid a data system"
-        log.fatal(msg)
-        raise OSError(msg, help_msg)
-    # roughly check all items in systems are valid
-    for ii in systems:
-        ii = DPPath(ii)
-        if not ii.is_dir():
-            msg = f"dir {ii} is not a valid dir"
-            log.fatal(msg)
-            raise OSError(msg, help_msg)
-        if not (ii / "type.raw").is_file():
-            msg = f"dir {ii} is not a valid data system dir"
-            log.fatal(msg)
-            raise OSError(msg, help_msg)
+    assert systems, "No systems provided"
     return systems
 
 

deepmd/utils/path.py

@@ -14,6 +14,7 @@
 from typing import (
     ClassVar,
     Optional,
+    Union,
 )
 
 import h5py
@@ -151,25 +152,24 @@
             If true, no error will be raised if the target directory already exists.
         """
 
 
 class DPOSPath(DPPath):
     """The OS path class to data system (DeepmdData) for real directories.
 
     Parameters
     ----------
-    path : str
+    path : Union[str, Path]
         path
     mode : str, optional
         mode, by default "r"
     """
 
-    def __init__(self, path: str, mode: str = "r") -> None:
+    def __init__(self, path: Union[str, Path], mode: str = "r") -> None:
         super().__init__()
         self.mode = mode
-        if isinstance(path, Path):
-            self.path = path
-        else:
-            self.path = Path(path)
+        self.path = Path(path)
+        if not self.path.exists():
+            raise FileNotFoundError(f"{self.path} not found")
 
     def load_numpy(self) -> np.ndarray:
         """Load NumPy array.
@@ -300,6 +300,8 @@
         # so we do not support file names containing #...
         s = path.split("#")
         self.root_path = s[0]
+        if not os.path.isfile(self.root_path):
+            raise FileNotFoundError(f"{self.root_path} not found")
         self.root = self._load_h5py(s[0], mode)
         # h5 path: default is the root path
         self._name = s[1] if len(s) > 1 else "/"