From 077da60065edf0b7687a5e16135e8a17bde78ad2 Mon Sep 17 00:00:00 2001 From: umsangdon Date: Wed, 29 Jul 2026 19:41:58 +0900 Subject: [PATCH] auto: 2026-07-29 19:41 (EOMSANGDON-HOME) --- .../B05_wf2_Route_Engine_Watershed_Trace.py | 84 ++++++++++++++++--- 1 file changed, 73 insertions(+), 11 deletions(-) diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Trace.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Trace.py index c5877c23..8604b8ca 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Trace.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Trace.py @@ -291,6 +291,11 @@ def _contour_chain_boundary( 그 등고선 위에서 두 세류(우리 상류망·인접 세류) 모두로부터 가장 먼 지점(능선 꼭짓점)을 정제해 포인트를 얻고, 링 위 위치 순으로 연결한다. 등고선이 없는 구간은 원래 링 정점으로 메운다. 제약: 우리 세류 침범·인접 세류 이탈 금지. + + 2차 보정(2026-07-29 사용자 채택, "2번 방식"): 능선은 등고선의 **직교 궤적**이므로, + 각 포인트를 등고선 위에서 미세 이동해 경계선이 그 등고선과 수직으로 교차하도록 + 반복 조정한다(TOPOG/TAPES-C 계열 개념). 능선 점수(세류 최소거리)가 꼭짓점 대비 + 크게 떨어지는 이동은 막는다. """ ring = LineString(region.exterior.coords) @@ -303,8 +308,9 @@ def _contour_chain_boundary( ) return to_network, to_other - # 링을 가로지르는 등고선 교차점마다 능선 꼭짓점 1개. - chained: list[tuple[float, float, float]] = [] # (링 위치 s, x, y) + # ① 링을 가로지르는 등고선 교차점마다 능선 꼭짓점 1개. + # entry = [링 위치 s, Point, geom_index(-1=링 정점), 등고선 파라미터 t, 꼭짓점 점수] + entries: list[list[Any]] = [] for index in contour_index.query(ring.buffer(1.0)): geom = contour_index.geoms[index] try: @@ -315,6 +321,7 @@ def _contour_chain_boundary( s = ring.project(crossing) t0 = geom.project(crossing) best = None + best_t = t0 best_value = -1.0 steps = int(RIDGE_WALK_M / 10.0) for offset in [0.0] + [ @@ -332,23 +339,25 @@ def _contour_chain_boundary( if min(to_network, to_other) > best_value: best_value = min(to_network, to_other) best = candidate + best_t = t if best is not None: - chained.append((s, best.x, best.y)) - if len(chained) < 4: + entries.append([s, best, index, best_t, best_value]) + if len(entries) < 4: return region - chained.sort() - # 등고선 공백 구간(교차점 사이가 먼 곳)은 원래 링 정점으로 메운다. - positions = [s for s, _, _ in chained] - filled: list[tuple[float, float, float]] = list(chained) + entries.sort(key=lambda entry: entry[0]) + # ② 등고선 공백 구간(교차점 사이가 먼 곳)은 원래 링 정점으로 메운다. + positions = [entry[0] for entry in entries] for x, y in list(region.exterior.coords)[:-1]: s = ring.project(Point(x, y)) slot = bisect.bisect_left(positions, s) before = positions[slot - 1] if slot > 0 else positions[-1] - ring.length after = positions[slot] if slot < len(positions) else positions[0] + ring.length if min(s - before, after - s) > VALLEY_CELL_M * 2.5: - filled.append((s, x, y)) - filled.sort() - polygon = Polygon([(x, y) for _, x, y in filled]).buffer(0) + entries.append([s, Point(x, y), -1, 0.0, 0.0]) + entries.sort(key=lambda entry: entry[0]) + # ③ 직교 보정: 경계 진행방향과 등고선 접선이 수직이 되도록 포인트를 미세 이동. + entries = _orthogonalize_chain(entries, contour_index, _score) + polygon = Polygon([(entry[1].x, entry[1].y) for entry in entries]).buffer(0) if polygon.geom_type == "MultiPolygon": polygon = max(polygon.geoms, key=lambda part: part.area) if polygon.is_empty or polygon.geom_type != "Polygon": @@ -356,6 +365,59 @@ def _contour_chain_boundary( return polygon +def _orthogonalize_chain( + entries: list[list[Any]], + contour_index: Any, + score: Any, +) -> list[list[Any]]: + """체인 포인트를 등고선 위에서 이동해 경계가 등고선과 직교하게 만든다. + + 각 포인트에서 |등고선 접선 · 체인 진행방향| (수직이면 0)을 최소화한다. 이동 허용 + 조건: 세류 침범·이탈 금지 + 능선 점수(두 세류 최소거리)가 꼭짓점 값의 70% 이상. + 2회 반복으로 이웃 이동의 영향을 수렴시킨다. + """ + count = len(entries) + for _ in range(2): + for i, entry in enumerate(entries): + geom_index = entry[2] + if geom_index < 0: + continue + geom = contour_index.geoms[geom_index] + previous = entries[i - 1][1] + following = entries[(i + 1) % count][1] + dx, dy = following.x - previous.x, following.y - previous.y + norm = (dx * dx + dy * dy) ** 0.5 + if norm < 1.0: + continue + dx, dy = dx / norm, dy / norm + floor = 0.7 * entry[4] + best_t = entry[3] + best_point = entry[1] + best_dot = None + for offset in range(-int(RIDGE_SNAP_M), int(RIDGE_SNAP_M) + 1, 5): + t = min(max(entry[3] + float(offset), 0.0), geom.length) + candidate = geom.interpolate(t) + ahead = geom.interpolate(min(t + 4.0, geom.length)) + behind = geom.interpolate(max(t - 4.0, 0.0)) + tx, ty = ahead.x - behind.x, ahead.y - behind.y + tangent_norm = (tx * tx + ty * ty) ** 0.5 + if tangent_norm < 0.5: + continue + to_network, to_other = score(candidate) + if to_network < STREAM_JOIN_TOL_M or to_other < to_network: + continue + if min(to_network, to_other) < floor: + continue + dot = abs((tx * dx + ty * dy) / tangent_norm) + if best_dot is None or dot < best_dot: + best_dot = dot + best_t = t + best_point = candidate + entry[1] = best_point + entry[3] = best_t + return entries + + def trace_upstream_network( crossing: Point, stream_features: list[dict[str, Any]],