pbf/tasks/WeatherServer_VectorTileGenerator.md

NavSea Weather Server

Task: WeatherServer_VectorTileGenerator Architecture: NavSea V11 Codex: codex6 Status: TODO

---

1 任务目标

实现 Vector Tile Generator。

功能：

将 Weather Grid JSON 转换为 MapLibre Vector Tile (PBF)。

输入：

data/grid/*.json

输出：

output/weather/{time}/{z}/{x}/{y}.pbf

用于 NavSea 客户端加载天气图层。

---

2 使用库

需要安装：

pip install mercantile pip install mapbox-vector-tile

---

3 输入数据

GridBuilder v2 生成：

data/grid/

grid_20260312_00_f000.json grid_20260312_00_f003.json

结构：

{ "time": "...", "grid": { "lat": [...], "lon": [...], "wind_speed": ..., "wind_dir": ..., "rain": ..., "temp": ..., "pressure": ... } }

grid 尺寸：

121 × 121

---

4 Tile Zoom 设计

Weather tile 只需要 4 个 zoom：

2 4 6 8

---

5 Tile 输出结构

output/weather/

time/ z/ x/ y.pbf

示例：

output/weather/20260312_00_f000/4/10/7.pbf

---

6 Feature 结构

每个 grid 点 → 一个 feature

geometry：

POINT(lon lat)

properties：

{ "ws": wind_speed "wd": wind_dir "r": rain "t": temp "p": pressure }

字段缩写减少 tile 大小。

---

7 创建文件

weather_server/tiles/vector_tile_generator.py

---

8 实现代码

import os
import json
import mercantile
import mapbox_vector_tile

GRID_DIR = "data/grid"
OUTPUT_DIR = "output/weather"

ZOOMS = [2,4,6,8]

def load_grid(path):

    with open(path) as f:
        data = json.load(f)

    return data["time"], data["grid"]


def grid_to_features(grid):

    lat = grid["lat"]
    lon = grid["lon"]

    ws = grid["wind_speed"]
    wd = grid["wind_dir"]
    rain = grid["rain"]
    temp = grid["temp"]
    pres = grid["pressure"]

    features = []

    for i in range(len(lat)):
        for j in range(len(lon)):

            feature = {
                "geometry":{
                    "type":"Point",
                    "coordinates":[lon[j], lat[i]]
                },
                "properties":{
                    "ws":ws[i][j],
                    "wd":wd[i][j],
                    "r":rain[i][j],
                    "t":temp[i][j],
                    "p":pres[i][j]
                }
            }

            features.append(feature)

    return features


def generate_tiles(time, features):

    for z in ZOOMS:

        tiles = mercantile.tiles(120,20,150,50,z)

        for tile in tiles:

            bounds = mercantile.bounds(tile)

            tile_features = []

            for f in features:

                lon, lat = f["geometry"]["coordinates"]

                if (
                    bounds.west <= lon <= bounds.east
                    and bounds.south <= lat <= bounds.north
                ):
                    tile_features.append(f)

            if not tile_features:
                continue

            layer = {
                "weather": tile_features
            }

            tile_data = mapbox_vector_tile.encode(layer)

            path = os.path.join(
                OUTPUT_DIR,
                time,
                str(z),
                str(tile.x)
            )

            os.makedirs(path, exist_ok=True)

            filename = os.path.join(
                path,
                f"{tile.y}.pbf"
            )

            with open(filename,"wb") as f:
                f.write(tile_data)

            print("tile", time, z, tile.x, tile.y)


def main():

    for file in os.listdir(GRID_DIR):

        if not file.endswith(".json"):
            continue

        path = os.path.join(GRID_DIR, file)

        print("processing", file)

        time, grid = load_grid(path)

        features = grid_to_features(grid)

        generate_tiles(time, features)


if __name__ == "__main__":
    main()