"""세류선 교차점 · 자동 세류선 관 라벨(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]) -> set[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: set[int] = set() for _, li, ci in pairs: if li in used_labels or ci in used: continue used_labels.add(li) used.add(ci) 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)