File size: 2,877 Bytes
4c4d99c | 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 | """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()
|