239 lines
14 KiB
Python
239 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Generate and validate offline provider-neutral V3 canonical fixtures."""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import shutil
|
||
import struct
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
OUT = ROOT / "integration/particle-v3-fields"
|
||
WIND = ROOT / "platform/android/MapLibreAndroidTestApp/src/main/assets/integration/particle-weatherv2/ecmwf-hres-wind-20260807"
|
||
DATASETS = Path("/root/sourceserver/weatherV2/src/dev/_runtime/dev_pipeline_data/datasets")
|
||
WAVE = DATASETS / "ecmwf-wam/waves/2026090912/wave_bin"
|
||
TIDE = DATASETS / "cmems/tide/2026090900/current_bin"
|
||
NODATA = -9999.0
|
||
|
||
|
||
def sha(path: Path) -> str:
|
||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||
|
||
|
||
def valid_time(value: str) -> int:
|
||
return int(datetime.strptime(value, "%Y%m%d%H").replace(tzinfo=timezone.utc).timestamp() * 1000)
|
||
|
||
|
||
def write(path: Path, data: bytes) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_bytes(data)
|
||
|
||
|
||
def float_bytes(values: list[float]) -> bytes:
|
||
return struct.pack("<%sf" % len(values), *values)
|
||
|
||
|
||
def meta(field: str, target: str, generation: int, width: int, height: int, bounds: dict,
|
||
scalar: bool, vector: bool, semantics: str | None, source: dict) -> dict:
|
||
output = {"schema": "maplibre-particle-field/v2", "fieldId": field, "generation": generation,
|
||
"dataVersion": generation, "validTime": valid_time(target), "bounds": bounds,
|
||
"grid": {"width": width, "height": height},
|
||
"validity": {"encoding": "uint8", "byteLength": width * height}, "source": source,
|
||
"canonical": {"origin": "north-west", "rowOrder": "north-to-south", "columnOrder": "west-to-east"}}
|
||
if scalar:
|
||
output["scalar"] = {"encoding": "float32-le", "byteLength": width * height * 4, "noData": NODATA}
|
||
if vector:
|
||
output["vector"] = {"layout": "interleaved-uv", "encoding": "float32-le", "byteLength": width * height * 8,
|
||
"noData": NODATA, "semantics": semantics, "uPositive": "east", "vPositive": "north"}
|
||
return output
|
||
|
||
|
||
def finish(directory: Path, metadata: dict, validity: list[int], scalar: list[float] | None = None,
|
||
vector: list[float] | None = None) -> None:
|
||
write(directory / "validity.u8", bytes(validity))
|
||
if scalar is not None:
|
||
write(directory / "scalar.f32", float_bytes(scalar))
|
||
if vector is not None:
|
||
write(directory / "vector.f32", float_bytes(vector))
|
||
filenames = ["validity.u8"] + (["scalar.f32"] if scalar is not None else []) + (["vector.f32"] if vector is not None else [])
|
||
metadata["planes"] = {name: {"path": name, "byteLength": (directory / name).stat().st_size,
|
||
"sha256": sha(directory / name)} for name in filenames}
|
||
(directory / "metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n")
|
||
(directory / "conversion_summary.json").write_text(json.dumps({
|
||
"grid": metadata["grid"], "bounds": metadata["bounds"], "validTime": metadata["validTime"],
|
||
"generation": metadata["generation"], "source": metadata["source"], "planes": metadata["planes"],
|
||
"validCount": sum(validity), "invalidCount": len(validity) - sum(validity)}, ensure_ascii=False, indent=2) + "\n")
|
||
|
||
|
||
def frame(label: str, target: str, scalar: bool, vector: bool) -> dict:
|
||
return {"label": label, "path": target, "metadata": "metadata.json", "validity": "validity.u8",
|
||
**({"scalar": "scalar.f32"} if scalar else {}), **({"vector": "vector.f32"} if vector else {})}
|
||
|
||
|
||
def manifest(name: str, provenance: str, field_source: str, particle: bool, semantics: str | None,
|
||
scale: float, frames: list[dict]) -> None:
|
||
(OUT / name / "dataset.json").write_text(json.dumps({
|
||
"schema": "maplibre-particle-test-dataset/v2", "datasetId": name, "sourceProvenance": provenance,
|
||
"frames": frames, "expectedFieldValueSource": field_source, "particleExpected": particle,
|
||
"vectorSemantics": semantics, "suggestedAdvectionScale": scale,
|
||
"说明": "离线 canonical fixture；Runtime 只接收 metadata 与本地 plane URI。"}, ensure_ascii=False, indent=2) + "\n")
|
||
|
||
|
||
def read_bin(path: Path, magic: bytes, float_planes: int) -> tuple[dict, list[list[float]], list[int]]:
|
||
raw = path.read_bytes()
|
||
assert raw[:len(magic)] == magic, f"{path}: magic"
|
||
header_length = struct.unpack_from("<I", raw, len(magic))[0]
|
||
start = len(magic) + 4
|
||
header = json.loads(raw[start:start + header_length])
|
||
sidecar = json.loads(Path(str(path) + ".header.json").read_text())
|
||
assert header == sidecar, f"{path}: header sidecar mismatch"
|
||
count, offset = header["width"] * header["height"], start + header_length
|
||
planes = []
|
||
for _ in range(float_planes):
|
||
planes.append(list(struct.unpack_from("<%sf" % count, raw, offset)))
|
||
offset += count * 4
|
||
validity = list(raw[offset:offset + count])
|
||
assert len(validity) == count and offset + count == len(raw), f"{path}: payload length"
|
||
return header, planes, validity
|
||
|
||
|
||
def provenance(path: Path, header: dict, reference_time: str) -> dict:
|
||
sidecar = Path(str(path) + ".header.json")
|
||
return {"kind": "WeatherV2 formal BIN decoded offline", "sourcePath": str(path), "sourceSha256": sha(path),
|
||
"headerPath": str(sidecar), "headerSha256": sha(sidecar), "referenceTime": reference_time,
|
||
"targetTime": header["target_time"], "grid": {"width": header["width"], "height": header["height"]},
|
||
"bounds": header["bounds"], "binMagic": header["magic"], "binFields": header["fields"]}
|
||
|
||
|
||
def generate_wind() -> None:
|
||
frames = []
|
||
for index, target in enumerate(["2026080700", "2026080706"]):
|
||
old = json.loads((WIND / target / "metadata.json").read_text())
|
||
count = old["grid"]["width"] * old["grid"]["height"]
|
||
vector = list(struct.unpack("<%sf" % (count * 2), (WIND / target / "field.f32").read_bytes()))
|
||
validity = [int(math.isfinite(vector[2 * i]) and math.isfinite(vector[2 * i + 1]) and vector[2 * i] != NODATA and vector[2 * i + 1] != NODATA) for i in range(count)]
|
||
assert any(validity), f"{target}: empty wind"
|
||
source = {"kind": "WeatherV2 ECMWF HRES WVECBIN1 canonical v1 upgraded offline", "referenceTime": "2026080600",
|
||
"targetTime": target, "originalPayloadSha256": sha(WIND / target / "field.f32"),
|
||
"validityDerivation": "finite U/V and both components != -9999.0"}
|
||
finish(OUT / "wind-real" / target, meta("wind-real", target, old["generation"], old["grid"]["width"], old["grid"]["height"], old["bounds"], False, True, "VELOCITY", source), validity, vector=vector)
|
||
frames.append(frame(f"T{index}", target, False, True))
|
||
manifest("wind-real", "真实本地 WeatherV2 Wind canonical v1 输出；validity 由 U/V noData 与有限性派生", "VECTOR_MAGNITUDE", True, "VELOCITY", 1.0, frames)
|
||
|
||
|
||
def generate_wave() -> None:
|
||
frames = []
|
||
for generation, target in enumerate(["2026090918", "2026090919"], start=1):
|
||
path = WAVE / target / "waves_particles.bin"
|
||
header, planes, source_validity = read_bin(path, b"WAVEBIN1", 5)
|
||
height_m, _period_s, _direction_from, unit_u, unit_v = planes
|
||
validity, scalar, vector = [], [], []
|
||
for raw_valid, height, u, v in zip(source_validity, height_m, unit_u, unit_v):
|
||
length = math.hypot(u, v)
|
||
valid = bool(raw_valid) and math.isfinite(height) and math.isfinite(u) and math.isfinite(v) and length > 1e-5
|
||
validity.append(int(valid)); scalar.append(height if valid else NODATA)
|
||
vector.extend((u / length, v / length) if valid else (NODATA, NODATA))
|
||
assert any(validity), f"{path}: empty wave"
|
||
finish(OUT / "wave-real" / target, meta("wave-real", target, generation, header["width"], header["height"], header["bounds"], True, True, "DIRECTION", provenance(path, header, "2026090912")), validity, scalar, vector)
|
||
frames.append(frame(f"T{generation - 1}", target, True, True))
|
||
manifest("wave-real", "真实本地 WeatherV2 WAVEBIN1；height_m + 正规化 unit_to_u/v，period 不进入 canonical", "SCALAR", True, "DIRECTION", 1.75, frames)
|
||
|
||
|
||
def generate_tide() -> None:
|
||
frames = []
|
||
for generation, target in enumerate(["2026090901", "2026090902"], start=1):
|
||
path = TIDE / target / "current_vectors.bin"
|
||
header, planes, source_validity = read_bin(path, b"WVECBIN1", 3)
|
||
u_values, v_values, _speed = planes
|
||
validity, vector = [], []
|
||
for raw_valid, u, v in zip(source_validity, u_values, v_values):
|
||
valid = bool(raw_valid) and math.isfinite(u) and math.isfinite(v)
|
||
validity.append(int(valid)); vector.extend((u, v) if valid else (NODATA, NODATA))
|
||
assert any(validity), f"{path}: empty tide"
|
||
finish(OUT / "tide-real" / target, meta("tide-real", target, generation, header["width"], header["height"], header["bounds"], False, True, "VELOCITY", provenance(path, header, "2026090900")), validity, vector=vector)
|
||
frames.append(frame(f"T{generation - 1}", target, False, True))
|
||
manifest("tide-real", "真实本地 WeatherV2 CMEMS WVECBIN1；仅 U/V VELOCITY", "VECTOR_MAGNITUDE", True, "VELOCITY", 1.0, frames)
|
||
|
||
|
||
def generate_rain() -> None:
|
||
frames, width, height = [], 6, 4
|
||
bounds = {"west": 122.0, "south": 20.0, "east": 150.0, "north": 48.0}
|
||
for generation, target in enumerate(["2026090912", "2026090918"], start=1):
|
||
values = [float((i % width) + generation) for i in range(width * height)]
|
||
source = {"kind": "decoder-contract fallback", "notReal": True, "referenceTime": "2026090912", "targetTime": target,
|
||
"说明": "本机未实现 PMTiles reader；此数值不是正式 WeatherV2 栅格。"}
|
||
finish(OUT / "rain-decoder-fixture" / target, meta("rain-decoder-fixture", target, generation, width, height, bounds, True, False, None, source), [1] * (width * height), scalar=values)
|
||
frames.append(frame(f"T{generation - 1}", target, True, False))
|
||
manifest("rain-decoder-fixture", "非真实场值：仅 decoder-contract fallback，不能当作 WeatherV2 实测栅格", "SCALAR", False, None, 1.0, frames)
|
||
|
||
|
||
def generate_synthetic_wave() -> None:
|
||
name, width, height, frames = "wave-synthetic-independence", 6, 4, []
|
||
bounds = {"west": 122.0, "south": 20.0, "east": 150.0, "north": 48.0}
|
||
for generation, target in enumerate(["2026010100", "2026010106"], start=1):
|
||
scalar = [float((i % width) + generation) for i in range(width * height)]
|
||
vector = [value for _ in range(width * height) for value in (0.0, 1.0)]
|
||
finish(OUT / name / target, meta(name, target, generation, width, height, bounds, True, True, "DIRECTION", {"kind": "synthetic independence oracle", "notReal": True, "targetTime": target}), [1] * (width * height), scalar, vector)
|
||
frames.append(frame(f"T{generation - 1}", target, True, True))
|
||
manifest(name, "固定 synthetic；height 与 direction 独立", "SCALAR", True, "DIRECTION", 2.5, frames)
|
||
|
||
|
||
def generate() -> None:
|
||
if OUT.exists():
|
||
shutil.rmtree(OUT)
|
||
generate_wind(); generate_rain(); generate_wave(); generate_tide(); generate_synthetic_wave()
|
||
|
||
|
||
def validate() -> None:
|
||
for dataset in OUT.glob("*/dataset.json"):
|
||
frames = json.loads(dataset.read_text())["frames"]
|
||
generations = []
|
||
for spec in frames:
|
||
directory = dataset.parent / spec["path"]
|
||
metadata = json.loads((directory / "metadata.json").read_text())
|
||
generations.append(metadata["generation"])
|
||
count = metadata["grid"]["width"] * metadata["grid"]["height"]
|
||
validity_path = directory / "validity.u8"
|
||
validity = validity_path.read_bytes()
|
||
assert len(validity) == count and any(validity), f"{directory}: validity"
|
||
assert validity_path.stat().st_size == metadata["planes"]["validity.u8"]["byteLength"]
|
||
assert sha(validity_path) == metadata["planes"]["validity.u8"]["sha256"]
|
||
decoded = {}
|
||
for key, filename, channels in (("scalar", "scalar.f32", 1), ("vector", "vector.f32", 2)):
|
||
if key in spec:
|
||
path = directory / filename
|
||
assert path.stat().st_size == count * channels * 4 and sha(path) == metadata["planes"][filename]["sha256"]
|
||
decoded[key] = struct.unpack("<%sf" % (count * channels), path.read_bytes())
|
||
for cell, flag in enumerate(validity):
|
||
if not flag:
|
||
continue
|
||
if "scalar" in decoded:
|
||
value = decoded["scalar"][cell]
|
||
assert math.isfinite(value) and value != metadata["scalar"]["noData"], f"{directory}: scalar valid cell {cell}"
|
||
if "vector" in decoded:
|
||
u, v = decoded["vector"][2 * cell:2 * cell + 2]
|
||
assert math.isfinite(u) and math.isfinite(v) and u != metadata["vector"]["noData"] and v != metadata["vector"]["noData"], f"{directory}: vector valid cell {cell}"
|
||
if metadata["vector"]["semantics"] == "DIRECTION":
|
||
assert abs(math.hypot(u, v) - 1) < 2e-4, f"{directory}: direction valid cell {cell}"
|
||
assert all(current > previous for previous, current in zip(generations, generations[1:])), f"{dataset}: non-monotonic generation"
|
||
root = OUT / "wave-synthetic-independence" / "2026010100"
|
||
scalar, vector = struct.unpack("<24f", (root / "scalar.f32").read_bytes()), struct.unpack("<48f", (root / "vector.f32").read_bytes())
|
||
assert all(scalar[row * 6] < scalar[row * 6 + 5] for row in range(4))
|
||
assert all(vector[cell * 2] == 0 and vector[cell * 2 + 1] == 1 for cell in range(24))
|
||
print("PASS: canonical fixtures, valid-cell planes, direction normalization, and wave scalar/vector independence")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("--write-reference-fixtures", action="store_true")
|
||
parser.add_argument("--validate", action="store_true")
|
||
args = parser.parse_args()
|
||
if args.write_reference_fixtures:
|
||
generate()
|
||
if args.validate:
|
||
validate()
|