| """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() |
|
|
