feat(476-1 ③): 추천 후보 초기 종단 미리 셈 — 응답 칸 initial_profile · 저장본

- 새 `B05_Profile_Engine_InitialProfile` — 후보 노드 → 화면 펴기 짝(`draft_nodes` ↔ TS flattenServerPlan) → 노선 짜기(csv 4자리) → 관 · 구조물 · 추가 측점 따라감(새 세류선 교차 관 포함) → build_sections(438 1차 계획선) → 켬 = 1 m 균형 종단 얹기(seat_base_pvi)
- 새 `B05_Profile_Router_RouteInitial_Initial.initial_profile_of` — [적용] 체인과 같은 입력 모으기 · 지문(후보 · 지형 · 페이지 설정 · 시설 파일 · 측점 옵션 · 균형 인자 · 코드) · 저장본 `route_recommend/initial/<키>.json` · 실패 = {error}
- compute_initial 응답에 `initial_profile`(stations_m · ground_m · plan_m · line · pvi_count · curve_count · straightened · max_grade_pct · curve_length_m · warnings · with_profile · seconds)
- 시험 test_476_1_recommend_initial — 미리 셈 ↔ 체인 파일 길 꼭짓점 차 0 mm(끔 · 노선 바뀜 + 옹벽 + 추가 측점 · 켬) · 지문 · 응답 칸
- 위키 calculations N2 · rules 줄

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

resources/tester/helper_476_1_flatten.cjs

@@ -0,0 +1,39 @@
+/* 476-1 추천 [적용] 초안 — 화면 `flattenServerPlan`(logic/B05_Profile_UI_RouteEdit_Edits.ts) +
+ * `routeDraftOf` 칸 꼴(곡선 켬 = !== false · R ?? null)을 Node 로 돌려 JSON 한 줄
+ * (`test_476_1_recommend_initial.py` 가 파이썬 `draft_nodes` 와 견줌).
+ *   node helper_476_1_flatten.cjs <후보 {nodes, curves} JSON 파일>
+ */
+const fs = require("fs");
+const path = require("path");
+const ts = require(path.join(__dirname, "..", "..", "config", "node_modules", "typescript"));
+
+const ROOT = path.join(__dirname, "..", "..");
+const cache = new Map();
+
+function loadTs(file) {
+  const full = path.resolve(file);
+  if (cache.has(full)) return cache.get(full);
+  const js = ts.transpileModule(fs.readFileSync(full, "utf8"), {
+    compilerOptions: { module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2020 },
+  }).outputText;
+  const box = { exports: {} };
+  cache.set(full, box.exports);
+  const localRequire = (request) =>
+    request.startsWith(".")
+      ? loadTs(path.join(path.dirname(full), `${request}.ts`))
+      : require(request);
+  new Function("exports", "module", "require", js)(box.exports, box, localRequire);
+  cache.set(full, box.exports);
+  return box.exports;
+}
+
+const edits = loadTs(path.join(ROOT, "B05_Profile", "logic", "B05_Profile_UI_RouteEdit_Edits.ts"));
+const body = JSON.parse(fs.readFileSync(process.argv[2], "utf8"));
+const flat = edits.flattenServerPlan(body.nodes, body.curves);
+process.stdout.write(
+  JSON.stringify({
+    nodes: flat.planned.map(([x, y]) => [x, y]),
+    curveOn: flat.planned.map((_, i) => flat.curveOn[i] !== false),
+    curveRadius: flat.planned.map((_, i) => flat.curveRadius[i] ?? null),
+  }),
+);

resources/tester/test_476_1_recommend_initial.py

