File size: 3,013 Bytes
43dce84 | 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 | """Generate compact AVHRR-like daily SST data for engineering validation."""
from datetime import date, timedelta
from pathlib import Path
import numpy as np
import yaml
ROOT = Path(__file__).resolve().parents[1]
def main():
config = yaml.safe_load((ROOT / "conf/config.yaml").read_text())
data_config = config["data"]
n = int(data_config["samples"])
height, width = map(int, data_config["grid"])
if height != width or height < 16 or height % 16 or n < 4:
raise ValueError("DINCAE engineering grids must be square, divisible by 16, and contain at least four days")
rng = np.random.default_rng(int(config["seed"]))
yy, xx = np.mgrid[-1:1:complex(height), -1:1:complex(width)]
longitude = (12.0 + 6.0 * xx).astype(np.float32)
latitude = (38.0 + 6.0 * yy).astype(np.float32)
ocean = ((xx + 0.28) ** 2 + (yy - 0.10) ** 2 > 0.055) & (xx - 0.75 * yy < 1.12)
sst = np.empty((n, height, width), np.float32)
observed = np.empty_like(sst)
precision = np.empty_like(sst)
start = date.fromisoformat(data_config["start_date"])
timestamps = []
for t in range(n):
current = start + timedelta(days=t)
timestamps.append(current.isoformat())
phase = 2 * np.pi * t / max(n, 16)
eddy = 1.9 * np.exp(-((xx - 0.35 * np.sin(phase)) ** 2 +
(yy - 0.25 * np.cos(phase)) ** 2) / 0.055)
field = 18.0 - 3.1 * yy + 0.7 * np.sin(3 * xx + phase) + eddy
field += rng.normal(0, 0.06, (height, width))
field[~ocean] = np.nan
cloud_noise = rng.normal(size=(height, width))
cloud = (np.sin(5 * xx + phase) + np.cos(4 * yy - phase) + cloud_noise > 1.05)
valid = ocean & ~cloud
sigma = (0.22 + 0.08 * (1 + np.sin(2 * xx - yy + phase))).astype(np.float32)
obs = field + rng.normal(0, sigma)
obs[~valid] = np.nan
sst[t], observed[t] = field, obs
precision[t] = np.where(valid, 1.0 / sigma ** 2, 0.0)
climatology = np.zeros((height, width), np.float32)
climatology[ocean] = np.mean(sst[:, ocean], axis=0)
anomaly = sst - climatology
observed_anomaly = observed - climatology
output = ROOT / data_config["root"]
output.mkdir(parents=True, exist_ok=True)
np.savez_compressed(output / data_config["file"],
format_version=np.array(data_config["format_version"]),
sst_anomaly=anomaly, observed_anomaly=observed_anomaly,
precision=precision, climatology=climatology,
ocean_mask=ocean, longitude=longitude, latitude=latitude,
timestamps=np.asarray(timestamps), units=np.array("degree_Celsius"),
grid=np.array([height, width], np.int32),
source=np.array("synthetic AVHRR-like engineering data"))
print(f"generated={n} grid={height}x{width} observed_fraction={np.isfinite(observed).mean():.4f}")
if __name__ == "__main__":
main()
|