85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
"""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 functools import partial
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
from common_util.common_util_derived_store import write_derived
|
|
from E02_StabGround.E02_StabGround_Model import station_groundwater
|
|
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"]
|
|
if conditions_ground is not None:
|
|
kw["groundwater_for"] = partial(station_groundwater, conditions_ground)
|
|
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()
|