fix(507-2 B05 관): 자동 세류선 관 라벨 · 노선 바꾼 [저장] 관 겹 생성 막음 · 지운 자동 관 지움 표지 — 요구 311-3
- 관 라벨 stream_label = 세류선 UFID · 교차 차례 · 처음 교차점 좌표 · 끌기 · 옵션 고침에도 남음 · 관 PUT 때 id 로 되붙임 - 노선 [저장] = 라벨(관 · 지움 표지)이 같은 세류선 가장 가까운 새 교차점을 짝으로 잡음 · 짝 잡힌 교차엔 관 안 더함 - 지운 자동 관 = pipe_points.json dismissed_streams · 라벨 없던 옛 관은 자리만 → 다음 노선 [저장] 때 라벨 - 라벨 없는 옛 stream 관 = 옛 노선 누가거리 5 m 안 교차 라벨 · 겹친 관 안 지움(로그 수) - 추천 [적용] = 표지 없음 · [초기화] = 초기값 파일 그대로 - 세류선 교차 셈을 common_util_stream_crossing 으로 옮김(common_util_route_geometry 700줄 넘음) - 시험 test_507_2_stream_pipe_label · test_494_2 가짜 저장 인자 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -54,11 +54,8 @@ from B04_PreProcess.B04_PreProcess_Engine_Watershed_Stream import (
|
||||
PrimaryRegion,
|
||||
build_primary_region,
|
||||
)
|
||||
from common_util.common_util_route_geometry import (
|
||||
RouteVertex,
|
||||
StructureCandidate,
|
||||
find_stream_crossings,
|
||||
)
|
||||
from common_util.common_util_route_geometry import RouteVertex, StructureCandidate
|
||||
from common_util.common_util_stream_crossing import find_stream_crossings
|
||||
from config.config_system import (
|
||||
DRAINAGE_ARROW_BLOCK_M,
|
||||
DRAINAGE_ARROW_MIN_AGREEMENT,
|
||||
|
||||
@@ -47,13 +47,17 @@ from common_util.common_util_drainage_pipes import (
|
||||
clear_pipe_points,
|
||||
load_pipe_points,
|
||||
parse_pipe_points,
|
||||
pipe_points_path,
|
||||
pipe_signature,
|
||||
read_dismissed,
|
||||
read_pipe_points_file,
|
||||
route_signature,
|
||||
save_detail_basins,
|
||||
save_pipe_points,
|
||||
)
|
||||
from common_util.common_util_json import LONLAT_DIGITS, round_floats
|
||||
from common_util.common_util_route_geometry import StructureCandidate
|
||||
from common_util.common_util_stream_crossing import keep_labels
|
||||
from config.config_system import (
|
||||
DRAINAGE_PIPE_MAX_SPACING_M,
|
||||
DRAINAGE_PIPE_MIN_SPACING_M,
|
||||
@@ -611,9 +615,13 @@ async def put_pipe_points(
|
||||
context, detail, points, _, stale = resolved
|
||||
|
||||
signature = route_signature(context.vertices)
|
||||
# 자동 세류선 관 라벨 되붙임 · 지운 자동 관 = 지움 표지(507-2)
|
||||
path = pipe_points_path(context.stored_path)
|
||||
dismissed = read_dismissed(path)
|
||||
keep_labels(points, read_pipe_points_file(path), dismissed)
|
||||
# 좌표를 같이 남긴다 — 다른 선(B05 최적 경로)으로 읽어도 그 자리에 되놓는다.
|
||||
saved = await asyncio.to_thread(
|
||||
save_pipe_points, context.stored_path, signature, points, context.vertices
|
||||
save_pipe_points, context.stored_path, signature, points, context.vertices, dismissed
|
||||
)
|
||||
if not stale: # 낡은 분석 — 관 자리 · 속성만 정본으로 · 세부 유역 산출물은 안 씀
|
||||
await asyncio.to_thread(
|
||||
|
||||
@@ -171,19 +171,29 @@ def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]:
|
||||
|
||||
|
||||
def followed_pipes(
|
||||
old_pipes: list, old_xy: list[list[float]], points: list[dict[str, float]], streams: list
|
||||
old_pipes: list,
|
||||
old_xy: list[list[float]],
|
||||
points: list[dict[str, float]],
|
||||
streams: list,
|
||||
dismissed: list | None = None,
|
||||
) -> tuple[list, int, list]:
|
||||
"""관 따라감 + 새 세류선 교차점 기본 관(저장 안 함) → (짝, 교차 수, 새 노선 정점) — 체인 ·
|
||||
추천 초기 종단(476) 한 벌."""
|
||||
추천 초기 종단(476) 한 벌. `dismissed` = 지움 표지(507-2 · 옛 노선 라벨 붙임을 제자리에)."""
|
||||
from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes
|
||||
from B04_PreProcess.B04_PreProcess_Router_Basins import _apply_new_defaults
|
||||
from common_util.common_util_drainage_pipes import fill_pipe_coordinates
|
||||
from common_util.common_util_stream_crossing import adopt_labels
|
||||
|
||||
vertices = vertices_of([(p["x"], p["y"]) for p in points])
|
||||
fill_pipe_coordinates(old_pipes, vertices_of(old_xy))
|
||||
held = [] if dismissed is None else dismissed
|
||||
if len(old_xy) >= 2: # 라벨 없는 옛 자동 관 · 지움 자리 = 옛 노선 교차 라벨(507-2)
|
||||
overlap = adopt_labels(old_pipes, held, _base_pipes(vertices_of(old_xy), streams), old_xy)
|
||||
if overlap:
|
||||
logger.warning("관 라벨: 세류선 교차 5 m 안 겹친 자동 관 %d개(안 지움)", overlap)
|
||||
pairs = follow_pipes(old_pipes, vertices)
|
||||
total = float(vertices[-1].chainage_m)
|
||||
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total)
|
||||
crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total, held)
|
||||
# 새 교차 관만 B04 와 같은 등록부 기본값(54-39 · 494-2) — 옛 관 빈칸은 미확정 그대로
|
||||
_apply_new_defaults([pipe for index, pipe in pairs if index < 0])
|
||||
return pairs, crossing, vertices
|
||||
@@ -204,6 +214,7 @@ async def _place_pipes(
|
||||
old_pipes: list,
|
||||
old_xy: list[list[float]],
|
||||
points: list[dict[str, float]],
|
||||
dismissed: list | None = None,
|
||||
) -> tuple[list, dict[str, int]]:
|
||||
"""관 따라감 → 새 교차점 기본 관 → 정본 저장(새 노선 지문) · 간격 보충 없음(369).
|
||||
초기화(492)면 부르는 쪽이 옛 관 · 옛 선을 빈 목록으로 줌 = 새 노선 교차 관만.
|
||||
@@ -218,8 +229,9 @@ async def _place_pipes(
|
||||
|
||||
prepared = await _prepare(project_id)
|
||||
streams = [] if isinstance(prepared, JSONResponse) else prepared["streams"] # 도엽 없음
|
||||
held = [] if dismissed is None else dismissed
|
||||
pairs, crossing, vertices = await asyncio.to_thread(
|
||||
followed_pipes, old_pipes, old_xy, points, streams
|
||||
followed_pipes, old_pipes, old_xy, points, streams, held
|
||||
)
|
||||
await asyncio.to_thread(
|
||||
save_pipe_points,
|
||||
@@ -227,6 +239,7 @@ async def _place_pipes(
|
||||
route_signature(vertices),
|
||||
[pipe for _, pipe in pairs],
|
||||
vertices,
|
||||
held,
|
||||
)
|
||||
line = [(p["x"], p["y"]) for p in points]
|
||||
outside = sum(route_outside(pipe.x, pipe.y, line) for _, pipe in pairs) # 459 · 측점 안 세움
|
||||
@@ -454,7 +467,11 @@ async def run_route_save_chain(
|
||||
from B06_Section.B06_Section_Router_Confirm import confirm_sections
|
||||
from B06_Section.B06_Section_Router_Design import stored_standard_cross_section
|
||||
from B06_Section.B06_Section_Schema import SectionConfirmRequest
|
||||
from common_util.common_util_drainage_pipes import pipe_points_path, read_pipe_points_file
|
||||
from common_util.common_util_drainage_pipes import (
|
||||
pipe_points_path,
|
||||
read_dismissed,
|
||||
read_pipe_points_file,
|
||||
)
|
||||
from common_util.common_util_initial_snapshot import clear_designing, mark_designing
|
||||
from common_util.common_util_workflow_state import get_workflow_state
|
||||
|
||||
@@ -511,8 +528,14 @@ async def run_route_save_chain(
|
||||
marks.append(("노선 풀기 · 구조물 따라감", time.perf_counter()))
|
||||
|
||||
# 초기화 = 옛 관(옵션 · 사용자 관) 버림 · 새 노선 세류선 교차 관만(492)
|
||||
# 지움 표지도 — [적용] = 새 초기값(표지 없음 · 507-2)
|
||||
pairs, pipes = await _place_pipes(
|
||||
project_id, stored_path, [] if initial else old_pipes, [] if initial else old_xy, points
|
||||
project_id,
|
||||
stored_path,
|
||||
[] if initial else old_pipes,
|
||||
[] if initial else old_xy,
|
||||
points,
|
||||
[] if initial else read_dismissed(pipe_points_path(stored_path)),
|
||||
)
|
||||
marks.append(("관 따라감 · 교차점", time.perf_counter()))
|
||||
|
||||
|
||||
@@ -6,7 +6,9 @@ DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름
|
||||
|
||||
- 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음
|
||||
(사용자 관 편집 지킴).
|
||||
- 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음).
|
||||
- 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음) ·
|
||||
같은 세류선 라벨 관(옮겼어도) · 지움 표지가 짝으로 잡은 교차점엔 안 더함(507-2 ·
|
||||
`common_util_stream_crossing`).
|
||||
- 관 최대 간격 보충 없음(369) — 간격 보충은 배수유역 [자동 제안]만.
|
||||
- 측점 열쇠 = `common_util_station_key.station_key` 한 벌(sub2 짝) — 일반 `grid:k` ·
|
||||
관 `pipe:<id>` · 구조물 `structure:<id>` · 시종점 `end:BP` · `end:EP` ·
|
||||
@@ -38,6 +40,7 @@ from common_util.common_util_station_key import (
|
||||
station_key,
|
||||
station_seat,
|
||||
)
|
||||
from common_util.common_util_stream_crossing import claim_crossings, crossing_labels
|
||||
from common_util.common_util_structure_station import plan_stations, snap_structure_station
|
||||
from config.config_system import DRAINAGE_PIPE_MIN_SPACING_M
|
||||
|
||||
@@ -78,18 +81,33 @@ def add_crossing_pipes(
|
||||
crossings: Sequence[Any],
|
||||
old_xy: Sequence[Sequence[float]] | None,
|
||||
total_m: float,
|
||||
dismissed: Sequence[dict[str, Any]] = (),
|
||||
) -> int:
|
||||
"""옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관을 더함 · 더한 수."""
|
||||
"""옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관(새 라벨)을 더함 · 더한 수."""
|
||||
old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
|
||||
held = [pipe.stream_label for _, pipe in pairs if pipe.stream_label]
|
||||
claimed = claim_crossings([*held, *(mark for mark in dismissed if "stream" in mark)], crossings)
|
||||
labels = crossing_labels(crossings)
|
||||
added = 0
|
||||
for crossing in crossings:
|
||||
for index, crossing in enumerate(crossings):
|
||||
if index in claimed:
|
||||
continue # 같은 세류선 라벨 관(옮겼어도) · 지운 자동 관(507-2)
|
||||
if not _moved_away(old_line, crossing.x, crossing.y):
|
||||
continue # 옛 교차점 — 사용자가 지운 관을 되살리지 않음
|
||||
at = snap_structure_station(float(crossing.chainage_m), [total_m])
|
||||
if any(abs(pipe.chainage_m - at) < DRAINAGE_PIPE_MIN_SPACING_M for _, pipe in pairs):
|
||||
continue
|
||||
pairs.append(
|
||||
(-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_STREAM, x=crossing.x, y=crossing.y))
|
||||
(
|
||||
-1,
|
||||
PipePoint(
|
||||
chainage_m=at,
|
||||
source=PIPE_SOURCE_STREAM,
|
||||
x=crossing.x,
|
||||
y=crossing.y,
|
||||
stream_label=labels[index],
|
||||
),
|
||||
)
|
||||
)
|
||||
added += 1
|
||||
pairs.sort(key=lambda pair: pair[1].chainage_m)
|
||||
|
||||
@@ -105,6 +105,9 @@ class PipePoint:
|
||||
# 관 고정 id(347 덧) — 측점 열쇠 `pipe:<id>`(common_util_station_key) · 옮겨도 · 노선이
|
||||
# 바뀌어도 그대로 · 빈 id 는 `ensure_pipe_ids` 가 읽기 · 되붙이기 · 저장 때 채움.
|
||||
id: str | None = None
|
||||
# 자동 세류선 관 라벨(507-2 · `common_util_stream_crossing`) `{stream, order, x, y}` — 옮기고
|
||||
# 고쳐도 남음 · 화면은 모름(저장 때 id 로 되붙임)
|
||||
stream_label: dict[str, Any] | None = None
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
# 구 형식 저장분이 확장 필드 없이 그대로 다시 저장되도록 기본값은 생략한다.
|
||||
@@ -132,6 +135,8 @@ class PipePoint:
|
||||
payload["summary_item"] = self.summary_item
|
||||
if self.group_items:
|
||||
payload["group_items"] = self.group_items
|
||||
if self.stream_label:
|
||||
payload["stream_label"] = self.stream_label
|
||||
return payload
|
||||
|
||||
|
||||
@@ -437,6 +442,11 @@ def parse_pipe_points(values: Any) -> list[PipePoint]:
|
||||
summary_item=parse_summary_item(item.get("summary_item")),
|
||||
group_items=parse_group_items(item.get("group_items")),
|
||||
id=str(item["id"]) if item.get("id") else None,
|
||||
stream_label=(
|
||||
dict(item["stream_label"])
|
||||
if isinstance(item.get("stream_label"), dict)
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
points.sort(key=lambda point: point.chainage_m)
|
||||
@@ -500,9 +510,12 @@ def save_pipe_points(
|
||||
signature: str,
|
||||
points: list[PipePoint],
|
||||
vertices: list[RouteVertex] | None = None,
|
||||
dismissed: list[dict[str, Any]] | None = None,
|
||||
) -> int:
|
||||
"""관 지점을 정본 파일에 쓴다. 저장된 개수를 돌려준다.
|
||||
|
||||
`dismissed` = 지운 자동 관 지움 표지(507-2) — None 이면 파일의 것 그대로.
|
||||
|
||||
`vertices`를 주면 좌표가 빈 관을 그 노선 위 좌표로 채워 둔다 — 다음에 다른 선으로
|
||||
읽어도 그 자리에 되놓을 수 있다. 좌표가 그 누가거리를 안 낳으면(누가거리만 옮긴 낡은 좌표)
|
||||
노선 위로 · 낳으면 땅 위 자리 그대로(493-4 · 구조물 `fill_structure_coordinates` · TS
|
||||
@@ -516,17 +529,29 @@ def save_pipe_points(
|
||||
ensure_pipe_ids(points)
|
||||
path = pipe_points_path(stored_path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
atomic_write_json(
|
||||
path,
|
||||
{
|
||||
"route_signature": signature,
|
||||
"points": [point.as_dict() for point in points],
|
||||
},
|
||||
)
|
||||
document: dict[str, Any] = {
|
||||
"route_signature": signature,
|
||||
"points": [point.as_dict() for point in points],
|
||||
}
|
||||
marks = read_dismissed(path) if dismissed is None else dismissed
|
||||
if marks:
|
||||
document["dismissed_streams"] = marks
|
||||
atomic_write_json(path, document)
|
||||
logger.info("배수유역: 관 지점 %d개를 저장했습니다 (%s).", len(points), path.name)
|
||||
return len(points)
|
||||
|
||||
|
||||
def read_dismissed(path: Path) -> list[dict[str, Any]]:
|
||||
"""지운 자동 관 지움 표지(507-2 `dismissed_streams`) — 없거나 깨졌으면 빈 목록."""
|
||||
try:
|
||||
marks = json.loads(path.read_text(encoding="utf-8")).get("dismissed_streams")
|
||||
except (OSError, ValueError, AttributeError):
|
||||
return []
|
||||
return (
|
||||
[dict(mark) for mark in marks if isinstance(mark, dict)] if isinstance(marks, list) else []
|
||||
)
|
||||
|
||||
|
||||
def read_pipe_points_file(path: Path) -> list[PipePoint]:
|
||||
"""저장분을 **노선 지문과 무관하게** 그대로 읽는다. 파일이 없거나 깨졌으면 빈 목록.
|
||||
|
||||
@@ -564,6 +589,7 @@ def append_pipe_points_file(path: Path, points: list[PipePoint]) -> int | None:
|
||||
atomic_write_json(
|
||||
path,
|
||||
{
|
||||
**document, # 지움 표지(507-2) 그대로
|
||||
"route_signature": str(document.get("route_signature") or ""),
|
||||
"points": [*stored, *(point.as_dict() for point in points)],
|
||||
},
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""계획 노선 기하 공용 유틸 — 정점·누가거리·세류 교차점.
|
||||
"""계획 노선 기하 공용 유틸 — 정점·누가거리(세류 교차점 = `common_util_stream_crossing`).
|
||||
|
||||
배수유역 분석(B04)과 관 편집·세부유역(B05)이 같은 노선 표현을 써야 하므로 여기 한 곳에만
|
||||
정의한다. 어느 한쪽 페이지 폴더에 두면 반대 방향 import가 생긴다.
|
||||
@@ -19,7 +19,7 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from shapely.geometry import LineString, Point, shape
|
||||
from shapely.geometry import LineString
|
||||
|
||||
from config.config_system import SURFACE_ROUTE_EDGE_TRIM_M
|
||||
|
||||
@@ -53,6 +53,8 @@ class StructureCandidate:
|
||||
# "stream"=세류 교차, "spacing"=최대 간격 규칙 보충, "confirmed"=사용자 확정
|
||||
reason: str
|
||||
stream_name: str | None = None
|
||||
# 세류선 지물 id(UFID · 507-2 관 라벨) — 없으면 None
|
||||
stream_id: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -623,67 +625,6 @@ def is_uphill_at(vertices: list[RouteVertex], chainage_m: float, window_m: float
|
||||
return forward_z >= back_z
|
||||
|
||||
|
||||
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 feature in 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,
|
||||
)
|
||||
)
|
||||
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 []
|
||||
|
||||
|
||||
def _first_text(row: dict[str, Any], keys: tuple[str, ...]) -> str | None:
|
||||
for key in keys:
|
||||
value = row.get(key)
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
"""세류선 교차점 · 자동 세류선 관 라벨(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)
|
||||
@@ -81,7 +81,9 @@ def test_추천_적용_체인_관_저장(monkeypatch):
|
||||
|
||||
monkeypatch.setattr(watershed, "_prepare", prepare)
|
||||
monkeypatch.setattr(
|
||||
drainage_pipes, "save_pipe_points", lambda path, sig, pipes, verts: written.extend(pipes)
|
||||
drainage_pipes,
|
||||
"save_pipe_points",
|
||||
lambda path, sig, pipes, verts, *_: written.extend(pipes),
|
||||
)
|
||||
pairs, counts = asyncio.run(save._place_pipes(None, "unused", [], [], _points(NEW)))
|
||||
assert counts["crossing"] == 1
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""507-2 자동 세류선 관 라벨 — 노선 바꾼 [저장] 에 관이 여럿 생기던 흠(요구 311-3).
|
||||
|
||||
못박는 것
|
||||
- 라벨 = 세류선 지물 id · 교차 차례 · 처음 교차점 좌표 · 끌어 source 가 user 로 바뀌어도 남음 ·
|
||||
파일 · 화면 PUT(라벨 모름)을 지나도 id 로 되붙음.
|
||||
- 노선 바꾼 [저장] 두 번 = 같은 세류선 라벨 관이 있으면(옮겼어도) 안 더함 · 관 수 그대로.
|
||||
- 지운 자동 관 = 지움 표지 · 다시 안 생김(라벨 없던 옛 관도 자리로 남겨 다음 [저장] 때 라벨로).
|
||||
- 초기값([적용]) = 지움 표지 없음 · 새 교차 관 · save_pipe_points 표지 None = 파일 것 그대로.
|
||||
- 라벨 없는 옛 stream 관 = 옛 노선 교차 5 m 안이면 라벨 · 겹친 관은 안 지우고 수만 셈.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
if str(ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
import B05_Profile.B05_Profile_Route_Save as save # noqa: E402
|
||||
import common_util.common_util_drainage_pipes as drainage_pipes # noqa: E402
|
||||
from B05_Profile.B05_Profile_Route_Save_Parts import vertices_of # noqa: E402
|
||||
from common_util.common_util_drainage_pipes import ( # noqa: E402
|
||||
PIPE_SOURCE_STREAM,
|
||||
PipePoint,
|
||||
parse_pipe_points,
|
||||
)
|
||||
from common_util.common_util_stream_crossing import ( # noqa: E402
|
||||
adopt_labels,
|
||||
find_stream_crossings,
|
||||
keep_labels,
|
||||
)
|
||||
|
||||
OLD = [[0.0, 0.0], [100.0, 0.0], [200.0, 0.0], [300.0, 0.0], [400.0, 0.0]]
|
||||
# 앞 측점 노선을 북으로 — 세류선 교차가 30 m · 45 m 옮음(5 m 솎기 넘음)
|
||||
NEW1 = [[0.0, 0.0], [100.0, 0.0], [200.0, 60.0], [300.0, 0.0], [400.0, 0.0]]
|
||||
NEW2 = [[0.0, 0.0], [100.0, 0.0], [200.0, 90.0], [300.0, 0.0], [400.0, 0.0]]
|
||||
STREAMS = [
|
||||
{
|
||||
"type": "Feature",
|
||||
"properties": {"UFID": "S1", "구분": "세류"},
|
||||
"geometry": {"type": "LineString", "coordinates": [[150.0, -50.0], [150.0, 120.0]]},
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def _points(xy) -> list[dict[str, float]]:
|
||||
return [{"x": x, "y": y} for x, y in xy]
|
||||
|
||||
|
||||
def _save(old: list[PipePoint], old_xy, new_xy, dismissed=None) -> tuple[list[PipePoint], int]:
|
||||
pairs, crossing, _ = save.followed_pipes(old, old_xy, _points(new_xy), STREAMS, dismissed)
|
||||
return [pipe for _, pipe in pairs], crossing
|
||||
|
||||
|
||||
def _legacy() -> PipePoint:
|
||||
"""라벨 없는 옛 자동 관(옛 노선 교차 (150, 0))."""
|
||||
return PipePoint(chainage_m=150.0, source=PIPE_SOURCE_STREAM, x=150.0, y=0.0, id="p15000")
|
||||
|
||||
|
||||
def test_교차점에_세류선_id():
|
||||
found = find_stream_crossings(vertices_of(OLD), STREAMS)
|
||||
assert [(c.stream_id, round(c.x), round(c.y)) for c in found] == [("S1", 150, 0)]
|
||||
|
||||
|
||||
def test_노선_바꾼_저장_두_번_관_수_그대로():
|
||||
pipes, crossing = _save([_legacy()], OLD, NEW1)
|
||||
assert crossing == 0 and len(pipes) == 1 # 고치기 전 = 옛 관 + 새 관 둘
|
||||
label = pipes[0].stream_label
|
||||
assert label == {"stream": "S1", "order": 1, "x": 150.0, "y": 0.0} # 처음 교차점
|
||||
# 파일을 지나도 라벨 그대로 · 두 번째 노선 [저장]
|
||||
again = parse_pipe_points([pipe.as_dict() for pipe in pipes])
|
||||
pipes2, crossing2 = _save(again, NEW1, NEW2)
|
||||
assert crossing2 == 0 and len(pipes2) == 1 and pipes2[0].stream_label == label
|
||||
|
||||
|
||||
def test_끌어_user_로_바뀌어도_라벨_남음():
|
||||
pipes, _ = _save([_legacy()], OLD, NEW1)
|
||||
moved = pipes[0]
|
||||
moved.source, moved.x, moved.y = "user", 170.0, 40.0 # 끌기 = user
|
||||
pipes2, crossing2 = _save(parse_pipe_points([moved.as_dict()]), NEW1, NEW2)
|
||||
assert crossing2 == 0 and len(pipes2) == 1 and pipes2[0].stream_label["stream"] == "S1"
|
||||
|
||||
|
||||
def test_화면_PUT_라벨_되붙임():
|
||||
stored = [
|
||||
PipePoint(chainage_m=150.0, id="p15000", stream_label={"stream": "S1", "x": 1, "y": 2})
|
||||
]
|
||||
request = [PipePoint(chainage_m=152.0, id="p15000", source="user")] # 화면은 라벨 모름
|
||||
dismissed: list = []
|
||||
keep_labels(request, stored, dismissed)
|
||||
assert request[0].stream_label == stored[0].stream_label and dismissed == []
|
||||
|
||||
|
||||
def test_지운_자동_관_다시_안_생김():
|
||||
pipes, _ = _save([_legacy()], OLD, NEW1)
|
||||
dismissed: list = []
|
||||
keep_labels([], pipes, dismissed) # 사용자가 지우고 [저장]
|
||||
assert dismissed == [pipes[0].stream_label]
|
||||
after, crossing = _save([], NEW1, NEW2, dismissed)
|
||||
assert crossing == 0 and after == []
|
||||
|
||||
|
||||
def test_라벨_없던_옛_관_지움도_다시_안_생김():
|
||||
dismissed: list = []
|
||||
keep_labels([], [_legacy()], dismissed)
|
||||
assert dismissed == [{"x": 150.0, "y": 0.0}] # 자리만
|
||||
after, crossing = _save([], OLD, NEW1, dismissed)
|
||||
assert crossing == 0 and after == []
|
||||
assert dismissed[0]["stream"] == "S1" # 다음 [저장] 때 라벨로
|
||||
|
||||
|
||||
def test_사용자_관_지움은_표지_없음():
|
||||
dismissed: list = []
|
||||
keep_labels([], [PipePoint(chainage_m=50.0, x=50.0, y=0.0)], dismissed)
|
||||
assert dismissed == []
|
||||
|
||||
|
||||
def test_초기값_적용_새_교차_관():
|
||||
pipes, crossing = _save([], [], NEW1, [])
|
||||
assert crossing == 1 and pipes[0].stream_label["stream"] == "S1"
|
||||
assert pipes[0].stream_label["y"] == 30.0
|
||||
|
||||
|
||||
def test_지움_표지_파일(monkeypatch, tmp_path):
|
||||
path = tmp_path / "pipe_points.json"
|
||||
monkeypatch.setattr(drainage_pipes, "pipe_points_path", lambda _stored: path)
|
||||
mark = [{"stream": "S1", "order": 1, "x": 150.0, "y": 0.0}]
|
||||
drainage_pipes.save_pipe_points("x", "sig", [], None, mark)
|
||||
drainage_pipes.save_pipe_points("x", "sig", [PipePoint(chainage_m=5.0)]) # None = 그대로
|
||||
assert json.loads(path.read_text("utf-8"))["dismissed_streams"] == mark
|
||||
assert drainage_pipes.read_dismissed(path) == mark
|
||||
drainage_pipes.save_pipe_points("x", "sig", [], None, []) # [적용] = 표지 없음
|
||||
assert "dismissed_streams" not in json.loads(path.read_text("utf-8"))
|
||||
|
||||
|
||||
def test_겹친_옛_관_안_지우고_셈():
|
||||
twin = PipePoint(chainage_m=152.0, source=PIPE_SOURCE_STREAM, x=152.0, y=0.0, id="p15200")
|
||||
old = [_legacy(), twin]
|
||||
overlap = adopt_labels(old, [], find_stream_crossings(vertices_of(OLD), STREAMS), OLD)
|
||||
assert overlap == 1
|
||||
assert old[0].stream_label and not old[1].stream_label # 가까운 관만 라벨
|
||||
pipes, crossing = _save(parse_pipe_points([p.as_dict() for p in old]), OLD, NEW1)
|
||||
assert crossing == 0 and len(pipes) == 2 # 둘 다 그대로 · 새 관 없음
|
||||
|
||||
|
||||
def test_옆_다른_선_위_옛_관도_라벨():
|
||||
"""옛 관 좌표가 다른 선 위(옆 15 m · 검증_sub3 실측) — 옛 노선 누가거리로 5 m 안이면 라벨."""
|
||||
side = PipePoint(chainage_m=150.0, source=PIPE_SOURCE_STREAM, x=151.0, y=15.0, id="p15000")
|
||||
pipes, crossing = _save([side], OLD, NEW1)
|
||||
assert crossing == 0 and len(pipes) == 1 and pipes[0].stream_label["stream"] == "S1"
|
||||
Reference in New Issue
Block a user