多相态数据集生成

ML/Training/Runner/runner_main.cpp

@@ -150,9 +150,19 @@ static void applySampledParamsAndMaybeOverrideRate(HX_NWTM_MODEL_INPUT& in, cons
 			qw.resize(nQ);
 
 			std::copy(p.timeQ.begin(), p.timeQ.end(), tt.begin());
-			std::copy(p.q.begin(), p.q.end(), qo.begin());
+			std::fill(qo.begin(), qo.end(), 0.0);
 			std::fill(qg.begin(), qg.end(), 0.0);
 			std::fill(qw.begin(), qw.end(), 0.0);
+
+			if (in.T == 3 || in.T == 4) {
+				std::copy(p.q.begin(), p.q.end(), qw.begin());
+			}
+			else if (in.T == 5) {
+				std::copy(p.q.begin(), p.q.end(), qg.begin());
+			}
+			else {
+				std::copy(p.q.begin(), p.q.end(), qo.begin());
+			}
 		}
 	}
 

ML/nmWTAI-ML/configs/data_gen.yaml

@@ -16,6 +16,9 @@ cpp:
   license_dat: "../../Bin/Res/license/HXNWTM_license.dat"
 
 dataset_runtime:
+  scene_bin: "scene.bin"  # 相对路径以 paths.temp_dir 为基准，也可填写绝对路径
+  dataset_bin: "dataset.bin"
+  grid_cache_bin: "grid_cache.bin"
   auto_build_dataset_bin: true  # 缺少 dataset.bin 时是否自动调用 training.exe 生成
   use_runner_server: true  # 是否使用常驻 runner 进程加速批量正演
   runner_timeout_sec: 240  # 单次 C++ 正演求解超时时间，单位秒

ML/nmWTAI-ML/configs/data_gen_case_neighborhood.yaml

@@ -16,6 +16,9 @@ cpp:
   license_dat: "../../Bin/Res/license/HXNWTM_license.dat"
 
 dataset_runtime:
+  scene_bin: "scene.bin"  # 相对路径以 paths.temp_dir 为基准，也可填写绝对路径
+  dataset_bin: "dataset.bin"
+  grid_cache_bin: "grid_cache.bin"
   auto_build_dataset_bin: true  # 缺少 dataset.bin 时是否自动调用 training.exe 生成
   use_runner_server: true  # 是否使用常驻 runner 进程加速批量正演
   runner_timeout_sec: 240  # 单次 C++ 正演求解超时时间，单位秒

ML/nmWTAI-ML/configs/data_gen_family_random.yaml

@@ -16,6 +16,9 @@ cpp:
   license_dat: "../../Bin/Res/license/HXNWTM_license.dat"
 
 dataset_runtime:
+  scene_bin: "scene.bin"  # 相对路径以 paths.temp_dir 为基准，也可填写绝对路径
+  dataset_bin: "dataset.bin"
+  grid_cache_bin: "grid_cache.bin"
   auto_build_dataset_bin: true  # 缺少 dataset.bin 时是否自动调用 training.exe 生成
   use_runner_server: true  # 是否使用常驻 runner 进程加速批量正演
   runner_timeout_sec: 240  # 单次 C++ 正演求解超时时间，单位秒

ML/nmWTAI-ML/configs/data_gen_family_random_hard.yaml

@@ -16,6 +16,9 @@ cpp:
   license_dat: "../../Bin/Res/license/HXNWTM_license.dat"
 
 dataset_runtime:
+  scene_bin: "scene.bin"  # 相对路径以 paths.temp_dir 为基准，也可填写绝对路径
+  dataset_bin: "dataset.bin"
+  grid_cache_bin: "grid_cache.bin"
   auto_build_dataset_bin: true  # 缺少 dataset.bin 时是否自动调用 training.exe 生成
   use_runner_server: true  # 是否使用常驻 runner 进程加速批量正演
   runner_timeout_sec: 240  # 单次 C++ 正演求解超时时间，单位秒

ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml

