File size: 3,692 Bytes
0d2f59e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """Generate dCDB9 collocation points and grids, with no CFD training data.
The filename follows the requested package interface. The generated data are
physical coordinates and exact initial conditions, not fabricated CFD fields.
"""
import numpy as np
from common import arguments, load_config, output_path, save_json, sha256
def initial_values(points, physics):
target = np.zeros((len(points), 3), dtype=np.float32)
inside = np.sum(points[:, :2] ** 2, axis=1) <= physics['dam_radius'] ** 2
target[:, 0] = np.where(inside, physics['depth_inside'], physics['depth_outside'])
return target
def sample_points(counts, physics, seed):
rng = np.random.default_rng(seed)
lower = np.array([physics['x_range'][0], physics['y_range'][0], physics['t_range'][0]])
upper = np.array([physics['x_range'][1], physics['y_range'][1], physics['t_range'][1]])
interior = rng.uniform(lower, upper, (counts['interior'], 3)).astype(np.float32)
initial = rng.uniform(lower, upper, (counts['initial'], 3)).astype(np.float32)
initial[:, 2] = physics['t_range'][0]
if counts['boundary'] % 4:
raise ValueError('Boundary count must be divisible by four for per-side losses.')
per_side = counts['boundary'] // 4
boundary = rng.uniform(lower, upper, (counts['boundary'], 3)).astype(np.float32)
pair = boundary.copy()
for side in range(4):
axis, high = side // 2, side % 2
sl = slice(side * per_side, (side + 1) * per_side)
boundary[sl, axis] = upper[axis] if high else lower[axis]
pair[sl, axis] = lower[axis] if high else upper[axis]
return {'interior': interior, 'initial': initial,
'initial_target': initial_values(initial, physics),
'boundary': boundary, 'boundary_pair': pair,
'boundary_side': np.repeat(np.arange(4, dtype=np.int64), per_side)}
def generate(config, root):
samples_path = output_path(root, config['paths']['samples'])
grid_path = output_path(root, config['paths']['grid'])
samples_path.parent.mkdir(parents=True, exist_ok=True)
train = sample_points(config['sampling'], config['physics'], config['seed'])
val = sample_points(config['validation'], config['physics'], config['seed'] + 1)
arrays = {**train, **{'val_' + k: v for k, v in val.items()}}
np.savez_compressed(samples_path, **arrays)
spacing = config['evaluation']['grid_spacing']
xmin, xmax = config['physics']['x_range']
ymin, ymax = config['physics']['y_range']
nx, ny = round((xmax - xmin) / spacing), round((ymax - ymin) / spacing)
x = xmin + (np.arange(nx) + 0.5) * spacing
y = ymin + (np.arange(ny) + 0.5) * spacing
xx, yy = np.meshgrid(x, y, indexing='xy')
np.savez_compressed(grid_path, x=x, y=y, xy=np.column_stack((xx.ravel(), yy.ravel())),
times=np.array(config['evaluation']['times']), spacing=spacing)
manifest = {'source': 'paper Eq.55 and periodic boundary geometry only',
'external_training_data': False, 'seed': config['seed'],
'train': {k: list(v.shape) for k, v in train.items()},
'validation': {k: list(v.shape) for k, v in val.items()},
'grid_shape': [ny, nx], 'samples_sha256': sha256(samples_path),
'grid_sha256': sha256(grid_path), 'units': ['m', 'm', 's'],
'initial_target_units': ['m', 'm/s', 'm/s']}
save_json(samples_path.parent / 'sampling_manifest.json', manifest)
print('SAMPLING', manifest, flush=True)
return samples_path
if __name__ == '__main__':
args = arguments(__doc__).parse_args()
config, root = load_config(args.config)
generate(config, root)
|