From 8bf24c268f5ab54acc9caf65c44af874d33ee759 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Mon, 14 Sep 2026 22:43:10 +0900 Subject: [PATCH] =?UTF-8?q?fix(b06):=20=EB=AF=B8=EA=B5=90=EC=B0=A8=20?= =?UTF-8?q?=EC=B8=A1=EC=A0=90=20=EC=A7=80=EB=B0=98=20=EC=83=98=ED=94=8C?= =?UTF-8?q?=EC=9D=84=20=EC=A7=80=ED=91=9C=EB=A9=B4=20=EB=81=9D=EA=B9=8C?= =?UTF-8?q?=EC=A7=80=20=EB=84=93=ED=9E=98(=E3=89=B3=20(=EA=B0=80)=20=C2=B7?= =?UTF-8?q?=20=EB=B8=8C=EB=A0=88=EC=9D=B8=20=ED=8C=90=EC=A0=95=20=E3=80=8C?= =?UTF-8?q?+5m=20=EC=94=A9=20=C2=B7=20=EC=83=81=ED=95=9C=20=3D=20=EC=A7=80?= =?UTF-8?q?=ED=91=9C=EB=A9=B4=20=EB=81=9D=E3=80=8D=20=C2=B7=20716=20?= =?UTF-8?q?=E3=80=8C+5m=20=ED=95=9C=20=EB=B2=88=E3=80=8D=20=EA=B0=88?= =?UTF-8?q?=EC=9D=8C)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 서버 재계산([저장]·[확정]·자동설계 체인) 맨 앞에서 미교차 측점만 열린 쪽을 +5m 씩 확정 지표면에서 다시 떠 계산 · 닫히면 멈춤 · 지표면 밖 샘플이 섞이면 그 걸음은 안 붙이고 미교차로 남김 - 넓힌 샘플은 횡단 파일에 남겨 화면·Node·B08 이 같은 지반을 봄 · 계산식은 안 바뀜(입력만 넓어짐) · 뒤 보정(포장·세월교·사토장)은 넓힌 지반 위에서 - 종전엔 반폭 20m 끝에서 잘려 성토가 작게 섰음 - 936be972 서버 재계산 전→후: 미교차 12 → 1(840 지표면 끝) · 성토 단면 460 19.87→35.11 · 480 48.04→84.35 · 500 37.15→52.47 · 700 24.74→54.84 · 1000 56.30→80.40 · 840 41.60→110.35 · 258.12 25.11→31.44 · 유토곡선 성토 13,350.64 → 17,248.52㎥ · 토취 7,259.84 → 11,100.86㎥ · 내역 본체 129,498,518 → 154,857,626원 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_016VBGFXB9AbJBwXP19z75Qq --- .../B06_Section_Engine_SampleExtend.py | 190 ++++++++++++++++++ .../B06_Section_Server_Calc_Prebuild.py | 34 +++- resources/tester/test_b06_sample_extend.py | 83 ++++++++ 3 files changed, 305 insertions(+), 2 deletions(-) create mode 100644 B06_Section/B06_Section_Engine_SampleExtend.py create mode 100644 resources/tester/test_b06_sample_extend.py diff --git a/B06_Section/B06_Section_Engine_SampleExtend.py b/B06_Section/B06_Section_Engine_SampleExtend.py new file mode 100644 index 00000000..4e1d5d84 --- /dev/null +++ b/B06_Section/B06_Section_Engine_SampleExtend.py @@ -0,0 +1,190 @@ +"""미교차 측점의 지반 샘플을 **지표면 끝까지** 넓힘 — ㉳ (가) (2026-09-14 브레인 판정). + +반폭(기본 20m) 샘플 끝에서 사면이 원지반과 안 만나면 면적이 거기서 잘려 성토가 **작게** 섰다 +(936be972 미교차 12곳 중 11곳이 넓히면 닫힘 · 성토 +2,559.9㎥). ⇒ **열린 쪽만** +5m 씩 확정 지표면에서 +다시 떠 계산하고, 닫히면 멈춘다. 새 걸음에 지표면 밖(무효) 샘플이 섞이면 그 걸음은 안 붙이고 멈춘다 — +그 측점은 「지표면 끝까지 넓혀도 안 만남」으로 미교차가 남는다. +⚠ 상한 숫자를 두지 않는다 — 지표면이 실제로 있는 끝이라 근거가 필요 없음(인위 기본값 금지). +⚠ 716 판정 「미교차 측점만 반폭 +5m 한 번」을 갈음(그것으론 4곳만 닫혔음). +⚠ 계산식은 안 바뀐다 — 샘플(입력)만 넓어진다. 넓힌 샘플은 횡단 파일에 남겨 화면(TS)·Node·B08 이 + 같은 지반을 본다(한 벌). +""" + +from __future__ import annotations + +import json +import logging +from pathlib import Path +from typing import Any, Callable + +import numpy as np + +from B05_Profile.B05_Profile_Engine_Sections import cross_filename +from B06_Section.B06_Section_Engine_Design import ( + _SLOPE_CLOSE_TOLERANCE_M, + compute_cross_design, + curve_widening_args, +) +from common_util.common_util_json import atomic_write_json + +logger = logging.getLogger(__name__) + +#: 한 걸음 — 브레인 판정 「+5m 씩」. +EXTEND_STEP_M = 5.0 +#: 지표면 끝까지 넓혀도 안 닫힌 측점의 사유(경고가 그대로 씀). +SURFACE_END_REASON = "지표면 끝까지 넓혀도 성토 비탈이 원지반과 안 만남" + + +def _valid_sorted(samples: list[dict[str, Any]]) -> list[tuple[float, float]]: + return sorted( + (float(s["offset_m"]), float(s["elevation_m"])) + for s in samples + if s.get("valid") is not False and s.get("elevation_m") is not None + ) + + +def _open_ends(samples: list[dict[str, Any]], design: dict[str, Any]) -> tuple[bool, bool]: + """(우 끝 열림, 좌 끝 열림) — 샘플 끝의 지반과 설계선 높이차가 허용오차를 넘나.""" + ground = _valid_sorted(samples) + line = sorted( + (float(p["offset_m"]), float(p["elevation_m"])) for p in design.get("design_line") or [] + ) + if not ground or not line: + return False, False + right = abs(ground[0][1] - line[0][1]) > _SLOPE_CLOSE_TOLERANCE_M + left = abs(ground[-1][1] - line[-1][1]) > _SLOPE_CLOSE_TOLERANCE_M + return right, left + + +def extend_unclosed( + samples: list[dict[str, Any]], + frame: dict[str, Any], + compute: Callable[[list[dict[str, Any]]], dict[str, Any]], + sampler: Any, + step_m: float = EXTEND_STEP_M, +) -> tuple[list[dict[str, Any]], dict[str, Any]] | None: + """넓힌 샘플과 `{left_m, right_m, surface_end}` — 이미 닫혔으면 `None`.""" + design = compute(samples) + if not design.get("slope_unclosed"): + return None + offsets = [offset for offset, _z in _valid_sorted(samples)] + spacing = min(b - a for a, b in zip(offsets, offsets[1:]) if b - a > 1e-9) + count = max(int(round(step_m / spacing)), 1) + origin = np.array([float(frame["origin"]["x"]), float(frame["origin"]["y"])]) + left_axis = np.array([float(v) for v in frame["left_xy"]]) + info: dict[str, Any] = {"left_m": 0.0, "right_m": 0.0, "surface_end": False} + samples = list(samples) + while design.get("slope_unclosed"): + right_open, left_open = _open_ends(samples, design) + added: list[dict[str, Any]] = [] + for side, is_open, sign in (("right", right_open, -1.0), ("left", left_open, 1.0)): + if not is_open: + continue + edge = min(offsets) if sign < 0 else max(offsets) + new_offsets = np.array([edge + sign * spacing * k for k in range(1, count + 1)]) + xy = origin[None, :] + left_axis[None, :] * new_offsets[:, None] + z, valid = sampler.sample_xy(xy) + if not np.all(valid): + info["surface_end"] = True + continue + added.extend( + { + "offset_m": round(float(o), 6), + "x": round(float(p[0]), 6), + "y": round(float(p[1]), 6), + "z": round(float(e), 6), + "elevation_m": round(float(e), 6), + "valid": True, + } + for o, p, e in zip(new_offsets, xy, z) + ) + info[f"{side}_m"] += step_m + if not added: + break + samples = sorted(samples + added, key=lambda s: float(s["offset_m"])) + offsets = [offset for offset, _z in _valid_sorted(samples)] + design = compute(samples) + return samples, info + + +def extend_unclosed_sections( + longitudinal: dict[str, Any], + cross_sections: list[dict[str, Any]], + project_root: Path, + standard: dict[str, Any] | None, + sampler: Any, +) -> int: + """저장 설계가 미교차인 측점을 넓혀 샘플·설계를 자리에서 갈고 횡단 파일에 남긴다. 넓힌 수.""" + from B06_Section.B06_Section_Router_Design import ( + USER_TOUCHED_KEYS, + ford_drop_at, + ford_surface_drops, + stored_berm, + stored_cut_slope, + ) + from common_util.common_util_route_profile import design_elevation_from_longitudinal + from config.config_system import STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M + + if sampler is None: + return 0 + drops = ford_surface_drops(project_root) + cross_dir = project_root / "B06_Section" / "cross_sections" + changed = 0 + for section in cross_sections: + design = section.get("design") + if not isinstance(design, dict) or not design.get("slope_unclosed"): + continue + chainage = float(section.get("chainage_m", 0.0)) + + def compute(samples: list[dict[str, Any]], design=design, chainage=chainage) -> dict: + return compute_cross_design( + samples, + design_elevation_from_longitudinal(longitudinal, chainage), + ground_type=str(design.get("ground_type") or "ripping_rock"), + section_mode=str(design.get("section_mode") or "left_cut"), + ditch_side=design.get("ditch_side"), + ditch_type=str(design.get("ditch_type") or "standard"), + paved=bool(design.get("paved", False)), + standard=standard, + rock_boundary_offset_m=design.get( + "rock_boundary_offset_m", STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M + ), + two_stage_slope=bool(design.get("two_stage_slope", True)), + cut_slope_ratio=stored_cut_slope(design), + ditch_enabled=design.get("ditch_enabled"), + ditch_choice=design.get("ditch_choice"), + surface_drop_m=ford_drop_at(chainage, drops), + berm=stored_berm(design), + **curve_widening_args(section), + ) + + try: + result = extend_unclosed( + section.get("samples") or [], section["frame"], compute, sampler + ) + except (ValueError, KeyError): + continue + if result is None: + continue + samples, info = result + if not info["left_m"] and not info["right_m"]: + continue # 첫 걸음부터 지표면 밖 — 샘플이 그대로라 갈 것이 없음 + recomputed = compute(samples) + for key in ("status", "pavement_suggested", *USER_TOUCHED_KEYS): + if design.get(key) is not None: + recomputed[key] = design[key] + section["samples"] = samples + section["design"] = recomputed + cross_path = cross_dir / cross_filename(chainage) + if cross_path.is_file(): + stored = json.loads(cross_path.read_text(encoding="utf-8")) + atomic_write_json(cross_path, {**stored, "samples": samples}) + logger.info( + "미교차 샘플 넓힘: 측점 %.3f 좌 +%sm 우 +%sm 지표면 끝 %s", + chainage, + info["left_m"], + info["right_m"], + info["surface_end"], + ) + changed += 1 + return changed diff --git a/B06_Section/B06_Section_Server_Calc_Prebuild.py b/B06_Section/B06_Section_Server_Calc_Prebuild.py index be311607..545dd448 100644 --- a/B06_Section/B06_Section_Server_Calc_Prebuild.py +++ b/B06_Section/B06_Section_Server_Calc_Prebuild.py @@ -44,6 +44,7 @@ from common_util.common_util_project_settings import ( quantity_settings, ) from common_util.common_util_storage import resolve_stored_project_path +from common_util.common_util_surface_confirmation import get_surface_confirmation_params from config.config_db import get_db_pool, run_with_connection from config.config_system import ( EARTHWORK_CONVERSION_FACTORS, @@ -154,23 +155,43 @@ def _mass_haul_context( } +def _surface_sampler(project_root: Path, params: dict[str, Any] | None) -> Any: + """확정 지표면 표고 조회기 — 못 열면 `None`(넓힘을 건너뛰고 종전대로 · 비치명).""" + from common_util.common_util_surface_sampler import build_surface_sampler + + try: + return build_surface_sampler( + project_root / "B04_PreProcess" / "models", + str((params or {})["source_filter"]), + str((params or {})["method"]), + bool((params or {})["smooth"]), + ) + except (FileNotFoundError, KeyError, OSError, ValueError) as exc: + logger.warning("서버 재계산: 확정 지표면을 못 열어 미교차 넓힘을 건너뜀 — %s", exc) + return None + + def _enforce_stored_designs( longitudinal: dict[str, Any], sections: list[dict[str, Any]], project_root: Path, standard: dict[str, Any] | None, + sampler: Any = None, ) -> None: """저장분 설계를 **쓰는 시점에** 바로잡는다 — 포장 구간·세월교 노면 하강. 예전에는 상세를 **읽을 때마다** 돌려 화면이 볼 때만 맞았다(저장분은 낡은 채로). 2026-09-06 사용자 확정대로 「읽기는 영구저장소에서 가져오기만」이므로 이쪽으로 옮겼다. """ + from B06_Section.B06_Section_Engine_SampleExtend import extend_unclosed_sections from B06_Section.B06_Section_Engine_SpoilFill import enforce_spoil_fills from B06_Section.B06_Section_Router_Design import ( enforce_ford_surface_drops, enforce_pavement_ranges, ) + # ⚠ 미교차 샘플 넓힘이 **맨 앞**이다 — 뒤의 보정들이 넓힌 지반 위에서 다시 계산해야 한다(㉳ (가)). + extend_unclosed_sections(longitudinal, sections, project_root, standard, sampler) enforce_pavement_ranges(longitudinal, sections, project_root, standard) enforce_ford_surface_drops(longitudinal, sections, project_root, standard) # ⚠ 사토장은 **맨 뒤**다 — 앞의 두 보정이 설계를 다시 계산하면서 사토장 칸을 지운다. @@ -198,11 +219,12 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int: pool = get_db_pool() # 구조물 몫(채집석 공제·구조물 잔토)도 함께 받아 온다 — **B08 이 낸 값**이고, 안 넘기면 # 통로만 있고 값이 안 흐른다(2026-09-09 실측: 공제가 늘 `None` 이라 사토가 안 줄었다). - response, stored_path, longitudinal_row, haul_inputs = await asyncio.gather( + response, stored_path, longitudinal_row, haul_inputs, surface = await asyncio.gather( get_section_detail(project_uuid, route_id), run_with_connection(get_project_storage_relative_path, project_uuid), run_with_connection(get_longitudinal_section, project_uuid, route_id), haul_inputs_for(project_uuid), + run_with_connection(get_surface_confirmation_params, str(project_uuid)), ) marks.append(("종횡단 상세+DB 조회(병렬)", time.perf_counter())) payload = getattr(response, "model_dump", None) @@ -217,8 +239,16 @@ async def _recompute(project_id: UUID | str, route_id: int) -> int: standard = stored_standard_cross_section(longitudinal_row) # 포장 구간·세월교 보정 — 고쳐진 설계 위에서 면적·유토곡선이 나와야 한다. before = [json.dumps(item.get("design"), sort_keys=True, default=str) for item in sections] + # 확정 지표면 — 미교차 측점이 있을 때만 연다(샘플 넓힘 ㉳ (가) · 없으면 여는 비용을 안 씀). + unclosed = any((item.get("design") or {}).get("slope_unclosed") for item in sections) + sampler = await asyncio.to_thread(_surface_sampler, project_root, surface) if unclosed else None await asyncio.to_thread( - _enforce_stored_designs, detail.get("longitudinal") or {}, sections, project_root, standard + _enforce_stored_designs, + detail.get("longitudinal") or {}, + sections, + project_root, + standard, + sampler, ) fixed = [ item diff --git a/resources/tester/test_b06_sample_extend.py b/resources/tester/test_b06_sample_extend.py new file mode 100644 index 00000000..a7574b09 --- /dev/null +++ b/resources/tester/test_b06_sample_extend.py @@ -0,0 +1,83 @@ +"""㉳ (가) 미교차 측점 지반 샘플을 지표면 끝까지 넓힘 — 2026-09-14 브레인 판정. + +반폭 20m 끝에서 성토 비탈이 원지반과 안 만나면 면적이 잘려 성토가 작게 섰음(936be972 12곳 중 11곳). +⇒ 열린 쪽만 +5m 씩 확정 지표면에서 다시 떠 계산 · 닫히면 멈춤 · 지표면 밖(무효 샘플)이면 멈추고 미교차 그대로. +상한 숫자 없음(지표면이 실제로 있는 끝) · 716 「미교차만 +5m 한 번」 갈음. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pytest + +ROOT = Path(__file__).resolve().parents[2] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from B06_Section.B06_Section_Engine_Design import compute_cross_design # noqa: E402 +from B06_Section.B06_Section_Engine_SampleExtend import extend_unclosed # noqa: E402 +from common_util.common_util_surface_sampler import CallableSurfaceSampler # noqa: E402 + +FRAME = {"origin": {"x": 0.0, "y": 0.0}, "left_xy": [1.0, 0.0]} + + +def _ground(x: np.ndarray) -> np.ndarray: + """좌(+)는 30m 까지 1:1 로 떨어지다 완만 · 우(−)는 오르막(절토가 금방 닫힘).""" + x = np.asarray(x, dtype=float) + left = np.where(x < 30.0, 100.0 - x, 70.0 - 0.1 * (x - 30.0)) + return np.where(x >= 0, left, 100.0 - 0.8 * x) + + +def _samples(half: float = 20.0) -> list[dict]: + offsets = np.arange(-half, half + 0.25, 0.5) + return [ + {"offset_m": float(o), "elevation_m": float(z), "valid": True} + for o, z in zip(offsets, _ground(offsets)) + ] + + +def _compute(samples: list[dict]) -> dict: + return compute_cross_design(samples, 100.0, ground_type="soil", section_mode="right_cut") + + +def _sampler(limit: float | None = None) -> CallableSurfaceSampler: + def function(xy: np.ndarray) -> np.ndarray: + values = _ground(xy[:, 0]) + if limit is not None: + values = np.where(xy[:, 0] > limit, np.nan, values) + return values + + return CallableSurfaceSampler(function) + + +def test_열린_쪽만_5m_씩_넓혀_닫히면_멈춤() -> None: + before = _compute(_samples()) + assert before["slope_unclosed"], "기준 단면이 20m 에서 안 닫혀야 시험이 뜻이 있음" + samples, info = extend_unclosed(_samples(), FRAME, _compute, _sampler()) + after = _compute(samples) + assert not after["slope_unclosed"] + assert info["left_m"] > 0 and info["left_m"] % 5 == 0 and info["right_m"] == 0 + assert info["surface_end"] is False + assert max(s["offset_m"] for s in samples) == pytest.approx(20.0 + info["left_m"]) + assert min(s["offset_m"] for s in samples) == pytest.approx(-20.0) # 닫힌 쪽은 안 넓힘 + assert after["fill_area_m2"] > before["fill_area_m2"] # 잘린 성토가 제 크기로 + + +def test_지표면_끝이면_멈추고_미교차_그대로() -> None: + samples, info = extend_unclosed(_samples(), FRAME, _compute, _sampler(limit=32.0)) + assert info["surface_end"] is True + assert max(s["offset_m"] for s in samples) == pytest.approx(30.0) # 무효가 섞인 걸음은 안 붙임 + assert _compute(samples)["slope_unclosed"] + + +def test_닫힌_측점은_그대로() -> None: + closed = _samples(60.0) + assert extend_unclosed(closed, FRAME, _compute, _sampler()) is None + + +def test_서버_재계산이_부름() -> None: + source = (ROOT / "B06_Section" / "B06_Section_Server_Calc_Prebuild.py").read_text("utf-8") + assert "extend_unclosed_sections(" in source