File size: 4,856 Bytes
ef2ae28 | 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 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 | """Generate small structured wildfire sequences with physical correlations."""
import argparse
from pathlib import Path
import numpy as np
import yaml
ROOT = Path(__file__).resolve().parents[1]
def make_split(path, count, config, seed, day_offset):
rng = np.random.default_rng(seed)
data_config = config["data"]
time = int(data_config["sequence_days"])
height = int(data_config["patch_height"])
width = int(data_config["patch_width"])
yy, xx = np.mgrid[:height, :width].astype(np.float32)
inputs = np.empty((count, time, 25, height, width), dtype=np.float32)
danger_scores = np.empty(count, dtype=np.float32)
for sample in range(count):
center_y = height / 2 + rng.uniform(-3, 3)
center_x = width / 2 + rng.uniform(-3, 3)
hotspot = np.exp(-((yy - center_y) ** 2 + (xx - center_x) ** 2) / (2 * rng.uniform(4, 7) ** 2))
elevation = np.clip(0.25 + 0.018 * yy + 0.012 * xx + rng.normal(0, 0.02, (height, width)), 0, 1)
slope = np.clip(np.hypot(*np.gradient(elevation)) * 15, 0, 1)
road = np.clip(np.abs(xx - rng.uniform(5, 20)) / 20, 0, 1)
water = np.clip(np.abs(yy - (height / 2 + 2 * np.sin(xx / 4))) / 18, 0, 1)
population = np.exp(-((xx - rng.uniform(5, 20)) ** 2 + (yy - rng.uniform(5, 20)) ** 2) / 60)
cover_logits = rng.normal(0, 0.8, (10, height, width))
cover_logits += np.stack([np.sin((xx + index) / (3 + index / 3)) for index in range(10)])
cover = np.exp(cover_logits - cover_logits.max(axis=0, keepdims=True))
cover /= cover.sum(axis=0, keepdims=True)
weather = rng.normal(0, 0.45)
for day in range(time):
weather = 0.82 * weather + rng.normal(0, 0.25)
drying = day / max(time - 1, 1)
spatial_noise = rng.normal(0, 0.025, (height, width))
temperature = 0.50 + 0.16 * weather + 0.20 * drying + 0.16 * hotspot + spatial_noise
wind = 0.32 + 0.12 * weather + 0.10 * hotspot + rng.normal(0, 0.035, (height, width))
humidity = 0.62 - 0.19 * weather - 0.20 * drying - 0.14 * hotspot + spatial_noise
precipitation = np.clip(0.30 - 0.13 * weather - 0.18 * drying - 0.10 * hotspot + spatial_noise, 0, 1)
dewpoint = 0.55 * temperature + 0.40 * humidity
pressure = 0.55 - 0.06 * weather + 0.02 * hotspot + spatial_noise
ndvi = np.clip(0.62 - 0.14 * drying - 0.08 * hotspot + 0.08 * cover[2], 0, 1)
day_lst = np.clip(temperature + 0.10 * hotspot, 0, 1)
night_lst = np.clip(temperature - 0.16 + 0.04 * hotspot, 0, 1)
soil_moisture = np.clip(0.58 * humidity + 0.42 * precipitation - 0.10 * drying, 0, 1)
dynamic = [temperature, wind, humidity, precipitation, dewpoint, pressure,
ndvi, day_lst, night_lst, soil_moisture]
static = [road, water, population, elevation, slope, *cover]
inputs[sample, day] = np.stack(dynamic + static).astype(np.float32)
cy, cx = height // 2, width // 2
latest = inputs[sample, -1, :, cy, cx]
danger_scores[sample] = (1.7 * latest[0] + 1.1 * latest[1] - 1.5 * latest[2]
- 1.2 * latest[9] - 0.35 * latest[10]
+ 0.25 * latest[12] + rng.normal(0, 0.12))
labels = (danger_scores >= np.median(danger_scores)).astype(np.float32)[:, None]
timestamps = (np.datetime64("2018-06-01") + (np.arange(count) + day_offset).astype("timedelta64[D]"))
timestamps = timestamps.astype("datetime64[s]").astype(np.int64)
latitude = rng.uniform(34.0, 43.0, count).astype(np.float32)
longitude = rng.uniform(19.0, 30.0, count).astype(np.float32)
np.savez_compressed(
path, inputs=inputs, labels=labels, timestamps_unix_s=timestamps,
coords=np.column_stack((latitude, longitude)).astype(np.float32),
format_version=np.asarray(data_config["format_version"]),
data_source=np.asarray("structured_synthetic"), input_layout=np.asarray("BTCHW"),
)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--force", action="store_true")
args = parser.parse_args()
config = yaml.safe_load((ROOT / "conf/config.yaml").read_text())
output = ROOT / config["data"]["root"]
output.mkdir(parents=True, exist_ok=True)
splits = (("train.npz", int(config["data"]["train_samples"]), 0),
("test.npz", int(config["data"]["test_samples"]), 1000))
for offset, (name, count, day_offset) in enumerate(splits):
path = output / name
if args.force or not path.exists():
make_split(path, count, config, int(config["seed"]) + offset, day_offset)
print(f"generated={path.relative_to(ROOT)} samples={count} shape={count},10,25,25,25")
if __name__ == "__main__":
main()
|