@@ -12,14 +12,70 @@ paths:
 cpp:
   training_exe: "../Training/Release/training.exe"
   runner_exe: "../Training/Release/runner.exe"
-  hx_dll: "../../3rd/Pebi/V1/bin/HX_NWTM.dll"
+  hx_dll: "../Training/Release/HX_NWTM.dll"
   license_dat: "../../Bin/Res/license/HXNWTM_license.dat"
 
 dataset_runtime:
+  scene_bin: "scene.bin"  # 相对路径以 paths.temp_dir 为基准，也可填写绝对路径
+  dataset_bin: "dataset.bin"
+  grid_cache_bin: "grid_cache.bin"
   auto_build_dataset_bin: true  # 缺少 dataset.bin 时是否自动调用 training.exe 生成
   use_runner_server: true  # 是否使用常驻 runner 进程加速批量正演
   runner_timeout_sec: 240  # 单次 C++ 正演求解超时时间，单位秒
 
+dataset_cases:
+  - name: "T1_oil_const_pvt"
+    solver_type: 1
+    dataset_runtime:
+      scene_bin: "scene_T1_oil_const_pvt.bin"
+      dataset_bin: "dataset_T1_oil_const_pvt.bin"
+      grid_cache_bin: "grid_cache_T1_oil_const_pvt.bin"
+    params:
+      active_param_names: ["k", "skin", "wellboreC", "phi", "h"]
+      log_params: ["k", "wellboreC", "h"]
+      fixed_params:
+        Ct: {enabled: true, value: 0.01}
+        Cf: {enabled: true, value: 4.3e-4}
+
+  - name: "T2_oil_variable_pvt"
+    solver_type: 2
+    dataset_runtime:
+      scene_bin: "scene_T2_oil_variable_pvt.bin"
+      dataset_bin: "dataset_T2_oil_variable_pvt.bin"
+      grid_cache_bin: "grid_cache_T2_oil_variable_pvt.bin"
+    params:
+      active_param_names: ["k", "skin", "wellboreC", "phi", "h", "Cf"]
+      log_params: ["k", "wellboreC", "h", "Cf"]
+      fixed_params:
+        Ct: {enabled: true, value: 0.01}
+        Cf: {enabled: false}
+
+  - name: "T3_water_const_pvt"
+    solver_type: 3
+    dataset_runtime:
+      scene_bin: "scene_T3_water_const_pvt.bin"
+      dataset_bin: "dataset_T3_water_const_pvt.bin"
+      grid_cache_bin: "grid_cache_T3_water_const_pvt.bin"
+    params:
+      active_param_names: ["k", "skin", "wellboreC", "phi", "h"]
+      log_params: ["k", "wellboreC", "h"]
+      fixed_params:
+        Ct: {enabled: true, value: 0.01}
+        Cf: {enabled: true, value: 4.3e-4}
+
+  - name: "T4_water_variable_pvt"
+    solver_type: 4
+    dataset_runtime:
+      scene_bin: "scene_T4_water_variable_pvt.bin"
+      dataset_bin: "dataset_T4_water_variable_pvt.bin"
+      grid_cache_bin: "grid_cache_T4_water_variable_pvt.bin"
+    params:
+      active_param_names: ["k", "skin", "wellboreC", "phi", "h", "Cf"]
+      log_params: ["k", "wellboreC", "h", "Cf"]
+      fixed_params:
+        Ct: {enabled: true, value: 0.01}
+        Cf: {enabled: false}
+
 streaming_hdf5:  # HDF5 样本文件流式写入设置
   write_batch_size: 2000  # HDF5 每累计多少条样本批量写入一次
   chunk_rows: 2048  # HDF5 数据分块行数，影响读写效率
@@ -39,7 +95,8 @@ generation:  # 数据集生成规模与随机性设置
   max_fail_examples_per_reason: 50  # 每类失败原因最多保存的示例数量
 
 params:  # 储层与井筒物理参数采样设置
