feat: add WeatherV2 wind canonical converter
This commit is contained in:
OpenAI Codex
37
tools/weatherv2_wind_canonical/README.md
Normal file
37
tools/weatherv2_wind_canonical/README.md
Normal file
@@ -0,0 +1,37 @@
|
||||
# WeatherV2 风场转 MapLibre canonical 测试数据工具
|
||||
|
||||
此工具只用于开发期离线转换：将 WeatherV2 的完整 `wind_vectors.bin`（`WVECBIN1`）转成 STEP-053 已冻结的 `maplibre-particle-field/v1`，不属于 MapLibre Runtime，也不访问网络、不扫描下载目录、更不修改 WeatherV2 原始数据。
|
||||
|
||||
## 输入
|
||||
|
||||
输入为一个 WeatherV2 reference/run 目录及其中一个 `YYYYMMDDHH` valid time：
|
||||
|
||||
```text
|
||||
<run-dir>/<valid-time>/wind_vectors.bin
|
||||
<run-dir>/<valid-time>/wind_vectors.bin.header.json
|
||||
```
|
||||
|
||||
当前已核对的 `WVECBIN1` 布局为：`magic(8)`、little-endian `uint32 headerLength`、UTF-8 JSON header、row-major `u float32-le` 平面、`v float32-le` 平面、`speed float32-le` 平面、`valid uint8` 平面。WeatherV2 `bin_export.py` 将同一 `decode_wind()` U/V grid 写出；QBIN 和 NSFT 也从同一 decoder U/V 生成。QBIN metadata 明确其 U/V 单位为 m/s，因此本工具保持 m/s，不进行 kt 换算。
|
||||
|
||||
源 grid 的 y 来自 Web Mercator tile y，向南增加；因此输出的 row-major rows 是 north-to-south，columns 是 west-to-east。U=eastward-positive，V=northward-positive。
|
||||
|
||||
## 输出
|
||||
|
||||
每个输出目录包含：
|
||||
|
||||
- `metadata.json`：现有 `LocalParticleFieldFeed` 读取的 `maplibre-particle-field/v1` metadata。
|
||||
- `field.f32`：`float32-le interleaved-uv`，顺序为 `U0,V0,U1,V1,...`。
|
||||
- `conversion_summary.json`：源/输出 bounds、尺寸、统计值、valid/noData 计数、抽样一致性与 SHA-256。
|
||||
|
||||
本工具不重采样。每个源 `valid=0` 或非有限 U/V 的 cell 会写为既有 canonical `noData=-9999.0` 的 U/V 对；不会错误变成静风 `(0,0)`。有效样本按 float32 bit-exact 复制。
|
||||
|
||||
## 命令示例
|
||||
|
||||
```bash
|
||||
python3 tools/weatherv2_wind_canonical/convert_weatherv2_wind.py \
|
||||
--run-dir /root/sourceserver/weatherV2/src/dev/_runtime/dev_pipeline_data/datasets/ecmwf-hres/wind/2026080600 \
|
||||
--valid-time 2026080700 \
|
||||
--output platform/android/MapLibreAndroidTestApp/src/main/assets/integration/particle-weatherv2/ecmwf-hres-wind-20260807/2026080700
|
||||
```
|
||||
|
||||
相同输入会产生相同 payload SHA-256。错误会输出明确消息并以非零状态退出。
|
||||
190
tools/weatherv2_wind_canonical/convert_weatherv2_wind.py
Normal file
190
tools/weatherv2_wind_canonical/convert_weatherv2_wind.py
Normal file
@@ -0,0 +1,190 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Convert one WeatherV2 WVECBIN1 wind snapshot to maplibre-particle-field/v1."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
import struct
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
MAGIC = b"WVECBIN1"
|
||||
NO_DATA = np.float32(-9999.0)
|
||||
|
||||
|
||||
def fail(message: str) -> "None":
|
||||
raise ValueError(message)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--run-dir", required=True, type=Path, help="WeatherV2 reference-time directory")
|
||||
parser.add_argument("--valid-time", required=True, help="target time in YYYYMMDDHH UTC")
|
||||
parser.add_argument("--output", required=True, type=Path, help="canonical snapshot output directory")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def require_mapping(value: object, name: str) -> dict:
|
||||
if not isinstance(value, dict):
|
||||
fail(f"{name} must be an object")
|
||||
return value
|
||||
|
||||
|
||||
def read_source(run_dir: Path, valid_time: str) -> tuple[dict, np.ndarray, np.ndarray, np.ndarray, np.ndarray, int]:
|
||||
source_dir = run_dir / valid_time
|
||||
payload_path = source_dir / "wind_vectors.bin"
|
||||
sidecar_path = source_dir / "wind_vectors.bin.header.json"
|
||||
if not payload_path.is_file() or not sidecar_path.is_file():
|
||||
fail(f"missing WeatherV2 snapshot for {valid_time}: {source_dir}")
|
||||
data = payload_path.read_bytes()
|
||||
if len(data) < 12 or data[:8] != MAGIC:
|
||||
fail(f"unsupported wind_vectors.bin magic in {payload_path}")
|
||||
header_length = struct.unpack_from("<I", data, 8)[0]
|
||||
header_end = 12 + header_length
|
||||
if header_end > len(data):
|
||||
fail("WVECBIN1 header length exceeds file length")
|
||||
header = json.loads(data[12:header_end].decode("utf-8"))
|
||||
sidecar = json.loads(sidecar_path.read_text(encoding="utf-8"))
|
||||
if header != sidecar:
|
||||
fail("embedded WVECBIN1 header differs from .header.json sidecar")
|
||||
if header.get("format") != "weather-v2-particle-bin-v1" or header.get("layer") != "wind":
|
||||
fail("source is not a WeatherV2 wind particle-bin-v1 snapshot")
|
||||
if header.get("target_time") != valid_time:
|
||||
fail(f"source target_time mismatch: {header.get('target_time')!r} != {valid_time!r}")
|
||||
width, height = int(header.get("width", 0)), int(header.get("height", 0))
|
||||
if width <= 0 or height <= 0:
|
||||
fail("source width/height must be positive")
|
||||
expected_fields = [
|
||||
{"name": "u", "dtype": "float32"},
|
||||
{"name": "v", "dtype": "float32"},
|
||||
{"name": "speed", "dtype": "float32"},
|
||||
{"name": "valid", "dtype": "uint8"},
|
||||
]
|
||||
if header.get("fields") != expected_fields:
|
||||
fail(f"unsupported WVECBIN1 fields: {header.get('fields')!r}")
|
||||
count = width * height
|
||||
expected_bytes = header_end + count * 4 * 3 + count
|
||||
if len(data) != expected_bytes:
|
||||
fail(f"WVECBIN1 byte length mismatch: expected {expected_bytes}, got {len(data)}")
|
||||
offset = header_end
|
||||
u = np.frombuffer(data, dtype="<f4", count=count, offset=offset).copy().reshape(height, width)
|
||||
offset += count * 4
|
||||
v = np.frombuffer(data, dtype="<f4", count=count, offset=offset).copy().reshape(height, width)
|
||||
offset += count * 4
|
||||
speed = np.frombuffer(data, dtype="<f4", count=count, offset=offset).copy().reshape(height, width)
|
||||
offset += count * 4
|
||||
valid = np.frombuffer(data, dtype=np.uint8, count=count, offset=offset).copy().reshape(height, width).astype(bool)
|
||||
bounds = require_mapping(header.get("bounds"), "source bounds")
|
||||
if not all(math.isfinite(float(bounds[name])) for name in ("west", "south", "east", "north")):
|
||||
fail("source bounds must be finite")
|
||||
if not float(bounds["west"]) < float(bounds["east"]) or not float(bounds["south"]) < float(bounds["north"]):
|
||||
fail("source bounds are invalid")
|
||||
if not np.isfinite(u[valid]).all() or not np.isfinite(v[valid]).all():
|
||||
fail("valid WeatherV2 samples contain non-finite U/V")
|
||||
if not np.isfinite(speed[valid]).all():
|
||||
fail("valid WeatherV2 samples contain non-finite speed")
|
||||
return header, u, v, speed, valid, len(data)
|
||||
|
||||
|
||||
def epoch_millis(valid_time: str) -> int:
|
||||
try:
|
||||
return int(datetime.strptime(valid_time, "%Y%m%d%H").replace(tzinfo=timezone.utc).timestamp() * 1000)
|
||||
except ValueError as error:
|
||||
fail(f"valid time must be YYYYMMDDHH UTC: {error}")
|
||||
|
||||
|
||||
def stats(values: np.ndarray, valid: np.ndarray) -> dict:
|
||||
chosen = values[valid]
|
||||
if not chosen.size:
|
||||
fail("source has no valid samples")
|
||||
return {"min": float(chosen.min()), "max": float(chosen.max()), "mean": float(chosen.mean())}
|
||||
|
||||
|
||||
def sample_indices(valid: np.ndarray) -> list[tuple[int, int]]:
|
||||
height, width = valid.shape
|
||||
candidates = [(height // 2, width // 2), (1, 1), (1, width - 2), (height - 2, 1), (height - 2, width - 2)]
|
||||
selected: list[tuple[int, int]] = []
|
||||
for row, col in candidates:
|
||||
if valid[row, col] and (row, col) not in selected:
|
||||
selected.append((row, col))
|
||||
for row, col in zip(*np.nonzero(valid)):
|
||||
if (int(row), int(col)) not in selected:
|
||||
selected.append((int(row), int(col)))
|
||||
if len(selected) == 8:
|
||||
break
|
||||
return selected
|
||||
|
||||
|
||||
def write_output(header: dict,
|
||||
u: np.ndarray,
|
||||
v: np.ndarray,
|
||||
speed: np.ndarray,
|
||||
valid: np.ndarray,
|
||||
source_bytes: int,
|
||||
reference_time: str,
|
||||
output: Path) -> dict:
|
||||
width, height = int(header["width"]), int(header["height"])
|
||||
valid = valid & np.isfinite(u) & np.isfinite(v)
|
||||
payload = np.empty((height, width, 2), dtype="<f4")
|
||||
payload[:, :, 0] = np.where(valid, u, NO_DATA)
|
||||
payload[:, :, 1] = np.where(valid, v, NO_DATA)
|
||||
payload_bytes = payload.tobytes(order="C")
|
||||
payload_hash = hashlib.sha256(payload_bytes).hexdigest()
|
||||
decoded = np.frombuffer(payload_bytes, dtype="<f4").reshape(height, width, 2)
|
||||
if not np.array_equal(decoded[:, :, 0][valid], u[valid]) or not np.array_equal(decoded[:, :, 1][valid], v[valid]):
|
||||
fail("source-to-canonical U/V conversion is not lossless")
|
||||
if not np.all(decoded[:, :, 0][~valid] == NO_DATA) or not np.all(decoded[:, :, 1][~valid] == NO_DATA):
|
||||
fail("invalid samples were not preserved as canonical noData")
|
||||
bounds = header["bounds"]
|
||||
timestamp = epoch_millis(str(header["target_time"]))
|
||||
metadata = {
|
||||
"schema": "maplibre-particle-field/v1",
|
||||
"fieldId": f"ecmwf-hres-wind-{header['target_time']}",
|
||||
"fieldType": "WIND",
|
||||
"displayDomain": "GLOBAL",
|
||||
"generation": timestamp,
|
||||
"validTime": timestamp,
|
||||
"dataVersion": timestamp,
|
||||
"bounds": {name: float(bounds[name]) for name in ("west", "south", "east", "north")},
|
||||
"grid": {"width": width, "height": height},
|
||||
"vector": {"layout": "interleaved-uv", "encoding": "float32-le", "noData": float(NO_DATA), "unit": "m/s", "origin": "north-west", "rowOrder": "north-to-south", "columnOrder": "west-to-east", "uPositive": "east", "vPositive": "north"},
|
||||
"payload": {"byteLength": len(payload_bytes), "sha256": payload_hash},
|
||||
"source": {"kind": "WeatherV2 ECMWF HRES WVECBIN1", "referenceTime": reference_time, "targetTime": header["target_time"], "sourceBytes": source_bytes, "resampled": False},
|
||||
}
|
||||
samples = []
|
||||
for row, col in sample_indices(valid):
|
||||
samples.append({"row": row, "column": col, "sourceU": float(u[row, col]), "sourceV": float(v[row, col]), "canonicalU": float(decoded[row, col, 0]), "canonicalV": float(decoded[row, col, 1])})
|
||||
summary = {
|
||||
"source": {"format": header["format"], "magic": header["magic"], "targetTime": header["target_time"], "width": width, "height": height, "bounds": metadata["bounds"], "dtype": "float32 little-endian u plane, v plane, speed plane; uint8 valid plane", "arrayOrder": "row-major; WeatherV2 tile y increases southward, so rows are north-to-south", "uVUnit": "m/s", "validCount": int(valid.sum()), "invalidCount": int(valid.size - valid.sum()), "bytes": source_bytes},
|
||||
"output": {"width": width, "height": height, "bounds": metadata["bounds"], "bytes": len(payload_bytes), "sha256": payload_hash, "validCount": int(valid.sum()), "invalidCount": int(valid.size - valid.sum()), "resampled": False},
|
||||
"statistics": {"u": stats(u, valid), "v": stats(v, valid), "speed": stats(speed, valid)},
|
||||
"sourceToCanonical": {"lossless": True, "maxAbsErrorU": 0.0, "maxAbsErrorV": 0.0, "samples": samples},
|
||||
}
|
||||
output.mkdir(parents=True, exist_ok=True)
|
||||
(output / "field.f32").write_bytes(payload_bytes)
|
||||
(output / "metadata.json").write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
(output / "conversion_summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||||
return summary
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
try:
|
||||
header, u, v, speed, valid, source_bytes = read_source(args.run_dir, args.valid_time)
|
||||
summary = write_output(header, u, v, speed, valid, source_bytes, args.run_dir.name, args.output)
|
||||
print(json.dumps({"ok": True, "targetTime": header["target_time"], "output": str(args.output), "sha256": summary["output"]["sha256"], "validCount": summary["output"]["validCount"], "invalidCount": summary["output"]["invalidCount"]}, ensure_ascii=False))
|
||||
return 0
|
||||
except (OSError, ValueError, json.JSONDecodeError) as error:
|
||||
print(f"error: {error}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user