scripts/fake_data.py

from __future__ import annotations

import argparse
from pathlib import Path

import h5py
import numpy as np

from common import DEFAULT_CONFIG, load_config, prepare_config


def generate_darcy_file(path: Path, samples: int, grid_size: int, seed: int, overwrite: bool) -> None:
    if path.exists() and not overwrite:
        print(f"Fake data already exists: {path}")
        return

    path.parent.mkdir(parents=True, exist_ok=True)
    rng = np.random.default_rng(seed)
    x = np.linspace(0.0, 1.0, grid_size, dtype=np.float32)
    y = np.linspace(0.0, 1.0, grid_size, dtype=np.float32)
    xx, yy = np.meshgrid(x, y, indexing="ij")

    nu = rng.uniform(0.05, 0.25, size=(samples, grid_size, grid_size)).astype(np.float32)
    tensor = np.empty((samples, 2, grid_size, grid_size), dtype=np.float32)
    base = np.sin(np.pi * xx) * np.sin(np.pi * yy)
    for idx in range(samples):
        tensor[idx, 0] = base + 0.05 * rng.standard_normal(base.shape)
        tensor[idx, 1] = tensor[idx, 0] * (1.0 + nu[idx]) + 0.01 * rng.standard_normal(base.shape)

    with h5py.File(path, "w") as handle:
        handle.create_dataset("tensor", data=tensor)
        handle.create_dataset("nu", data=nu)
        handle.create_dataset("x-coordinate", data=x)
        handle.create_dataset("y-coordinate", data=y)
        handle.create_dataset("t-coordinate", data=np.array([0.0, 1.0], dtype=np.float32))

    print(f"Fake Darcy HDF5 written: {path}")
    print(f"tensor={tensor.shape}, nu={nu.shape}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Generate a small PDENNEval FNO fake dataset.")
    parser.add_argument("--config", default=str(DEFAULT_CONFIG), help="Path to conf/config.yaml")
    parser.add_argument("--data-dir", default=None, help="Override datapipe.source.data_dir")
    parser.add_argument("--samples", type=int, default=20)
    parser.add_argument("--grid-size", type=int, default=16)
    parser.add_argument("--seed", type=int, default=0)
    parser.add_argument("--overwrite", action="store_true")
    args = parser.parse_args()

    cfg = prepare_config(load_config(args.config), data_dir=args.data_dir)
    data_dir = Path(cfg.datapipe.source.data_dir)
    file_path = data_dir / cfg.datapipe.source.file_name
    generate_darcy_file(file_path, args.samples, args.grid_size, args.seed, args.overwrite)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())