feat(도구): 등고선 벡터 조각 미리 만들기 — z8~12 · 이어 하기 · 일꾼 둘 · 낮은 우선순위 (PLAN 615-3)
- tools/precompute_contour_tiles.py — 조각 목록 = 612 전국 도엽 목록이 닿는 z/x/y(z8~12 2,923 조각) · 창고에 있으면 건너뜀 · --status 남은 수 · 크기 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,103 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""등고선 벡터 조각 미리 만들기(PLAN 615-3) — 줌아웃 조각(z8~12)을 공용 창고에 미리.
|
||||
|
||||
실행: ./venv/Scripts/python.exe tools/precompute_contour_tiles.py --workers 2
|
||||
--zooms 8-12 줌 범위 · --status 남은 수 · 크기만
|
||||
|
||||
- 조각 목록 = 전국 등고선 도엽 목록(612 `_precompute_sheets.json`)의 도엽 상자가 닿는 z/x/y ·
|
||||
낮은 줌부터.
|
||||
- 셈 = `common_util_contour_tiles.ensure_tile`(창고에 있으면 건너뜀 = 이어 하기 · 잠금 ·
|
||||
`.part` → 바꿈).
|
||||
- 낮은 우선순위(BELOW_NORMAL) · 일꾼 여럿.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import math
|
||||
import sys
|
||||
import time
|
||||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from B04_PreProcess.B04_PreProcess_Engine_MapSheet import sheet5k_to_bounds # noqa: E402
|
||||
from common_util import common_util_contour_tiles as ct # noqa: E402
|
||||
from tools.precompute_terrain import below_normal, sheet_index # noqa: E402
|
||||
|
||||
|
||||
def tiles_for(sheets: list[str], zooms: range) -> list[tuple[int, int, int]]:
|
||||
"""도엽 상자가 닿는 조각(낮은 줌부터 · 같은 줌은 x · y 차례)."""
|
||||
found: set[tuple[int, int, int]] = set()
|
||||
for no in sheets:
|
||||
lon0, lat0, lon1, lat1 = sheet5k_to_bounds(no)
|
||||
for z in zooms:
|
||||
n = 2**z
|
||||
|
||||
def tx(lon: float) -> int:
|
||||
return min(n - 1, math.floor((lon + 180) / 360 * n))
|
||||
|
||||
def ty(lat: float) -> int:
|
||||
r = math.radians(lat)
|
||||
y = (1 - math.log(math.tan(r) + 1 / math.cos(r)) / math.pi) / 2 * n
|
||||
return min(n - 1, math.floor(y))
|
||||
|
||||
for x in range(tx(lon0), tx(lon1) + 1):
|
||||
for y in range(ty(lat1), ty(lat0) + 1):
|
||||
found.add((z, x, y))
|
||||
return sorted(found)
|
||||
|
||||
|
||||
def _path(folder: Path, tile: tuple[int, int, int]) -> Path:
|
||||
z, x, y = tile
|
||||
return folder / str(z) / str(x) / f"{y}.pbf"
|
||||
|
||||
|
||||
def build(tile: tuple[int, int, int]) -> tuple[tuple[int, int, int], int, float]:
|
||||
started = time.monotonic()
|
||||
size = len(ct.ensure_tile(*tile))
|
||||
return tile, size, round(time.monotonic() - started, 2)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description="등고선 벡터 조각 미리 만들기(PLAN 615-3)")
|
||||
parser.add_argument("--zooms", default="8-12")
|
||||
parser.add_argument("--workers", type=int, default=2)
|
||||
parser.add_argument("--status", action="store_true")
|
||||
args = parser.parse_args(argv)
|
||||
lo, _, hi = args.zooms.partition("-")
|
||||
zooms = range(int(lo), int(hi or lo) + 1)
|
||||
if zooms.start < ct.MIN_ZOOM or zooms.stop - 1 > ct.MAX_ZOOM:
|
||||
parser.error(f"줌 = {ct.MIN_ZOOM}~{ct.MAX_ZOOM}")
|
||||
|
||||
below_normal()
|
||||
folder = ct._folder()
|
||||
if folder is None or not ct.NATIONAL_GPKG.is_file():
|
||||
raise SystemExit(f"전국 등고선 파일 없음: {ct.NATIONAL_GPKG}")
|
||||
tiles = [t for t in tiles_for(list(sheet_index()), zooms) if ct.in_bounds(*t)]
|
||||
todo = [t for t in tiles if not _path(folder, t).is_file()]
|
||||
print(f"조각 {len(tiles)} · 남음 {len(todo)} · 창고 {folder} · 일꾼 {args.workers}", flush=True)
|
||||
if not args.status and todo:
|
||||
started = time.monotonic()
|
||||
with ProcessPoolExecutor(args.workers, initializer=below_normal) as pool:
|
||||
futures = [pool.submit(build, t) for t in todo]
|
||||
for n, future in enumerate(as_completed(futures), 1):
|
||||
(z, x, y), size, seconds = future.result()
|
||||
if n % 50 == 0 or n == len(todo):
|
||||
elapsed = time.monotonic() - started
|
||||
print(
|
||||
f"{n}/{len(todo)} z{z}/{x}/{y} {size} B {seconds} s · {elapsed:.0f} s",
|
||||
flush=True,
|
||||
)
|
||||
for z in zooms:
|
||||
paths = [_path(folder, t) for t in tiles if t[0] == z]
|
||||
have = [p.stat().st_size for p in paths if p.is_file()]
|
||||
print(f"z{z} 조각 {len(paths)} · 있음 {len(have)} · {sum(have) / 1e6:.1f} MB", flush=True)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user