# -*- coding: utf-8 -*-
"""전국 미리 계산 일괄 도구(PLAN 612-1 · 요구 493) — 공용 지형 창고에 도엽마다 미리 셈.

실행:  ./venv/Scripts/python.exe tools/precompute_terrain.py --sheets 36901 --workers 4
       --steps dem,zero,e07  단계(차례 고정 = 지형 → 0차곡 누적 → E07 · 뒤에 tiles)
       --sheets 36901,36712089  도엽 번호 또는 앞자리(1:50,000 묶음) · 주면 높이 차 거름 안 씀
       --box lon0,lat0,lon1,lat1  도엽 가운데가 이 경위도 상자 안
       --min-relief 150  도엽 안 높이 차(m) 이상만(산지) · --limit N 앞 N 장만
       --workers 4  일꾼 수 · --status 진행 기록 요약만(모든 PC) · --rebuild-index 도엽 목록 다시
       --part 0-69/100  두 PC 나눔(도엽 번호 crc32 % 100 이 0~69 인 것만 · PLAN 618)
       --cpu-cap 70  전체 CPU % 가 넘으면 새 도엽 안 줌 · --e-cores 12-19 --e-workers 3
       일꾼 셋만 그 논리 CPU 로 지정 · --root 창고 자리(시험용 임시 폴더)

- 도엽 목록 = 전국 등고선 파일 꼭짓점이 닿는 도엽(0.025°) — `_precompute_sheets.json`(전국 파일
  지문이 같으면 다시 안 셈) · 차례 = 등고선 길이 큰 것(산지)부터 · 같으면 높이 차.
- 좌표계 = 도엽 가운데 경도의 TM 띠(`tm_epsg_for_lon` · 5185~5187) · 격자 2 m.
- 셈 = 공용 함수만 부름(창고 지형 · D 0차곡 누적 · E07 도엽) — 저장 열쇠가 같으면 함수가 건너뜀 ·
  쓰기 = 함수 쪽 원자 쓰기(끊겨도 반쯤 쓴 파일 없음).
- 이어 하기 = PC 별 `_precompute_progress_<컴퓨터 이름>.json`(단계 · 도엽마다 성공 · 시간 ·
  시놀로지가 두 PC 사이 옮김 → 한 파일은 한 PC 만 씀) · 옛 `_precompute_progress.json` 의 성공은
  읽어서 이어 받음(안 씀) · 성공한 도엽은 다시 안 부름 · 실패는 다음에 다시 · 전국 파일이 바뀌면
  기록을 새로 시작.
- 나눔 = 지형 단계가 목록 도엽 + 0차곡이 쓸 이웃(목록 밖 · 띠 넘음 포함)을 해시 몫대로 셈 ·
  0차곡은 남의 몫 이웃 지형이 창고에 없으면 미뤘다 다시(다른 PC 셈 · 시놀로지 동기 기다림).
- 시작 때 자기 몫 도엽의 죽은 셈 잠금(주인 프로세스 없음) · 오래된 임시 파일을 치움 ·
  낮은 우선순위(BELOW_NORMAL).
"""

from __future__ import annotations

import argparse
import json
import math
import multiprocessing
import os
import platform
import sqlite3
import sys
import threading
import time
import zlib
from concurrent.futures import FIRST_COMPLETED, ProcessPoolExecutor, wait
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import numpy as np  # noqa: E402

from B04_PreProcess.B04_PreProcess_Engine_MapSheet import (  # noqa: E402
    SHEET5K_SIZE_DEG,
    latlon_to_sheet5k,
    neighbors_3x3,
    sheet5k_center,
)
from common_util import common_util_terrain_store as store  # noqa: E402
from common_util.common_util_crs import tm_epsg_for_lon  # noqa: E402
from common_util.common_util_json import atomic_write_json  # noqa: E402