-  all_physical_param_names: ["k", "skin", "wellboreC", "phi", "h", "Ct", "Cf"]  # 写入数据集的完整物理参数列表
+  all_physical_param_names: ["k", "skin", "wellboreC", "phi", "h", "Ct", "Cf"]  # 写入params.bin的完整求解器参数
+  model_param_names: ["k", "skin", "wellboreC", "phi", "h", "Cf", "solverType"]  # 写入H5并交给模型的参数
   active_param_names: ["k", "skin", "wellboreC", "phi", "h", "Ct"]  # 参与采样变化的物理参数列表
   log_params: ["k", "wellboreC", "h", "Ct"]  # 采样时采用对数尺度的物理参数
   ranges:  # 各物理参数采样范围

ML/nmWTAI-ML/scripts/generate_dataset.py

@@ -38,6 +38,12 @@ def main() -> None:
     parser.add_argument("--n-workers", type=int, default=None)
     parser.add_argument("--seed", type=int, default=None)
     parser.add_argument("--method", type=str, default=None)
+    parser.add_argument(
+        "--dataset-case",
+        type=str,
+        default=None,
+        help="Generate only the named dataset_cases entry; omit to generate all cases",
+    )
     parser.add_argument(
         "--dataset-tag",
         type=str,
@@ -50,19 +56,39 @@ def main() -> None:
     if config_path is None:
         config_path = str(config_for_stage(args.stage) or Path("configs/data_gen.yaml"))
 
-    # stage 用来选择预设配置；命令行参数继续覆盖样本数、并行数和随机种子。
+    # 没有dataset_cases时保持原来的单case行为；有case时按声明顺序依次生成。
-    cfg = Config(config_path)
+    base_cfg = Config(config_path)
-    cfg.ensure_dirs()
+    case_names = [str(case.get("name", "")).strip() for case in base_cfg.dataset_cases]
-    path = ParallelDatasetGenerator(
+    if any(not name for name in case_names) or len(case_names) != len(set(case_names)):
-        cfg=cfg,
+        raise ValueError("dataset_cases中的name必须非空且不能重复")
-        n_workers=args.n_workers,
+
-    ).generate(
+    if args.dataset_case is None:
-        n_samples=args.n_samples,
+        selected_cases = case_names or [None]
-        method=args.method,
+    elif args.dataset_case not in case_names:
-        random_seed=args.seed,
+        available = ", ".join(case_names) if case_names else "(none)"
-        dataset_tag=args.dataset_tag,
+        parser.error(f"unknown --dataset-case {args.dataset_case!r}; available: {available}")
-    )
+    else:
-    print(path)
+        selected_cases = [args.dataset_case]
+
+    for case_name in selected_cases:
+        cfg = Config(config_path, dataset_case=case_name)
+        cfg.ensure_dirs()
+        if case_name is None:
+            output_tag = args.dataset_tag
+        else:
+            output_tag = f"{args.dataset_tag}_{case_name}" if args.dataset_tag else case_name
+            print(f"\n=== Generating dataset case: {case_name} ===")
+
+        path = ParallelDatasetGenerator(
+            cfg=cfg,
+            n_workers=args.n_workers,
+        ).generate(
+            n_samples=args.n_samples,
+            method=args.method,
+            random_seed=args.seed,
+            dataset_tag=output_tag,
+        )
+        print(path)
 
 
 if __name__ == "__main__":

ML/nmWTAI-ML/src/common/config.py

@@ -12,6 +12,7 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -29,14 +30,33 @@ class ProjectPaths:
     models_dir: Path
 
 
+def _deep_update(target: dict[str, Any], overrides: dict[str, Any]) -> None:
+    """递归合并case配置，使未覆盖的基础YAML设置继续生效。"""
+    for key, value in overrides.items():
+        if isinstance(value, dict) and isinstance(target.get(key), dict):
+            _deep_update(target[key], value)
+        else:
+            target[key] = deepcopy(value)
+
+
 class Config:
     # pylint: disable=too-many-instance-attributes
     """读取 YAML 配置，并把项目根目录、数据目录、模型目录等派生为可复用属性。"""
-    def __init__(self, config_path: str | Path) -> None:
+    def __init__(self, config_path: str | Path, dataset_case: str | None = None) -> None:
         """读取 YAML 配置，解析项目根目录以及训练、求解器和数据文件路径。"""
         self.path = Path(config_path)
         with open(self.path, "r", encoding="utf-8") as f:
-            self.raw: dict[str, Any] = yaml.safe_load(f)
+            loaded: dict[str, Any] = yaml.safe_load(f)
+
+        self.dataset_cases = list(loaded.get("dataset_cases", []) or [])
+        self.dataset_case = dataset_case
+        self.raw = deepcopy(loaded)
+        if dataset_case is not None:
+            matched = [case for case in self.dataset_cases if case.get("name") == dataset_case]
+            if len(matched) != 1:
+                raise ValueError(f"dataset case not found or duplicated: {dataset_case}")
+            overrides = {key: value for key, value in matched[0].items() if key != "name"}
+            _deep_update(self.raw, overrides)
 
         # project_root 允许写成相对配置文件路径，便于整套项目目录整体移动。
         raw_root = Path(self.raw["paths"]["project_root"])
@@ -58,8 +78,16 @@ class Config:
         self.hx_dll = (root / self.raw["cpp"]["hx_dll"]).resolve()
         self.license_dat = (root / self.raw["cpp"]["license_dat"]).resolve()
 
-        # 这些二进制文件名与 C++ 求解器约定一致，Python 侧只负责确定临时目录。
+        def resolve_runtime_file(key: str, default_name: str) -> Path:
-        self.dataset_bin = self.paths.temp_dir / "dataset.bin"
+            value = Path(self.raw.get("dataset_runtime", {}).get(key, default_name))
+            if value.is_absolute():
+                return value.resolve()
+            return (self.paths.temp_dir / value).resolve()
+
+        # scene/dataset/grid_cache 可按求解器类型分别配置；相对路径以 temp_dir 为基准。
+        self.scene_bin = resolve_runtime_file("scene_bin", "scene.bin")
+        self.dataset_bin = resolve_runtime_file("dataset_bin", "dataset.bin")
+        self.grid_cache_bin = resolve_runtime_file("grid_cache_bin", "grid_cache.bin")
         self.params_bin = self.paths.temp_dir / "params.bin"
         self.result_bin = self.paths.temp_dir / "result.bin"
 
@@ -73,6 +101,8 @@ class Config:
             self.paths.models_dir,
         ]:
             p.mkdir(parents=True, exist_ok=True)
