eomsangdonandClaude Opus 5.5
164024ab43
- analyze2d 가 측점마다 벽(walls_by_station · 복사 지문으로 기억) · 그려지는 설계선을 넘김 · 벽 있는 측점만 지문에 벽 · 벽 없는 측점 = 종전 지문 - 측점 보기 = 같은 벽 모형 · 수위선 = E02 고른 지하수 갈래(station_groundwater) - 화면 — 「반영 전」 안내 걷음 · 벽 회색 빗금 · 결과 「벽 있음」 · 목록 벽 표시 - result2d · stab/ground 에 missing_stations 칸 덧붙임 · 화면 위 띠(펼침 목록) - E03 지반 조건 = E02 화면과 같은 upgrade(옛 정본 지하수 칸 기본값) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
77 lines
3.0 KiB
Python
77 lines
3.0 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, items: [{fp, meta, layers, cross, design, walls, ground}]}.
|
|
"""
|
|
|
|
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]) -> 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"]
|
|
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), 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): 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()
|