STEPS = ("dem", "zero", "e07")
CELL_M = 2.0
PROGRESS_NAME = "_precompute_progress.json"  # 옛 한 벌 기록(읽기만)
PC_NAME = platform.node() or "pc"
CPU_SAMPLE_S = 3.0  # 전체 CPU 평균 잼 · 상한 쓸 때 새 도엽 주는 간격
DEFER_WAIT_S = 600.0  # 0차곡 이웃 지형 기다림 한 바퀴
INDEX_NAME = "_precompute_sheets.json"
INDEX_CHUNK = 50_000  # 가닥 묶음(약 0.5 GB 읽기)
INDEX_STRIDE = 8  # 꼭짓점 솎음(도엽 2 km 에 비해 넉넉)
SAVE_EVERY_S = 10.0
# 단계별 창고 파일 앞머리(크기 셈)
STEP_FILES = {"dem": ("dem_",), "zero": ("zero_acc_",), "e07": ("e07_", "e07si_")}
_ENVELOPE = {0: 0, 1: 32, 2: 48, 3: 48, 4: 64}


def below_normal() -> None:
    """이 프로세스를 낮은 우선순위로(Windows) — 다른 일 · 서버를 막지 않게."""
    import psutil

    if os.name == "nt":
        psutil.Process().nice(psutil.BELOW_NORMAL_PRIORITY_CLASS)


def epsg_of(job: str) -> int:
    """도엽(또는 `도엽@좌표계` 일감)의 좌표계 — 기본 = 도엽 가운데 경도 띠."""
    no, _, epsg = job.partition("@")
    return int(epsg) if epsg else tm_epsg_for_lon(sheet5k_center(no)[1])


# ── 나눔 · CPU ───────────────────────────────────────────────────────────────

Part = tuple[int, int, int]


def parse_part(text: str) -> Part | None:
    """`A-B/N` → (A, B, N) · 빈 글 = 나눔 없음."""
    if not text:
        return None
    span, _, n = text.partition("/")
    a, _, b = span.partition("-")
    part = (int(a), int(b or a), int(n or 100))
    if not 0 <= part[0] <= part[1] < part[2]:
        raise ValueError(f"--part = A-B/N (0 ≤ A ≤ B < N): {text}")
    return part


def in_part(job: str, part: Part | None) -> bool:
    """도엽 번호 crc32 % N 이 A~B — PC 와 실행에 상관없이 같은 몫(겹침 0)."""
    if part is None:
        return True
    value = zlib.crc32(job.partition("@")[0].encode()) % part[2]
    return part[0] <= value <= part[1]


def parse_cpus(text: str) -> list[int]:
    """`12-19` · `12,14,16-17` → 논리 CPU 번호."""
    cpus: list[int] = []
    for piece in filter(None, (p.strip() for p in text.split(","))):
        a, _, b = piece.partition("-")
        cpus.extend(range(int(a), int(b or a) + 1))
    return cpus


class CpuMeter:
    """전체 CPU %(시스템 · 다른 일 포함) — 뒤 실이 `CPU_SAMPLE_S` 평균을 잼 · 부르면 마지막 값."""

    def __init__(self) -> None:
        import psutil

        self.value = psutil.cpu_percent(interval=CPU_SAMPLE_S)

        def loop() -> None:
            while True:
                self.value = psutil.cpu_percent(interval=CPU_SAMPLE_S)

        threading.Thread(target=loop, daemon=True).start()

    def __call__(self) -> float:
        return self.value


def init_worker(root: str, counter: Any, e_cpus: list[int], e_workers: int) -> None:
    """일꾼 시작 — 창고 자리 · 낮은 우선순위 · 앞선 `e_workers` 명은 `e_cpus` 에만(저전력 코어)."""
    store.TERRAIN_ROOT = Path(root)
    below_normal()
    if counter is None:
        return
    with counter.get_lock():
        slot = counter.value
        counter.value += 1
    if slot < e_workers:
        import psutil

        psutil.Process().cpu_affinity(e_cpus)


# ── 도엽 목록 ────────────────────────────────────────────────────────────────