+        self.dataset_bin.parent.mkdir(parents=True, exist_ok=True)
+        self.grid_cache_bin.parent.mkdir(parents=True, exist_ok=True)
 
     def get(self, *keys: str, default=None):
         """按点号路径读取嵌套配置项，例如 training.batch_size；缺失时返回默认值。"""

ML/nmWTAI-ML/src/data/dataset_generation.py

@@ -59,6 +59,7 @@ from src.data.params import Params, Schedule, generate_params_dataset, write_par
 from src.data.schedule_encoding import encode_schedule_to_timegrid
 
 _WORKER_CFG_PATH: Optional[str] = None
+_WORKER_DATASET_CASE: Optional[str] = None
 _WORKDIR: Optional[Path] = None
 _PARAMS_BIN: Optional[Path] = None
 _RESULT_BIN: Optional[Path] = None
@@ -72,6 +73,26 @@ def _seed32(x: int) -> int:
     return int(x) & 0xFFFFFFFF
 
 
+def _model_param_names(cfg: Config) -> list[str]:
+    """返回写入H5并交给代理模型的参数列，允许与求解器完整参数不同。"""
+    params_cfg = cfg.raw["params"]
+    return list(params_cfg.get("model_param_names", params_cfg["all_physical_param_names"]))
+
+
+def _build_model_param_vector(cfg: Config, params: Params) -> np.ndarray:
+    """构造模型参数向量；solverType来自当前dataset case，Ct仍只传给求解器。"""
+    values = []
+    for name in _model_param_names(cfg):
+        if name == "solverType":
+            solver_type = cfg.get("solver_type", default=None)
+            if solver_type is None:
+                raise ValueError("model_param_names包含solverType，但当前case未配置solver_type")
+            values.append(float(solver_type))
+        else:
+            values.append(float(getattr(params, name)))
+    return np.asarray(values, dtype=np.float32)
+
+
 def _read_result_bin(result_bin_path: Path) -> Optional[Dict[str, Any]]:
     """读取 C++ 求解器输出的二进制结果文件，并解析时间、压力和双对数曲线。"""
     try:
