eomsangdonandClaude Opus 5.5
7c94294f85
- 단위 = plan_route_pieces 한 줄 · 조각 없는 연도 행 = 연도 하나 · 확정은 연도 단위 그대로 - 요약 pieces(이름 「본선 0+0~12+10」) · 업체 행 piece_id · 지정 회사 = 자기 조각만 - 내보내기 폴더 = 연도_선_조각 · 전체 = 선마다 레코드 · 좌표표 선 · 조각 칸 - 확정 메일 = 업체마다 자기 조각 zip · 연결 · 가져오기 = 조각 꺾은선 - 받은 예정노선 목록 = 조각 이름 · 길이 - 시험 531 둘 = 조각 없음 가짜 더함 · 새 563_2 지킴 시험 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
299 lines
12 KiB
Python
299 lines
12 KiB
Python
"""D04 예정노선 내보내기 — 조각(구역 × 갈래)마다 · 옛 프로젝트는 연도마다 노선(꺾은선)을
|
|
용화 예정노선 양식 shapefile 한 벌로 만든다.
|
|
|
|
양식 = `resources/knowledge/original/실무문서/2026년 임도 용화/예정노선/대상지_일월.용화.2.2.*`
|
|
(분석 345-C): 2D 폴리라인 1레코드 · UTM-K(EPSG 5179) · cp949 · 칸 `대상지` C(20) ·
|
|
`길이` N(11,3) km + 덧칸 `연도` N(4) · `시점누가` · `종점누가` N(11,2) m.
|
|
`.prj` 는 용화 원문 그대로 · `.cpg` 는 `949`.
|
|
|
|
계산 자리 ③ 서버 단독 — 화면이 만지지 않는다. 순수 함수(값 → 바이트)만 두고
|
|
파일은 임시 폴더에서만 쓴다.
|
|
|
|
짝: `station_label` ↔ `common_util/common_util_station.ts` `stationLabel(…, 1)`
|
|
(거울 시험 `test_545_1_station_label_mirror.py`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import math
|
|
import re
|
|
import tempfile
|
|
import zipfile
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import geopandas as gpd
|
|
from openpyxl import Workbook
|
|
from pyproj import Transformer
|
|
from shapely.geometry import LineString
|
|
|
|
EXPORT_EPSG = 5179
|
|
SHP_SUFFIXES = (".shp", ".shx", ".dbf", ".prj", ".cpg")
|
|
MIME_ZIP = "application/zip"
|
|
MIME_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
|
|
|
# 용화 `.prj` 원문(ESRI WKT) 그대로 — 양식 일치(시험이 용화 파일과 바이트 비교).
|
|
_YONGHWA_PRJ = (
|
|
'PROJCS["PCS_ITRF2000_UTM_K",GEOGCS["GCS_ITRF_2000",DATUM["D_ITRF_2000",'
|
|
'SPHEROID["GRS_1980",6378137.0,298.257222101]],PRIMEM["Greenwich",0.0],'
|
|
'UNIT["Degree",0.0174532925199433]],PROJECTION["Transverse_Mercator"],'
|
|
'PARAMETER["False_Easting",1000000.0],PARAMETER["False_Northing",2000000.0],'
|
|
'PARAMETER["Central_Meridian",127.5],PARAMETER["Scale_Factor",0.9996],'
|
|
'PARAMETER["Latitude_Of_Origin",38.0],UNIT["METER",1.0]]'
|
|
)
|
|
_SCHEMA = {
|
|
"geometry": "LineString",
|
|
"properties": {
|
|
"대상지": "str:20",
|
|
"길이": "float:11.3",
|
|
"연도": "int:4",
|
|
"시점누가": "float:11.2",
|
|
"종점누가": "float:11.2",
|
|
},
|
|
}
|
|
_BAD_FILENAME = re.compile(r'[\\/:*?"<>|\r\n\t]')
|
|
STATION_INTERVAL_M = 20.0 # D03 `PLAN_STATION_INTERVAL_M` 와 같음
|
|
|
|
|
|
def station_label(chainage_m: float, interval_m: float = STATION_INTERVAL_M) -> str:
|
|
"""측점 표기 `61+14.5` — TS `stationLabel(chainage, interval, 1)` 짝(소수 한 자리 · 반올림 위로)."""
|
|
safe = interval_m if interval_m > 0 else 1.0
|
|
number = math.floor((chainage_m + 1e-6) / safe)
|
|
rest = chainage_m - number * safe
|
|
if abs(rest) < 0.05:
|
|
rest = 0.0
|
|
if rest >= safe - 0.05:
|
|
number, rest = number + 1, 0.0
|
|
tenth = math.floor(rest * 10 + 0.5) / 10
|
|
return f"{number}+{int(tenth) if tenth == int(tenth) else tenth}"
|
|
|
|
|
|
@dataclass
|
|
class YearRoute:
|
|
"""한 조각(또는 옛 연도)의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점).
|
|
|
|
누가거리 두 값이 있으면 그대로 칸에 쓰고(조각 = 선마다 No.0), 없으면 연도 순서로 이어 셈.
|
|
`piece` = 조각 id(옛 연도 노선은 None) · `line` = 선 이름(파일 이름 · 전체 묶음 열쇠).
|
|
"""
|
|
|
|
year: int
|
|
points: list[tuple[float, float]]
|
|
crs_epsg: int
|
|
start_station_m: float | None = None
|
|
end_station_m: float | None = None
|
|
piece: str | None = None
|
|
line: str | None = None
|
|
|
|
@property
|
|
def key(self) -> str | int:
|
|
return self.piece or self.year
|
|
|
|
|
|
@dataclass
|
|
class ExportBundle:
|
|
"""`by_year[열쇠]` = 그 조각(열쇠 = 조각 id · 옛은 연도) shp 한 벌 + 좌표표 ·
|
|
`all` = 선마다 이은 레코드 한 벌 + 좌표표."""
|
|
|
|
by_year: dict[int, dict[str, bytes]] = field(default_factory=dict)
|
|
all: dict[str, bytes] = field(default_factory=dict)
|
|
lengths_m: dict[int, float] = field(default_factory=dict)
|
|
|
|
|
|
def safe_name(text: str) -> str:
|
|
"""파일명에 못 쓰는 글자를 `_` 로."""
|
|
return _BAD_FILENAME.sub("_", text).strip() or "_"
|
|
|
|
|
|
def polyline_length_m(points: list[tuple[float, float]]) -> float:
|
|
return sum(math.dist(a, b) for a, b in zip(points, points[1:], strict=False))
|
|
|
|
|
|
def to_utmk(points: list[tuple[float, float]], crs_epsg: int) -> list[tuple[float, float]]:
|
|
"""프로젝트 좌표 → UTM-K(5179). 같은 좌표계면 그대로."""
|
|
pts = [(float(p[0]), float(p[1])) for p in points]
|
|
if crs_epsg == EXPORT_EPSG:
|
|
return pts
|
|
tf = Transformer.from_crs(crs_epsg, EXPORT_EPSG, always_xy=True)
|
|
return [tf.transform(x, y) for x, y in pts]
|
|
|
|
|
|
def _clip_bytes(text: str, limit: int = 20) -> str:
|
|
"""dbf 문자 칸(cp949 바이트) 한도에 글자 단위로 맞춤."""
|
|
while len(text.encode("cp949", errors="replace")) > limit:
|
|
text = text[:-1]
|
|
return text
|
|
|
|
|
|
def _target_text(region: str, length_m: float) -> str:
|
|
return f"{region}.{length_m / 1000:.1f}"
|
|
|
|
|
|
Record = tuple[list[tuple[float, float]], int | None, tuple[float, float]]
|
|
|
|
|
|
def _write_shapefile(records: list[Record], region: str) -> dict[str, bytes]:
|
|
"""5179 꺾은선 레코드(점 · 연도 · 누가 두 값) → `.shp/.shx/.dbf/.prj/.cpg` 바이트
|
|
(파일명은 호출부가 붙임: 키 = 확장자)."""
|
|
lengths = [polyline_length_m(points) for points, _, _ in records]
|
|
frame = gpd.GeoDataFrame(
|
|
{
|
|
"대상지": [_clip_bytes(_target_text(region, m)) for m in lengths],
|
|
"길이": [round(m / 1000, 3) for m in lengths],
|
|
"연도": [year for _, year, _ in records],
|
|
"시점누가": [round(stations[0], 2) for _, _, stations in records],
|
|
"종점누가": [round(stations[1], 2) for _, _, stations in records],
|
|
},
|
|
geometry=[LineString(points) for points, _, _ in records],
|
|
crs=EXPORT_EPSG,
|
|
)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
out = Path(tmp) / "r.shp"
|
|
frame.to_file(
|
|
out, driver="ESRI Shapefile", encoding="cp949", engine="fiona", schema=_SCHEMA
|
|
)
|
|
out.with_suffix(".prj").write_text(_YONGHWA_PRJ, encoding="ascii")
|
|
out.with_suffix(".cpg").write_text("949", encoding="ascii")
|
|
return {suffix: out.with_suffix(suffix).read_bytes() for suffix in SHP_SUFFIXES}
|
|
|
|
|
|
def _coordinate_xlsx(project_name: str, rows: list[dict]) -> bytes:
|
|
"""사람이 읽는 좌표표 — 연도별 시점 · 종점(UTM-K · 위경도) · 측점 · 누가 · 연장."""
|
|
to_ll = Transformer.from_crs(EXPORT_EPSG, 4326, always_xy=True)
|
|
book = Workbook()
|
|
sheet = book.active
|
|
sheet.title = "좌표표"
|
|
sheet.append([f"{project_name} 예정노선 좌표표 (UTM-K EPSG:5179)"])
|
|
sheet.append(
|
|
[
|
|
"연도",
|
|
"시점 X",
|
|
"시점 Y",
|
|
"종점 X",
|
|
"종점 Y",
|
|
"시점 경도",
|
|
"시점 위도",
|
|
"종점 경도",
|
|
"종점 위도",
|
|
"시점 측점",
|
|
"종점 측점",
|
|
"시점누가(m)",
|
|
"종점누가(m)",
|
|
"연장(m)",
|
|
"선",
|
|
"조각",
|
|
]
|
|
)
|
|
for row in rows:
|
|
(sx, sy), (ex, ey) = row["points"][0], row["points"][-1]
|
|
(slon, slat), (elon, elat) = to_ll.transform(sx, sy), to_ll.transform(ex, ey)
|
|
sheet.append(
|
|
[
|
|
row["year"],
|
|
round(sx, 3),
|
|
round(sy, 3),
|
|
round(ex, 3),
|
|
round(ey, 3),
|
|
round(slon, 7),
|
|
round(slat, 7),
|
|
round(elon, 7),
|
|
round(elat, 7),
|
|
station_label(row["start_station"]),
|
|
station_label(row["end_station"]),
|
|
round(row["start_station"], 2),
|
|
round(row["end_station"], 2),
|
|
round(row["length_m"], 2),
|
|
row["line"],
|
|
row["piece"],
|
|
]
|
|
)
|
|
buffer = io.BytesIO()
|
|
book.save(buffer)
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _named(base: str, parts: dict[str, bytes]) -> dict[str, bytes]:
|
|
return {f"{base}{suffix}": data for suffix, data in parts.items()}
|
|
|
|
|
|
def build_export(
|
|
project_name: str,
|
|
region: str,
|
|
years: list[YearRoute],
|
|
first_station_m: float = 0.0,
|
|
with_xlsx: bool = True,
|
|
) -> ExportBundle:
|
|
"""조각(옛 = 연도) 노선들 → 조각마다 한 벌 + 전체 한 벌(+ 좌표표).
|
|
|
|
누가거리 = 값이 없으면 `first_station_m` 에서 시작해 연도 순서로 이어 셈(직전 연도 종점 =
|
|
다음 연도 시점). shp 안 측점은 칸으로만 보관(B03 은 칸을 무시).
|
|
"""
|
|
ordered = sorted(years, key=lambda y: y.year)
|
|
rows: list[dict] = []
|
|
station = float(first_station_m)
|
|
for item in ordered:
|
|
points = to_utmk(item.points, item.crs_epsg)
|
|
if len(points) < 2:
|
|
raise ValueError(f"{item.year}년 노선 점이 2개 미만입니다.")
|
|
length_m = polyline_length_m(points)
|
|
if item.start_station_m is not None and item.end_station_m is not None:
|
|
start, end = float(item.start_station_m), float(item.end_station_m)
|
|
else:
|
|
start, end = station, station + length_m
|
|
rows.append(
|
|
{
|
|
"year": item.year,
|
|
"points": points,
|
|
"length_m": length_m,
|
|
"start_station": start,
|
|
"end_station": end,
|
|
"key": item.key,
|
|
"piece": item.piece,
|
|
"line": item.line,
|
|
}
|
|
)
|
|
station = end
|
|
|
|
bundle = ExportBundle()
|
|
region_name = safe_name(region)
|
|
xlsx_name = f"좌표표_{safe_name(project_name)}.xlsx"
|
|
for row in rows:
|
|
base = f"대상지_{region_name}.{row['length_m'] / 1000:.1f}_{row['year']}"
|
|
if row["piece"]:
|
|
base += f"_{safe_name(row['line'] or '')}_{safe_name(row['piece'])}"
|
|
stations = (row["start_station"], row["end_station"])
|
|
files = _named(base, _write_shapefile([(row["points"], row["year"], stations)], region))
|
|
if with_xlsx:
|
|
files[xlsx_name] = _coordinate_xlsx(project_name, [row])
|
|
bundle.by_year[row["key"]] = files
|
|
bundle.lengths_m[row["key"]] = row["length_m"]
|
|
|
|
# 전체: 선마다 한 레코드 · 앞 조각 종점 = 다음 조각 시점이라 겹치는 점 하나는 뺌.
|
|
by_line: dict[str | None, list[dict]] = {}
|
|
for row in rows:
|
|
by_line.setdefault(row["line"], []).append(row)
|
|
records: list[Record] = []
|
|
for parts in by_line.values():
|
|
parts.sort(key=lambda r: r["start_station"])
|
|
merged: list[tuple[float, float]] = []
|
|
for row in parts:
|
|
same = merged and math.dist(merged[-1], row["points"][0]) < 1e-6
|
|
merged.extend(row["points"][1:] if same else row["points"])
|
|
records.append((merged, None, (parts[0]["start_station"], parts[-1]["end_station"])))
|
|
total_m = sum(polyline_length_m(points) for points, _, _ in records)
|
|
bundle.lengths_m[0] = total_m
|
|
base = f"대상지_{region_name}.{total_m / 1000:.1f}_전체"
|
|
bundle.all = _named(base, _write_shapefile(records, region))
|
|
if with_xlsx:
|
|
bundle.all[xlsx_name] = _coordinate_xlsx(project_name, rows)
|
|
return bundle
|
|
|
|
|
|
def company_zip(project_name: str, label: str | int, files: dict[str, bytes]) -> tuple[str, bytes]:
|
|
"""업체에 보낼 `<프로젝트명>_예정노선_<연도 | 연도_선_조각>.zip` (이름, 바이트)."""
|
|
buffer = io.BytesIO()
|
|
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
|
|
for name, data in files.items():
|
|
archive.writestr(name, data)
|
|
return f"{safe_name(project_name)}_예정노선_{safe_name(str(label))}.zip", buffer.getvalue()
|