bjkim1
21ebb16fb7
- SPipe_server_2026-10-30.zip 내용을 풀어서 등록 (압축 파일은 올리지 않음) - README에 데모 v2 서버 업데이트 순서 추가, 저장소에 없는 문서 목차 정리 - 패치내역을 docs/로 이동, _worklog.md 추가 - 임시 파일(backend/temp.*)과 API 키가 든 파일은 제외
112 lines
5.4 KiB
Python
112 lines
5.4 KiB
Python
"""GIS layer export / reprojection / geometry operations (/api/gis/export, /op, /reproject)."""
|
|
import io
|
|
import os
|
|
import tempfile
|
|
import zipfile
|
|
|
|
import numpy as np
|
|
import pyogrio
|
|
import pytest
|
|
import shapely
|
|
from fastapi.testclient import TestClient
|
|
from pyproj import Transformer
|
|
|
|
from backend import gisops
|
|
from backend.batch import BatchError
|
|
from backend.main import app
|
|
|
|
EPSG = "EPSG:5186"
|
|
TO_WGS = Transformer.from_crs(EPSG, "EPSG:4326", always_xy=True)
|
|
|
|
|
|
def wgs(x, y):
|
|
return list(TO_WGS.transform(x, y))
|
|
|
|
|
|
def fc(features):
|
|
return {"type": "FeatureCollection", "features": features}
|
|
|
|
|
|
SQUARE = fc([{"type": "Feature", "properties": {"name": "블록A", "dia": 150}, "geometry": {"type": "Polygon", "coordinates": [[wgs(200000, 500000), wgs(200100, 500000), wgs(200100, 500100), wgs(200000, 500100), wgs(200000, 500000)]]}}])
|
|
LINES = fc([
|
|
{"type": "Feature", "properties": {"id": "P1", "dia": 150}, "geometry": {"type": "LineString", "coordinates": [wgs(200000, 500000), wgs(200200, 500000)]}},
|
|
{"type": "Feature", "properties": {"id": "P2", "dia": None}, "geometry": {"type": "LineString", "coordinates": [wgs(200200, 500000), wgs(200200, 500200)]}},
|
|
])
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
with TestClient(app, raise_server_exceptions=False) as c:
|
|
yield c
|
|
|
|
|
|
def _area_m2(feature_collection):
|
|
from shapely.geometry import shape
|
|
g = gisops._project(np.array([shape(f["geometry"]) for f in feature_collection["features"]], dtype=object), gisops.WGS84, gisops.CRS.from_user_input(EPSG))
|
|
return float(sum(x.area for x in g))
|
|
|
|
|
|
def test_export_shapefile_in_target_crs_with_korean_fields():
|
|
data, name, media = gisops.write_vector(LINES, EPSG, "shp", "관로")
|
|
assert name == "관로.zip" and media == "application/zip"
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
zipfile.ZipFile(io.BytesIO(data)).extractall(tmp)
|
|
shp = [f for f in os.listdir(tmp) if f.endswith(".shp")][0]
|
|
df_info = pyogrio.read_info(os.path.join(tmp, shp))
|
|
assert "5186" in df_info["crs"] and df_info["features"] == 2
|
|
meta, _, geom, fields = pyogrio.raw.read(os.path.join(tmp, shp))
|
|
x0 = shapely.from_wkb(geom[0]).coords[0][0]
|
|
assert x0 == pytest.approx(200000, abs=0.01)
|
|
assert list(meta["fields"]) == ["id", "dia"]
|
|
|
|
|
|
def test_export_gpkg_dxf_and_mixed_shapefile_refused():
|
|
gpkg, name, _ = gisops.write_vector(SQUARE, EPSG, "gpkg", "blocks")
|
|
assert name == "blocks.gpkg" and gpkg[:15] == b"SQLite format 3"
|
|
dxf, _, _ = gisops.write_vector(LINES, EPSG, "dxf", "lines")
|
|
assert b"SECTION" in dxf
|
|
mixed = fc(SQUARE["features"] + LINES["features"])
|
|
with pytest.raises(BatchError):
|
|
gisops.write_vector(mixed, EPSG, "shp", "x")
|
|
|
|
|
|
def test_reproject_fixes_a_layer_read_in_the_wrong_crs():
|
|
# Numbers really in EPSG:5186 but read as EPSG:5187: reassign -> lands on the 5186 position.
|
|
wrong = Transformer.from_crs("EPSG:5187", "EPSG:4326", always_xy=True)
|
|
bad = fc([{"type": "Feature", "properties": {}, "geometry": {"type": "Point", "coordinates": list(wrong.transform(200000, 500000))}}])
|
|
fixed = gisops.reproject_fc(bad, "EPSG:5187", EPSG)
|
|
assert fixed["features"][0]["geometry"]["coordinates"] == pytest.approx(wgs(200000, 500000), abs=1e-8)
|
|
|
|
|
|
def test_buffer_clip_difference_dissolve():
|
|
buf = gisops.vector_op("buffer", {"crs": EPSG, "layer": LINES, "distance": 10})
|
|
assert len(buf["features"]) == 2 and buf["features"][0]["properties"]["id"] == "P1"
|
|
# P1 buffered: 200 x 20 rectangle + two half-discs of r=10
|
|
from shapely.geometry import shape
|
|
a = gisops._project(np.array([shape(buf["features"][0]["geometry"])], dtype=object), gisops.WGS84, gisops.CRS.from_user_input(EPSG))[0].area
|
|
assert a == pytest.approx(200 * 20 + np.pi * 100, rel=0.01)
|
|
clipped = gisops.vector_op("clip", {"crs": EPSG, "layer": LINES, "other": SQUARE})
|
|
assert len(clipped["features"]) == 1 # only P1 crosses the square
|
|
diff = gisops.vector_op("difference", {"crs": EPSG, "layer": SQUARE, "other": buf})
|
|
assert _area_m2(diff) < 100 * 100
|
|
dis = gisops.vector_op("dissolve", {"crs": EPSG, "layer": buf, "field": None})
|
|
assert len(dis["features"]) == 1 and dis["features"][0]["properties"]["count"] == 2
|
|
|
|
|
|
def test_voronoi_and_grid():
|
|
pts = [[200000, 500000], [200100, 500000], [200000, 500100], [200100, 500100], [200050, 500050]]
|
|
v = gisops.vector_op("voronoi", {"crs": EPSG, "points": pts, "ids": ["A", "B", "C", "D", "E"]})
|
|
assert sorted(f["properties"]["node"] for f in v["features"]) == ["A", "B", "C", "D", "E"]
|
|
g = gisops.vector_op("grid", {"crs": EPSG, "bounds": [200000, 500000, 200100, 500050], "cell": 25})
|
|
assert len(g["features"]) == 8
|
|
with pytest.raises(BatchError):
|
|
gisops.vector_op("grid", {"crs": "EPSG:4326", "bounds": [0, 0, 1, 1], "cell": 1})
|
|
|
|
|
|
def test_http(client):
|
|
r = client.post("/api/gis/export", json={"featureCollection": LINES, "crs": EPSG, "format": "geojson", "name": "관로"})
|
|
assert r.status_code == 200 and "attachment" in r.headers["content-disposition"]
|
|
r = client.post("/api/gis/op", json={"op": "buffer", "crs": EPSG, "layer": LINES, "distance": 5})
|
|
assert r.status_code == 200 and len(r.json()["features"]) == 2
|
|
assert client.post("/api/gis/op", json={"op": "nope", "crs": EPSG, "layer": LINES}).status_code == 400
|