@@ -378,12 +399,13 @@ def _resolve_section_indices(cfg: Config, timeQ, q, rng):
     raise ValueError(f"Unknown section_policy.mode: {mode}")
 
 
-def _init_process_worker(cfg_path: str):
+def _init_process_worker(cfg_path: str, dataset_case: str | None):
     """初始化并行工作进程中的全局配置、临时目录和求解器路径。"""
-    global _WORKER_CFG_PATH, _WORKDIR, _PARAMS_BIN, _RESULT_BIN, _SUBPROC_ENV
+    global _WORKER_CFG_PATH, _WORKER_DATASET_CASE, _WORKDIR, _PARAMS_BIN, _RESULT_BIN, _SUBPROC_ENV
 
     _WORKER_CFG_PATH = cfg_path
-    cfg = Config(cfg_path)
+    _WORKER_DATASET_CASE = dataset_case
+    cfg = Config(cfg_path, dataset_case=dataset_case)
     cfg.ensure_dirs()
 
     base_worker_dir = cfg.paths.temp_dir / "parallel_workers"
@@ -405,7 +427,7 @@ def _init_process_worker(cfg_path: str):
 def _worker_simulate_parallel(args):
     """在工作进程中写入参数、调用求解器、清洗曲线并返回一个训练样本。"""
     task_idx, params_dict = args
-    cfg = Config(_WORKER_CFG_PATH)
+    cfg = Config(_WORKER_CFG_PATH, dataset_case=_WORKER_DATASET_CASE)
 
     try:
         # 子进程之间不共享 Python 对象，这里从可序列化字典重建 Params/Schedule。
@@ -486,10 +508,7 @@ def _worker_simulate_parallel(args):
             n_sections=len(sch.timeQ),
         )
 
