Aislo/common_util/common_util_calc_runner.py

"""계산 등록부 파이썬 실행기(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` · 묶음 단계 `STAGE_IDS`(511-B) · 화면 길 `PAGE_IDS`(B-4) 도 받음.
"""

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",
        "profile.base",
        "profile.alignment",
        "profile.carry",
        "facility.plus_follow",
        "facility.follow",
        "cross.design",
        "cross.berm_spans",
        "facility.rebase",
        "stations.carry",
        "facility.attach",
        "cross.enforce",
        "stations.list",
        "stations.sections",
        "route.polyline",
        "stations.plan",
    }
)
# 창구(`/calc`)가 안 받는 id — 서버 파일 경로를 입력으로 받음(측점 생성 · 511-B B-2)
SERVER_ONLY_IDS = frozenset({"stations.sections"})
# 시험 길(`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",
    }
)
# 묶음 단계(`design_calc_node_stages.ts` `STAGE_IDS` 와 같음 · 511-B) — 한 요청 = 묶음 한 바퀴
STAGE_IDS = frozenset(
    {"stage.route", "stage.follow", "stage.stations", "stage.profile", "stage.cross"}
)
# 화면 길(`design_calc_node_cross.ts` `PAGE_IDS` · 511-B B-4) — cross = server 저장 노선 다시 셈
PAGE_IDS = frozenset({"cross.refresh"})
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)}")
    # 늘 떠 있는 Node(511-B B-4 · 부름당 새 프로세스 0.1~0.2 s 덞) · 못 띄우면 새 프로세스 한 번
    answer = _SERVED.call({"run": list(ids), "inputs": inputs})
    if answer is None:
        answer = run_bundle_json(BUNDLE, NPM_SCRIPT, {"run": list(ids), "inputs": inputs})
    return answer.get("results") if isinstance(answer, 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 | STAGE_IDS | PAGE_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