diff --git a/dpgen2/op/run_dp_train.py b/dpgen2/op/run_dp_train.py index c0cf2d4d..b4bf8f6a 100644 --- a/dpgen2/op/run_dp_train.py +++ b/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() diff --git a/tests/op/test_run_dp_train.py b/tests/op/test_run_dp_train.py index 45ba950c..8926e21a 100644 --- a/tests/op/test_run_dp_train.py +++ b/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):