File size: 2,510 Bytes
9f29df6 | 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 | """Generate a compact, full-resolution synthetic RYDL-like sequence."""
from datetime import datetime, timedelta
from pathlib import Path
import sys
import h5py
import numpy as np
import yaml
ROOT = Path(__file__).resolve().parents[1]
def load_config():
with (ROOT / "conf/config.yaml").open(encoding="utf-8") as handle:
return yaml.safe_load(handle)
def precipitation_frame(y, x, step, rng):
cells = (
(220 + 4.2 * step, 260 + 6.0 * step, 70 + 0.5 * step, 45, 1.25),
(610 - 3.2 * step, 590 + 2.7 * step, 58, 82 - 0.4 * step, 0.85),
(430 + 1.5 * step, 710 - 4.0 * step, 42 + 0.6 * step, 55, 0.55),
)
field = np.zeros_like(x, dtype=np.float32)
for cy, cx, sy, sx, amplitude in cells:
evolution = 1.0 + 0.12 * np.sin((step + amplitude) / 3.0)
field += amplitude * evolution * np.exp(
-0.5 * (((x - cx) / sx) ** 2 + ((y - cy) / sy) ** 2)
)
# A smooth perturbation evolves with the cells without decorrelating frames.
phase = rng.uniform(-0.03, 0.03)
field *= 1.0 + 0.025 * np.sin(x / 35.0 + step / 4.0 + phase) * np.cos(y / 47.0)
return np.maximum(field, 0).astype(np.float32)
def main():
config = load_config()
data = config["data"]
np.random.seed(config["seed"])
rng = np.random.default_rng(config["seed"])
total = data["train_frames"] + data["val_frames"] + data["test_frames"]
height, width = data["raw_height"], data["raw_width"]
yy, xx = np.mgrid[:height, :width].astype(np.float32)
path = ROOT / data["path"]
path.parent.mkdir(parents=True, exist_ok=True)
start = datetime(2017, 1, 1)
keys = []
with h5py.File(path, "w") as handle:
handle.attrs["units"] = "mm/5min"
handle.attrs["interval_minutes"] = data["interval_minutes"]
for index in range(total):
key = (start + timedelta(minutes=index * data["interval_minutes"])).strftime(
"%Y%m%d%H%M"
)
keys.append(key)
handle.create_dataset(key, data=precipitation_frame(yy, xx, index, rng))
with h5py.File(path, "r") as handle:
first = handle[keys[0]][...]
print(f"Number of frames: {len(keys)}")
print(f"First key: {keys[0]}")
print(f"Last key: {keys[-1]}")
print(f"Frame shape: {first.shape}")
print(f"dtype: {first.dtype}")
print(f"min: {first.min():.8f}")
print(f"max: {first.max():.8f}")
print(f"mean: {first.mean():.8f}")
if __name__ == "__main__":
main()
|