@@ -0,0 +1,409 @@
+# -*- coding: utf-8 -*-
+"""476-1 추천 후보 초기 종단 — 미리 셈 = [적용] 체인 종단 · 화면 펴기 짝 · 응답 칸 · 저장본 지문.
+
+고정 자료 = `fixtures/golden_472`(사본) · 체인 쪽 = `run_route_save_chain` 파일 길 차례(노선 csv →
+구조물 · 추가 측점 따라감 → 노선 geojson → 관 따라감 · 정본 → `run_section_generation` →
+켬이면 `_set_base_pvi` + `_apply_alignment_edits`) · 미리 셈 쪽 = 메모리 조각(파일 안 씀).
+화면 펴기 = TS `flattenServerPlan`(Node `helper_476_1_flatten.cjs`) ↔ 파이썬 `draft_nodes`.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import json
+import subprocess
+import sys
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+TESTER = Path(__file__).resolve().parent
+for path in (ROOT, TESTER):
+    if str(path) not in sys.path:
+        sys.path.insert(0, str(path))
+
+import helper_472_1_paths as P  # noqa: E402
+from test_224_3_recommend_store import _ask, wired  # noqa: E402, F401 — 가짜 세계 fixture
+
+from B05_Profile import B05_Profile_Engine_InitialProfile as ip  # noqa: E402
+
+RADIUS = 12.0
+HELPER = TESTER / "helper_476_1_flatten.cjs"
+NODES_CSV = "B05_Profile/route/planned_route_initial_nodes.csv"
+WALL = {
+    "structure_id": "wall600",
+    "type_id": "retaining_wall",
+    "placement": "interval",
+    "chainage_m": 600.0,
+    "start_m": 595.0,
+    "end_m": 605.0,
+    "options": {"wall_form": "콘크리트", "length_m": 10, "height_m": 2},
+    "memo": "",
+    "placement_source": "manual",
+    "status": "draft",
+    "revision": 0,
+    "geometry": None,
+    "summary_item": None,
+    "group_items": None,
+    "x": 208785.614,
+    "y": 367845.1028,
+}
+
+
+def _candidate(root: Path, shift: float = 0.0) -> dict:
+    """엔진 결과 꼴 후보(`nodes` · `curves` as_dict) — 교점 9 를 동으로 `shift` m."""
+    from B05_Profile.B05_Profile_Router_Replan import _vertices_of
+    from common_util.common_util_route_polyline import build_planned_polyline
+
+    nodes = [tuple(p) for p in _vertices_of(root / NODES_CSV)]
+    nodes[9] = (nodes[9][0] + shift, nodes[9][1])
+    outline = build_planned_polyline(nodes, min_radius_m=RADIUS, simplify=False)
+    return {
+        "nodes": [n.as_dict() for n in outline.nodes],
+        "curves": [c.as_dict() for c in outline.curves],
+    }
+
+
+def _browser_draft(candidate: dict, work: Path) -> dict:
+    if not (ROOT / "config" / "node_modules" / "typescript").is_dir():
+        pytest.skip("프론트엔드 의존성 미설치")
+    body = work / "candidate.json"
+    body.write_text(json.dumps(candidate), encoding="utf-8")
+    out = subprocess.run(
+        ["node", str(HELPER), str(body)], capture_output=True, text=True, check=True
+    )
+    return json.loads(out.stdout)
+
+
+def _settings() -> dict:
+    from B05_Profile.B05_Profile_Engine_Grade import GradeDesignOptions
+    from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
+
+    summary = P.read("summary.json")["longitudinal"]["data"]
+    options = {k: v for k, v in summary["options"].items() if k != "extra_stations"}
+    return {
+        "options": SectionGenerationOptions(**options),
+        "grade_options": GradeDesignOptions(**summary["grade_options"]),
+        "crs": f"EPSG:{P.read('context.json')['crs_epsg']}",
+    }
+
+
+def _old_xy(root: Path) -> list[list[float]]:
+    coords = json.loads((root / P.ROUTE).read_text("utf-8"))["geometry"]["coordinates"]
+    return [[c[0], c[1]] for c in coords]
+
+
+def _seed(root: Path, extras: bool) -> None:
+    from B05_Profile.B05_Profile_Plus_Stations import save_plus_stations
+    from common_util.common_util_json import atomic_write_json
+
+    if extras:
+        atomic_write_json(
+            root / "B05_Profile" / "route" / "structures.json",
+            {"revision": 1, "structures": [copy.deepcopy(WALL)], "definition_version": None},
+        )
+        save_plus_stations(root, [510], _old_xy(root))
+
+
+def _chain(root: Path, draft: dict, target) -> dict:
+    """`run_route_save_chain` 파일 길 차례(DB 없이) — [적용] 뒤 종단 정본."""
+    from B05_Profile.B05_Profile_Engine import _route_geojson
+    from B05_Profile.B05_Profile_Engine_AsPlanned import _vertex_z
+    from B05_Profile.B05_Profile_Engine_Sections import run_section_generation
+    from B05_Profile.B05_Profile_Plus_Stations import follow_plus_stations
+    from B05_Profile.B05_Profile_Route_Save import followed_pipes
+    from B05_Profile.B05_Profile_Router import _apply_alignment_edits
+    from B05_Profile.B05_Profile_Router_Replan import _vertices_of, _write_planned_polyline
+    from B05_Profile.B05_Profile_Router_RouteInitial import _set_base_pvi
+    from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
+    from common_util.common_util_drainage_pipes import (
+        ensure_pipe_ids,
+        fill_pipe_coordinates,
+        pipe_points_path_in,
+        read_pipe_points_file,
+        route_signature,
+    )
+    from common_util.common_util_json import atomic_write_json
+    from common_util.common_util_route_geometry import planned_route_working_path
+
+    old_xy = _old_xy(root)
+    old_pipes = read_pipe_points_file(pipe_points_path_in(root))
+    working = planned_route_working_path(root)
+    count = len(draft["nodes"])
+    _write_planned_polyline(
+        working,
+        [tuple(n) for n in draft["nodes"]],
+        RADIUS,
+        simplify=False,
+        curve_flags=draft["curveOn"],
+        radii=draft["curveRadius"],
+        station_snap=True,
+        station_snap_step=None,
+        locks=[None] * count,
+        arcs=[None] * count,
+    )
+    points = [{"x": x, "y": y} for x, y in _vertices_of(working)]
+    follow_structures_to_route(root, P.ROUTE, points)
+    follow_plus_stations(root, [(p["x"], p["y"]) for p in points])
+    xy = [(p["x"], p["y"]) for p in points]
+    z = _vertex_z(root, *P.SURFACE, xy)
+    atomic_write_json(root / P.ROUTE, _route_geojson([[x, y, h] for (x, y), h in zip(xy, z)]))
+    pairs, _crossing, vertices = followed_pipes(old_pipes, old_xy, points, [])
+    pipes = [pipe for _, pipe in pairs]
+    fill_pipe_coordinates(pipes, vertices)
+    ensure_pipe_ids(pipes)
+    atomic_write_json(
+        pipe_points_path_in(root),
+        {"route_signature": route_signature(vertices), "points": [p.as_dict() for p in pipes]},
+    )
+    settings = _settings()
+    run_section_generation(root, P.ROUTE, *P.SURFACE, **settings)
+    if target is not None:
+        _set_base_pvi(root / P.LONG, *target)
+        _apply_alignment_edits(root, P.LONG, {})
+    return json.loads((root / P.LONG).read_text("utf-8"))
+
+
+def _preview(root: Path, candidate: dict, target) -> tuple[dict, list, list]:
+    from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
+    from common_util.common_util_surface_sampler import build_surface_sampler
+
+    line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS)
+    facilities = ip.followed_facilities(
+        root,
+        line,
+        P.ROUTE,
+        _old_xy(root),
+        read_pipe_points_file(pipe_points_path_in(root)),
+        [],
+    )
+    sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE)
+    longitudinal = ip.initial_longitudinal(
+        root,
+        line,
+        curves,
+        sampler,
+        facilities,
+        source_snapshot={"filter": "classification", "method": "dtm", "smooth": True},
+        target=target,
+        **_settings(),
+    )
+    return longitudinal, line, curves
+
+
+def _target(root: Path, line, curves):
+    from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import route_profile
+    from common_util.common_util_surface_sampler import build_surface_sampler
+
+    sampler = build_surface_sampler(root / "B04_PreProcess" / "models", *P.SURFACE)
+    profile = route_profile(line, sampler, max_grade=0.14, design_speed=20, design_curves=curves)
+    return profile["stations_m"], profile["plan_m"]
+
+
+CASES = {
+    "same_off": {"shift": 0.0, "extras": False, "balanced": False},
+    "moved_off": {"shift": 5.0, "extras": True, "balanced": False},
+    "moved_on": {"shift": 5.0, "extras": True, "balanced": True},
+}
+
+
+@pytest.fixture(scope="module", params=list(CASES))
+def pair(request, tmp_path_factory):
+    case = CASES[request.param]
+    work = tmp_path_factory.mktemp(request.param)
+    chain_root = P.stage_project(work / "chain")
+    preview_root = P.stage_project(work / "preview")
+    for root in (chain_root, preview_root):
+        _seed(root, case["extras"])
+    candidate = _candidate(preview_root, case["shift"])
+    draft = _browser_draft(candidate, work)
+    target = None
+    if case["balanced"]:
+        line, curves = ip.candidate_line(candidate["nodes"], candidate["curves"], RADIUS)
+        target = _target(preview_root, line, curves)
+    preview, line, curves = _preview(preview_root, candidate, target)
+    chain = _chain(chain_root, draft, target)
+    return {"preview": preview, "chain": chain, "line": line, "curves": curves, "root": chain_root}
+
+
+def test_화면_펴기_짝(tmp_path):
+    root = P.stage_project(tmp_path / "p")
+    candidate = _candidate(root, 5.0)
+    draft = _browser_draft(candidate, tmp_path)
+    points, flags, radii = ip.draft_nodes(candidate["nodes"], candidate["curves"])
+    assert [list(p) for p in points] == draft["nodes"]
+    assert flags == draft["curveOn"] and radii == draft["curveRadius"]
+
+
+def test_노선_곡선표_같음(pair):
+    from B05_Profile.B05_Profile_Engine_Sections import _load_design_curves
+    from B05_Profile.B05_Profile_Router_Replan import _vertices_of
+    from common_util.common_util_route_geometry import planned_route_working_path
+
+    root = pair["root"]
+    assert [tuple(p) for p in _vertices_of(planned_route_working_path(root))] == pair["line"]
+    assert _load_design_curves(root) == pair["curves"]
+
+
+def test_미리_셈_종단_체인과_1mm(pair):
+    preview, chain = pair["preview"], pair["chain"]
+    got = [float(s["chainage_m"]) for s in preview["stations"]]
+    want = [float(s["chainage_m"]) for s in chain["stations"]]
+    assert got == pytest.approx(want, abs=1e-6)
+    a = preview["profile_alignment"]["base_pvi"]
+    b = chain["profile_alignment"]["base_pvi"]
+    assert len(a) == len(b)
+    for p, q in zip(a, b):
+        assert abs(p["chainage_m"] - q["chainage_m"]) <= 1e-3
+        assert abs(p["elevation_m"] - q["elevation_m"]) <= 1e-3
+    assert ip.profile_summary(preview)["plan_m"] == pytest.approx(
+        ip.profile_summary(chain)["plan_m"], abs=1e-3
+    )
+
+
+def test_초기_종단_438_규칙(pair):
+    summary = ip.profile_summary(pair["preview"])
+    assert summary["curve_length_m"] == 20.0  # 432 L 20
+    assert summary["straightened"] > 0  # R > 500 꼭짓점 묶음 = 직선 위(438)
+    curves = pair["preview"]["profile_alignment"]["curves"]
+    assert not [c for c in curves if not c["omitted"] and c["r_m"] > 500.0 + 1e-6]
+    assert len(summary["line"]) > len(summary["stations_m"])  # 종단곡선 포함 선
+
+
+# ── 응답 칸 · 저장본 지문 ──────────────────────────────────────────────────────
+
+
+class _Pool:
+    def acquire(self):
+        pool = self
+
+        class _Acquire:
+            async def __aenter__(self):
+                return pool
+
+            async def __aexit__(self, *exc):
+                return False
+
+        return _Acquire()
+
+    def cursor(self, *args):
+        return self.acquire()
+
+
+def _async(value):
+    async def fake(*args, **kwargs):
+        return value() if callable(value) else value
+
+    return fake
+
+
+@pytest.fixture
+def gather(tmp_path, monkeypatch):
+    import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed
+    import B05_Profile.B05_Profile_Repository as b05_repo
+    import B05_Profile.B05_Profile_Router_Replan as replan
+    import B05_Profile.B05_Profile_Router_RouteInitial as route_initial
+    import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer
+    import B06_Section.B06_Section_Repository as b06_repo
+    import common_util.common_util_drainage_pipes as drainage
+    import common_util.common_util_workflow_state as workflow
+
+    root = P.stage_project(tmp_path / "project")
+    summary = P.read("summary.json")["longitudinal"]["data"]
+    grade = summary["grade_options"]
+    params = {
+        "options": {
+            "grade_class": grade.get("grade_class") or "trunk",
+            "terrain_type": grade.get("terrain_type") or "special",
+            "design_speed_kph": grade.get("design_speed_kph"),
+        },
+        "max_grade_pct": grade.get("max_grade_pct"),
+    }
+    state = {"stages": [{"stage_no": 2, "params": params}]}
+    pipes_file = root / "B04_PreProcess" / "drainage" / "edits" / "pipe_points.json"
+    monkeypatch.setattr(gatherer, "get_db_pool", lambda: _Pool())
+    monkeypatch.setattr(replan, "_plan_criteria", _async((RADIUS, 0.0, 0.0)))
+    monkeypatch.setattr(replan, "_road_settings", _async(("trunk", 20, "special")))
+    monkeypatch.setattr(b05_repo, "get_latest_route", _async({"id": 1, "route_data_path": P.ROUTE}))
+    monkeypatch.setattr(b06_repo, "get_route_generation_source", _async({"crs_epsg": 5186}))
+    monkeypatch.setattr(b06_repo, "get_latest_section_options", _async(summary["options"]))
+    monkeypatch.setattr(workflow, "get_workflow_state", _async(lambda: copy.deepcopy(state)))
+    monkeypatch.setattr(route_initial, "project_standard", _async(None))
+    monkeypatch.setattr(watershed, "_prepare", _async({"streams": []}))
+    monkeypatch.setattr(watershed, "_sheet_dir", lambda stored: root / "none")
+    monkeypatch.setattr(drainage, "pipe_points_path", lambda stored: pipes_file)
+    request = replan.RouteInitialRequest(mode="follow", with_profile=False)
+    results = root / "B05_Profile" / "route_recommend"
+    results.mkdir(parents=True, exist_ok=True)
+    (results / "k1.json").write_text("{}", "utf-8")  # 결과 저장본 짝(없으면 초기 종단 파일 지움)
+    candidate = _candidate(root)
+    selection = {"source_filter": "classification", "method": "dtm", "smooth": True}
+
+    def run(key: str = "k1") -> dict:
+        return asyncio.run(
+            gatherer.initial_profile_of(uuid4(), root, "p", candidate, request, selection, key)
+        )
+
+    return {"run": run, "root": root, "state": state, "pipes": pipes_file}
+
+
+def test_응답_칸(gather):
+    got = gather["run"]()
+    assert "error" not in got, got
+    for name in (
+        "stations_m",
+        "ground_m",
+        "plan_m",
+        "line",
+        "pvi_count",
+        "curve_count",
+        "straightened",
+        "max_grade_pct",
+        "curve_length_m",
+        "warnings",
+    ):
+        assert name in got, name
+    assert got["with_profile"] is False and got["curve_length_m"] == 20.0
+    assert len(got["stations_m"]) == len(got["plan_m"]) == len(got["ground_m"])
+
+
+def test_지문_같으면_꺼냄_다르면_다시(gather, monkeypatch):
+    first = gather["run"]()
+    assert "error" not in first, first
+
+    def broken(*args, **kwargs):
+        raise RuntimeError("다시 셈")
+
+    monkeypatch.setattr(ip, "initial_longitudinal", broken)
+    assert gather["run"]() == first  # 같은 입력 = 저장본
+    gather["state"]["stages"][0]["params"]["max_grade_pct"] = 9.0  # 페이지 설정 최대 기울기
+    assert "다시 셈" in gather["run"]()["error"]
+    gather["state"]["stages"][0]["params"].pop("max_grade_pct")
+    doc = json.loads(gather["pipes"].read_text("utf-8"))
+    doc["points"][0]["x"] = float(doc["points"][0]["x"]) + 1.0  # 관 하나 옮김
+    gather["pipes"].write_text(json.dumps(doc), "utf-8")
+    assert "다시 셈" in gather["run"]()["error"]
+
+
+def test_코드_바뀌면_다시(gather, monkeypatch):
+    assert "error" not in gather["run"]()
+    monkeypatch.setattr(ip, "source_fingerprint", lambda: "other")
+    monkeypatch.setattr(ip, "initial_longitudinal", lambda *a, **k: {})
+    assert "error" in gather["run"]()  # 438 등 소스 바뀜 = 다시 셈(빈 종단 = 실패 글)
+
+
+def test_추천_응답에_initial_profile(wired, monkeypatch):
+    import B05_Profile.B05_Profile_Router_RouteInitial_Initial as gatherer
+
+    seen: list = []
+
+    async def fake(project_id, root, stored, result, request, selection, key):
+        seen.append(key)
+        return {"pvi_count": 3}
+
+    monkeypatch.setattr(gatherer, "initial_profile_of", fake)
+    body = _ask(mode="earthwork")  # 224-3 가짜 세계
+    assert body["initial_profile"] == {"pvi_count": 3} and seen  # 옛 코드 = 칸 없음