pbf/tasks/pipline.md

NavSea Weather Server

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

---

1 任务目标

创建 Weather Pipeline。

Pipeline 负责自动执行整个天气数据生产流程：

Downloader → Grid Builder → Vector Tile Generator

输入：

NOAA GFS

输出：

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

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

---

2 Pipeline 结构

weather_server/

downloader/ grid/ tiles/ pipeline/

创建文件：

weather_server/pipeline/weather_pipeline.py

---

3 Pipeline 执行流程

step1 下载 GRIB

step2 生成 Weather Grid

step3 生成 Vector Tiles

---

4 创建 Pipeline 文件

weather_server/pipeline/weather_pipeline.py

---

5 实现代码

import os
import subprocess
import time

DOWNLOADER = "weather_server/downloader/gfs_downloader.py"
GRID_BUILDER = "weather_server/grid/grid_builder_v2.py"
TILE_GENERATOR = "weather_server/tiles/vector_tile_generator.py"


def run_step(name, script):

    print("\n==========================")
    print("Running:", name)
    print("==========================\n")

    start = time.time()

    result = subprocess.run(
        ["python", script],
        capture_output=True,
        text=True
    )

    print(result.stdout)

    if result.returncode != 0:
        print("ERROR:", result.stderr)
        raise RuntimeError(name + " failed")

    end = time.time()

    print("\nFinished:", name)
    print("Time:", round(end-start,2),"seconds")


def main():

    print("\nNavSea Weather Pipeline\n")

    run_step("Downloader", DOWNLOADER)

    run_step("Grid Builder", GRID_BUILDER)

    run_step("Vector Tile Generator", TILE_GENERATOR)

    print("\nWeather pipeline completed.\n")


if __name__ == "__main__":
    main()