"""E03 측점 해석 일꾼 — 서버 밖 프로세스(PLAN 532-3).

`python -m E03_StabSection.E03_StabSection_Worker <job.json>` — 서버(`Router.run_all`)가 띄움.
측점마다 프로세스로 나눠 돌리고 끝날 때마다 파생 저장본을 쓰고 한 줄(JSON)을 찍음 → 서버 진행률.
서버 프로세스에서 바로 프로세스 풀을 열면 자식마다 `main.py` 를 다시 읽어 이 모듈을 입구로 둠.
job = {root, keep, kwargs, workers, conditions_ground, items: [{fp, meta, layers, cross, design,
walls, ground}]} · conditions_ground = E02 지반 조건(지하수 갈래 → 측점 수위).
"""

from __future__ import annotations

import json
import logging
import sys
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Callable

from common_util.common_util_derived_store import write_derived
from E03_StabSection.E03_StabSection_Engine_Station import analyze_station

KIND_STATION = "stab2d_station"
logger = logging.getLogger(__name__)


def _analyze(
    item: dict[str, Any], kwargs: dict[str, Any], conditions_ground: dict | None = None
) -> dict[str, Any]:
    kw = {**kwargs, "layers": item["layers"], "ground": [tuple(p) for p in item["ground"]]}
    if item.get("walls"):
        kw["walls"] = item["walls"]
    return analyze_station(item["cross"], item["design"], **kw)


def _value(item: dict[str, Any], out: dict | None, error: str | None) -> dict[str, Any]:
    return {
        "meta": item["meta"],
        "inputs": {"layers": item["layers"], "ground": item["ground"]},
        "out": out,
        "error": error,
    }


def run_job(job: dict[str, Any], emit: Callable[[str], None]) -> None:
    """측점마다 해석 → 저장본 · emit(열쇠). 한 측점 실패는 그 줄에 까닭만 남기고 이어 감."""
    root, keep, kwargs = Path(job["root"]), int(job["keep"]), job["kwargs"]
    conditions_ground = job.get("conditions_ground")
    items = job["items"]

    def save(item: dict[str, Any], out: dict | None, error: str | None) -> None:
        write_derived(root, KIND_STATION, item["fp"], _value(item, out, error), keep=keep)
        emit(item["meta"]["key"])

    workers = max(1, min(int(job.get("workers") or 1), len(items)))
    if workers == 1:
        for item in items:
            try:
                save(item, _analyze(item, kwargs, conditions_ground), None)
            except Exception as error:  # noqa: BLE001
                logger.exception("E03 측점 해석 실패: %s", item["meta"]["key"])
                save(item, None, str(error))
        return
    with ProcessPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(_analyze, item, kwargs, conditions_ground): item for item in items}
        for future in as_completed(futures):
            item = futures[future]
            try:
                save(item, future.result(), None)
            except Exception as error:  # noqa: BLE001
                logger.exception("E03 측점 해석 실패: %s", item["meta"]["key"])
                save(item, None, str(error))


def main() -> None:
    job = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
    run_job(job, lambda key: print(json.dumps({"done": key}), flush=True))


if __name__ == "__main__":
    main()