feat(data): add dpdata format conversion

deepmd/dpmodel/train/data.py

@@ -22,6 +22,9 @@
     DataConfig,
     data_group_configs,
 )
+from deepmd.utils.data_system import (
+    close_data_systems,
+)
 from deepmd.utils.stat_file import (
     StatFileMode,
     StatFileSpec,
@@ -106,15 +109,23 @@ def make_task_maps(
     config: Mapping[str, Any],
     factory: Callable[[TrainingTaskConfig], tuple[Any, Any | None, Any | None]],
 ) -> tuple[dict[str, Any], dict[str, Any | None], dict[str, Any | None]]:
-    """Build training, validation, and stat maps from normalized task configs."""
+    """Build task maps, closing completed data objects if a later task fails.
+
+    Factories own partial results until they return and must clean those up
+    themselves. Successful maps transfer ownership to the caller.
+    """
     training_data: dict[str, Any] = {}
     validation_data: dict[str, Any | None] = {}
     stat_data: dict[str, Any | None] = {}
-    for task_config in iter_training_task_configs(config):
-        train_item, valid_item, stat_item = factory(task_config)
-        training_data[task_config.key] = train_item
-        validation_data[task_config.key] = valid_item
-        stat_data[task_config.key] = stat_item
+    try:
+        for task_config in iter_training_task_configs(config):
+            train_item, valid_item, stat_item = factory(task_config)
+            training_data[task_config.key] = train_item
+            validation_data[task_config.key] = valid_item
+            stat_data[task_config.key] = stat_item
+    except BaseException:
+        close_data_systems(training_data, validation_data)
+        raise
     return training_data, validation_data, stat_data
 
 

deepmd/entrypoints/test.py

@@ -118,6 +118,8 @@ def test(
                     group["systems"],
                     patterns=group.get("rglob_patterns"),
                     root_dir=Path(input_json).parent,
+                    fmt=group.get("format"),
+                    out_fmt=group.get("out_format", group.get("output_format")),
                 )
             )
     elif datafile is not None:

deepmd/jax/entrypoints/train.py

@@ -36,6 +36,7 @@
 )
 from deepmd.utils import random as dp_random
 from deepmd.utils.data_system import (
+    close_data_systems,
     get_data,
 )
 from deepmd.utils.summary import SummaryPrinter as BaseSummaryPrinter
@@ -189,23 +190,32 @@ def factory(
                 None,
             )
             valid_data = None
-            if task_config.validation_data_params is not None:
-                valid_data = get_data(
-                    dict(task_config.validation_data_params),
-                    task_model.get_rcut(),
-                    train_data.type_map,
-                    None,
-                )
-            return train_data, valid_data, None
+            try:
+                if task_config.validation_data_params is not None:
+                    valid_data = get_data(
+                        dict(task_config.validation_data_params),
+                        task_model.get_rcut(),
+                        train_data.type_map,
+                        None,
+                    )
+                return train_data, valid_data, None
+            except BaseException:
+                # The factory owns these objects until make_task_maps records
+                # them; validation setup or stat-file validation may fail.
+                close_data_systems(train_data, valid_data)
+                raise
 
         train_data_map, valid_data_map, _ = make_task_maps(config, factory)
-        print_data_summaries(train_data_map, valid_data_map)
+        try:
+            print_data_summaries(train_data_map, valid_data_map)
 
