eomsangdonandClaude Opus 5.5
29512c001b
- D02 걷음 — 길 · 단계(PLAN 단계 둘 = 노선 분석 · 확정) · [>] · 대시보드 단계 점 · 서버 길 · 폴더 만들기 · 캐시 등록 · 글 · 가이드 폴더 - 등록 [등록] · 받은 예정노선 [보기] = D03 으로 곧바로 - D03 지점 = 긋기에 붙는 점 · 목록 칸(여섯 색 · 「지점 N」 · 🗑) · 핀 끌기 = 그 자리 노선 노드 같이 · [저장] 몸에 실음 · [초기화] = 선만 지우고 지점 남김 - 지도 우클릭 메뉴(부품 상자) — 여기서 긋기 시작(빈 땅 = 새 본선) · 갈래 시작 · 이어 긋기 · 지점 놓기 · 지점 지우기 · 긋기 끝 - 받기 — 처음 = 등록 기준점 둘레 ± 1 km(지점 · 선 없을 때) · 지점 · 선 · 지도 보기 가운데가 받은 범위 밖이면 조용히 이어 받음(단추 · 막는 창 걷음 · 실패만 [다시 받기]) - 좌표계 = 파일 머리 · 없으면 기준점 경도 TM 띠 - 시험: 529_3(D02 지점 API) 지움 → 591_2 로 갈음 · 280_2 · 528 · 530_2 · 547_1 · 548 · 553_4 · 556_2 · 560_4 · 401 · 54_2 = D02 단계 · 페이지 걷음에 맞춤 · 590_1 · 530_2_analyze = 사업연도 · 기준점 한 번에 읽음 · 591_1 D04 씨앗 시험 = 590-1 이 씨앗 줄 걷음에 맞춤 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
396 lines
16 KiB
Python
396 lines
16 KiB
Python
"""D 예정노선 2단계 노선 정본 — 손으로 그은 직선 · 갈래 · 구역
|
|
(요구 389 6 · 11 · 13 · 14 · 391 · 452 · PLAN 562-2 · 590-1).
|
|
|
|
- `GET /api/projects/{id}/plan/network` — 설계 내용 파일(`D03_PlanAnalysis_Store`) + 서버가 다시 센
|
|
길이 · 측점 · 조각 · 파일이 없으면 옛 DB 칸을 옮겨 보임(쓰지 않음)
|
|
- `PUT …/plan/network` — [저장] · [초기화](빈 선) — 선 · 마크 · 조각 연도만 받아 서버가 정본으로
|
|
다시 셈 → 파일(원자 쓰기) · 연도 상태 줄(`plan_route_years` 지문 · 확정) · 사업연도 끝
|
|
|
|
길이 · 측점 · 높이는 저장하지 않음(늘 다시 셈 · 높이는 지형에서). 측점 = 선마다 No.0 부터.
|
|
조각 = 선의 마크 사이 구간(구역 × 갈래) · 업체 지정은 조각 단위(`plan_route_assignees.piece_id`).
|
|
지점(`points` · 긋기에 붙는 점) = 같은 파일 · [저장] 몸에 실림(빈 값 = 그대로) ·
|
|
기준점(`projects.site_*`) = 처음 자료 받는 기준일 뿐(노선 · 붙음 무관 · PLAN 591-2).
|
|
계산 자리 = ① 파이썬·TS 짝(`D03_PlanAnalysis_Draw`) — 화면은 그리는 즉시 · [저장]은 여기서 다시 셈.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
from common_util.common_util_auth import is_forest_service, require_company
|
|
from common_util.common_util_crs import tm_epsg_for_lon, tm_to_lonlat
|
|
from common_util.common_util_save_guard import project_lock
|
|
from config.config_db import get_db_pool
|
|
from D03_PlanAnalysis import D03_PlanAnalysis_Draw as draw
|
|
from D03_PlanAnalysis import D03_PlanAnalysis_Store as store
|
|
|
|
router = APIRouter(prefix="/api/projects", tags=["D03 Plan Network"])
|
|
|
|
ID = r"^[A-Za-z0-9_-]{1,64}$"
|
|
SAME_M = 0.001 # 1 mm 안 = 같은 점 · 같은 마크
|
|
PIECE_TOL_M = 0.01 # 조각 끝 = 마크 경계(1 cm)
|
|
STATION_M = 20.0 # 측점 간격(지식DB 설계제원_총괄)
|
|
MAX_POINTS = 40 # 지점 상한(옛 기준 지정 시점 20 + 종점 20)
|
|
|
|
|
|
class Branch(BaseModel):
|
|
line: str = Field(pattern=ID)
|
|
vertex: int = Field(ge=0)
|
|
|
|
|
|
class Line(BaseModel):
|
|
id: str = Field(pattern=ID)
|
|
vertices: list[tuple[float, float]] = Field(min_length=1)
|
|
branch: Branch | None = None
|
|
|
|
|
|
class Mark(BaseModel):
|
|
line: str = Field(pattern=ID)
|
|
at_m: float
|
|
|
|
|
|
class Piece(BaseModel):
|
|
id: str = Field(pattern=ID)
|
|
line: str = Field(pattern=ID)
|
|
from_m: float
|
|
to_m: float
|
|
year: int
|
|
|
|
|
|
class Network(BaseModel):
|
|
epsg: int
|
|
# 지점(작업 좌표계 m) — 빈 값(None) = 저장된 지점 그대로
|
|
points: list[tuple[float, float]] | None = Field(default=None, max_length=MAX_POINTS)
|
|
lines: list[Line] = []
|
|
marks: list[Mark] = []
|
|
pieces: list[Piece] = []
|
|
|
|
|
|
class NetworkError(ValueError):
|
|
pass
|
|
|
|
|
|
def _check_points(points: list[tuple[float, float]] | None) -> None:
|
|
if points is not None and not all(math.isfinite(v) for p in points for v in p):
|
|
raise HTTPException(status_code=400, detail="지점 좌표가 수가 아닙니다.")
|
|
|
|
|
|
def _check(ok: bool, message: str) -> None:
|
|
if not ok:
|
|
raise NetworkError(message)
|
|
|
|
|
|
def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, Any]:
|
|
"""선 · 마크 · 조각 → 선 길이 · 측점 · 조각(경계 = 마크 · 꺾은선 · 측점) ·
|
|
틀리면 `NetworkError`.
|
|
|
|
`year_from` = 사업연도 시작(그 앞 연도 조각 거절 · None = 안 봄)."""
|
|
lines: dict[str, list] = {}
|
|
for line in network["lines"]:
|
|
vertices = line["vertices"]
|
|
_check(line["id"] not in lines, f"선 id 가 겹칩니다: {line['id']}")
|
|
_check(
|
|
all(math.isfinite(v) for point in vertices for v in point),
|
|
f"선 {line['id']} 점이 수가 아닙니다.",
|
|
)
|
|
branch = line.get("branch")
|
|
if branch:
|
|
parent = lines.get(branch["line"])
|
|
_check(parent is not None, f"갈래 {line['id']} 의 부모 선이 앞에 없습니다.")
|
|
_check(branch["vertex"] < len(parent), f"갈래 {line['id']} 의 부모 점이 없습니다.")
|
|
_check(
|
|
draw.segment_length(vertices[0], parent[branch["vertex"]]) <= SAME_M,
|
|
f"갈래 {line['id']} 첫 점이 부모 선 점과 다릅니다.",
|
|
)
|
|
lines[line["id"]] = vertices
|
|
lengths = {key: draw.line_length(vertices) for key, vertices in lines.items()}
|
|
cuts: dict[str, list[float]] = {key: [] for key in lines}
|
|
for mark in network["marks"]:
|
|
_check(mark["line"] in lines, f"마크의 선이 없습니다: {mark['line']}")
|
|
at = mark["at_m"]
|
|
_check(
|
|
math.isfinite(at) and SAME_M < at < lengths[mark["line"]] - SAME_M,
|
|
f"마크가 선 {mark['line']} 밖입니다: {at}",
|
|
)
|
|
cuts[mark["line"]].append(at)
|
|
tiles: dict[str, list[tuple[float, float]]] = {}
|
|
for key, marks in cuts.items():
|
|
edges = [0.0, *sorted(marks), lengths[key]]
|
|
_check(
|
|
all(b - a > SAME_M for a, b in zip(edges[1:-1], edges[2:-1])),
|
|
f"선 {key} 마크가 겹칩니다.",
|
|
)
|
|
tiles[key] = list(zip(edges, edges[1:])) if lengths[key] > SAME_M else []
|
|
order = {key: i for i, key in enumerate(lines)}
|
|
stations = {key: draw.station_chainages(lengths[key], STATION_M) for key in lines}
|
|
used: set[tuple[str, int]] = set()
|
|
pieces = []
|
|
for piece in network["pieces"]:
|
|
key = piece["line"]
|
|
_check(key in lines, f"조각 {piece['id']} 의 선이 없습니다.")
|
|
_check(year_from is None or piece["year"] >= year_from, f"조각 연도가 사업연도 앞: {piece}")
|
|
hit = next(
|
|
(
|
|
i
|
|
for i, (a, b) in enumerate(tiles[key])
|
|
if abs(a - piece["from_m"]) <= PIECE_TOL_M and abs(b - piece["to_m"]) <= PIECE_TOL_M
|
|
),
|
|
None,
|
|
)
|
|
_check(hit is not None, f"조각 {piece['id']} 끝이 마크 경계와 다릅니다.")
|
|
_check((key, hit) not in used, f"조각이 겹칩니다: {piece['id']}")
|
|
used.add((key, hit))
|
|
a, b = tiles[key][hit]
|
|
pieces.append(
|
|
{
|
|
"id": piece["id"],
|
|
"line": key,
|
|
"year": int(piece["year"]),
|
|
"from_m": a,
|
|
"to_m": b,
|
|
"length_m": b - a,
|
|
"planned": [list(p) for p in draw.slice_line(lines[key], a, b)],
|
|
"stations": sorted(
|
|
{a, b, *(c for c in stations[key] if a - 1e-6 <= c <= b + 1e-6)}
|
|
),
|
|
}
|
|
)
|
|
_check(len({p["id"] for p in pieces}) == len(pieces), "조각 id 가 겹칩니다.")
|
|
pieces.sort(key=lambda p: (order[p["line"]], p["from_m"]))
|
|
return {
|
|
"lines": [
|
|
{"id": key, "length_m": lengths[key], "stations": stations[key]} for key in lines
|
|
],
|
|
"pieces": pieces,
|
|
}
|
|
|
|
|
|
def year_rows(pieces: list[dict[str, Any]]) -> dict[int, dict[str, Any]]:
|
|
"""조각 → 연도 묶음 — `planned` = 첫 조각 · `pieces` = 그 연도 전부 · `fingerprint` = 연도 상태
|
|
줄(`plan_route_years`) 지문."""
|
|
by_year: dict[int, list[dict[str, Any]]] = {}
|
|
for piece in pieces:
|
|
by_year.setdefault(piece["year"], []).append(piece)
|
|
out = {}
|
|
for year, items in sorted(by_year.items()):
|
|
first = items[0]
|
|
parts = [
|
|
{
|
|
"id": p["id"],
|
|
"line": p["line"],
|
|
"planned": p["planned"],
|
|
"start_chainage_m": p["from_m"],
|
|
"end_chainage_m": p["to_m"],
|
|
"length_m": p["length_m"],
|
|
}
|
|
for p in items
|
|
]
|
|
route = {
|
|
"mode": "manual",
|
|
"planned": first["planned"],
|
|
"start_chainage_m": first["from_m"],
|
|
"end_chainage_m": first["to_m"],
|
|
"length_m": sum(p["length_m"] for p in items),
|
|
"pieces": parts,
|
|
}
|
|
out[year] = {
|
|
"route": route,
|
|
"fingerprint": _sha1(parts),
|
|
"start": first["planned"][0],
|
|
"end": first["planned"][-1],
|
|
}
|
|
return out
|
|
|
|
|
|
def _sha1(value: Any) -> str:
|
|
return hashlib.sha1(json.dumps(value, sort_keys=True).encode()).hexdigest() # noqa: S324
|
|
|
|
|
|
def _json(value: Any) -> Any:
|
|
if isinstance(value, (str, bytes)):
|
|
try:
|
|
return json.loads(value)
|
|
except ValueError:
|
|
return None
|
|
return value
|
|
|
|
|
|
def legacy_network(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
|
|
"""옛 연도 노선(연도 차례 · 직전 종점 = 다음 시점) → 본선 하나 + 연도 경계 마크 + 연도 조각."""
|
|
vertices: list[list[float]] = []
|
|
pieces: list[dict[str, Any]] = []
|
|
epsg = None
|
|
for row in rows:
|
|
route = _json(row.get("route_json"))
|
|
planned = route.get("planned") if isinstance(route, dict) else None
|
|
if not planned or len(planned) < 2:
|
|
continue
|
|
epsg = epsg or row.get("epsg")
|
|
points = [[float(p[0]), float(p[1])] for p in planned]
|
|
if vertices and draw.segment_length(vertices[-1], points[0]) <= SAME_M:
|
|
points = points[1:]
|
|
start = draw.line_length(vertices)
|
|
vertices.extend(points)
|
|
end = draw.line_length(vertices)
|
|
pieces.append(
|
|
{
|
|
"id": f"y{row['year']}",
|
|
"line": "main",
|
|
"from_m": start,
|
|
"to_m": end,
|
|
"year": row["year"],
|
|
}
|
|
)
|
|
if not pieces:
|
|
return None
|
|
return {
|
|
"epsg": int(epsg) if epsg is not None else None,
|
|
"lines": [{"id": "main", "vertices": vertices, "branch": None}],
|
|
"marks": [{"line": "main", "at_m": p["to_m"]} for p in pieces[:-1]],
|
|
"pieces": pieces,
|
|
}
|
|
|
|
|
|
def work_epsg(plan: dict[str, Any], site: dict[str, Any] | None) -> int | None:
|
|
"""작업 좌표계 — 파일 머리 · 없으면 기준점 경도의 TM 띠(첫 [저장] · 지형 받기 때 굳음)."""
|
|
if plan.get("epsg") is not None:
|
|
return int(plan["epsg"])
|
|
return tm_epsg_for_lon(site["lon"]) if site else None
|
|
|
|
|
|
def base_view(plan: dict[str, Any], site: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""{epsg, points: [{lon, lat, x, y}], site} — 지점(TM) + 경위도 · 기준점(경위도 · 주소)."""
|
|
epsg = work_epsg(plan, site)
|
|
points = [
|
|
{"lon": lon, "lat": lat, "x": float(p["x"]), "y": float(p["y"])}
|
|
for p in plan["points"]
|
|
if epsg is not None
|
|
for lon, lat in [tm_to_lonlat(float(p["x"]), float(p["y"]), epsg)]
|
|
]
|
|
return {"epsg": epsg, "points": points, "site": site}
|
|
|
|
|
|
def _view(plan: dict[str, Any], project: dict[str, Any]) -> dict[str, Any]:
|
|
network = {key: plan[key] for key in ("epsg", "lines", "marks", "pieces")}
|
|
return {
|
|
"status": "success",
|
|
"base": base_view(plan, project["site"]),
|
|
"year_start": project["year_start"],
|
|
"network": network if plan["lines"] else None,
|
|
"derived": derive(network) if plan["lines"] else None,
|
|
"legacy": False,
|
|
}
|
|
|
|
|
|
async def project_info(project_id: UUID) -> dict[str, Any]:
|
|
"""{year_start, site: {lon, lat, address} | None} — 사업연도 시작 · 등록 기준점(591-1)."""
|
|
async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT project_year, site_lon, site_lat, site_address FROM projects WHERE id = %s",
|
|
(str(project_id),),
|
|
)
|
|
row = await cursor.fetchone()
|
|
if not row:
|
|
return {"year_start": None, "site": None}
|
|
year, lon, lat, address = row
|
|
site = (
|
|
{"lon": float(lon), "lat": float(lat), "address": address}
|
|
if lon is not None and lat is not None
|
|
else None
|
|
)
|
|
return {"year_start": year, "site": site}
|
|
|
|
|
|
@router.get("/{project_id}/plan/network", response_model=None)
|
|
async def get_plan_network(project_id: UUID) -> dict[str, Any]:
|
|
_, plan = await store.load_plan(project_id)
|
|
try:
|
|
return _view(plan, await project_info(project_id))
|
|
except NetworkError as error:
|
|
raise HTTPException(status_code=409, detail=f"저장된 노선이 틀립니다 — {error}") from error
|
|
|
|
|
|
async def _store_status(
|
|
connection: Any, project_id: UUID, pieces: list[dict[str, Any]], years: dict[int, Any]
|
|
) -> None:
|
|
"""연도 상태 줄 — 있는 연도만(지문이 바뀐 연도만 확정 풂) · 없는 연도 줄 지움 · 사업연도 끝."""
|
|
pid = str(project_id)
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"SELECT DISTINCT piece_id FROM plan_route_assignees"
|
|
" WHERE project_id = %s AND piece_id IS NOT NULL",
|
|
(pid,),
|
|
)
|
|
lost = sorted({r[0] for r in await cursor.fetchall()} - {p["id"] for p in pieces})
|
|
if lost:
|
|
raise HTTPException(
|
|
status_code=409, detail=f"업체가 지정된 조각을 지울 수 없습니다: {lost}"
|
|
)
|
|
for year, row in years.items():
|
|
# confirmed_at 을 fingerprint 보다 먼저 셈
|
|
await cursor.execute(
|
|
"INSERT INTO plan_route_years (project_id, year, fingerprint) VALUES (%s, %s, %s)"
|
|
" ON DUPLICATE KEY UPDATE"
|
|
" confirmed_at = IF(fingerprint <=> VALUES(fingerprint), confirmed_at, NULL),"
|
|
" fingerprint = VALUES(fingerprint)",
|
|
(pid, year, row["fingerprint"]),
|
|
)
|
|
marks = ", ".join(["%s"] * len(years))
|
|
await cursor.execute(
|
|
"DELETE FROM plan_route_years WHERE project_id = %s"
|
|
+ (f" AND year NOT IN ({marks})" if years else ""),
|
|
(pid, *years),
|
|
)
|
|
if years:
|
|
await cursor.execute(
|
|
"UPDATE projects SET project_year_end = %s WHERE id = %s", (max(years), pid)
|
|
)
|
|
|
|
|
|
@router.put("/{project_id}/plan/network", response_model=None)
|
|
async def put_plan_network(
|
|
project_id: UUID,
|
|
payload: Network,
|
|
session: dict[str, Any] = Depends(require_company),
|
|
) -> dict[str, Any]:
|
|
"""[저장] · [초기화](lines 빈 것) — 서버가 정본으로 다시 셈 → 파일 · 연도 상태 줄.
|
|
|
|
파일은 DB 커밋 직전에 씀 — 파일 쓰기가 깨지면 DB 되돌림 · 커밋이 깨지면 다음 [저장]이 바로잡음
|
|
(확정은 지문이 같을 때만 보이므로 노선이 틀리게 확정으로 보이지 않음)."""
|
|
if not is_forest_service(session):
|
|
raise HTTPException(status_code=403, detail="예정노선은 산림청만 고칠 수 있습니다.")
|
|
network = payload.model_dump()
|
|
points = network.pop("points")
|
|
async with project_lock(str(project_id)):
|
|
root, plan = await store.load_plan(project_id) # 없음 404 · 예정노선 아님 400
|
|
if plan["epsg"] is not None and plan["epsg"] != network["epsg"]:
|
|
raise HTTPException(status_code=400, detail="좌표계가 저장된 노선과 다릅니다.")
|
|
project = await project_info(project_id)
|
|
try:
|
|
derived = derive(network, project["year_start"])
|
|
except NetworkError as error:
|
|
raise HTTPException(status_code=400, detail=str(error)) from error
|
|
_check_points(points)
|
|
plan = {**plan, **network}
|
|
if points is not None:
|
|
plan["points"] = [{"x": float(x), "y": float(y)} for x, y in points]
|
|
async with get_db_pool().acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
await _store_status(
|
|
connection, project_id, derived["pieces"], year_rows(derived["pieces"])
|
|
)
|
|
store.save_plan(root, plan)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return {**_view(plan, project), "derived": derived}
|