common_util/common_util_drainage_detail.py

@@ -460,46 +460,43 @@ def assign_road_cells_to_pipes(
 ) -> np.ndarray:
     """도로 셀마다 물이 실제로 흘러가는 담당 관 번호를 정한다.
 
-    노면 물은 측구를 타고 종단 내리막으로 흐르므로, 종단 계획선을 1차원 지형으로 보고
-    같은 방식(내리막 추적 + 관에서 흡수)으로 푼다. 관이 없는 사그(저점)에 갇힌 구간은
-    가장 가까운 관이 받는 것으로 본다.
+    노면 물은 측구를 타고 종단 내리막으로 흐르므로 종단 계획선을 1차원 지형으로 본다.
+    1차원에서는 물이 **마루(구간 최고점)를 넘지 못한다** — 이웃한 두 관 사이의 최고점이
+    곧 분수령이고, 그 왼쪽은 앞 관이, 오른쪽은 뒤 관이 받는다. 첫 관 앞과 마지막 관 뒤는
+    그 관이 받는다.
+
+    옛 방식(한 칸 이웃만 보는 국소 하강 + 관 없는 저점은 최근접 관)은 계획고의 미세
+    요철에 걸려 멈췄다. 용화 실측: 측점 2,138개 중 1,898개(88.8%)가 저점에 갇혀 흐름이
+    아니라 **누가거리 최근접**으로 배정됐고, 그 결과 도로 셀 43.2%가 자기보다 높은 관에
+    배정됐다(최대 6.82m 오르막). 마루 기준은 미세 요철을 타지 않으므로 오르막 배정이
+    구조적으로 생기지 않는다(2026-09-03 사용자 확정).
     """
     total_length = vertices[-1].chainage_m
     step = max(DRAINAGE_DITCH_SAMPLE_M, 0.5)
     stations = np.arange(0.0, total_length + step, step)
-    heights = np.array([interpolate_vertex(vertices, float(s))[2] for s in stations])
     pipe_chainages = np.array([pipe.chainage_m for pipe in pipes])
-    pipe_station = np.clip(np.round(pipe_chainages / step).astype(np.int64), 0, stations.size - 1)
-
-    # 앞뒤 이웃 중 더 낮은 쪽으로 흘려보낸다(양쪽 다 높으면 사그 = 제자리).
-    back_z = np.full(stations.size, np.inf)
-    back_z[1:] = heights[:-1]
-    forward_z = np.full(stations.size, np.inf)
-    forward_z[:-1] = heights[1:]
-    go_back = (back_z < heights) & (back_z <= forward_z)
-    go_forward = (forward_z < heights) & ~go_back
-    receiver = np.arange(stations.size, dtype=np.int64)
-    receiver[go_back] -= 1
-    receiver[go_forward] += 1
-    receiver[pipe_station] = pipe_station  # 관은 물을 흡수한다
-
-    owner = np.full(stations.size, -1, dtype=np.int64)
-    owner[pipe_station] = np.arange(pipe_chainages.size)
-    jump = receiver
-    for _ in range(40):
-        next_jump = jump[jump]
-        if np.array_equal(next_jump, jump):
-            break
-        jump = next_jump
-    resolved = owner[jump]
-    # 관 없는 사그에 갇힌 구간은 가장 가까운 관에 붙인다.
-    orphan = resolved < 0
-    if orphan.any() and pipe_chainages.size:
-        nearest = np.abs(stations[orphan, None] - pipe_chainages[None, :]).argmin(axis=1)
-        resolved[orphan] = nearest
-
     slot_station = np.clip(np.round(road_chainage / step).astype(np.int64), 0, stations.size - 1)
-    return resolved[slot_station].astype(np.int32)
+    if pipe_chainages.size == 0:
+        return np.full(road_chainage.size, -1, dtype=np.int32)
+
+    heights = np.array([interpolate_vertex(vertices, float(s))[2] for s in stations])
+    # 관 순서는 호출자가 준 그대로 돌려줘야 한다 — 누가거리로 정렬해 풀고 끝에 되돌린다.
+    order = np.argsort(pipe_chainages, kind="stable")
+    pipe_station = np.clip(
+        np.round(pipe_chainages[order] / step).astype(np.int64), 0, stations.size - 1
+    )
+
+    owner = np.full(stations.size, pipe_station.size - 1, dtype=np.int64)  # 마지막 관 뒤
+    owner[: pipe_station[0] + 1] = 0  # 첫 관 앞
+    for index in range(pipe_station.size - 1):
+        left = pipe_station[index]
+        right = pipe_station[index + 1]
+        if right <= left:
+            continue
+        ridge = left + int(np.argmax(heights[left : right + 1]))
+        owner[left : ridge + 1] = index
+        owner[ridge + 1 : right + 1] = index + 1
+    return order[owner][slot_station].astype(np.int32)
 
 
 # ── ⑩ 세부유역 조립 ────────────────────────────────────────────────────────