-        start_time = time.time()
-        model.train(train_data_map, valid_data_map)
-        end_time = time.time()
-        log.info("finished training")
-        log.info(f"wall time: {(end_time - start_time):.3f} s")
+            start_time = time.time()
+            model.train(train_data_map, valid_data_map)
+            end_time = time.time()
+            log.info("finished training")
+            log.info(f"wall time: {(end_time - start_time):.3f} s")
+        finally:
+            close_data_systems(train_data_map, valid_data_map)
 
 
 def train(
@@ -296,9 +306,12 @@ def update_sel(
                 type_map,
                 None,  # not used
             )
-            updated_model, task_min_nbor_dist = BaseModel.update_sel(
-                train_data, type_map, dict(task_config.model_params)
-            )
+            try:
+                updated_model, task_min_nbor_dist = BaseModel.update_sel(
+                    train_data, type_map, dict(task_config.model_params)
+                )
+            finally:
+                close_data_systems(train_data)
             if multi_task:
                 jdata_cpy["model"]["model_dict"][task_config.key] = updated_model
                 min_nbor_dist[task_config.key] = task_min_nbor_dist

deepmd/pd/entrypoints/main.py

@@ -75,6 +75,7 @@
 )
 from deepmd.utils.data_system import (
     get_data,
+    validate_backend_data_config,
 )
 from deepmd.utils.model_preset import (
     expand_model_preset,
@@ -294,6 +295,18 @@ def train(
     config = update_deepmd_input(config, warning=True, dump="input_v2_compat.json")
     config = normalize(config, multi_task=multi_task)
 
+    # Validate every task and validation source before neighbor-stat loading
+    # can run the converter. Paddle has no LMDB reader.
+    task_data = (
+        config["training"]["data_dict"].values() if multi_task else [config["training"]]
+    )
+    for data_config in task_data:
+        for key in ("training_data", "validation_data"):
+            if data_config.get(key) is not None:
+                validate_backend_data_config(
+                    data_config[key], backend_name="Paddle", lmdb_supported=False
+                )
+
     # do neighbor stat
     min_nbor_dist = None
     if not skip_neighbor_stat:
@@ -304,7 +317,11 @@ def train(
         if not multi_task:
             type_map = config["model"].get("type_map")
             train_data = get_data(
-                config["training"]["training_data"], 0, type_map, None
+                config["training"]["training_data"],
+                0,
+                type_map,
+                None,
+                lmdb_supported=False,
             )
             config["model"], min_nbor_dist = BaseModel.update_sel(
                 train_data, type_map, config["model"]
@@ -318,6 +335,7 @@ def train(
                     0,
                     type_map,
                     None,
+                    lmdb_supported=False,
                 )
                 config["model"]["model_dict"][model_item], min_nbor_dist[model_item] = (
                     BaseModel.update_sel(

deepmd/pt/utils/dataloader.py

@@ -72,6 +72,7 @@
     system_probabilities,
 )
 from deepmd.utils.data_system import (
+    close_data_systems,
     print_summary,
 )
 
@@ -361,49 +362,70 @@ def create_dataset(
     modifier: BaseModifier | None = None,
 ) -> DpLoaderSet | LmdbDataset | GroupedDataset:
     """Construct native readers from the shared recursive group configuration."""
+    from deepmd.utils.data_system import (
+        conversion_will_write_lmdb,
+        validate_lmdb_sampling_options,
+    )
+
+    for params in data_group_configs(config):
+        if conversion_will_write_lmdb(params):
+            validate_lmdb_sampling_options(params, modifier=modifier)
+            if not type_map:
+                raise ValueError(
+                    "LMDB training data requires model/type_map to be set."
+                )
     groups = resolve_data_groups(config)
     configs = data_group_configs(config)
     datasets = []
-    for group, params in zip(groups, configs, strict=True):
-        sources = resolve_data_sources(group)
-        has_lmdb = any(source.format == "lmdb" for source in sources)
-        children = []
-        child_configs = []
-        for source in sources:
-            if source.format == "deepmd":
-                child = DpLoaderSet(
-                    list(source.systems),
-                    source.batch_size,
-                    type_map,
-                    seed=seed,
-                    modifier=modifier,
+    acquired = []
+    try:
+        for group, params in zip(groups, configs, strict=True):
+            sources = resolve_data_sources(group)
+            has_lmdb = any(source.format == "lmdb" for source in sources)
+            if has_lmdb:
+                validate_lmdb_sampling_options(params, modifier=modifier)
+            children = []
+            child_configs = []
+            for source in sources:
+                if source.format == "deepmd":
+                    child = DpLoaderSet(
+                        list(source.systems),
+                        source.batch_size,
+                        type_map,
+                        seed=seed,
+                        modifier=modifier,
+                    )
+                else:
+                    child = LmdbDataset(
+                        source.systems[0],
+                        type_map,
+                        source.batch_size,
+                        auto_prob_style=group.auto_prob,
+                        seed=seed,
+                    )
+                children.append(child)
+                acquired.append(child)
+                child_configs.append(
+                    {**params, "sys_probs": group.sys_probs if not has_lmdb else None}
                 )
-            else:
-                child = LmdbDataset(
-                    source.systems[0],
-                    type_map,
-                    source.batch_size,
-                    auto_prob_style=group.auto_prob,
+            datasets.append(
+                children[0]
+                if len(children) == 1
+                else GroupedDataset(
+                    children,
+                    child_configs,
+                    [None] * len(children),
+                    uniform=group.auto_prob == "prob_uniform",
                 )
-            children.append(child)
-            child_configs.append(
-                {**params, "sys_probs": group.sys_probs if not has_lmdb else None}
-            )
-        datasets.append(
-            children[0]
-            if len(children) == 1
-            else GroupedDataset(
-                children,
-                child_configs,
-                [None] * len(children),
-                uniform=group.auto_prob == "prob_uniform",
             )
+        return (
+            datasets[0]
+            if len(datasets) == 1
+            else GroupedDataset(datasets, configs, [group.prob for group in groups])
         )
-    return (
-        datasets[0]
-        if len(datasets) == 1
-        else GroupedDataset(datasets, configs, [group.prob for group in groups])
-    )
+    except BaseException:
+        close_data_systems(acquired)
+        raise
 
 
 def create_dataloader(

deepmd/pt/utils/lmdb_dataset.py

@@ -11,6 +11,7 @@
     Any,
 )
 
+import numpy as np
 import torch
 from torch.utils.data import (
     DataLoader,
@@ -206,6 +207,8 @@ class LmdbDataset(Dataset):
           are padded with phantom atoms.
     auto_prob_style : str, optional
         ``auto_prob`` string used to reweight the original systems.
+    seed : int or list[int], optional
+        Seed for shuffled batches, including the worker rank when distributed.
     """
 
     def __init__(
@@ -214,8 +217,14 @@ def __init__(
         type_map: list[str],
         batch_size: int | str = "auto",
         auto_prob_style: str | None = None,
+        seed: int | list[int] | None = None,
     ) -> None:
         self._reader = LmdbDataReader(lmdb_path, type_map, batch_size)
+        # A batch has one periodicity: zero/missing cells must be separated
+        # exactly, including in statistics and full-validation snapshots.
+        self._reader.add_data_requirement(
+            [DataRequirementItem("box", 9, atomic=False, must=False)]
+        )
         self._collate = functools.partial(
             _collate_lmdb_batch, config=self._reader.decode_config
         )
@@ -236,9 +245,17 @@ def __init__(
                     f"nsystems={self._reader.nsystems}"
                 )
 
+        # The common sampler advances integer seeds by epoch. Fold the
+        # rank/seed pair into one reproducible integer before handing it off.
+        sampler_seed = (
+            int(np.random.SeedSequence(seed).generate_state(1)[0])
+            if isinstance(seed, list)
+            else seed
+        )
         sampler = LmdbBatchSampler(
             self._reader,
             shuffle=True,
+            seed=sampler_seed,
             block_targets=self._block_targets,
         )
         self._batch_sampler = _LmdbBatchSamplerTorch(sampler)