scripts/fake_data.py

import pickle
import sys
from pathlib import Path

import numpy as np


PROJECT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_ROOT))

from onescience.utils.YParams import YParams


def resolve_path(path_value):
    path = Path(path_value)
    return path if path.is_absolute() else PROJECT_ROOT / path


def make_fields(num_samples, height, width, seed):
    rng = np.random.default_rng(seed)
    y_axis = np.linspace(-1.0, 1.0, height, dtype=np.float32)
    x_axis = np.linspace(-1.0, 1.0, width, dtype=np.float32)
    yy, xx = np.meshgrid(y_axis, x_axis, indexing="ij")

    x = np.empty((num_samples, 3, height, width), dtype=np.float32)
    target = np.empty_like(x)

    for i in range(num_samples):
        cx = 0.2 * np.sin(i)
        cy = 0.2 * np.cos(i * 0.7)
        radius = 0.25 + 0.03 * (i % 3)
        distance = np.sqrt((xx - cx) ** 2 + (yy - cy) ** 2) - radius
        obstacle = (distance < 0.0).astype(np.float32)
        inlet = np.ones_like(xx, dtype=np.float32) * (1.0 + 0.05 * i)

        x[i, 0] = obstacle
        x[i, 1] = distance
        x[i, 2] = inlet

        ux = inlet * (1.0 - obstacle) + 0.1 * np.sin(np.pi * yy)
        uy = -0.2 * yy * np.exp(-4.0 * np.maximum(distance, 0.0))
        pressure = 0.5 * (1.0 - xx) + 0.1 * obstacle

        noise = 0.005 * rng.standard_normal((3, height, width)).astype(np.float32)
        target[i, 0] = ux
        target[i, 1] = uy
        target[i, 2] = pressure
        target[i] += noise

    return x, target


def main():
    cfg = YParams(str(PROJECT_ROOT / "config" / "config.yaml"), "root")
    data_dir = resolve_path(cfg.datapipe.source.data_dir)
    data_dir.mkdir(parents=True, exist_ok=True)

    x, y = make_fields(
        num_samples=cfg.fake_data.num_samples,
        height=cfg.fake_data.height,
        width=cfg.fake_data.width,
        seed=cfg.fake_data.seed,
    )

    with open(data_dir / cfg.datapipe.source.data_x_name, "wb") as f:
        pickle.dump(x, f)
    with open(data_dir / cfg.datapipe.source.data_y_name, "wb") as f:
        pickle.dump(y, f)

    print(f"Fake DeepCFD data written to {data_dir}")
    print(f"x shape={x.shape}, y shape={y.shape}, dtype={x.dtype}")


if __name__ == "__main__":
    main()