feat(B04·B05 배수): 369-1 관 자리 기본 = 세류선 교차만 · auto 칸(all · add) · 노선 저장 오목 보충 끔

- build_detail auto — 없음 = 세류선 교차만 · all = 세류선 + 100 m 보충으로 다시 · add = 지금 목록 + 빈 구간 보충 · 후보는 늘 셈
- 배수유역 라우터 — 요청 auto 칸(틀리면 400) · 같은 자리(0.05 m) 관만 속성 · id · 사유 이어받음 · 나머지 새 관 = 등록부 기본값 · 저장본 열쇠에 auto
- 노선 [저장] · 추천 [적용] — 오목 지점 보충 지움(응답 concave 칸 · 화면 한 줄 같이)
- 시험 — test_369_drainage_auto 세 모드 · test_346_3 · 312 · 342_3 고쳐 씀

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VmC34fsxxbweEG3RUjs3Mn
This commit is contained in:
2026-10-17 11:54:30 +09:00
co-authored by Claude Opus 5.5
parent cd27e7d367
commit c52a98a115
15 changed files with 208 additions and 165 deletions
+1 -1
View File
@@ -105,7 +105,7 @@ async def _prepare_drainage_pipes_and_reprofile(
marks.append(("배수유역 분석", time.perf_counter()))
# 2) 기본 관(도로 × 상류 세류선) + 최대 간격 자동 보충을 정본으로 저장한다.
# 2) 기본 관(도로 × 상류 세류선 교차)만 정본으로 저장한다 — 간격 보충은 [자동 제안](369).
pipes = await put_pipe_points(project_id, None)
if isinstance(pipes, JSONResponse):
logger.error(
@@ -342,7 +342,7 @@ def build_flow_arrows(analysis: Any, flow: Any) -> list[tuple[float, float, floa
def _base_pipes(
vertices: list[RouteVertex], stream_features: list[dict[str, Any]]
) -> list[StructureCandidate]:
"""도로 × 세류선 교차점을 기본 관 위치로 삼는다. 300m 보충 배치는 다음 단계다."""
"""도로 × 세류선 교차점을 기본 관 위치로 삼는다. 간격 보충은 [자동 제안] 때만(369)."""
pipes: list[StructureCandidate] = []
for candidate in find_stream_crossings(vertices, stream_features):
if pipes and candidate.chainage_m - pipes[-1].chainage_m < DRAINAGE_PIPE_MIN_SPACING_M:
+45 -18
View File
@@ -2,7 +2,7 @@
격자 해석은 하지 않는다. 배수유역 분석이 저장해 둔 `03_road_routing` 산출물을 읽어
· 기본 관(도로 × 상류 세류선 교차점) + 최대 간격 자동 보충으로 관 목록을 만들고
· 기본 관(도로 × 상류 세류선 교차점)으로 관 목록을 만들고([자동 제안] = 최대 간격 보충 · 369)
· 계획노선 종단 Z로 노면 물이 어느 관으로 가는지 정해 세부유역을 나눈다
계산 알고리즘은 B05 사용자 화면과 공용이다(`common_util_drainage_detail`) — 같은 이름의
@@ -67,6 +67,8 @@ router = APIRouter(prefix="/api/projects", tags=["B04 Surface Drainage Basins"])
_NO_ANALYSIS = "배수유역 분석 결과가 없습니다. [유역 분석]을 먼저 실행하세요."
# 노선 지문이 분석 때와 다름(옛 저장분은 지문 없음) — 옛 격자로 나누면 틀린 유역이 나온다(342-3).
# [자동 제안] 방식(369) — None = 지금 목록 그대로(없으면 세류선 교차만) · all · add
AUTO_MODES = (None, "all", "add")
STALE_MESSAGE = (
"배수유역 낡음 — 노선이 분석 때와 다릅니다. 유역을 다시 분석해야 세부 유역을 나눕니다."
)
@@ -78,20 +80,22 @@ def _build(
points: list[PipePoint] | None,
routing: Any = None,
upstream_lonlat: list | None = None,
auto: str | None = None,
) -> tuple[DrainageDetail | None, list[PipePoint]]:
"""세부유역을 계산하고, 그 결과에 쓰인 관 목록을 함께 돌려준다.
`points`가 None 이면 B04 기본 관에 최대 간격 규칙으로 자동 보충한 목록이 만들어진다.
있으면 그 자리를 그대로 쓰되, 관이 왜 거기 있는지(기본/자동/수동)는 넘겨받은 목록의
표시를 유지한다 — 계산기는 확정 좌표만 보므로 그대로 두면 전부 "수동"이 된다.
`points`가 None 이면 B04 기본 관(세류선 교차)만. 있으면 그 자리를 그대로 쓰되, 관이 왜 거기
있는지(기본/자동/수동)는 넘겨받은 목록의 표시를 유지한다 — 계산기는 확정 좌표만 보므로 그대로
두면 전부 "수동"이 된다. `auto`([자동 제안] · 369) = "all" 자리 새로 · "add" 빈 구간 보충 —
같은 자리(0.05 m) 관만 속성 · id 를 이어받고 나머지는 새 관(등록부 기본값).
"""
directory = drainage_dir(stored_path)
chainages = [point.chainage_m for point in points] if points is not None else None
detail = build_detail(directory, context.vertices, chainages, routing, upstream_lonlat)
detail = build_detail(directory, context.vertices, chainages, routing, upstream_lonlat, auto)
if detail is None:
return None, []
if points is not None:
_retag(detail.pipes, points)
_retag(detail.pipes, points, keep_new=auto is not None)
rebuilt = _rebuilt(detail)
# 원본에 없던 자리 = 이번에 새로 놓인 관(자동 배치 · 간격 보충) — 등록부 기본값을 받을 관
fresh = [
@@ -132,11 +136,14 @@ _DETAIL_CACHE_SIZE = 4
DETAIL_CACHE_FILENAME = "04_detail_cache.pkl"
def _detail_key(context: DrainageContext, points: list[PipePoint] | None) -> str:
def _detail_key(
context: DrainageContext, points: list[PipePoint] | None, auto: str | None = None
) -> str:
directory = drainage_dir(context.stored_path)
digest = hashlib.sha1(usedforsecurity=False)
listed = None if points is None else [point.as_dict() for point in points]
digest.update(json.dumps(listed, sort_keys=True, ensure_ascii=False).encode())
# auto 를 열쇠에 — 같은 관 목록이라도 [자동 제안] 결과와 섞이지 않게(369 · 옛 저장본도 비껴감)
digest.update(json.dumps([listed, auto], sort_keys=True, ensure_ascii=False).encode())
for vertex in context.vertices:
digest.update(f"{vertex.x:.3f},{vertex.y:.3f},{vertex.z:.3f};".encode())
routing = STAGES["road_routing"]
@@ -167,10 +174,10 @@ def _remember(key: str, value: tuple[DrainageDetail, list[PipePoint]]) -> None:
def _build_cached(
context: DrainageContext, points: list[PipePoint] | None
context: DrainageContext, points: list[PipePoint] | None, auto: str | None = None
) -> tuple[DrainageDetail | None, list[PipePoint]]:
"""`_build` + 저장본 — 관 목록은 사본으로 돌려준다(저장이 좌표를 채워도 저장본은 그대로)."""
key = _detail_key(context, points)
key = _detail_key(context, points, auto)
hit = _DETAIL_CACHE.get(key)
if hit is None:
try:
@@ -180,7 +187,7 @@ def _build_cached(
except (OSError, pickle.PickleError, EOFError, AttributeError, KeyError, TypeError):
hit = None
if hit is None:
detail, carried = _build(context.stored_path, context, points)
detail, carried = _build(context.stored_path, context, points, auto=auto)
if detail is None:
return None, []
hit = (detail, carried)
@@ -276,8 +283,20 @@ def ford_height_m(flow_m3s: float | None, width_m: float) -> float | None:
return math.ceil(depth * 100 - 1e-9) / 100
def _retag(pipes: list[StructureCandidate], points: list[PipePoint]) -> None:
"""확정 좌표로 되돌아온 관에 원래의 생성 사유를 다시 붙인다."""
def _retag(
pipes: list[StructureCandidate], points: list[PipePoint], keep_new: bool = False
) -> None:
"""확정 좌표로 되돌아온 관에 원래의 생성 사유를 다시 붙인다.
`keep_new`([자동 제안]) = 같은 자리(0.05 m) 관만 · 새로 놓인 관은 계산기 사유
(stream · spacing) 그대로.
"""
if keep_new:
picks = carry_indices([pipe.chainage_m for pipe in pipes], [p.chainage_m for p in points])
for pipe, pick in zip(pipes, picks):
if pick is not None:
pipe.reason = points[pick].source
return
by_chainage = {round(point.chainage_m, 2): point.source for point in points}
for pipe in pipes:
source = by_chainage.get(round(pipe.chainage_m, 2))
@@ -452,11 +471,16 @@ async def _resolve(
돌려준다 — 관 목록도 없으면(자동 생성 자리) 409(342-3). 마지막 값 = 낡음.
빈 목록(`[]`)은 「관 0개」로 그대로 쓴다 — 자동 생성은 요청에 칸이 없고 저장분도 없을(None) 때만(312).
요청 `auto`([자동 제안] · 369) = "all" · "add" — `_build` 참고 · 없으면 None.
"""
context, reason = await load_drainage_context(project_id)
if context is None:
return JSONResponse(status_code=404, content={"status": "error", "message": reason})
auto = (payload or {}).get("auto")
if auto not in AUTO_MODES:
message = f"auto 는 {', '.join(m for m in AUTO_MODES if m)} 중 하나입니다."
return JSONResponse(status_code=400, content={"status": "error", "message": message})
try:
raw = (payload or {}).get("points")
requested = None if raw is None else parse_pipe_points(raw)
@@ -479,9 +503,11 @@ async def _resolve(
# [저장]은 정본으로 다시 셈 — 저장본은 GET · POST 만
if fresh:
result = await asyncio.to_thread(_build, context.stored_path, context, points)
result = await asyncio.to_thread(
lambda: _build(context.stored_path, context, points, auto=auto)
)
else:
result = await asyncio.to_thread(_build_cached, context, points)
result = await asyncio.to_thread(_build_cached, context, points, auto)
detail, resolved = result
if detail is None:
return JSONResponse(status_code=404, content={"status": "error", "message": _NO_ANALYSIS})
@@ -511,8 +537,9 @@ async def post_detail_basins(
) -> dict[str, Any] | JSONResponse:
"""화면에서 편집 중인 관 목록으로 세부유역을 다시 나눈다(저장하지 않는다).
`points` 칸 없이 부르면 저장분을 무시하고 기본 관 + 자동 보충으로 되돌린다 —
노선이 바뀌어 전체 분석을 다시 돌린 직후의 경로다. 빈 목록은 관 0개(312).
`points` 칸 없이 부르면 저장분을 무시하고 기본 관(세류선 교차)만으로 되돌린다 —
[배관 배치 초기화] · 노선이 바뀌어 전체 분석을 다시 돌린 직후의 경로다. 빈 목록은 관 0개(312).
`auto` = [자동 제안](369) — "all" 전체 자동 · "add" 지금 관 + 빈 구간 보충.
"""
resolved = await _resolve(project_id, payload, use_stored=False)
if isinstance(resolved, JSONResponse):
@@ -523,7 +550,7 @@ async def post_detail_basins(
@router.delete("/{project_id}/drainage/pipe-points", response_model=None)
async def delete_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
"""저장된 관 지점을 버리고 기본 관 + 자동 보충 배치로 되돌린다("초기화").
"""저장된 관 지점을 버리고 기본 관(세류선 교차) 배치로 되돌린다("초기화").
화면만 되돌리면 다시 들어왔을 때 옛 관이 살아난다 — 저장분과 파생 산출물까지 지운다
(2026-08-02 사용자 보고).
+2 -2
View File
@@ -129,8 +129,8 @@ export interface RouteReplanResponse {
route_id: number | null;
total_length_m: number | null;
vertex_count?: number;
/** 노선 저장 결과(346-3 · 서버 `B05_Profile_Route_Save` 머리) — 관 지킴 · 새 교차 · 오목 보충. */
pipes?: { kept: number; crossing: number; concave: number };
/** 노선 저장 결과(346-3 · 서버 `B05_Profile_Route_Save` 머리) — 관 지킴 · 새 교차. */
pipes?: { kept: number; crossing: number };
structures?: { moved: number; far: number };
/** 새 노선에 못 넘긴 횡단 고침(측점이 사라짐) 수. */
cross_dropped?: number;
+6 -39
View File
@@ -9,14 +9,14 @@ API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan
stationSnap, terrain}` · 오면 `vertices` · `station_snap` · `terrain` 대신 이것.
`as_initial` — 추천 노선 [적용](초기값 다시 뜸) · 손으로 고친 노선 저장은 false(초기값 그대로).
응답 200 `{status: "success", project_id, route_id, total_length_m, vertex_count, current_initial,
initial_refreshed, pipes: {kept, crossing, concave}, structures: {moved, far}, cross_dropped,
initial_refreshed, pipes: {kept, crossing}, structures: {moved, far}, cross_dropped,
drainage_stale: true, seconds}`.
오류 400(노드 둘 미만 · 초안 칸 길이 어긋남) · 404(프로젝트 없음) ·
409 `{status: "error", message}`(체인 실패 — 노선 · 관 · 구조물 · 종횡단 파일 · 새 노선 행을
옛것으로 되살린 뒤).
규칙:
- 관 정본을 지우지 않음 — 좌표를 따라감 · 새 세류선 교차점 기본 관 · 오목 지점 보충만 더함
- 관 정본을 지우지 않음 — 좌표를 따라감 · 새 세류선 교차점 기본 관만 더함
(`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`).
- 계획고 = 바뀐 구간만 새 지반 첫 계획선 · 그 밖 사용자 계획고 지킴
(`B05_Profile_Engine_Grade_Carry` · 바뀐 구간 = 옛 · 새 **계획 폴리라인** — 화면 planline 과
@@ -37,15 +37,12 @@ from pathlib import Path
from typing import Any, Literal
from uuid import UUID
import numpy as np
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from B05_Profile.B05_Profile_Route_Save_Parts import (
add_crossing_pipes,
carry_cross_inputs,
concave_chainages,
fill_concave_pipes,
follow_pipes,
station_keys,
vertices_of,
@@ -161,40 +158,14 @@ def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]:
return []
def _ground_along(project_root: Path, selection: dict[str, Any], vertices: list) -> tuple:
"""새 노선 지반 종단(2 m) — 오목 지점 찾기. 지표면을 못 열면 빈 배열(보충 건너뜀)."""
from shapely.geometry import LineString
from B05_Profile.B05_Profile_Route_Save_Parts import CONCAVE_STEP_M
from common_util.common_util_surface_sampler import build_surface_sampler
try:
sampler = build_surface_sampler(
project_root / "B04_PreProcess" / "models",
str(selection.get("source_filter")),
str(selection.get("method") or "dtm"),
bool(selection.get("smooth")),
)
except (FileNotFoundError, KeyError, OSError, ValueError):
logger.warning("노선 저장: 지표면을 못 열어 오목 지점 보충을 건너뜀")
return np.zeros(0), np.zeros(0)
line = LineString([(vertex.x, vertex.y) for vertex in vertices])
chainage = np.arange(0.0, line.length + 1e-9, CONCAVE_STEP_M)
xy = np.array([line.interpolate(float(s)).coords[0] for s in chainage], dtype=np.float64)
z, valid = sampler.sample_xy(xy)
return chainage[valid], np.asarray(z, dtype=np.float64)[valid]
async def _place_pipes(
project_id: UUID,
project_root: Path,
stored_path: str,
selection: dict[str, Any],
old_pipes: list,
old_xy: list[list[float]],
points: list[dict[str, float]],
) -> tuple[list, dict[str, int]]:
"""관 따라감 → 새 교차점 기본 관 → 오목 지점 보충 → 정본 저장(새 노선 지문)."""
"""관 따라감 → 새 교차점 기본 관 → 정본 저장(새 노선 지문) · 간격 보충 없음(369)."""
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes
from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare
from common_util.common_util_drainage_pipes import route_signature, save_pipe_points
@@ -207,8 +178,6 @@ async def _place_pipes(
pairs = follow_pipes(old_pipes, vertices)
total = float(vertices[-1].chainage_m)
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total)
chainage, ground = await asyncio.to_thread(_ground_along, project_root, selection, vertices)
concave = fill_concave_pipes(pairs, concave_chainages(chainage, ground), vertices, old_xy)
await asyncio.to_thread(
save_pipe_points,
stored_path,
@@ -216,7 +185,7 @@ async def _place_pipes(
[pipe for _, pipe in pairs],
vertices,
)
return pairs, {"kept": len(old_pipes), "crossing": crossing, "concave": concave}
return pairs, {"kept": len(old_pipes), "crossing": crossing}
async def _carry_profile(
@@ -423,10 +392,8 @@ async def run_route_save_chain(
new_xy = _xy(project_root, solved.route_data_path)
marks.append(("노선 풀기 · 구조물 따라감", time.perf_counter()))
pairs, pipes = await _place_pipes(
project_id, project_root, stored_path, selection, old_pipes, old_xy, points
)
marks.append(("관 따라감 · 교차점 · 오목 보충", time.perf_counter()))
pairs, pipes = await _place_pipes(project_id, stored_path, old_pipes, old_xy, points)
marks.append(("관 따라감 · 교차점", time.perf_counter()))
seed = asdict(_section_options(request, old_section_options))
seed.pop("extra_stations", None)
+3 -60
View File
@@ -1,4 +1,4 @@
"""노선 저장 길(346-3) 순수 부품 — 관 따라감 · 새 교차점 기본 관 · 오목 지점 보충 · 측점 열쇠 ·
"""노선 저장 길(346-3) 순수 부품 — 관 따라감 · 새 교차점 기본 관 · 측점 열쇠 ·
횡단 이월.
DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름). 차례 · 저장 · 되살림은
@@ -7,7 +7,7 @@ DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름
- 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음
(사용자 관 편집 지킴).
- 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음).
- 관 최대 간격 보충 = 옛 노선에서 벗어난 자리의 **오목 지점**(지반 종단 낮은 곳)에만 — 격자 없음.
- 관 최대 간격 보충 없음(369) — 간격 보충은 배수유역 [자동 제안]만.
- 측점 열쇠 = `common_util_station_key.station_key` 한 벌(sub2 짝) — 일반 `grid:k` ·
관 `pipe:<id>` · 구조물 `structure:<id>` · 시종점 `end:BP` · `end:EP` ·
id 없으면 `at:<누가거리 0.01 m>`.
@@ -22,11 +22,9 @@ import copy
import math
from typing import Any, Sequence
import numpy as np
from shapely.geometry import LineString, Point
from common_util.common_util_drainage_pipes import (
PIPE_SOURCE_SPACING,
PIPE_SOURCE_STREAM,
PipePoint,
project_pipe_points,
@@ -35,14 +33,10 @@ from common_util.common_util_route_geometry import RouteVertex
from common_util.common_util_station_key import chainage_key as station_chainage_key
from common_util.common_util_station_key import station_key
from common_util.common_util_structure_station import plan_stations, snap_structure_station
from config.config_system import DRAINAGE_PIPE_MAX_SPACING_M, DRAINAGE_PIPE_MIN_SPACING_M
from config.config_system import DRAINAGE_PIPE_MIN_SPACING_M
# 옛 노선에서 이만큼(m) 넘게 떨어진 자리 = 노선이 바뀐 자리(교차점 · 보충 대상).
MOVED_AWAY_M = 0.5
# 오목 지점을 찾을 지반 표본 간격(m) · 고름 창(표본 수) — DTM 잔물결을 오목으로 안 읽게.
# ponytail: 이동 평균 + 국소 최저 · 계곡이 얕아 놓치면 창을 줄이거나 흐름 강도와 섞을 것.
CONCAVE_STEP_M = 2.0
CONCAVE_WINDOW = 5
_SAME_M = 0.01
@@ -96,57 +90,6 @@ def add_crossing_pipes(
return added
def concave_chainages(chainage: np.ndarray, ground: np.ndarray) -> list[tuple[float, float]]:
"""지반 종단의 오목 지점(이동 평균 국소 최저) → [(누가거리, 고른 지반고)]."""
if ground.size < 3:
return []
kernel = np.ones(CONCAVE_WINDOW) / CONCAVE_WINDOW
pad = CONCAVE_WINDOW // 2
smooth = np.convolve(np.pad(ground, pad, mode="edge"), kernel, mode="valid")
return [
(float(chainage[i]), float(smooth[i]))
for i in range(1, smooth.size - 1)
if smooth[i] < smooth[i - 1] and smooth[i] <= smooth[i + 1]
]
def fill_concave_pipes(
pairs: list[tuple[int, PipePoint]],
concave: list[tuple[float, float]],
vertices: list[RouteVertex],
old_xy: Sequence[Sequence[float]] | None,
) -> int:
"""관 사이(시종점 포함)가 최대 간격을 넘으면 옛 노선 밖 오목 지점 중 가장 낮은 곳에 관."""
line = LineString([(vertex.x, vertex.y) for vertex in vertices])
old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
total = float(line.length)
added = 0
while True:
edges = [0.0, *(pipe.chainage_m for _, pipe in pairs), total]
best: tuple[float, float] | None = None
for start, end in zip(edges, edges[1:]):
if end - start <= DRAINAGE_PIPE_MAX_SPACING_M:
continue
for at, z in concave:
if not (
start + DRAINAGE_PIPE_MIN_SPACING_M < at < end - DRAINAGE_PIPE_MIN_SPACING_M
):
continue
point = line.interpolate(at)
if _moved_away(old_line, point.x, point.y) and (best is None or z < best[1]):
best = (at, z)
if best is None:
return added
at = snap_structure_station(best[0], [total])
point = line.interpolate(at)
pairs.append(
(-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_SPACING, x=point.x, y=point.y))
)
pairs.sort(key=lambda pair: pair[1].chainage_m)
concave = [item for item in concave if abs(item[0] - best[0]) > _SAME_M]
added += 1
def _facility_owners(
pipes: Sequence[PipePoint], structures: Sequence[Any]
) -> dict[float, tuple[str, str | None]]:
@@ -128,7 +128,6 @@ export function routeSaveSummary(saved: RouteReplanResponse, outside = 0): strin
`구조물 옮김 ${saved.structures?.moved ?? 0}`,
`멀어짐 ${saved.structures?.far ?? 0}`,
];
if (saved.pipes?.concave) parts.splice(2, 0, `오목 보충 관 ${saved.pipes.concave}`);
const dropped = (saved.cross_dropped ?? 0) + outside;
if (dropped) parts.push(`노선 밖 고침 ${dropped} 건`);
if (saved.drainage_stale) parts.push("배수유역 낡음");
+19 -8
View File
@@ -3,7 +3,8 @@
**격자 해석은 하지 않는다.** B04 배수유역 분석이 미리 돌려 저장한 결과를 읽어, 사용자가
실제로 손대는 두 가지만 처리한다(2026-07-31 사용자 지시).
⑨ 관 간격이 최대치를 넘는 구간에 **최소 개수**로 관을 보충
⑨ 관 간격이 최대치를 넘는 구간에 **최소 개수**로 관을 보충([자동 제안] 때만 · 기본 = 세류선
교차만)
⑩ 측구 흐름으로 도로 셀 → 담당 관을 정하고, 셀이 도달한 도로 셀의 담당 관을 그대로
그 셀의 유역 번호로 삼아 세부유역을 나눈다
⑪ 사용자가 관을 옮기거나 추가하면 ⑩만 다시 돈다 — 격자 해석은 재사용한다
@@ -167,21 +168,25 @@ def build_detail(
confirmed_chainages: list[float] | None = None,
routing: RoadRouting | None = None,
upstream_lonlat: list[list[list[float]]] | None = None,
auto: str | None = None,
) -> DrainageDetail | None:
"""B04 분석 결과를 읽어 관을 보충하고 세부유역을 나눈다.
"""B04 분석 결과를 읽어 관 자리를 정하고 세부유역을 나눈다.
`confirmed_chainages`를 주면 그 위치를 관으로 확정하고(사용자 편집 · 빈 목록 = 관 0개), None 이면
B04의 기본 관에 최대 간격 규칙으로 최소 개수만 보충한다. 어느 쪽이든 격자 해석은 하지 않는다.
B04의 기본 관(세류선 교차)만. 어느 쪽이든 격자 해석은 하지 않는다.
`auto`([자동 제안] · 369) — None = 위 그대로 · "all" = 기본 관 + 최대 간격 보충으로 다시
(확정 목록 무시) · "add" = 확정 목록 + 빈 구간 보충. 관 자리 후보는 어느 길이든 셈.
`routing` · `upstream_lonlat` 을 주면 파일 대신 그것(「지금 노선으로 셈」 메모리 결과 · 347-2).
"""
routing = routing or read_road_routing(directory)
if routing is None or len(vertices) < 2:
return None
candidates: list[StructureCandidate] = []
if confirmed_chainages is not None:
if confirmed_chainages is not None and auto != "all":
pipes = pipes_from_chainages(vertices, confirmed_chainages)
candidates = suggest_pipe_positions(vertices, pipes, routing.strength_curve)
if auto == "add":
pipes = [*pipes, *suggest_pipe_positions(vertices, pipes, routing.strength_curve)]
pipes.sort(key=lambda pipe: pipe.chainage_m)
else:
# 저장분의 기본 관은 누가거리만 신뢰한다 — 좌표는 현재 노선 위로 다시 찍는다.
base = [
@@ -194,7 +199,9 @@ def build_detail(
for pipe in routing.base_pipes
]
# 자동 관 자리 = 정수 m · 일반 측점 1 m 안이면 그 측점(234-1) — 확정본(위)은 단추로만
pipes = snap_candidates(place_pipes(vertices, base, routing.strength_curve), vertices)
placed = place_pipes(vertices, base, routing.strength_curve, fill=auto is not None)
pipes = snap_candidates(placed, vertices)
candidates = suggest_pipe_positions(vertices, pipes, routing.strength_curve)
detail = DrainageDetail(
route_lonlat=routing.route_lonlat,
@@ -386,10 +393,12 @@ def place_pipes(
vertices: list[RouteVertex],
base_pipes: list[StructureCandidate],
strength_curve: np.ndarray,
fill: bool = True,
) -> list[StructureCandidate]:
"""B04가 정한 기본 관(세류 교차점)에, 최대 간격을 넘는 구간만 최소 개수로 보충한다.
기본 관은 여기서 다시 찾지 않는다 — B04 산출물에 이미 들어 있다.
기본 관은 여기서 다시 찾지 않는다 — B04 산출물에 이미 들어 있다. `fill=False` = 보충 없이
기본 관만(최소 간격 겹침만 걸러 냄 · 369 기본).
"""
total_length = vertices[-1].chainage_m
base: list[StructureCandidate] = []
@@ -397,6 +406,8 @@ def place_pipes(
if base and candidate.chainage_m - base[-1].chainage_m < DRAINAGE_PIPE_MIN_SPACING_M:
continue
base.append(candidate)
if not fill:
return base
filled: list[StructureCandidate] = []
previous = 0.0
+1 -1
View File
@@ -56,7 +56,7 @@ const STUBS = {
if (failReplan) throw new Error("체인 실패");
return {
route_id: 9,
pipes: { kept: 3, crossing: 1, concave: 2 },
pipes: { kept: 3, crossing: 1 },
structures: { moved: 1, far: 0 },
cross_dropped: 1,
drainage_stale: true,
+1 -1
View File
@@ -148,7 +148,7 @@ def test_칸_없음_파일_없음_DELETE_노선_바뀜은_자동_배치(api) ->
def test_빈_목록과_None_을_가르는_자리() -> None:
src = _read("common_util/common_util_drainage_detail.py")
assert "if confirmed_chainages is not None:" in src
assert 'if confirmed_chainages is not None and auto != "all":' in src
router = _read("B04_PreProcess/B04_PreProcess_Router_Basins.py")
assert "points = requested if requested is not None else stored" in router
assert "requested or stored" not in router
+1 -1
View File
@@ -24,7 +24,7 @@ def setup(tmp_path, monkeypatch):
basins._DETAIL_CACHE.clear()
calls = SimpleNamespace(count=0)
def build(stored_path, context, points):
def build(stored_path, context, points, auto=None):
calls.count += 1
chainages = [p.chainage_m for p in points] if points else [50.0]
return DrainageDetail(), [PipePoint(chainage_m=c, source="stream") for c in chainages]
@@ -52,13 +52,13 @@ def _patch(monkeypatch, analysis_signature, stored):
monkeypatch.setattr(basins, "load_pipe_points", lambda *args: stored)
built = SimpleNamespace(called=False)
def build(stored_path, context, points):
def build(stored_path, context, points, auto=None):
built.called = True
return SimpleNamespace(pipes=[], basins=["유역"]), points or []
monkeypatch.setattr(basins, "_build", build)
monkeypatch.setattr(
basins, "_build_cached", lambda context, points: build("x", context, points)
basins, "_build_cached", lambda context, points, auto=None: build("x", context, points)
)
return built
+8 -29
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 오목 보충 · 이월 열쇠 ·
"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 보충 없음(369) · 이월 열쇠 ·
초기값 규칙 ·
배수유역 낡음 · 실패 때 옛 노선 되살림.
@@ -41,14 +41,11 @@ import common_util.common_util_workflow_state as workflow # noqa: E402
from B05_Profile.B05_Profile_Route_Save_Parts import ( # noqa: E402
add_crossing_pipes,
carry_cross_inputs,
concave_chainages,
fill_concave_pipes,
follow_pipes,
station_keys,
vertices_of,
)
from common_util.common_util_drainage_pipes import ( # noqa: E402
PIPE_SOURCE_SPACING,
PIPE_SOURCE_STREAM,
PipePoint,
analysis_route_signature,
@@ -60,7 +57,7 @@ OLD = [(0.0, 0.0), (100.0, 0.0), (200.0, 0.0), (300.0, 0.0), (400.0, 0.0)]
NEW = [(0.0, 0.0), (100.0, 0.0), (200.0, 60.0), (300.0, 0.0), (400.0, 0.0)]
# ── 관 따라감 · 교차점 · 오목 보충 ─────────────────────────────────────────────
# ── 관 따라감 · 교차점 ─────────────────────────────────────────────
def test_관은_지우지_않고_좌표를_따라감():
@@ -91,29 +88,11 @@ def test_새_교차점에만_기본_관():
assert add_crossing_pipes(pairs, crossings, OLD, 456.0) == 0
def test_오목_지점_보충은_새_자리의_오목만():
vertices = vertices_of(NEW)
total = vertices[-1].chainage_m
chainage = np.arange(0.0, total, 2.0)
ground = 100.0 + np.abs(chainage - 220.0) * 0.05 # 220 m 에 골(새 자리)
ground[chainage < 60] = (
100.0 + np.abs(chainage[chainage < 60] - 30.0) * 0.05
) # 30 m 골(옛 자리)
concave = concave_chainages(chainage, ground)
assert any(abs(at - 220.0) < 3 for at, _ in concave)
pairs: list = []
added = fill_concave_pipes(pairs, concave, vertices, OLD)
places = [pipe.chainage_m for _, pipe in pairs]
assert added >= 1 and all(pipe.source == PIPE_SOURCE_SPACING for _, pipe in pairs)
assert any(abs(at - 220.0) <= 3 for at in places)
assert not any(at < 60 for at in places) # 옛 노선 위 오목엔 안 둠
def test_노선_저장은_간격_보충_없음():
"""369 — 오목 · 100 m 간격 보충은 배수유역 [자동 제안]만 · 노선 저장 부품에 없음."""
import B05_Profile.B05_Profile_Route_Save_Parts as parts
def test_간격이_좁으면_보충_없음():
vertices = vertices_of(NEW)
pairs = [(0, PipePoint(chainage_m=at)) for at in (0.0, 90.0, 180.0, 270.0, 360.0, 450.0)]
concave = [(220.0, 99.0)]
assert fill_concave_pipes(pairs, concave, vertices, OLD) == 0
assert not hasattr(parts, "fill_concave_pipes") and not hasattr(save, "_ground_along")
# ── 측점 열쇠 · 횡단 이월 ──────────────────────────────────────────────────────
@@ -366,7 +345,6 @@ def chain_world(monkeypatch, tmp_path):
monkeypatch.setattr(watershed, "_prepare", prepare)
monkeypatch.setattr(drainage_pipes, "clear_pipe_points", watershed_must_not_run)
monkeypatch.setattr(drainage_pipes, "drainage_dir", lambda stored: drain)
monkeypatch.setattr(save, "_ground_along", lambda *a: (np.zeros(0), np.zeros(0)))
monkeypatch.setattr(b06_router, "regenerate_route_sections", regenerate)
monkeypatch.setattr(save, "_carry_profile", _async(None, order, "profile"))
monkeypatch.setattr(save, "_carry_cross", _async(2, order, "cross"))
@@ -416,7 +394,8 @@ def test_차례와_배수유역_뗌(chain_world):
"confirm",
]
assert result["cross_dropped"] == 2 and result["drainage_stale"] is True
assert result["pipes"] == {"kept": 1, "crossing": 0, "concave": 0}
# 456 m 노선 · 관 하나 — 100 m 간격 보충 없음(369)
assert result["pipes"] == {"kept": 1, "crossing": 0}
# 관 정본 = 새 노선 지문 · 배수유역 분석 지문(옛) 과 달라 B04 가 「낡음」
saved = json.loads(chain_world.pipes_file.read_text("utf-8"))
assert saved["route_signature"] == route_signature(vertices_of(NEW))
@@ -67,7 +67,7 @@ def test_저장_뒤_가게_초안_비움_결과_한_줄(out):
assert out["storeAfterSave"] is None
assert out["keptAfterSave"] == 1
assert out["summary"] == (
"노선 저장 — 관 지킴 3 · 새 교차 관 1 · 오목 보충 관 2 · 구조물 옮김 1 · 멀어짐 0 · "
"노선 저장 — 관 지킴 3 · 새 교차 관 1 · 구조물 옮김 1 · 멀어짐 0 · "
"노선 밖 고침 3 건 · 배수유역 낡음"
) # 서버 cross_dropped 1 + 화면 넘김 밖 2 = 한 줄
+117
View File
@@ -0,0 +1,117 @@
"""369 배수 구조물 자리 — 기본 = 세류선 교차만 · [자동 제안] 두 방식(all · add).
- 기본(auto 없음) = 세류선 교차 관만 · 100 m 보충 없음 · 관 자리 후보는 셈.
- all = 세류선 + 100 m 보충으로 다시(지금 목록 무시) · 같은 자리(0.05 m) 관은 속성 · id 유지 ·
나머지 = 새 관(등록부 기본값).
- add = 지금 목록 그대로 + 빈 구간 보충 · 더한 관 = 새 관(등록부 기본값).
- 틀린 auto = 400 · 저장본 열쇠에 auto.
"""
import asyncio
from pathlib import Path
from uuid import uuid4
import numpy as np
from B04_PreProcess import B04_PreProcess_Router_Basins as basins
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import GridSpec
from common_util.common_util_drainage_context import DrainageContext
from common_util.common_util_drainage_detail import RoadRouting, build_detail
from common_util.common_util_drainage_pipes import PipePoint
from common_util.common_util_route_geometry import RouteVertex, StructureCandidate
VERTICES = [RouteVertex(0.0, 0.0, 100.0, 0.0), RouteVertex(300.0, 0.0, 90.0, 300.0)]
CONTEXT = DrainageContext(stored_path="x", project_root=Path("."), vertices=VERTICES)
def _routing() -> RoadRouting:
spec = GridSpec(x_min=-5.0, y_max=5.0, cell_m=1.0, n_rows=10, n_cols=310)
road_index = np.arange(5 * 310 + 5, 5 * 310 + 305, dtype=np.int32)
road_slot = np.full(spec.size, -1, dtype=np.int32)
road_slot[road_index] = np.arange(road_index.size)
return RoadRouting(
spec=spec,
road_slot=road_slot,
path_length=np.zeros(spec.size, np.float32),
elevation=np.full(spec.size, 95.0, np.float32),
road_cell_index=road_index,
road_chainage=np.arange(road_index.size, dtype=np.float64),
strength=np.ones(road_index.size, np.int64),
base_pipes=[StructureCandidate(chainage_m=150.0, x=150.0, y=0.0, reason="stream")],
)
def _detail(tmp_path, chainages=None, auto=None):
return build_detail(tmp_path / "없음", VERTICES, chainages, _routing(), [], auto)
def test_기본은_세류선_교차만(tmp_path):
detail = _detail(tmp_path)
assert [(p.chainage_m, p.reason) for p in detail.pipes] == [(150.0, "stream")]
assert len(detail.pipe_candidates) == 2 # 0~150 · 150~300 이 100 m 넘음 — 제안만
def test_all_은_세류선_더하기_100m_보충_지금_목록_무시(tmp_path):
detail = _detail(tmp_path, [10.0, 150.02], "all")
reasons = [p.reason for p in detail.pipes]
places = [p.chainage_m for p in detail.pipes]
assert reasons.count("stream") == 1 and reasons.count("spacing") == 2
assert 150.0 in places and 10.0 not in places
gaps = np.diff([0.0, *places, 300.0])
assert gaps.max() <= 100.0 and not detail.pipe_candidates
def test_add_는_지금_목록_더하기_빈_구간_보충(tmp_path):
detail = _detail(tmp_path, [20.0], "add")
places = [p.chainage_m for p in detail.pipes]
assert 20.0 in places and len(places) == 3 # 20 ~ 300 = 280 m → 보충 둘
assert np.diff([0.0, *places, 300.0]).max() <= 100.0
# auto 없음 = 목록 그대로(보충은 후보만)
plain = _detail(tmp_path, [20.0])
assert [p.chainage_m for p in plain.pipes] == [20.0] and len(plain.pipe_candidates) == 2
def _build(tmp_path, monkeypatch, points, auto):
monkeypatch.setattr(basins, "drainage_dir", lambda stored: tmp_path / "없음")
return basins._build("x", CONTEXT, points, _routing(), [], auto=auto)
def test_all_같은_자리_관은_속성_id_유지_새_관은_기본값(tmp_path, monkeypatch):
kept = PipePoint(
chainage_m=150.03,
source="user",
facility="ford_bridge",
options={"ford_width_m": 4.0},
id="keep",
)
gone = PipePoint(chainage_m=10.0, source="user", options={"pipe_diameter_mm": 600}, id="gone")
_, carried = _build(tmp_path, monkeypatch, [gone, kept], "all")
ford = [p for p in carried if p.id == "keep"]
assert len(ford) == 1 and ford[0].facility == "ford_bridge" and ford[0].source == "user"
assert all(p.id != "gone" for p in carried)
fresh = [p for p in carried if p.id != "keep"]
assert len(fresh) == 2 and all(p.source == "spacing" for p in fresh)
assert all((p.options or {}).get("pipe_diameter_mm") for p in fresh) # 등록부 관경
def test_add_더한_관은_기본값_지금_관은_그대로(tmp_path, monkeypatch):
mine = PipePoint(chainage_m=20.0, source="user", id="mine")
_, carried = _build(tmp_path, monkeypatch, [mine], "add")
assert [p.id for p in carried].count("mine") == 1
assert next(p for p in carried if p.id == "mine").options is None # 옛 빈칸 = 미확정 유지
added = [p for p in carried if p.id != "mine"]
assert len(added) == 2 and all(p.source == "spacing" for p in added)
assert all((p.options or {}).get("pipe_diameter_mm") for p in added)
def test_틀린_auto_는_400_열쇠에_auto(tmp_path, monkeypatch):
async def load(project_id):
return CONTEXT, ""
monkeypatch.setattr(basins, "load_drainage_context", load)
monkeypatch.setattr(basins, "drainage_dir", lambda stored: tmp_path)
result = asyncio.run(basins._resolve(uuid4(), {"auto": "spacing"}, use_stored=False))
assert result.status_code == 400
points = [PipePoint(chainage_m=20.0)]
keys = {basins._detail_key(CONTEXT, points, auto) for auto in basins.AUTO_MODES}
assert len(keys) == 3