- tools/precompute_contour_tiles.py — 조각 목록 = 612 전국 도엽 목록이 닿는 z/x/y(z8~12 2,923 조각) · 창고에 있으면 건너뜀 · --status 남은 수 · 크기 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
# -*- 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())
|