def aggregate(wkbs: list[bytes], heights: np.ndarray) -> dict[int, list[float]]:
    """EPSG:5179 선(WKB) · 높이 → {칸 열쇠: [길이 m, 최저, 최고]} · 칸 = 0.025°(위도 · 경도)."""
    import shapely
    from pyproj import Transformer

    geoms = shapely.from_wkb(wkbs)
    coords, owner = shapely.get_coordinates(geoms, return_index=True)
    coords, owner = coords[::INDEX_STRIDE], owner[::INDEX_STRIDE]
    if not len(coords):
        return {}
    lon, lat = Transformer.from_crs("EPSG:5179", "EPSG:4326", always_xy=True).transform(
        coords[:, 0], coords[:, 1]
    )
    keys = np.floor(lat / SHEET5K_SIZE_DEG).astype(np.int64) * 100_000 + np.floor(
        lon / SHEET5K_SIZE_DEG
    ).astype(np.int64)
    seg = np.zeros(len(coords))
    same = owner[1:] == owner[:-1]
    seg[:-1] = np.where(same, np.hypot(*(coords[1:] - coords[:-1]).T), 0.0)
    z = np.asarray(heights, dtype=np.float64)[owner]
    uniq, inv = np.unique(keys, return_inverse=True)
    length = np.bincount(inv, weights=seg, minlength=len(uniq))
    lo = np.full(len(uniq), np.inf)
    hi = np.full(len(uniq), -np.inf)
    np.minimum.at(lo, inv, z)
    np.maximum.at(hi, inv, z)
    return {int(k): [float(a), float(b), float(c)] for k, a, b, c in zip(uniq, length, lo, hi)}


def _merge(total: dict[int, list[float]], part: dict[int, list[float]]) -> None:
    for k, (length, lo, hi) in part.items():
        if k in total:
            t = total[k]
            t[0] += length
            t[1] = min(t[1], lo)
            t[2] = max(t[2], hi)
        else:
            total[k] = [length, lo, hi]


def _gpkg_rows(gpkg: Path):
    """전국 등고선 가닥 묶음 — (WKB 목록 · 높이) · 읽기 전용."""
    con = sqlite3.connect(f"file:{gpkg.resolve().as_posix()}?mode=ro", uri=True)
    try:
        last = -1
        while True:
            rows = con.execute(
                "SELECT fid, geom, CTRLN_HG FROM contours WHERE fid > ? AND geom IS NOT NULL"
                " ORDER BY fid LIMIT ?",
                (last, INDEX_CHUNK),
            ).fetchall()
            if not rows:
                return
            last = rows[-1][0]
            wkbs = [g[8 + _ENVELOPE[(g[3] >> 1) & 7] :] for _, g, _ in rows]
            yield wkbs, np.array([h if h is not None else np.nan for _, _, h in rows])
    finally:
        con.close()


