| """Generate deterministic HLS-like four-timestamp samples for engineering validation.""" |
|
|
| from pathlib import Path |
|
|
| import numpy as np |
| import yaml |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def make_split(path, count, config, seed): |
| rng = np.random.default_rng(seed) |
| data = config["data"] |
| channels, frames, size = int(data["channels"]), int(data["frames"]), int(data["image_size"]) |
| means = np.asarray(data["mean"], np.float32) |
| stds = np.asarray(data["std"], np.float32) |
| y, x = np.mgrid[-1:1:complex(size), -1:1:complex(size)].astype(np.float32) |
| pixels = np.empty((count, channels, frames, size, size), np.float32) |
| temporal = np.empty((count, frames, 2), np.float32) |
| location = np.empty((count, 2), np.float32) |
| class_target = np.empty(count, np.int64) |
| regression_target = np.empty(count, np.float32) |
| for sample in range(count): |
| latitude, longitude = rng.uniform(-70, 70), rng.uniform(-180, 180) |
| start_day = int(rng.integers(1, 80)) |
| days = np.clip(start_day + np.arange(frames) * int(rng.integers(45, 100)), 1, 365) |
| temporal[sample, :, 0] = 2018 + sample % 5 |
| temporal[sample, :, 1] = days |
| location[sample] = (latitude, longitude) |
| phase = rng.uniform(0, 2 * np.pi) |
| class_target[sample] = int(np.sin(phase) > 0) |
| regression_target[sample] = np.cos(phase) + latitude / 180 |
| for step, day in enumerate(days): |
| seasonal = np.sin(2 * np.pi * day / 365 + phase) |
| landscape = np.sin(2.5 * np.pi * x + phase) * np.cos(2 * np.pi * y - phase) |
| landscape += 0.35 * x + 0.2 * y + 0.25 * seasonal |
| for channel in range(channels): |
| normalized = landscape + 0.12 * channel + rng.normal(0, 0.04, (size, size)) |
| pixels[sample, channel, step] = normalized * stds[channel] + means[channel] |
| payload = { |
| "format_version": np.asarray(data["format_version"]), |
| "data_source": np.asarray("synthetic_hls_like"), |
| "pixels": pixels, |
| "temporal_coords": temporal, |
| "location_coords": location, |
| "class_target": class_target, |
| "regression_target": regression_target, |
| } |
| np.savez_compressed(path, **payload) |
|
|
|
|
| def main(): |
| config = yaml.safe_load((ROOT / "conf/config.yaml").read_text()) |
| output = ROOT / config["data"]["root"] |
| output.mkdir(parents=True, exist_ok=True) |
| for offset, (filename, count) in enumerate(( |
| ("train.npz", int(config["data"]["train_samples"])), |
| ("test.npz", int(config["data"]["test_samples"])), |
| )): |
| path = output / filename |
| if not path.exists(): |
| make_split(path, count, config, int(config["seed"]) + offset) |
| print(f"generated={path.relative_to(ROOT)} samples={count} format={config['data']['format_version']}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