-        params_vec = np.asarray(
+        params_vec = _build_model_param_vector(cfg, p)
-            [float(getattr(p, name)) for name in cfg.raw["params"]["all_physical_param_names"]],
-            dtype=np.float32,
-        )
 
         schedule_parts = [
             np.asarray(enc.x_sched, dtype=np.float32).reshape(-1),
@@ -662,7 +681,12 @@ class ParallelDatasetGenerator:
         """执行并行数据生成主循环，直到收集到目标数量的有效样本。"""
         try:
             result = subprocess.run(
-                [str(self.cfg.training_exe)],
+                [
+                    str(self.cfg.training_exe),
+                    str(self.cfg.scene_bin),
+                    str(self.cfg.dataset_bin),
+                    str(self.cfg.grid_cache_bin),
+                ],
                 cwd=str(self.cfg.training_exe.parent),
                 timeout=60,
                 stdout=subprocess.DEVNULL,
@@ -674,12 +698,6 @@ class ParallelDatasetGenerator:
             if result.returncode == 0 and self.cfg.dataset_bin.exists():
                 return True
 
-            fallback = self.cfg.training_exe.parent / "dataset.bin"
-            if result.returncode == 0 and fallback.exists():
-                self.cfg.dataset_bin.parent.mkdir(parents=True, exist_ok=True)
-                shutil.copy2(fallback, self.cfg.dataset_bin)
-                return True
-
             return False
         except Exception:
             return False
@@ -718,7 +736,7 @@ class ParallelDatasetGenerator:
         sched_dim = n_u_points * n_channels + cfg.sec_feat_dim
         if bool(cfg.raw.get("schedule", {}).get("use_metadata_features_for_model", False)):
             sched_dim += len(cfg.raw.get("schedule", {}).get("metadata_features_for_model", []) or [])
-        param_names = list(cfg.raw["params"]["all_physical_param_names"])
+        param_names = _model_param_names(cfg)
         param_dim = len(param_names)
 
         # HDF5 布局与正演模型输入输出一致：
@@ -856,7 +874,7 @@ class ParallelDatasetGenerator:
                 with ProcessPoolExecutor(
                     max_workers=self.n_workers,
                     initializer=_init_process_worker,
-                    initargs=(str(cfg.path),),
+                    initargs=(str(cfg.path), cfg.dataset_case),
                 ) as executor:
                     task_iter = iter(tasks)
                     in_flight = deque()

ML/nmWTAI-ML/src/data/runner_client.py

@@ -52,7 +52,9 @@ class CppRunner:
         self.temp_dir = Path(temp_dir) if temp_dir is not None else cfg.paths.temp_dir
         self.temp_dir.mkdir(parents=True, exist_ok=True)
 
-        self.dataset_bin = self.temp_dir / "dataset.bin"
+        self.dataset_bin = (
+            cfg.dataset_bin if temp_dir is None else self.temp_dir / cfg.dataset_bin.name
+        )
         self.params_bin = self.temp_dir / "params.bin"
         self.result_bin = self.temp_dir / "result.bin"
 
@@ -63,10 +65,10 @@ class CppRunner:
         )
         self.proc: Optional[subprocess.Popen] = None
 
-        if auto_init and not self.dataset_bin.exists():
+        if auto_init and not cfg.dataset_bin.exists():
-            # dataset.bin 是 runner.exe 的输入数据库；缺失时尝试调用训练端生成。
+            # 配置指定的 dataset 是 runner.exe 的输入数据库；缺失时尝试调用训练端生成。
             if not self.run_training():
-                raise RuntimeError("Failed to generate dataset.bin")
+                raise RuntimeError(f"Failed to generate dataset: {cfg.dataset_bin}")
 
         if not self.dataset_bin.exists() and cfg.dataset_bin.exists():
             try:
@@ -107,7 +109,12 @@ class CppRunner:
     def run_training(self, timeout: int = 60) -> bool:
         """调用 C++ 端训练/预处理命令，并返回是否执行成功。"""
         result = subprocess.run(
-            [str(self.training_exe)],
+            [
+                str(self.training_exe),
+                str(self.cfg.scene_bin),
+                str(self.cfg.dataset_bin),
+                str(self.cfg.grid_cache_bin),
+            ],
             cwd=str(self.training_exe.parent),
             capture_output=True,
             text=True,
@@ -121,16 +128,11 @@ class CppRunner:
         if result.returncode != 0:
             return False
 
-        if self.dataset_bin.exists():
+        if self.cfg.dataset_bin.exists() and self.dataset_bin != self.cfg.dataset_bin:
-            return True
-
-        fallback = self.training_exe.parent / "dataset.bin"
-        if fallback.exists():
             self.dataset_bin.parent.mkdir(parents=True, exist_ok=True)
-            shutil.copy2(fallback, self.dataset_bin)
+            shutil.copy2(self.cfg.dataset_bin, self.dataset_bin)
-            return True
 
-        return False
+        return self.dataset_bin.exists()
 
     def _ensure_server_started(self) -> None:
         """按需启动常驻求解器服务，并等待健康检查通过。"""