eomsangdon/Aislo
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(D 예정노선): 저장 꼴에 지점 이름 · 선 이름 · 노드 기록 칸 · [저장] 몸 기준점 (PLAN 594-1 · 594-3 · 요구 463)

Browse Source 
- 파일 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>
This commit is contained in:
eomsangdon
2026-10-30 14:40:34 +09:00
co-authored by Claude Opus 5.5
parent 4cb77b2bcd
commit bf77b70be9
6 changed files with 319 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
D03_PlanAnalysis/D03_PlanAnalysis_Network.py
+130 -6
View File
@@ -9,7 +9,9 @@
길이 · 측점 · 높이는 저장하지 않음(늘 다시 셈 · 높이는 지형에서). 측점 = 선마다 No.0 부터.
조각 = 선의 마크 사이 구간(구역 × 갈래) · 업체 지정은 조각 단위(`plan_route_assignees.piece_id`).
지점(`points` · 긋기에 붙는 점) = 같은 파일 · [저장] 몸에 실림(빈 값 = 그대로) ·
기준점(`projects.site_*`) = 처음 자료 받는 기준일 뿐(노선 · 붙음 무관 · PLAN 591-2).
기준점(`projects.site_*`) = 처음 자료 받는 기준일 뿐(노선 · 붙음 무관 · PLAN 591-2) ·
[저장] 몸 `site` 가 바뀌었으면 같은 칸을 고침(594-3 · 등록과 한 칸).
지점 · 선 이름(20 자 · 파일 이름 못 쓰는 글자 막음) · 선 노드 기록 = 파일에 있을 때만(594-1).
계산 자리 = ① 파이썬·TS 짝(`D03_PlanAnalysis_Draw`) — 화면은 그리는 즉시 · [저장]은 여기서 다시 셈.
"""
@@ -18,12 +20,14 @@ from __future__ import annotations
import hashlib
import json
import math
from typing import Any
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
@@ -45,10 +49,29 @@ class Branch(BaseModel):
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):
@@ -68,20 +91,78 @@ 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)
@@ -100,6 +181,11 @@ def derive(network: dict[str, Any], year_from: int | None = None) -> dict[str, A
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"])
@@ -221,10 +307,17 @@ def work_epsg(plan: dict[str, Any], site: dict[str, Any] | None) -> int | 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, 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"])}
{
"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)]
@@ -308,6 +401,31 @@ async def _store_status(
)
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,
@@ -322,6 +440,8 @@ async def put_plan_network(
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"]:
@@ -332,15 +452,19 @@ async def put_plan_network(
except NetworkError as error:
raise HTTPException(status_code=400, detail=str(error)) from error
_check_points(points)
plan = {**plan, **network}
plan = {**plan, **network, "lines": _lines_out(network["lines"])}
if points is not None:
plan["points"] = [{"x": float(x), "y": float(y)} for x, y in points]
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: