140 lines
5.0 KiB
Python
140 lines
5.0 KiB
Python
"""계산 등록부 파이썬 실행기(PLAN 472-3 3-B) — 등록부 id 를 받아 서버 Node 입구 하나를 부름.
|
|
|
|
등록부 = `A00_Common/design/logic/design_calc_registry.ts` · 입구 = `design_calc_node.ts`
|
|
(`build:design-calc`). 부름은 `common_util_node_bundle.run_bundle_json` 한 길 · 계산 없음.
|
|
`NODE_IDS` = 등록부에서 실행기가 `node:` 인 줄(같은지 시험 `test_472_3_calc_runner.py`).
|
|
입력 꼴 = 입구 머리(`detail` · `route_points` · `context` · `excavation`).
|
|
`run_calc_served` = 계산 자리 「서버」 창구(511-A · `B06_Section_Router_Calc`) —
|
|
늘 떠 있는 Node 하나(`--serve` · 한 줄 요청 · 한 줄 답) · 죽음 · 묶음 다시 지음 = 다시 띄움 ·
|
|
시험 길 `TRIAL_IDS` 도 받음.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import atexit
|
|
import json
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from common_util.common_util_node_bundle import (
|
|
build_bundle,
|
|
bundle_stale,
|
|
node_env,
|
|
run_bundle_json,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BUNDLE = ROOT / "config" / "design_calc_node" / "design_calc_node.js"
|
|
NPM_SCRIPT = "build:design-calc"
|
|
NODE_IDS = frozenset(
|
|
{
|
|
"structure.layouts",
|
|
"masshaul.cumulative",
|
|
"haul.plan",
|
|
"excavation.trench",
|
|
"corridor.mesh",
|
|
"profile.seat",
|
|
"facility.plus_follow",
|
|
"facility.follow",
|
|
"cross.design",
|
|
"cross.berm_spans",
|
|
"facility.rebase",
|
|
"stations.carry",
|
|
}
|
|
)
|
|
# 시험 길(`design_calc_node_trial.ts` `TRIAL` 표와 같음) — 조작 중 브라우저 함수를 서버가 같은 TS 로
|
|
TRIAL_IDS = frozenset(
|
|
{
|
|
"masshaul.series",
|
|
"stations.carry_keyed",
|
|
"route.handle",
|
|
"route.station_fit",
|
|
"stations.swap",
|
|
"profile.tools",
|
|
"cross.fit",
|
|
"footprint.plan",
|
|
}
|
|
)
|
|
SERVE_TIMEOUT_S = 60
|
|
|
|
|
|
def run_calc(ids: Sequence[str], inputs: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""id 들을 한 번에 돌려 `{id: 값}` · 묶음 실패는 None · Node 실행기 아닌 id 는 ValueError."""
|
|
unknown = [item for item in ids if item not in NODE_IDS]
|
|
if unknown:
|
|
raise ValueError(f"Node 실행기 아닌 id: {', '.join(unknown)}")
|
|
output = run_bundle_json(BUNDLE, NPM_SCRIPT, {"run": list(ids), "inputs": inputs})
|
|
return output.get("results") if isinstance(output, dict) else None
|
|
|
|
|
|
class _Served:
|
|
"""늘 떠 있는 Node 하나 — 한 번에 한 요청(잠금) · 답 없음 · 죽음 = 끄고 다음 부름에 다시."""
|
|
|
|
def __init__(self) -> None:
|
|
self.lock = threading.Lock()
|
|
self.proc: subprocess.Popen[str] | None = None
|
|
self.stamp = 0
|
|
|
|
def _start(self) -> subprocess.Popen[str]:
|
|
self.stop()
|
|
self.proc = subprocess.Popen( # noqa: S603 — 고정 실행 파일 · 입력은 표준 입력 한 줄
|
|
["node", str(BUNDLE), "--serve"],
|
|
cwd=str(ROOT),
|
|
env=node_env(),
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
text=True,
|
|
encoding="utf-8",
|
|
)
|
|
self.stamp = BUNDLE.stat().st_mtime_ns
|
|
return self.proc
|
|
|
|
def stop(self) -> None:
|
|
if self.proc and self.proc.poll() is None:
|
|
self.proc.kill()
|
|
self.proc = None
|
|
|
|
def call(self, payload: dict[str, Any]) -> dict[str, Any] | None:
|
|
line = json.dumps(payload, default=float) + "\n"
|
|
with self.lock:
|
|
if bundle_stale(BUNDLE) and not build_bundle(NPM_SCRIPT):
|
|
return None
|
|
for _ in range(2): # 죽은 프로세스에 쓴 첫 부름 = 다시 띄워 한 번 더
|
|
proc = self.proc
|
|
if not proc or proc.poll() is not None or BUNDLE.stat().st_mtime_ns != self.stamp:
|
|
proc = self._start()
|
|
timer = threading.Timer(SERVE_TIMEOUT_S, proc.kill)
|
|
timer.start()
|
|
try:
|
|
proc.stdin.write(line) # type: ignore[union-attr]
|
|
proc.stdin.flush() # type: ignore[union-attr]
|
|
answer = proc.stdout.readline() # type: ignore[union-attr]
|
|
except OSError:
|
|
answer = ""
|
|
finally:
|
|
timer.cancel()
|
|
if answer:
|
|
return json.loads(answer)
|
|
self.stop()
|
|
return None
|
|
|
|
|
|
_SERVED = _Served()
|
|
atexit.register(_SERVED.stop)
|
|
|
|
|
|
def run_calc_served(ids: Sequence[str], inputs: dict[str, Any]) -> dict[str, Any] | None:
|
|
"""창구 길 — `{results, ms(Node 셈), wall_ms(파이썬 ↔ Node)}` · `{error}` · 못 돌림 None."""
|
|
unknown = [item for item in ids if item not in NODE_IDS | TRIAL_IDS]
|
|
if unknown:
|
|
raise ValueError(f"서버 계산 id 아님: {', '.join(unknown)}")
|
|
t0 = time.perf_counter()
|
|
answer = _SERVED.call({"run": list(ids), "inputs": inputs})
|
|
if answer is not None:
|
|
answer["wall_ms"] = round((time.perf_counter() - t0) * 1000, 2)
|
|
return answer
|