259 lines
9.4 KiB
Python
259 lines
9.4 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`.
|
|
|
|
계산 자리 ③ 서버 단독 — 화면이 만지지 않는다. 순수 함수(값 → 바이트)만 두고
|
|
파일은 임시 폴더에서만 쓴다.
|
|
"""
|
|
|
|
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]')
|
|
|
|
|
|
@dataclass
|
|
class YearRoute:
|
|
"""한 연도의 노선 — `points` 는 `crs_epsg` 좌표의 (x, y) 꺾은선(시점 → 종점).
|
|
|
|
누가거리 두 값이 있으면 그대로 칸에 쓰고(D03 이은 값), 없으면 연도 순서로 이어 셈.
|
|
"""
|
|
|
|
year: int
|
|
points: list[tuple[float, float]]
|
|
crs_epsg: int
|
|
start_station_m: float | None = None
|
|
end_station_m: float | None = None
|
|
|
|
|
|
@dataclass
|
|
class ExportBundle:
|
|
"""`by_year[연도]` = 그 연도 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}"
|
|
|
|
|
|
def _write_shapefile(
|
|
points: list[tuple[float, float]], year: int | None, stations: tuple[float, float], region: str
|
|
) -> dict[str, bytes]:
|
|
"""5179 꺾은선 → `.shp/.shx/.dbf/.prj/.cpg` 바이트 (파일명은 호출부가 붙임: 키 = 확장자)."""
|
|
length_m = polyline_length_m(points)
|
|
frame = gpd.GeoDataFrame(
|
|
{
|
|
"대상지": [_clip_bytes(_target_text(region, length_m))],
|
|
"길이": [round(length_m / 1000, 3)],
|
|
"연도": [year],
|
|
"시점누가": [round(stations[0], 2)],
|
|
"종점누가": [round(stations[1], 2)],
|
|
},
|
|
geometry=[LineString(points)],
|
|
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),
|
|
round(row["start_station"], 2),
|
|
round(row["end_station"], 2),
|
|
round(row["length_m"], 2),
|
|
]
|
|
)
|
|
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 안 측점은 파일마다 0 부터가 아니라 이 누가값을 칸으로만 보관(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,
|
|
}
|
|
)
|
|
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']}"
|
|
files = _named(
|
|
base,
|
|
_write_shapefile(
|
|
row["points"], row["year"], (row["start_station"], row["end_station"]), region
|
|
),
|
|
)
|
|
if with_xlsx:
|
|
files[xlsx_name] = _coordinate_xlsx(project_name, [row])
|
|
bundle.by_year[row["year"]] = files
|
|
bundle.lengths_m[row["year"]] = row["length_m"]
|
|
|
|
# 전체: 앞 연도 종점 = 다음 연도 시점이라 겹치는 점 하나는 뺌.
|
|
merged: list[tuple[float, float]] = []
|
|
for row in rows:
|
|
merged.extend(
|
|
row["points"][1:] if merged and merged[-1] == row["points"][0] else row["points"]
|
|
)
|
|
total_m = polyline_length_m(merged)
|
|
bundle.lengths_m[0] = total_m
|
|
base = f"대상지_{region_name}.{total_m / 1000:.1f}_전체"
|
|
bundle.all = _named(
|
|
base,
|
|
_write_shapefile(merged, None, (rows[0]["start_station"], rows[-1]["end_station"]), region),
|
|
)
|
|
if with_xlsx:
|
|
bundle.all[xlsx_name] = _coordinate_xlsx(project_name, rows)
|
|
return bundle
|
|
|
|
|
|
def company_zip(project_name: str, year: 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)}_예정노선_{year}.zip", buffer.getvalue()
|