eomsangdonandClaude Opus 5.5
bf77b70be9
- 파일 points [{x, y, name?}] · lines[].name? · lines[].nodes?(꼭짓점 수와 같음 · kind free/length/slope/both · length_m · slope_pct) — 칸 없는 옛 파일 그대로 읽힘
- PUT 몸 point_names · site{lon, lat, address?} — 기준점이 바뀌면 같은 트랜잭션에서 projects.site_* 고침(주소 = 서버 VWorld · 못 찾으면 몸 주소)
- 이름 20 자 · 파일 이름 못 쓰는 글자 막음 — 서버 400(name_error) ↔ TS nameError 짝 · 노드 기록 수 틀림 400
- 선 이름 = 붙인 이름 · 없으면 차례 이름(lineNames ↔ line_names 짝 · 내보내기 조각 이름에 이어짐)
- TS 꼴 NodeKind · NodeRecord · PlanLine.name/nodes · BasePoint.name · ServerNetwork.point_names/site · searchPlaces 틀
- 시험 530_3 이름 거울 · 591_2 왕복 · 400 · 기준점
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
474 lines
19 KiB
Python
474 lines
19 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`) + 서버가 다시 센
|
|
길이 · 측점 · 조각 · 파일이 없으면 빈 설계
|
|
- `PUT …/plan/network` — [저장] · [초기화](빈 선) — 선 · 마크 · 조각 연도만 받아 서버가 정본으로
|
|
다시 셈 → 파일(원자 쓰기) · 연도 상태 줄(`plan_route_years` 지문 · 확정) · 사업연도 끝
|
|
|
|
길이 · 측점 · 높이는 저장하지 않음(늘 다시 셈 · 높이는 지형에서). 측점 = 선마다 No.0 부터.
|
|
조각 = 선의 마크 사이 구간(구역 × 갈래) · 업체 지정은 조각 단위(`plan_route_assignees.piece_id`).
|
|
지점(`points` · 긋기에 붙는 점) = 같은 파일 · [저장] 몸에 실림(빈 값 = 그대로) ·
|
|
기준점(`projects.site_*`) = 처음 자료 받는 기준일 뿐(노선 · 붙음 무관 · PLAN 591-2) ·
|
|
[저장] 몸 `site` 가 바뀌었으면 같은 칸을 고침(594-3 · 등록과 한 칸).
|
|
지점 · 선 이름(20 자 · 파일 이름 못 쓰는 글자 막음) · 선 노드 기록 = 파일에 있을 때만(594-1).
|
|
계산 자리 = ① 파이썬·TS 짝(`D03_PlanAnalysis_Draw`) — 화면은 그리는 즉시 · [저장]은 여기서 다시 셈.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
from typing import Any, Literal
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel, Field
|
|
|
|
from B01_Dashboard.B01_Dashboard_Map import reverse_place
|
|
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 NodeRecord(BaseModel):
|
|
"""노드 기록(594-1) — 찍을 때 방식 · 길이(m) · 기울기(%) · 안 쓴 값 = None."""
|
|
|
|
kind: Literal["free", "length", "slope", "both"]
|
|
length_m: float | None = Field(default=None, allow_inf_nan=False)
|
|
slope_pct: float | None = Field(default=None, allow_inf_nan=False)
|
|
|
|
|
|
class Line(BaseModel):
|
|
id: str = Field(pattern=ID)
|
|
vertices: list[tuple[float, float]] = Field(min_length=1)
|
|
branch: Branch | None = None
|
|
# 594-1 — 빈 값 = 차례 이름 · 노드 기록 = 꼭짓점 수와 같음(없음 = 기록 없음)
|
|
name: str | None = None
|
|
nodes: list[NodeRecord | None] | None = None
|
|
|
|
|
|
class Site(BaseModel):
|
|
"""[저장] 몸 기준점(594-3) — 주소는 서버가 다시 찾음(못 찾으면 이 값)."""
|
|
|
|
lon: float = Field(ge=120, le=135)
|
|
lat: float = Field(ge=30, le=45)
|
|
address: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
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)
|
|
# 지점 이름(594-1) — points 와 같은 차례 · points 와 함께만 읽음 · 빈 칸 = 「지점 N」
|
|
point_names: list[str | None] | None = Field(default=None, max_length=MAX_POINTS)
|
|
lines: list[Line] = []
|
|
marks: list[Mark] = []
|
|
pieces: list[Piece] = []
|
|
site: Site | None = None
|
|
|
|
|
|
class NetworkError(ValueError):
|
|
pass
|
|
|
|
|
|
NAME_MAX = 20
|
|
NAME_BAD = re.compile(r'[\\/:*?"<>|\x00-\x1f]')
|
|
|
|
|
|
def name_error(name: str) -> str | None:
|
|
"""이름 틀림 까닭 글 · 맞으면 None(앞뒤 빈칸 뺀 값으로 봄 · 594-1).
|
|
|
|
⚠ 짝 = TS `logic/D03_PlanAnalysis_Plan.ts` `nameError` — 거울 시험 530-3."""
|
|
text = name.strip()
|
|
if len(text) > NAME_MAX:
|
|
return f"이름은 {NAME_MAX} 자까지입니다."
|
|
if NAME_BAD.search(text):
|
|
return '이름에 \\ / : * ? " < > | 를 쓸 수 없습니다.'
|
|
return None
|
|
|
|
|
|
def _clean_name(name: str | None, what: str) -> str | None:
|
|
"""검사한 이름 · 빈 값 = None · 틀리면 400."""
|
|
if name is None:
|
|
return None
|
|
if error := name_error(name):
|
|
raise HTTPException(status_code=400, detail=f"{what} — {error}")
|
|
return name.strip() or None
|
|
|
|
|
|
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 _points_out(
|
|
points: list[tuple[float, float]], names: list[str | None] | None
|
|
) -> list[dict[str, Any]]:
|
|
"""파일 지점 꼴 `{x, y, name?}` — 이름 칸은 있을 때만."""
|
|
names = names or []
|
|
if len(names) > len(points):
|
|
raise HTTPException(status_code=400, detail="지점 이름이 지점보다 많습니다.")
|
|
out = []
|
|
for i, (x, y) in enumerate(points):
|
|
name = _clean_name(names[i] if i < len(names) else None, f"지점 {i + 1} 이름")
|
|
out.append({"x": float(x), "y": float(y), **({"name": name} if name else {})})
|
|
return out
|
|
|
|
|
|
def _lines_out(lines: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""파일 선 꼴 — 이름 · 노드 기록 칸은 있을 때만(옛 꼴 그대로)."""
|
|
out = []
|
|
for line in lines:
|
|
name = _clean_name(line.get("name"), f"선 {line['id']} 이름")
|
|
base = {key: value for key, value in line.items() if key not in ("name", "nodes")}
|
|
out.append(
|
|
{
|
|
**base,
|
|
**({"name": name} if name else {}),
|
|
**({"nodes": line["nodes"]} if line.get("nodes") else {}),
|
|
}
|
|
)
|
|
return out
|
|
|
|
|
|
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']} 점이 수가 아닙니다.",
|
|
)
|
|
nodes = line.get("nodes")
|
|
_check(
|
|
nodes is None or len(nodes) == len(vertices),
|
|
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 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, name?}], site} — 지점(TM) + 경위도 · 이름 ·
|
|
기준점(경위도 · 주소)."""
|
|
epsg = work_epsg(plan, site)
|
|
points = [
|
|
{
|
|
"lon": lon,
|
|
"lat": lat,
|
|
"x": float(p["x"]),
|
|
"y": float(p["y"]),
|
|
**({"name": p["name"]} if p.get("name") else {}),
|
|
}
|
|
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,
|
|
}
|
|
|
|
|
|
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)
|
|
)
|
|
|
|
|
|
async def _moved_site(
|
|
current: dict[str, Any] | None, site: dict[str, Any] | None
|
|
) -> dict[str, Any] | None:
|
|
"""[저장] 몸 기준점(594-3) — 그대로면 None · 바뀌면 {lon, lat, address}(주소 = 서버 VWorld ·
|
|
못 찾으면 몸 주소)."""
|
|
if site is None or (
|
|
current
|
|
and abs(current["lon"] - site["lon"]) < 1e-9
|
|
and abs(current["lat"] - site["lat"]) < 1e-9
|
|
):
|
|
return None
|
|
found = (await reverse_place(site["lon"], site["lat"]))["address"]
|
|
address = found or (site.get("address") or "").strip() or None
|
|
return {"lon": float(site["lon"]), "lat": float(site["lat"]), "address": address}
|
|
|
|
|
|
async def _store_site(connection: Any, project_id: UUID, site: dict[str, Any]) -> None:
|
|
"""기준점 칸 — 등록(D01)과 같은 `projects.site_*`(사업 지역 · 이름은 안 바꿈)."""
|
|
async with connection.cursor() as cursor:
|
|
await cursor.execute(
|
|
"UPDATE projects SET site_lon = %s, site_lat = %s, site_address = %s WHERE id = %s",
|
|
(site["lon"], site["lat"], site["address"], str(project_id)),
|
|
)
|
|
|
|
|
|
@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")
|
|
point_names = network.pop("point_names")
|
|
site = network.pop("site")
|
|
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, "lines": _lines_out(network["lines"])}
|
|
if points is not None:
|
|
plan["points"] = _points_out(points, point_names)
|
|
site = await _moved_site(project["site"], site)
|
|
async with get_db_pool().acquire() as connection:
|
|
await connection.begin()
|
|
try:
|
|
await _store_status(
|
|
connection, project_id, derived["pieces"], year_rows(derived["pieces"])
|
|
)
|
|
if site:
|
|
await _store_site(connection, project_id, site)
|
|
project = {**project, "site": site}
|
|
store.save_plan(root, plan)
|
|
await connection.commit()
|
|
except Exception:
|
|
await connection.rollback()
|
|
raise
|
|
return {**_view(plan, project), "derived": derived}
|