diff --git a/common_util/common_util_drainage_detail.py b/common_util/common_util_drainage_detail.py index ceb52b29..21b90fbf 100644 --- a/common_util/common_util_drainage_detail.py +++ b/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) # ── ⑩ 세부유역 조립 ────────────────────────────────────────────────────────