"""세류선 교차점 · 자동 세류선 관 라벨(507-2). 교차점 = 노선 평면 선형 × 세류선(`find_stream_crossings`) — 지물 id(`stream_id`)를 함께 단다. 자동 세류선 관 라벨(`PipePoint.stream_label`) = `{stream, order, x, y}` — 세류선 지물 id · 그 세류선의 노선 위 교차 차례 · 처음 교차점 좌표. 끌거나 옵션을 고쳐 `source` 가 바뀌어도 남음. - 노선 바뀐 [저장](`B05_Profile_Route_Save.followed_pipes`) — 라벨(관 · 지움 표지)마다 같은 세류선의 새 교차점 중 라벨 좌표와 가장 가까운 것을 짝으로 잡음 · 짝 잡힌 교차점엔 관 안 더함. - 지움 표지(`pipe_points.json` `dismissed_streams`) = 사용자가 지운 자동 관 라벨 — 다시 안 생김. 라벨 없는 자동 관을 지우면 그 자리(x · y)만 남겨 두고 다음 노선 [저장] 때 라벨로 바꿈. - 라벨 없는 옛 관(`source` stream) = 노선 [저장] 때 옛 노선 교차점 5 m 안이면 그 교차 라벨 · 교차 하나에 관 하나(가장 가까운 관) · 나머지 겹친 관은 지우지 않고 수만 셈. - 추천 [적용] · [초기화] = 초기값 파일 그대로(지움 표지도 초기값). """ from __future__ import annotations import math from typing import Any, Sequence from shapely.geometry import LineString, Point, shape from common_util.common_util_route_geometry import RouteVertex, StructureCandidate from config.config_system import DRAINAGE_PIPE_MIN_SPACING_M def stream_key(feature: dict[str, Any], index: int) -> str: """세류선 지물 id — 도엽 UFID · 없으면 피처 id · 그것도 없으면 파일 안 차례.""" properties = feature.get("properties") or {} value = properties.get("UFID") or feature.get("id") return str(value) if value not in (None, "") else f"#{index}" def find_stream_crossings( vertices: list[RouteVertex], stream_features: list[dict[str, Any]], ) -> list[StructureCandidate]: """노선 평면 선형과 세류선의 교차 지점을 누가거리 순으로 찾는다.""" if len(vertices) < 2: return [] route_line = LineString([(vertex.x, vertex.y) for vertex in vertices]) candidates: list[StructureCandidate] = [] for index, feature in enumerate(stream_features): geometry = feature.get("geometry") if not geometry: continue try: stream = shape(geometry) except Exception: # noqa: BLE001 - 손상된 피처는 건너뛴다 continue if stream.is_empty: continue intersection = route_line.intersection(stream) if intersection.is_empty: continue name = _stream_name(feature) for point in _collect_points(intersection): candidates.append( StructureCandidate( chainage_m=route_line.project(point), x=point.x, y=point.y, reason="stream", stream_name=name, stream_id=stream_key(feature, index), ) ) candidates.sort(key=lambda item: item.chainage_m) return candidates def _stream_name(feature: dict[str, Any]) -> str | None: properties = feature.get("properties") or {} for key in ("명칭", "하천명", "NAME", "name"): value = properties.get(key) if value: return str(value) return None def _collect_points(geometry: Any) -> list[Point]: """교차 결과(Point/MultiPoint/LineString 등)에서 대표 점들을 뽑는다.""" if geometry.geom_type == "Point": return [geometry] if geometry.geom_type in {"MultiPoint", "GeometryCollection"}: points: list[Point] = [] for part in geometry.geoms: points.extend(_collect_points(part)) return points # 선분끼리 겹쳐 선으로 나온 경우는 중점을 대표로 쓴다. if geometry.geom_type in {"LineString", "MultiLineString"}: return [geometry.interpolate(0.5, normalized=True)] return [] # ── 자동 세류선 관 라벨(507-2) ──────────────────────────────────────────────── def crossing_labels(crossings: Sequence[Any]) -> list[dict[str, Any]]: """교차점(누가거리 순)마다 라벨 — 차례 = 같은 세류선 안에서 몇 번째 교차인가.""" seen: dict[Any, int] = {} labels = [] for crossing in crossings: stream = getattr(crossing, "stream_id", None) seen[stream] = seen.get(stream, 0) + 1 labels.append( { "stream": stream, "order": seen[stream], "x": round(float(crossing.x), 3), "y": round(float(crossing.y), 3), } ) return labels def _gap(label: dict[str, Any], x: float, y: float) -> float: return math.hypot(float(label["x"]) - x, float(label["y"]) - y) def claim_crossings(labels: Sequence[dict[str, Any]], crossings: Sequence[Any]) -> dict[int, int]: """라벨마다 같은 세류선 교차점 중 라벨 좌표와 가장 가까운 것 하나 — {교차점 번호: 라벨 번호}. 가까운 짝부터 하나씩(라벨 하나 · 교차점 하나) · 거리 한계 없음. """ pairs = sorted( (_gap(label, float(crossing.x), float(crossing.y)), li, ci) for li, label in enumerate(labels) if "stream" in label for ci, crossing in enumerate(crossings) if getattr(crossing, "stream_id", None) == label["stream"] ) used_labels: set[int] = set() used: dict[int, int] = {} for _, li, ci in pairs: if li in used_labels or ci in used: continue used_labels.add(li) used[ci] = li return used def adopt_labels( pipes: Sequence[Any], dismissed: list[dict[str, Any]], crossings: Sequence[Any], old_xy: Sequence[Sequence[float]], ) -> int: """옛 노선 교차점 중 라벨(관 · 지움 표지)이 안 잡은 것 — 옛 노선 누가거리 5 m 안 라벨 없는 stream 관(가장 가까운 것) · 없으면 자리만 남은 지움 표지에 그 교차 라벨(제자리) · 돌려줌 = 겹쳐 라벨 못 받은 stream 관 수. 거리 = 옛 노선에 투영한 누가거리 차(옛 관 좌표가 다른 선 위에 남은 저장분 · 옆 15 m 실측). """ from common_util.common_util_drainage_pipes import PIPE_SOURCE_STREAM line = LineString(old_xy) labels = crossing_labels(crossings) held = [pipe.stream_label for pipe in pipes if pipe.stream_label] + [ mark for mark in dismissed if "stream" in mark ] claimed = claim_crossings(held, crossings) loose = [ (line.project(Point(pipe.x, pipe.y)), pipe) for pipe in pipes if not pipe.stream_label and pipe.source == PIPE_SOURCE_STREAM and pipe.x is not None ] spots = [ (line.project(Point(float(mark["x"]), float(mark["y"]))), mark) for mark in dismissed if "stream" not in mark ] def nearest(items: list, at: float) -> int | None: gaps = [(abs(chainage - at), order) for order, (chainage, _) in enumerate(items)] best = min(gaps, default=None) return best[1] if best and best[0] <= DRAINAGE_PIPE_MIN_SPACING_M else None for index, (label, crossing) in enumerate(zip(labels, crossings)): if index in claimed: continue at = float(crossing.chainage_m) pick = nearest(loose, at) if pick is not None: loose.pop(pick)[1].stream_label = dict(label) continue pick = nearest(spots, at) if pick is not None: spots.pop(pick)[1].update(label) near = [float(crossing.chainage_m) for crossing in crossings] return sum( 1 for chainage, _ in loose if nearest([(c, None) for c in near], chainage) is not None ) def keep_labels( points: Sequence[Any], stored: Sequence[Any], dismissed: list[dict[str, Any]] ) -> None: """관 저장(PUT) — 화면이 모르는 라벨을 정본에서 id 로 되붙임 · 지운 자동 관 = 지움 표지(제자리). 라벨 있는 관 = 라벨 · 라벨 없는 stream 관 = 그 자리(x · y · 다음 노선 [저장] 때 라벨로). """ from common_util.common_util_drainage_pipes import PIPE_SOURCE_STREAM by_id = {pipe.id: pipe for pipe in stored if pipe.id} for point in points: old = by_id.get(point.id) if not point.stream_label and old is not None and old.stream_label: point.stream_label = dict(old.stream_label) kept = {point.id for point in points} for pipe in stored: if pipe.id in kept: continue if pipe.stream_label: mark = dict(pipe.stream_label) elif pipe.source == PIPE_SOURCE_STREAM and pipe.x is not None and pipe.y is not None: mark = {"x": round(float(pipe.x), 3), "y": round(float(pipe.y), 3)} else: continue if mark not in dismissed: dismissed.append(mark) # 지운 자동 관을 [자동 제안]으로 다시 넣음(510) — 그 지움 표지는 걷음 held = [point.stream_label for point in points if point.stream_label] dismissed[:] = [mark for mark in dismissed if mark not in held] def propose_pipes( vertices: Sequence[RouteVertex], crossings: Sequence[Any], pipes: Sequence[Any], dismissed: Sequence[dict[str, Any]], strength_curve: Any = None, ) -> list[dict[str, Any]]: """[자동 제안] 후보(510) — 세류선 교차(`_base_pipes` · 라벨) + 간격 보충(흐름 세기 · 분석이 지금 노선일 때) · 관 목록엔 안 넣음(사용자가 골라 관 넣기 길로). 상태 = held(라벨 짝 관 · 교차 5 m 안 관) · dismissed(지움 표지 짝 · 라벨 = 그 표지) · new. 짝 잡기 = 노선 [저장](`add_crossing_pipes`)과 같은 `claim_crossings`(관 라벨 + 지움 표지 함께). 간격 보충 = 지금 관 + 교차 후보(새 · 지운 관) 사이 최대 간격 넘는 구간(suggest_pipe_positions). """ from common_util.common_util_drainage_detail import suggest_pipe_positions from common_util.common_util_route_geometry import interpolate_vertex from common_util.common_util_structure_station import snap_structure_station total = float(vertices[-1].chainage_m) held = [pipe.stream_label for pipe in pipes if pipe.stream_label] marks = [mark for mark in dismissed if "stream" in mark] claimed = claim_crossings([*held, *marks], crossings) labels = crossing_labels(crossings) taken = [float(pipe.chainage_m) for pipe in pipes] found: list[dict[str, Any]] = [] for index, crossing in enumerate(crossings): at = snap_structure_station(float(crossing.chainage_m), [total]) label, state = labels[index], "new" if index in claimed: pick = claimed[index] state = "held" if pick < len(held) else "dismissed" if state == "dismissed": label = marks[pick - len(held)] elif any(abs(other - at) < DRAINAGE_PIPE_MIN_SPACING_M for other in taken): state = "held" found.append( { "chainage_m": at, "kind": "stream", "state": state, "stream_label": dict(label), "stream_name": getattr(crossing, "stream_name", None), } ) if strength_curve is not None: base = sorted({*taken, *(item["chainage_m"] for item in found if item["state"] != "held")}) placed = [ StructureCandidate(chainage_m=c, x=xy[0], y=xy[1], reason="confirmed") for c in base for xy in [interpolate_vertex(list(vertices), c)] ] for pipe in suggest_pipe_positions(list(vertices), placed, strength_curve): found.append({"chainage_m": float(pipe.chainage_m), "kind": "spacing", "state": "new"}) return sorted(found, key=lambda item: item["chainage_m"])