diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py index dbd3340a..3fcf332f 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Flow.py @@ -189,41 +189,46 @@ def burn_stream_flow( 것이 아니라 **그대로 따라야 한다**(2026-07-31 사용자 지시). 세류선 위 셀은 물길 방향의 다음 셀을 수신 셀로 삼는다. 그러면 세류선 셀은 물론, - 세류선으로 흘러드는 사면 셀까지 전부 도로에 도달한다. 도로 셀은 흡수점이므로 건드리지 않는다. + 세류선으로 흘러드는 사면 셀까지 전부 도로에 도달한다. 도로 셀은 흡수점이므로 수신 셀은 + 건드리지 않되, 세류 셀 목록에는 포함한다. - 돌려주는 값: (흐름이 새겨진 지형, 새긴 셀 마스크). + **표고 유무를 따지지 않는다.** 세류선은 확정된 자료이므로 등고선 TIN 껍질 밖이라 + 표고가 없는 셀이라도 물이 지나간다는 사실은 변하지 않는다. 표고를 조건으로 걸면 그런 + 셀이 새김에서 빠져 파랑·회색으로 남는다. + + 돌려주는 값: (흐름이 새겨진 지형, 세류선이 지나는 셀 마스크). """ spec = terrain.spec receiver = terrain.receiver.copy() step_length = terrain.step_length.copy() - valid = terrain.valid.reshape(-1) road_cells = road.mask.reshape(-1) burned = np.zeros(spec.size, dtype=bool) tails: list[int] = [] + # 셀이 몇 갈래에 속하는지 — 하류 끝이 합류 지점인지 판단하는 근거. + visits = np.zeros(spec.size, dtype=np.int32) for line in streams: chain = _line_cell_chain(spec, line) - if len(chain) < 2: + if not chain: continue + # 사슬의 **모든** 셀을 세류 셀로 표시한다 — 마지막 셀도 물길 위다. + burned[chain] = True + np.add.at(visits, np.unique(chain), 1) for current, following in zip(chain, chain[1:]): - if not valid[current] or not valid[following] or road_cells[current]: - continue - if not _is_neighbour(spec, current, following): + if road_cells[current] or not _is_neighbour(spec, current, following): continue receiver[current] = following step_length[current] = _cell_distance(spec, current, following) - burned[current] = True tails.append(chain[-1]) - # 하류 끝이 도로에도, 다른 세류 갈래에도 닿지 않은 갈래만 진짜 문제다. - # 지류가 본류 중간에 합류하는 경우는 끝점이 본류 셀이므로 정상이다. - detached = sum(1 for tail in tails if not road_cells[tail] and not burned[tail]) + # 하류 끝이 도로에도 닿지 않고 다른 갈래와도 겹치지 않으면 그 갈래는 떠 있는 것이다. + # 지류가 본류 중간에 합류하는 경우는 끝 셀을 두 갈래가 공유하므로 정상이다. + detached = sum(1 for tail in tails if not road_cells[tail] and visits[tail] < 2) if detached: logger.warning( - "배수유역: 세류망 %d갈래의 하류 끝이 도로·다른 세류 어디에도 닿지 않습니다.", - detached, + "배수유역: 세류망 %d갈래의 하류 끝이 도로·다른 세류 어디에도 닿지 않습니다.", detached ) - logger.info("배수유역: 세류망 흐름 새김 %d셀 (세류 %d갈래)", int(burned.sum()), len(streams)) + logger.info("배수유역: 세류망 셀 %d개 표시 (세류 %d갈래)", int(burned.sum()), len(streams)) return ( TerrainGrid( spec=spec, @@ -300,48 +305,49 @@ def classify_flow( 출발해 화살표가 가리키는 다음 셀, 그 셀이 가리키는 그다음 셀로 사슬처럼 이어 간다 (2026-07-31 사용자 지시). - 사슬이 **도로 셀 또는 확정된 세류망 셀**에 닿으면 그 사슬 전체가 적색이다. 세류망은 - 이미 "도로를 건너 하류로 빠지는 물길"로 확정된 자료이므로, 물이 세류에 합류한 시점에 - 도로 도달이 결정된다. 사슬이 싱크에서 멈추거나 해석 영역 밖으로 나가면 전체가 미도달 - (파랑 채움 + 백색 화살표)이다. + 판정 순서는 다음과 같다(2026-07-31 사용자 지시): - **한 번 판정한 셀은 다시 분석하지 않는다** — 사슬이 이미 색이 정해진 셀을 만나면 그 - 셀의 색을 그대로 물려받고 끝낸다. 최외곽 추적으로 안 닿은 내부 셀은 그다음에 따로 - 출발시킨다. + ⓪ **세류선과 겹치는 셀을 먼저 적색으로 확정한다.** 세류망은 이미 "도로를 건너 하류로 + 빠지는 물길"로 확정된 자료다. 표고가 있든 없든 물이 지나간다는 사실은 변하지 않으므로 + 추적 결과를 기다릴 이유가 없다. + ① 최외곽 셀에서 출발해 사슬을 따라간다. + ② 사슬이 도로 셀이나 세류 셀에 닿으면 사슬 전체가 적색, 싱크에서 멈추거나 해석 영역 + 밖으로 나가면 전체가 미도달(파랑 채움 + 백색 화살표)이다. + ③ 이미 색이 정해진 셀을 만나면 **그 셀의 색을 그대로 물려받고** 끝낸다. 판정된 셀은 + 다시 분석하지 않는다. + ④ 최외곽 추적에 안 걸린 내부 셀을 그다음에 따로 출발시킨다. - 화살표 방위는 D8(8방위)이 아니라 지형 최급강하 32방위를 쓴다. `burned`(세류망을 따라 - 흐름을 새긴 셀)는 확정된 물길 방향을 그대로 쓴다. + 화살표 방위는 D8(8방위)이 아니라 지형 최급강하 32방위를 쓴다. 세류 셀은 확정된 물길 + 방향을 그대로 쓴다. """ spec = terrain.spec valid = terrain.valid.reshape(-1) receiver = terrain.receiver # 도로 셀과 세류망 셀 둘 다 "여기 닿으면 적색"인 종결점이다. - absorbing = road.mask.reshape(-1) & valid - if burned is not None: - absorbing = absorbing | (burned & valid) + stream_cells = np.zeros(spec.size, dtype=bool) if burned is None else burned + absorbing = (road.mask.reshape(-1) & valid) | stream_cells direction = descent_azimuth(spec, terrain.elevation, terrain.valid, receiver, burned) reaches = np.zeros(spec.size, dtype=bool) # 0=미방문, 1=경로에 올라 있음, 2=판정 완료 state = np.zeros(spec.size, dtype=np.int8) + # ⓪ 세류선과 겹치는 셀을 먼저 적색으로 못박는다. + reaches[stream_cells] = True + state[stream_cells] = 2 + stream_done = int(stream_cells.sum()) + outer = np.flatnonzero(outermost_cells(terrain.valid)) remaining = np.flatnonzero(valid) outer_done = _walk_from(outer, receiver, valid, absorbing, state, reaches) interior_done = _walk_from(remaining, receiver, valid, absorbing, state, reaches) analyzed = state == 2 - if burned is not None: - stranded = int((burned & analyzed & ~reaches).sum()) - if stranded: - logger.warning( - "배수유역: 세류망 새김 셀 %d개가 도로에 닿지 않습니다 — 하류 끝이 도로 셀과 " - "이어지지 않았는지 확인이 필요합니다.", - stranded, - ) logger.info( - "배수유역: 흐름 판정 %d셀 (최외곽 출발 %d / 내부 보충 %d) — 도로 도달 %d, 미도달 %d", + "배수유역: 흐름 판정 %d셀 (세류 선확정 %d / 최외곽 출발 %d / 내부 보충 %d) — " + "도로 도달 %d, 미도달 %d", int(analyzed.sum()), + stream_done, outer_done, interior_done, int((reaches & analyzed).sum()), @@ -390,8 +396,12 @@ def _walk_from( verdict = True # 도로 또는 세류망에 합류 — 여기서 하류로 빠진다 break following = int(receiver[node]) - if following == node or not valid[following]: - verdict = False # 싱크에 갇히거나 해석 영역 밖으로 빠짐 + if following == node: + verdict = False # 싱크에 갇힘 + break + if not valid[following] and state[following] != 2: + # 해석 영역 밖으로 빠짐. 단, 이미 색이 정해진 셀(세류 선확정 등)이면 따라간다. + verdict = False break node = following for visited in path: diff --git a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py index b6498d83..1bad541d 100644 --- a/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py +++ b/B05_wf2_Route/B05_wf2_Route_Engine_Watershed_Grid.py @@ -545,7 +545,9 @@ def descent_azimuth( (np.abs(move_row) < 1e-12) & (np.abs(move_col) < 1e-12) ) code = np.where(is_sink, AZIMUTH_SINK, code) - code = np.where(valid, code, AZIMUTH_INVALID) + # 세류망을 따라 흐름을 새긴 셀은 표고가 없어도 방향이 확정돼 있다. + known = valid if forced is None else (valid | forced.reshape(rows, cols)) + code = np.where(known, code, AZIMUTH_INVALID) return code.reshape(-1).astype(np.int16)