diff --git a/ML/Training/Runner/runner_main.cpp b/ML/Training/Runner/runner_main.cpp index 6d52b24..e9ad15f 100644 --- a/ML/Training/Runner/runner_main.cpp +++ b/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()); + } } } diff --git a/ML/nmWTAI-ML/configs/data_gen.yaml b/ML/nmWTAI-ML/configs/data_gen.yaml index 868f9d7..8c138c7 100644 --- a/ML/nmWTAI-ML/configs/data_gen.yaml +++ b/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++ 正演求解超时时间，单位秒 diff --git a/ML/nmWTAI-ML/configs/data_gen_case_neighborhood.yaml b/ML/nmWTAI-ML/configs/data_gen_case_neighborhood.yaml index dce29d3..1b93208 100644 --- a/ML/nmWTAI-ML/configs/data_gen_case_neighborhood.yaml +++ b/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++ 正演求解超时时间，单位秒 diff --git a/ML/nmWTAI-ML/configs/data_gen_family_random.yaml b/ML/nmWTAI-ML/configs/data_gen_family_random.yaml index 74e19ee..23ab1a2 100644 --- a/ML/nmWTAI-ML/configs/data_gen_family_random.yaml +++ b/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++ 正演求解超时时间，单位秒 diff --git a/ML/nmWTAI-ML/configs/data_gen_family_random_hard.yaml b/ML/nmWTAI-ML/configs/data_gen_family_random_hard.yaml index 15ac856..45ef0a9 100644 --- a/ML/nmWTAI-ML/configs/data_gen_family_random_hard.yaml +++ b/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++ 正演求解超时时间，单位秒 diff --git a/ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml b/ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml index 8530763..ab166e2 100644 --- a/ML/nmWTAI-ML/configs/data_gen_family_random_v2_q.yaml +++ b/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: # 各物理参数采样范围 diff --git a/ML/nmWTAI-ML/scripts/generate_dataset.py b/ML/nmWTAI-ML/scripts/generate_dataset.py index f000ae1..fe68f4c 100644 --- a/ML/nmWTAI-ML/scripts/generate_dataset.py +++ b/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 用来选择预设配置；命令行参数继续覆盖样本数、并行数和随机种子。 - cfg = Config(config_path) - cfg.ensure_dirs() - path = ParallelDatasetGenerator( - cfg=cfg, - n_workers=args.n_workers, - ).generate( - n_samples=args.n_samples, - method=args.method, - random_seed=args.seed, - dataset_tag=args.dataset_tag, - ) - print(path) + # 没有dataset_cases时保持原来的单case行为；有case时按声明顺序依次生成。 + base_cfg = Config(config_path) + case_names = [str(case.get("name", "")).strip() for case in base_cfg.dataset_cases] + if any(not name for name in case_names) or len(case_names) != len(set(case_names)): + raise ValueError("dataset_cases中的name必须非空且不能重复") + + if args.dataset_case is None: + selected_cases = case_names or [None] + elif args.dataset_case not in case_names: + available = ", ".join(case_names) if case_names else "(none)" + parser.error(f"unknown --dataset-case {args.dataset_case!r}; available: {available}") + else: + 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__": diff --git a/ML/nmWTAI-ML/src/common/config.py b/ML/nmWTAI-ML/src/common/config.py index c653e0f..c283d1d 100644 --- a/ML/nmWTAI-ML/src/common/config.py +++ b/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 侧只负责确定临时目录。 - self.dataset_bin = self.paths.temp_dir / "dataset.bin" + def resolve_runtime_file(key: str, default_name: str) -> Path: + 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；缺失时返回默认值。""" diff --git a/ML/nmWTAI-ML/src/data/dataset_generation.py b/ML/nmWTAI-ML/src/data/dataset_generation.py index 329def1..14c447d 100644 --- a/ML/nmWTAI-ML/src/data/dataset_generation.py +++ b/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( - [float(getattr(p, name)) for name in cfg.raw["params"]["all_physical_param_names"]], - dtype=np.float32, - ) + params_vec = _build_model_param_vector(cfg, p) 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() diff --git a/ML/nmWTAI-ML/src/data/runner_client.py b/ML/nmWTAI-ML/src/data/runner_client.py index 968ead7..d6571a9 100644 --- a/ML/nmWTAI-ML/src/data/runner_client.py +++ b/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(): - # dataset.bin 是 runner.exe 的输入数据库；缺失时尝试调用训练端生成。 + if auto_init and not cfg.dataset_bin.exists(): + # 配置指定的 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(): - return True - - fallback = self.training_exe.parent / "dataset.bin" - if fallback.exists(): + if self.cfg.dataset_bin.exists() and self.dataset_bin != self.cfg.dataset_bin: self.dataset_bin.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(fallback, self.dataset_bin) - return True + shutil.copy2(self.cfg.dataset_bin, self.dataset_bin) - return False + return self.dataset_bin.exists() def _ensure_server_started(self) -> None: """按需启动常驻求解器服务，并等待健康检查通过。"""