fix: skip training with no expanded systems

dpgen2/op/run_dp_train.py

@@ -206,6 +206,18 @@ def execute(
             iter_data_new_exp = train_systems
             valid_data = append_valid_data(config, valid_data, valid_systems)
         iter_data_exp = iter_data_old_exp + iter_data_new_exp
+        if isinstance(init_data, dict):
+            if config["multitask"]:
+                has_init_training_data = len(init_data.get(config["head"], [])) > 0
+            else:
+                has_init_training_data = any(
+                    len(systems) > 0 for systems in init_data.values()
+                )
+        else:
+            has_init_training_data = len(init_data) > 0
+        # Initial data is expanded when the workflow is submitted, while a
+        # non-empty iteration artifact list may expand to zero systems here.
+        training_systems_empty = not has_init_training_data and len(iter_data_exp) == 0
         work_dir = Path(task_name)
         init_model_with_finetune = config["init_model_with_finetune"]
 
@@ -231,7 +243,7 @@ def execute(
             old_ratio = config["init_model_old_ratio"]
             if config["multitask"]:
                 head = config["head"]
-                len_init = len(init_data[head])
+                len_init = len(init_data.get(head, []))
             else:
                 len_init = len(init_data)
             numb_old = len_init + len(iter_data_old_exp)
@@ -269,7 +281,12 @@ def execute(
         )
 
         if RunDPTrain.skip_training(
-            work_dir, train_dict, init_model, iter_data, finetune_mode
+            work_dir,
+            train_dict,
+            init_model,
+            iter_data,
+            finetune_mode,
+            training_systems_empty=training_systems_empty,
         ):
             return OPIO(
                 {
@@ -462,18 +479,26 @@ def skip_training(
         init_model,
         iter_data,
         finetune_mode,
+        training_systems_empty=False,
     ):
         # do not skip if we do finetuning
         if finetune_mode is not None and finetune_mode == "finetune":
             return False
-        # we have init model and no iter data, skip training
-        if (init_model is not None) and (iter_data is None or len(iter_data) == 0):
+        # Reuse the supplied model when there is no new iteration data or when
+        # all configured inputs expand to zero actual training systems.
+        no_iter_data = iter_data is None or len(iter_data) == 0
+        if (init_model is not None) and (no_iter_data or training_systems_empty):
             with set_directory(work_dir):
                 with open(train_script_name, "w") as fp:
                     json.dump(train_dict, fp, indent=4)
+                reason = (
+                    "no expanded training systems"
+                    if training_systems_empty
+                    else "no iteration training data"
+                )
                 Path("train.log").write_text(
                     f"We have init model {init_model} and "
-                    f"no iteration training data. "
+                    f"{reason}. "
                     f"The training is skipped.\n"
                 )
                 Path("lcurve.out").touch()

tests/op/test_run_dp_train.py

@@ -1047,6 +1047,118 @@ def test_exec_v2_empty_dir(self, mocked_run):
             jdata = json.load(fp)
             self.assertDictEqual(jdata, self.expected_odict_v2)
 
+    @patch("dpgen2.op.run_dp_train.run_command")
+    def test_exec_v2_fully_empty_training_systems(self, mocked_run):
+        """Propagate the initial model instead of launching an empty train."""
+        mocked_run.side_effect = [(0, "foo\n", ""), (0, "bar\n", "")]
+
+        config = self.config.copy()
+        config["init_model_policy"] = "yes"
+
+        task_path = Path(self.task_path)
+        task_path.mkdir(exist_ok=True)
+        with open(task_path / train_script_name, "w") as fp:
+            json.dump(self.idict_v2, fp, indent=4)
+
+        empty_data = Path("foo")
+        empty_data.mkdir(exist_ok=True)
+        init_model = Path(self.init_model).absolute()
+        init_model.write_text("this is init model")
+        self.addCleanup(init_model.unlink, missing_ok=True)
+
+        out = RunDPTrain().execute(
+            OPIO(
+                {
+                    "config": config,
+                    "task_name": self.task_name,
+                    "task_path": task_path,
+                    "init_model": init_model,
+                    "init_data": [],
+                    "iter_data": [empty_data],
+                }
+            )
+        )
+
+        mocked_run.assert_not_called()
+        self.assertEqual(out["model"], init_model)
+        self.assertIn("no expanded training systems", out["log"].read_text())
+        with open(out["script"]) as fp:
+            train_dict = json.load(fp)
+        self.assertEqual(train_dict["training"]["training_data"]["systems"], [])
+        self.assertEqual(
+            train_dict["training"]["training_data"]["auto_prob"],
+            "prob_sys_size",
+        )
+
+    @patch("dpgen2.op.run_dp_train.run_command")
+    def test_exec_v2_empty_active_multitask_head(self, mocked_run):
+        """Ignore pretrained data belonging only to inactive heads."""
+        mocked_run.side_effect = [(0, "foo\n", ""), (0, "bar\n", "")]
+
+        config = self.config.copy()
+        config.update(
+            {
+                "init_model_policy": "yes",
+                "multitask": True,
+                "head": "A",
+            }
+        )
+        multitask_script = {
+            "training": {
+                "data_dict": {
+                    "A": {"training_data": {"systems": []}},
+                    "B": {"training_data": {"systems": []}},
+                }
+            },
+            "learning_rate": {"start_lr": 1.0},
+            "loss_dict": {
+                head: {
+                    "start_pref_e": 1.0,
+                    "start_pref_f": 1.0,
+                    "start_pref_v": 1.0,
+                }
+                for head in ("A", "B")
+            },
+        }
+
+        task_path = Path(self.task_path)
+        task_path.mkdir(exist_ok=True)
+        with open(task_path / train_script_name, "w") as fp:
+            json.dump(multitask_script, fp, indent=4)
+
+        empty_data = Path("foo")
+        empty_data.mkdir(exist_ok=True)
+        init_model = Path(self.init_model).absolute()
+        init_model.write_text("this is init model")
+        self.addCleanup(init_model.unlink, missing_ok=True)
+
+        out = RunDPTrain().execute(
+            OPIO(
+                {
+                    "config": config,
+                    "task_name": self.task_name,
+                    "task_path": task_path,
+                    "init_model": init_model,
+                    # Head A intentionally has no entry; only inactive head B
+                    # carries pretrained data.
+                    "init_data": {"B": [self.init_data[0]]},
+                    "iter_data": [empty_data],
+                }
+            )
+        )
+
+        mocked_run.assert_not_called()
+        self.assertEqual(out["model"], init_model)
+        self.assertIn("no expanded training systems", out["log"].read_text())
+        with open(out["script"]) as fp:
+            train_dict = json.load(fp)
+        data_dict = train_dict["training"]["data_dict"]
+        self.assertEqual(data_dict["A"]["training_data"]["systems"], [])
+        self.assertEqual(
+            data_dict["B"]["training_data"]["systems"],
+            [str(self.init_data[0])],
+        )
+
 
 class TestSplitValid(unittest.TestCase):
     def setUp(self):