def sheet_index(rebuild: bool = False) -> dict[str, list[float]]:
    """{도엽: [등고선 길이 km, 최저 m, 최고 m]} — 전국 파일 지문이 같으면 저장본."""
    path = store.TERRAIN_ROOT / INDEX_NAME
    stamp = store._gpkg_stamp()
    if not rebuild:
        try:
            saved = json.loads(path.read_text(encoding="utf-8"))
            if saved.get("gpkg") == stamp or stamp is None:
                return saved["sheets"]
        except (OSError, ValueError, KeyError):
            pass
    if stamp is None:
        raise SystemExit(f"전국 등고선 파일 없음: {store.NATIONAL_GPKG}")
    started = time.monotonic()
    cells: dict[int, list[float]] = {}
    for n, (wkbs, heights) in enumerate(_gpkg_rows(store.NATIONAL_GPKG), 1):
        _merge(cells, aggregate(wkbs, heights))
        elapsed = time.monotonic() - started
        print(f"도엽 목록 {n * INDEX_CHUNK:,} 가닥 · {elapsed:.0f} s", flush=True)
    step = SHEET5K_SIZE_DEG
    sheets = {
        latlon_to_sheet5k((k // 100_000 + 0.5) * step, (k % 100_000 + 0.5) * step): [
            round(v[0] / 1000, 2),
            v[1],
            v[2],
        ]
        for k, v in cells.items()
        if math.isfinite(v[1])
    }
    atomic_write_json(path, {"gpkg": stamp, "sheets": sheets})
    return sheets


def pick_sheets(
    index: dict[str, list[float]],
    *,
    sheets: list[str] | None = None,
    box: tuple[float, float, float, float] | None = None,
    min_relief: float = 150.0,
    limit: int | None = None,
) -> list[str]:
    """고른 도엽 — 등고선 길이 큰 것부터(같으면 높이 차) · 전국 파일이 안 덮는 도엽은 뺌."""
    if sheets:
        chosen = [no for no in index if any(no.startswith(p) for p in sheets)]
    else:
        chosen = [no for no in index if index[no][2] - index[no][1] >= min_relief]
    if box is not None:
        lon0, lat0, lon1, lat1 = box
        inside = []
        for no in chosen:
            lat, lon = sheet5k_center(no)
            if lon0 <= lon <= lon1 and lat0 <= lat <= lat1:
                inside.append(no)
        chosen = inside
    chosen.sort(key=lambda no: (-index[no][0], -(index[no][2] - index[no][1]), no))
    return chosen[:limit] if limit else chosen


# ── 셈(일꾼) ─────────────────────────────────────────────────────────────────


def _mtime(path: Path) -> int | None:
    try:
        return path.stat().st_mtime_ns
    except OSError:
        return None


def dem_jobs(sheets: list[str], part: Part | None) -> list[str]:
    """지형 단계 일감 — 목록 도엽 뒤 0차곡이 쓸 3x3 이웃(그 가운데 도엽 좌표계) · 자기 몫만.

    이웃이 가운데와 띠가 다르면 `도엽@좌표계`(옮겨 담기) — 목록 밖 이웃도 한 PC 만 셈.
    """
    jobs: dict[str, None] = {no: None for no in sheets if in_part(no, part)}
    for center in sheets:
        epsg = epsg_of(center)
        for no in neighbors_3x3(center):
            if in_part(no, part):
                jobs[no if epsg == epsg_of(no) else f"{no}@{epsg}"] = None
    return list(jobs)


def run_step(step: str, job: str, part: Part | None = None) -> tuple[str, str, dict[str, Any]]:
    """일감 하나 · 단계 하나 — 공용 함수만 부름 · fresh = 이번에 새로 셈 · wait = 이웃 기다림."""
    sheet_no = job.partition("@")[0]
    epsg = epsg_of(job)
    started = time.monotonic()
    try:
        if step == "dem":
            fresh = store._stored_ok(sheet_no, epsg, CELL_M) is None
            empty = store.ensure_sheet_terrain(sheet_no, epsg, CELL_M) is None
        elif step == "zero":
            from D03_PlanAnalysis import D03_PlanAnalysis_ZeroOrder as zero

            neighbors = neighbors_3x3(sheet_no)
            for no in neighbors:  # 남의 몫 이웃 지형이 아직 없음 = 미룸(누적 열쇠가 나중에 바뀜)
                if not in_part(no, part) and store._stored_ok(no, epsg, CELL_M) is None:
                    return step, job, {"ok": False, "wait": True, "error": f"이웃 지형 없음 {no}"}
            for no in neighbors:  # 여유 200 m 이웃 지형(누적 열쇠가 나중에 안 바뀌게)
                store.ensure_sheet_terrain(no, epsg, CELL_M)
            target = zero._acc_names(sheet_no, epsg, CELL_M)[0]
            before = _mtime(target)
            empty = zero.sheet_accumulation(sheet_no, epsg, CELL_M) is None
            fresh = _mtime(target) != before
        elif step == "e07":
            from E07_StabArea import E07_StabArea_Router as e07_router
            from E07_StabArea import E07_StabArea_Sheet as e07

            paths = e07._paths(sheet_no, epsg, CELL_M)
            before = (_mtime(paths["flow"]), _mtime(paths["si"]))
            empty = e07.ensure_sheet(sheet_no, epsg, CELL_M, e07_router.default_params()) is None
            fresh = (_mtime(paths["flow"]), _mtime(paths["si"])) != before
        else:
            raise ValueError(f"모르는 단계: {step}")
    except Exception as exc:  # noqa: BLE001 — 한 장이 죽어도 나머지는 감 · 기록에 까닭
        return step, job, {"ok": False, "error": f"{type(exc).__name__}: {exc}"[:300]}
    seconds = round(time.monotonic() - started, 2)
    return step, job, {"ok": True, "s": seconds, "fresh": fresh, "empty": empty}


# ── 진행 기록 · 정리 ─────────────────────────────────────────────────────────


def progress_path() -> Path:
    return store.TERRAIN_ROOT / f"_precompute_progress_{PC_NAME}.json"


def progress_files() -> list[Path]:
    """모든 PC 기록(옛 한 벌 · 시놀로지 충돌 사본 포함)."""
    return sorted(store.TERRAIN_ROOT.glob("_precompute_progress*.json"))


def _read_progress(path: Path, stamp: str | None) -> dict[str, Any] | None:
    try:
        saved = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    if not isinstance(saved, dict) or saved.get("gpkg") != stamp or saved.get("cell_m") != CELL_M:
        return None
    return saved


def load_progress(every_pc: bool = False) -> dict[str, Any]:
    """이 PC 기록 + 옛 한 벌의 성공(every_pc = 모든 PC 기록의 성공) — 쓰기는 이 PC 파일만."""
    stamp = store._gpkg_stamp()
    own_path = progress_path()
    saved = _read_progress(own_path, stamp) or {"gpkg": stamp, "cell_m": CELL_M, "steps": {}}
    for step in STEPS:
        saved["steps"].setdefault(step, {})
    others = progress_files() if every_pc else [store.TERRAIN_ROOT / PROGRESS_NAME]
    for path in others:
        other = None if path == own_path else _read_progress(path, stamp)
        for step, records in (other or {}).get("steps", {}).items():
            mine = saved["steps"].get(step)
            if mine is None:
                continue
            for job, info in records.items():
                if info.get("ok") and not mine.get(job, {}).get("ok"):
                    mine[job] = info
    return saved


def save_progress(progress: dict[str, Any]) -> None:
    atomic_write_json(progress_path(), progress)


def _alive(pid: int) -> bool:
    import psutil

    return psutil.pid_exists(pid)


def clean_dead(root: Path | None = None, part: Part | None = None) -> int:
    """자기 몫 도엽의 주인이 죽은 잠금 · 오래된 임시 파일을 지움(끊긴 일괄 셈이 남긴 것) → 지운 수.

    남의 몫 도엽 = 다른 PC 잠금(이 PC 에선 주인 번호가 없어 보임)이라 안 건드림.
    """
    root = root or store.TERRAIN_ROOT
    removed = 0
    if not root.is_dir():
        return 0
    now = time.time()
    for folder in root.iterdir():
        if not folder.is_dir() or not in_part(folder.name, part):
            continue
        for path in folder.glob(".*"):
            try:
                if path.suffix == ".lock":
                    pid = int(path.read_text(encoding="utf-8").split()[0])
                    if _alive(pid):
                        continue
                elif path.suffix != ".tmp" or now - path.stat().st_mtime < store.LOCK_STALE_S:
                    continue
                path.unlink(missing_ok=True)
                removed += 1
            except (OSError, ValueError, IndexError):
                continue
    return removed


def _execute(step: str, todo: list[str], workers: int, record, opts: dict[str, Any]) -> None:
    """일감 차례대로 일꾼에 줌 — 상한 쓰면 전체 CPU 가 넘는 동안 새 일 안 줌(도는 것은 그대로)."""
    part, cap, cpu = opts["part"], opts["cpu_cap"], opts["cpu"]

    def over() -> bool:
        return cap is not None and cpu() > cap

    if workers <= 1:
        for job in todo:
            while over():
                time.sleep(CPU_SAMPLE_S)
            record(run_step(step, job, part))
        return
    e_cpus, e_workers = opts["e_cpus"], opts["e_workers"]
    counter = multiprocessing.Value("i", 0) if e_cpus and e_workers else None
    initargs = (str(store.TERRAIN_ROOT), counter, e_cpus, e_workers)
    with ProcessPoolExecutor(workers, initializer=init_worker, initargs=initargs) as pool:
        queue = list(reversed(todo))
        running: set = set()
        try:
            while queue or running:
                while queue and len(running) < workers and not over():
                    running.add(pool.submit(run_step, step, queue.pop(), part))
                    if cap is not None:
                        break  # 상한 쓸 때 한 번에 하나 — 다음 잼을 보고 또
                if not running:
                    time.sleep(CPU_SAMPLE_S)
                    continue
                timeout = CPU_SAMPLE_S if cap is not None and queue else None
                finished, running = wait(running, timeout=timeout, return_when=FIRST_COMPLETED)
                for future in finished:
                    record(future.result())
        except BaseException:
            pool.shutdown(wait=False, cancel_futures=True)
            raise


def run(
    sheets: list[str],
    steps: list[str],
    workers: int,
    progress: dict[str, Any],
    say=print,
    *,
    part: Part | None = None,
    cpu_cap: float | None = None,
    cpu=None,
    e_cpus: list[int] | None = None,
    e_workers: int = 0,
    defer_wait: float = 0.0,
) -> dict[str, Any]:
    """단계 차례대로 · 단계 안은 일꾼 여럿 · 성공 기록된 일감은 건너뜀 → 단계별 벽시계 초.

    sheets = 나누기 전 목록(이웃 몫에 씀) · 미룬 0차곡 = defer_wait 초 뒤 다시(0 = 안 기다림).
    """
    if cpu_cap is not None and cpu is None:
        cpu = CpuMeter()
    opts = {"part": part, "cpu_cap": cpu_cap, "cpu": cpu, "e_cpus": e_cpus or []}
    opts["e_workers"] = e_workers
    walls: dict[str, float] = {}
    for step in [s for s in STEPS if s in steps]:
        done = progress["steps"][step]
        jobs = dem_jobs(sheets, part) if step == "dem" else [n for n in sheets if in_part(n, part)]
        todo = [job for job in jobs if not done.get(job, {}).get("ok")]
        say(f"[{step}] {len(todo)} 장 (건너뜀 {len(jobs) - len(todo)} · 몫 {len(jobs)})")
        started = time.monotonic()
        saved_at = started
        count = 0
        waiting: list[str] = []

        def record(result: tuple[str, str, dict[str, Any]]) -> None:
            nonlocal count, saved_at
            _, job, info = result
            done[job] = info
            count += 1
            if info.get("wait"):
                waiting.append(job)
            state = (
                f"{info['s']} s"
                if info["ok"]
                else f"{'기다림' if info.get('wait') else '실패'} {info['error']}"
            )
            say(f"[{step}] {count}/{len(todo)} {job} {state}")
            if time.monotonic() - saved_at > SAVE_EVERY_S:
                save_progress(progress)
                saved_at = time.monotonic()

        try:
            while True:
                _execute(step, todo, workers, record, opts)
                if not waiting or defer_wait <= 0:
                    break
                save_progress(progress)
                say(f"[{step}] 이웃 지형 기다림 {len(waiting)} 장 · {defer_wait:g} s 뒤 다시")
                time.sleep(defer_wait)
                todo, waiting[:], count = list(waiting), [], 0
        finally:
            save_progress(progress)
        walls[step] = round(time.monotonic() - started, 1)
    return walls


def summary(sheets: list[str], progress: dict[str, Any]) -> dict[str, Any]:
    """단계별 — 성공 · 새로 셈 · 실패 · 새로 셈 장당 초(가운데 · 평균 · 최대) · 장당 크기 MB."""
    out: dict[str, Any] = {}
    for step in STEPS:
        done = progress["steps"][step]
        infos = [done[no] for no in sheets if no in done]
        failed = {
            no: done[no]["error"]
            for no in sheets
            if done.get(no, {}).get("ok") is False and not done[no].get("wait")
        }
        fresh = [i["s"] for i in infos if i.get("ok") and i.get("fresh")]
        sizes = []  # 창고 실물(다른 PC 가 셈 · 시놀로지로 옴 포함)
        for no in sheets:
            stem = f"{epsg_of(no)}_{CELL_M:g}m"
            folder = store.TERRAIN_ROOT / no
            total = 0
            for head in STEP_FILES[step]:
                total += sum(p.stat().st_size for p in folder.glob(f"{head}{stem}*"))
            if total:
                sizes.append(total / 1e6)
        out[step] = {
            "ok": sum(1 for i in infos if i.get("ok")),
            "fresh": len(fresh),
            "failed": failed,
            "waiting": sum(1 for no in sheets if done.get(no, {}).get("wait")),
            "stored": len(sizes),
            "s_median": round(float(np.median(fresh)), 1) if fresh else None,
            "s_mean": round(float(np.mean(fresh)), 1) if fresh else None,
            "s_max": round(float(np.max(fresh)), 1) if fresh else None,
            "mb_mean": round(float(np.mean(sizes)), 2) if sizes else None,
            "mb_total": round(float(np.sum(sizes)), 1) if sizes else 0.0,
        }
    return out


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="공용 지형 창고 전국 미리 계산(PLAN 612)")
    parser.add_argument("--steps", default="dem,zero,e07")
    parser.add_argument("--sheets", default="")
    parser.add_argument("--box", default="")
    parser.add_argument("--min-relief", type=float, default=150.0)
    parser.add_argument("--limit", type=int, default=0)
    parser.add_argument("--workers", type=int, default=4)
    parser.add_argument("--status", action="store_true")
    parser.add_argument("--rebuild-index", action="store_true")
    parser.add_argument("--part", default="", help="A-B/N — 도엽 번호 crc32 %% N 이 A~B 인 것만")
    parser.add_argument("--cpu-cap", type=float, default=None, help="전체 CPU %% 상한")
    parser.add_argument("--e-cores", default="", help="저전력 논리 CPU(예 12-19)")
    parser.add_argument("--e-workers", type=int, default=0, help="그 CPU 로 지정할 일꾼 수")
    parser.add_argument("--defer-wait", type=float, default=DEFER_WAIT_S)
    parser.add_argument("--root", default="", help="창고 자리(시험용 임시 폴더)")
    args = parser.parse_args(argv)
    try:
        part = parse_part(args.part)
        e_cpus = parse_cpus(args.e_cores)
    except ValueError as exc:
        parser.error(str(exc))
    if (args.e_workers and not e_cpus) or args.e_workers > args.workers:
        parser.error("--e-workers 는 --e-cores 와 함께 · --workers 이하")
    if args.root:
        store.TERRAIN_ROOT = Path(args.root)
    steps = [s.strip() for s in args.steps.split(",") if s.strip()]
    unknown = [s for s in steps if s not in STEPS]
    if unknown:
        parser.error(f"모르는 단계 {unknown} — 지금 되는 단계 {STEPS}(tiles 는 다음)")
    box = tuple(float(v) for v in args.box.split(",")) if args.box else None
    if box is not None and len(box) != 4:
        parser.error("--box = lon0,lat0,lon1,lat1")

    below_normal()
    index = sheet_index(args.rebuild_index)
    wanted = [s.strip() for s in args.sheets.split(",") if s.strip()]
    sheets = pick_sheets(
        index, sheets=wanted or None, box=box, min_relief=args.min_relief, limit=args.limit or None
    )
    print(
        f"도엽 {len(sheets)} 장 (목록 {len(index)} 장) · 단계 {steps} · 일꾼 {args.workers}"
        f" · 몫 {args.part or '전부'} · PC {PC_NAME} · 상한 {args.cpu_cap}"
        f" · 지정 CPU {e_cpus} × {args.e_workers}"
    )
    mine = [no for no in sheets if in_part(no, part)]
    if args.status:
        out = summary(mine, load_progress(every_pc=True))
        stamp = store._gpkg_stamp()
        out["pc"] = {}
        for path in progress_files():
            recs = (_read_progress(path, stamp) or {}).get("steps", {})
            out["pc"][path.name] = {
                step: sum(1 for i in recs.get(step, {}).values() if i.get("ok")) for step in STEPS
            }
    else:
        progress = load_progress()
        print(f"치운 잠금 · 임시 파일 {clean_dead(part=part)}")
        started = time.monotonic()
        walls = run(
            sheets,
            steps,
            args.workers,
            progress,
            part=part,
            cpu_cap=args.cpu_cap,
            e_cpus=e_cpus,
            e_workers=args.e_workers,
            defer_wait=args.defer_wait,
        )
        print(f"벽시계 {walls} · 전체 {time.monotonic() - started:.1f} s")
        out = summary(mine, progress)
    print(json.dumps(out, ensure_ascii=False, indent=1))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())