refactor(510-2 서버 자동 관 옛 길 걷음): auto all · add 갈래 걷음(부르는 곳 0 · detail-basins · drainage/live · build_detail · 저장본 열쇠 · computeDetailBasins 셋째 인자) · 전처리 첫 자동 관 그대로 · test_369 옛 all/add 시험 걷음 · test_b05_structures_router 이관 5 = 가짜 함수를 (legacy, standing) 쌍으로 맞춤

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-22 13:14:13 +09:00
co-authored by Claude Sonnet 5.5
parent 59c2907b73
commit 7b72dffd60
9 changed files with 39 additions and 144 deletions
+2 -4
View File
@@ -624,16 +624,14 @@ export async function fetchDetailPipePoints(projectId: string): Promise<DetailBa
/** 편집 중인 관 목록으로 세부유역을 다시 나눈다(저장하지 않는다).
*
* 빈 목록 = 관 0개(312) · null = 세류선 교차 관만(369) · `auto`([자동 제안]) = "all" 전체 자동
* (같은 자리 관은 속성 유지) · "add" 지금 관 + 빈 구간 보충. */
* 빈 목록 = 관 0개(312) · null = 세류선 교차 관만(369). */
export async function computeDetailBasins(
projectId: string,
points: DetailPipeInput[] | null,
auto?: "all" | "add",
): Promise<DetailBasinResponse> {
return requestJson<DetailBasinResponse>(`/projects/${projectId}/drainage/detail-basins`, {
method: "POST",
body: JSON.stringify(auto ? { points, auto } : { points }),
body: JSON.stringify({ points }),
});
}
+13 -40
View File
@@ -73,8 +73,6 @@ router = APIRouter(prefix="/api/projects", tags=["B04 Surface Drainage Basins"])
_NO_ANALYSIS = "배수유역 분석 결과가 없습니다. [유역 분석]을 먼저 실행하세요."
# 노선 지문이 분석 때와 다름(옛 저장분은 지문 없음) — 옛 격자로 나누면 틀린 유역이 나온다(342-3).
# 머리 「배수유역 낡음」 = 화면 `STALE_PREFIX`(B05 Drainage_Live · 369) 짝.
# [자동 제안] 방식(369) — None = 지금 목록 그대로(없으면 세류선 교차만) · all · add
AUTO_MODES = (None, "all", "add")
STALE_MESSAGE = (
"배수유역 낡음 — 노선이 분석 때와 다릅니다. 유역을 다시 분석해야 세부 유역을 나눕니다."
)
@@ -86,22 +84,20 @@ 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 기본 관(세류선 교차)만. 있으면 그 자리를 그대로 쓰되, 관이 왜 거기
있는지(기본/자동/수동)는 넘겨받은 목록의 표시를 유지한다 — 계산기는 확정 좌표만 보므로 그대로
두면 전부 "수동"이 된다. `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, auto)
detail = build_detail(directory, context.vertices, chainages, routing, upstream_lonlat)
if detail is None:
return None, []
if points is not None:
_retag(detail.pipes, points, keep_new=auto is not None)
_retag(detail.pipes, points)
rebuilt = _rebuilt(detail)
# 원본에 없던 자리 = 이번에 새로 놓인 관(자동 배치 · 간격 보충) — 등록부 기본값을 받을 관
fresh = [
@@ -142,14 +138,11 @@ _DETAIL_CACHE_SIZE = 4
DETAIL_CACHE_FILENAME = "04_detail_cache.pkl"
def _detail_key(
context: DrainageContext, points: list[PipePoint] | None, auto: str | None = None
) -> str:
def _detail_key(context: DrainageContext, points: list[PipePoint] | 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]
# auto 를 열쇠에 — 같은 관 목록이라도 [자동 제안] 결과와 섞이지 않게(369 · 옛 저장본도 비껴감)
digest.update(json.dumps([listed, auto], sort_keys=True, ensure_ascii=False).encode())
digest.update(json.dumps(listed, 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"]
@@ -180,10 +173,10 @@ def _remember(key: str, value: tuple[DrainageDetail, list[PipePoint]]) -> None:
def _build_cached(
context: DrainageContext, points: list[PipePoint] | None, auto: str | None = None
context: DrainageContext, points: list[PipePoint] | None
) -> tuple[DrainageDetail | None, list[PipePoint]]:
"""`_build` + 저장본 — 관 목록은 사본으로 돌려준다(저장이 좌표를 채워도 저장본은 그대로)."""
key = _detail_key(context, points, auto)
key = _detail_key(context, points)
hit = _DETAIL_CACHE.get(key)
if hit is None:
try:
@@ -193,7 +186,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, auto=auto)
detail, carried = _build(context.stored_path, context, points)
if detail is None:
return None, []
hit = (detail, carried)
@@ -289,20 +282,8 @@ 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], 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
def _retag(pipes: list[StructureCandidate], points: list[PipePoint]) -> None:
"""확정 좌표로 되돌아온 관에 원래의 생성 사유를 다시 붙인다."""
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))
@@ -481,16 +462,11 @@ 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)
@@ -504,7 +480,7 @@ async def _resolve(
saved = requested is None and stored is not None
if analysis_route_signature(context.stored_path) != signature:
if points is None or auto is not None: # 자동 자리 = 지금 노선 분석이 있어야(369)
if points is None:
return JSONResponse(
status_code=409, content={"status": "stale", "message": STALE_MESSAGE}
)
@@ -513,11 +489,9 @@ async def _resolve(
# [저장]은 정본으로 다시 셈 — 저장본은 GET · POST 만
if fresh:
result = await asyncio.to_thread(
lambda: _build(context.stored_path, context, points, auto=auto)
)
result = await asyncio.to_thread(lambda: _build(context.stored_path, context, points))
else:
result = await asyncio.to_thread(_build_cached, context, points, auto)
result = await asyncio.to_thread(_build_cached, context, points)
detail, resolved = result
if detail is None:
return JSONResponse(status_code=404, content={"status": "error", "message": _NO_ANALYSIS})
@@ -549,7 +523,6 @@ async def post_detail_basins(
`points` 칸 없이 부르면 저장분을 무시하고 기본 관(세류선 교차)만으로 되돌린다 —
[배관 배치 초기화] · 노선이 바뀌어 전체 분석을 다시 돌린 직후의 경로다. 빈 목록은 관 0개(312).
`auto` = [자동 제안](369) — "all" 전체 자동 · "add" 지금 관 + 빈 구간 보충.
"""
resolved = await _resolve(project_id, payload, use_stored=False)
if isinstance(resolved, JSONResponse):
@@ -22,7 +22,7 @@ from fastapi.responses import JSONResponse
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import preview_stages
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Expand import AnalysisStopped
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
from B04_PreProcess.B04_PreProcess_Router_Basins import AUTO_MODES, _build, _payload
from B04_PreProcess.B04_PreProcess_Router_Basins import _build, _payload
from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare, router
from common_util.common_util_drainage_context import _open_sampler, load_drainage_context
from common_util.common_util_drainage_detail import RoadRouting
@@ -79,7 +79,7 @@ async def post_drainage_live(
project_id: UUID, payload: dict[str, Any] | None = None
) -> dict[str, Any] | JSONResponse:
"""몸통 = {generation, route: RouteDraft | null(저장 노선),
points: 관 목록 | null(세류선 교차만), auto: [자동 제안] 방식(all · add · 369) | 없음}."""
points: 관 목록 | null(세류선 교차만)}."""
body = payload or {}
generation = int(body.get("generation") or 0)
key = str(project_id)
@@ -88,9 +88,6 @@ async def post_drainage_live(
def superseded() -> bool:
return _LATEST_GENERATION.get(key) != generation
auto = body.get("auto")
if auto not in AUTO_MODES:
return JSONResponse(status_code=400, content={"status": "error", "message": "auto 틀림"})
try:
raw = body.get("points")
points = None if raw is None else parse_pipe_points(raw)
@@ -143,7 +140,7 @@ async def post_drainage_live(
[list(to_lonlat(x, y)) for x, y in line.coords] for line in preview.region.split.upstream
]
detail, carried = await asyncio.to_thread(
lambda: _build(context.stored_path, live, points, routing, upstream, auto=auto)
lambda: _build(context.stored_path, live, points, routing, upstream)
)
if detail is None:
return JSONResponse(
+3 -8
View File
@@ -172,25 +172,20 @@ 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 분석 결과를 읽어 관 자리를 정하고 세부유역을 나눈다.
`confirmed_chainages`를 주면 그 위치를 관으로 확정하고(사용자 편집 · 빈 목록 = 관 0개), None 이면
B04의 기본 관(세류선 교차)만. 어느 쪽이든 격자 해석은 하지 않는다.
`auto`([자동 제안] · 369) — None = 위 그대로 · "all" = 기본 관 + 최대 간격 보충으로 다시
(확정 목록 무시) · "add" = 확정 목록 + 빈 구간 보충. 관 자리 후보는 어느 길이든 셈.
관 자리 후보(`pipe_candidates`)는 어느 길이든 셈 — [자동 제안](510)이 이 후보를 씀.
`routing` · `upstream_lonlat` 을 주면 파일 대신 그것(「지금 노선으로 셈」 메모리 결과 · 347-2).
"""
routing = routing or read_road_routing(directory)
if routing is None or len(vertices) < 2:
return None
if confirmed_chainages is not None and auto != "all":
if confirmed_chainages is not None:
pipes = pipes_from_chainages(vertices, confirmed_chainages)
if auto == "add":
pipes = [*pipes, *suggest_pipe_positions(vertices, pipes, routing.strength_curve)]
pipes.sort(key=lambda pipe: pipe.chainage_m)
else:
# 저장분의 기본 관은 누가거리만 신뢰한다 — 좌표는 현재 노선 위로 다시 찍는다.
base = [
@@ -203,7 +198,7 @@ def build_detail(
for pipe in routing.base_pipes
]
# 자동 관 자리 = 정수 m · 일반 측점 1 m 안이면 그 측점(234-1) — 확정본(위)은 단추로만
placed = place_pipes(vertices, base, routing.strength_curve, fill=auto is not None)
placed = place_pipes(vertices, base, routing.strength_curve, fill=False)
pipes = snap_candidates(placed, vertices)
if confirmed_chainages:
kept = pipes_from_chainages(vertices, confirmed_chainages)
+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 and auto != "all":' in src
assert "if confirmed_chainages is not None:" 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, auto=None):
def build(stored_path, context, points):
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, auto=None):
def build(stored_path, context, points):
built.called = True
return SimpleNamespace(pipes=[], basins=["유역"]), points or []
monkeypatch.setattr(basins, "_build", build)
monkeypatch.setattr(
basins, "_build_cached", lambda context, points, auto=None: build("x", context, points)
basins, "_build_cached", lambda context, points: build("x", context, points)
)
return built
+10 -79
View File
@@ -1,10 +1,7 @@
"""369 배수 구조물 자리 — 기본 = 세류선 교차만 · [자동 제안] 두 방식(all · add).
"""369 배수 구조물 자리 — 기본 = 세류선 교차만 · 목록 있으면 그대로 · 낡은 분석은 관 자리만.
- 기본(auto 없음) = 세류선 교차 관만 · 100 m 보충 없음 · 관 자리 후보는 셈.
- all = 세류선 + 100 m 보충으로 다시(지금 목록 무시) · 같은 자리(0.05 m) 관은 속성 · id 유지 ·
나머지 = 새 관(등록부 기본값).
- add = 지금 목록 그대로 + 빈 구간 보충 · 더한 관 = 새 관(등록부 기본값).
- 틀린 auto = 400 · 저장본 열쇠에 auto.
- 목록 없음 = 세류선 교차 관만 · 100 m 보충 없음 · 관 자리 후보는 셈.
- 목록 있음 = 그 자리 그대로(보충은 후보만) · [자동 제안] = 510 (`test_510_auto_pipe_propose.py`).
"""
import asyncio
@@ -17,7 +14,6 @@ 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)]
@@ -41,8 +37,8 @@ def _routing() -> RoadRouting:
)
def _detail(tmp_path, chainages=None, auto=None):
return build_detail(tmp_path / "없음", VERTICES, chainages, _routing(), [], auto)
def _detail(tmp_path, chainages=None):
return build_detail(tmp_path / "없음", VERTICES, chainages, _routing(), [])
def test_기본은_세류선_교차만(tmp_path):
@@ -51,84 +47,19 @@ def test_기본은_세류선_교차만(tmp_path):
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.02 in places and 10.0 not in places # 세류선 150 m 0.5 m 안의 지금 관 = 그 자리
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 없음 = 목록 그대로(보충은 후보만)
def test_목록은_그대로_보충은_후보만(tmp_path):
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.3, # 세류선 150 m 에서 0.3 m — 노선 따라감으로 어긋난 같은 관
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 ford[0].chainage_m == 150.3 # 그 자리 그대로
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
def test_낡은_분석에_auto_는_409(monkeypatch):
def test_낡은_분석은_관_자리만_낡음으로(monkeypatch):
async def load(project_id):
return CONTEXT, ""
monkeypatch.setattr(basins, "load_drainage_context", load)
monkeypatch.setattr(basins, "analysis_route_signature", lambda stored: "옛 노선")
points = [{"chainage_m": 20.0}]
for auto in ("all", "add"):
result = asyncio.run(
basins._resolve(uuid4(), {"points": points, "auto": auto}, use_stored=False)
)
assert result.status_code == 409 # 말없이 그대로 돌려주지 않음 — 낡음 까닭
kept = asyncio.run(basins._resolve(uuid4(), {"points": points}, use_stored=False))
assert kept[-1] is True # auto 없음 = 관 자리만 낡음으로(그대로)
assert kept[-1] is True # 관 목록 있음 = 관 자리만 낡음으로
none = asyncio.run(basins._resolve(uuid4(), {}, use_stored=False))
assert none.status_code == 409 # 목록 없음 = 낡음 까닭
@@ -32,8 +32,9 @@ def _make_client(tmp_path, monkeypatch, invalidated_calls, *, stale_ok=True, rou
monkeypatch.setattr(router_module, "_invalidate_downstream", fake_invalidate)
async def fake_saved_irregular(project_id, root, route_id):
# 이관 원천 = 저장 종단 파일(452) — 본문 대신 여기서 같은 측점을 줌
return _migrate_payload()["stations"]
# 이관 원천 = 저장 종단 파일(452) — 본문 대신 여기서 같은 측점을 줌 (legacy, standing)
legacy = _migrate_payload()["stations"]
return legacy, [float(item["chainage_m"]) for item in legacy]
monkeypatch.setattr(router_module, "_route_length", fake_route_length)
monkeypatch.setattr(router_module, "_saved_irregular_stations", fake_saved_irregular)
@@ -306,7 +307,7 @@ def test_migrate_keeps_existing_structures(client, tmp_path):
def test_migrate_empty_list_is_noop(client, monkeypatch):
async def none_saved(project_id, root, route_id):
return []
return [], []
monkeypatch.setattr(router_module, "_saved_irregular_stations", none_saved)
response = client.post(