feat(B07): 토적도(유토곡선)·유역도(수리집수면적유역도)를 도면으로 낸다

납품 도면 2장을 B07 도면 목록에 새로 붙였다. 계산은 하지 않는다 —
유토곡선은 B06 확정 시 저장한 산출물(longitudinal_sections.data.mass_haul)을,
유역도는 B04 세부유역 GeoJSON과 도엽 등고선·세류선을 읽어 좌표만 종이 mm로 옮긴다.

- 도면 종류 확장: kind에 mass_haul·watershed 추가(Schema·Api_Fetch·openwebcad
  App.types), 12분류 라벨에 연결, 목록에 단장 도면 2건 상시 노출
- 엔진 신규: _Engine_Cad_MassHaul.py(축·곡선·평형선·띠 현·balloon·측점 테이블),
  _Engine_Cad_Basin.py(등고선·세류선 배경·노선·유역·구역별 정보표·방위표)
- 척도 상수: 유토곡선 H 1/2,000 · 세로 1mm=50㎥, 유역도 1/6,000 (A1 고정)
- 유토곡선 저장 payload에 띠·잔여의 기하 필드 추가 — 파이썬에 곡선 보간·토량 배분
  로직을 복제하지 않기 위해 값을 낳는 쪽(TS 엔진)에서 함께 남긴다
- 유역도 배경은 여러 도엽을 합쳐 받은 뒤 도곽 크기로 절취(clip_line_to_box)
- 배수규격은 소요 관경이 아니라 규격관(recommended_diameter_mm)으로 표기

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-30 13:01:39 +09:00
co-authored by Claude Opus 5
parent 14044de50e
commit a1dc0ee235
14 changed files with 1162 additions and 13 deletions
@@ -206,6 +206,8 @@ def _basin_features(
"relief_m": round(basin.relief_m, 2),
"flow_length_m": round(basin.flow_length_m, 1),
"pipe_diameter_mm": basin.pipe_diameter_mm,
# 규격관(도면 배수규격 표기용) — 소요 관경과 구분해 함께 남긴다.
"recommended_diameter_mm": basin.recommended_diameter_mm,
"tc_minutes": basin.tc_minutes,
"intensity_mm_hr": basin.intensity_mm_hr,
"design_flow_m3s": basin.design_flow_m3s,
@@ -66,8 +66,9 @@ logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/projects", tags=["B04 Surface Watershed"])
# 도엽 레이어 파일명 (B04 전처리 산출물과 같은 위치)
_CONTOUR_FILE = "도엽_등고선.geojson"
_STREAM_FILE = "도엽_하천중심선.geojson"
# B07 유역도가 같은 배경을 그리므로 파일명은 공개 상수로 둔다(사본 금지).
CONTOUR_FILE = "도엽_등고선.geojson"
STREAM_FILE = "도엽_하천중심선.geojson"
# 응답 형식 판(版). 응답에 항목을 더하거나 값의 의미를 바꾸면 이 값을 올린다 —
# 저장해 둔 옛 응답을 그대로 돌려주면 화면이 조용히 어긋나기 때문이다.
@@ -187,9 +188,9 @@ async def _prepare(project_id: UUID) -> dict[str, Any] | JSONResponse:
to_metric_transformer = Transformer.from_crs("EPSG:4326", source_crs, always_xy=True)
directory = _sheet_dir(stored_path)
streams = _reproject_features(_load_features(directory, _STREAM_FILE), to_metric_transformer)
streams = _reproject_features(_load_features(directory, STREAM_FILE), to_metric_transformer)
contour_features = _reproject_features(
_load_features(directory, _CONTOUR_FILE), to_metric_transformer
_load_features(directory, CONTOUR_FILE), to_metric_transformer
)
return {
"route_source": route_file.name,
@@ -4,7 +4,7 @@ import { API_BASE_URL, API_TIMEOUT_MS } from "@config/config_frontend";
export interface DesignDrawingItem {
id: string;
kind: "longitudinal" | "cross";
kind: "longitudinal" | "cross" | "mass_haul" | "watershed";
label: string;
chainage_m: number | null;
confirmed: boolean;
@@ -63,7 +63,7 @@ export interface DesignDrawingResponse {
project_id: string;
route_id: number;
id: string;
kind: "longitudinal" | "cross";
kind: "longitudinal" | "cross" | "mass_haul" | "watershed";
label: string;
drawing: CadDrawing;
confirmed: boolean;
@@ -0,0 +1,367 @@
"""B07 유역도(수리집수면적유역도) CAD 조립 — 배수 지점별 집수유역을 평면도로 옮긴다.
납품 양식(2026-08-30 사용자 제시 도면):
- 배경: 수치지형도의 등고선·세류선(도엽 GeoJSON).
- 노선: 계획 노선 중심선.
- 유역: 배수 지점마다 유역 경계를 다른 색으로, 안쪽에 번호.
- 구역별 정보표: 번호·No.측점·배수규격 / 유역면적(ha) / 유역표고(m) / 유하거리(m).
- 제목 수리집수면적유역도, 방위표, 척도(S=1/6,000).
값은 B04 배수유역 산출물(`detail_basins.geojson` 피처 속성) 그대로 읽는다
유역 계산·수리 계산을 여기서 다시 하지 않는다. 좌표는 라우터가 사업지 CRS(m)
바꿔 넘기고, 모듈은 종이 mm로만 옮긴다.
좌표 규약: 종이 mm = (사업지 좌표 m - 콘텐츠 최소점) x MM (1/6,000 -> 1 m = 1/6 mm).
"""
from typing import Any
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
DRAWING_FORMAT,
FRAME_LAYER_ID,
TABLE_LABEL_COLOR,
TABLE_VALUE_COLOR,
_format,
_layer,
_line_entity,
_text_entity,
polyline_entity,
station_no_label,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Template import (
compass_entities,
entities_bbox,
frame_entities,
usable_area,
)
from config.config_system import DRAINAGE_RECOMMEND_DIAMETERS_MM, DRAWING_SCALE_BASIN
# 도면 좌표 = 종이 mm. 실거리 1 m가 종이에서 차지하는 mm (1/6,000 -> 0.1667).
MM = 1000.0 / DRAWING_SCALE_BASIN
CONTOUR_LAYER_ID = "b08-basin-contour"
CONTOUR_COLOR = "#6b7684"
STREAM_LAYER_ID = "b08-basin-stream"
STREAM_COLOR = "#4d9dff"
ROUTE_LAYER_ID = "b08-basin-route"
ROUTE_COLOR = "#ffe066"
BASIN_LAYER_ID = "b08-basin-area"
TABLE_LAYER_ID = "b08-basin-table"
TITLE_LAYER_ID = "b08-basin-title"
# 유역마다 돌려 쓰는 색 (첨부 도면처럼 이웃 유역이 서로 구분되게).
BASIN_COLORS = (
"#ff4d4d",
"#4dff88",
"#4d9dff",
"#ffe066",
"#ff66ff",
"#66ffe0",
"#ff9d4d",
"#b794f6",
)
_FONT_SIZE = 2.2
_NUMBER_FONT_SIZE = 4.0
_TITLE_FONT_SIZE = 7.0
# 구역별 정보표 치수(mm).
_BOX_WIDTH = 46.0
_BOX_ROW_H = 4.5
_BOX_LABEL_W = 20.0
_BOX_GAP = 3.0
_BOX_MARGIN = 12.0 # 지형 콘텐츠 오른쪽 여백
_BOX_ROWS = (("유역면적", "area"), ("유역표고", "relief"), ("유하거리", "flow"))
_TITLE_BAND = 22.0 # 제목·척도가 차지하는 위쪽 띠(mm)
_CLIP_SLACK = 14.0 # 절취 경계 물림 여유(mm)
def map_area_mm() -> tuple[float, float]:
"""지형 배경이 차지할 수 있는 크기(mm) — A1 작도영역에서 정보표 칸과 제목 띠를 뺀다.
라우터는 크기를 축척으로 되돌려 등고선·세류선 절취 범위를 잡는다(정의처 ).
"""
width, height = usable_area()
# _CLIP_SLACK: 절취가 범위 밖 점을 하나씩 물고 나오므로 그만큼 미리 뺀다.
return (
width - (_BOX_MARGIN + _BOX_WIDTH) - _CLIP_SLACK,
height - _TITLE_BAND - _CLIP_SLACK,
)
def _number(value: Any, fallback: float = 0.0) -> float:
return float(value) if isinstance(value, (int, float)) else fallback
def _centroid(ring: list[tuple[float, float]]) -> tuple[float, float]:
"""폴리곤 무게중심(단순 평균) — 번호를 놓을 자리."""
if not ring:
return (0.0, 0.0)
return (
sum(x for x, _y in ring) / len(ring),
sum(y for _x, y in ring) / len(ring),
)
def _diameter_label(props: dict[str, Any]) -> str:
"""배수규격 표기. 다리(세월교·물넘이) 판정이면 관경 대신 시설명을 적는다.
`pipe_diameter_mm` 수리계산이 요구한 **소요 관경**(Φ929 같은 )이라 도면에 그대로
없다. B04가 고른 규격관(`recommended_diameter_mm`) 우선하고, 저장본이라
값이 없으면 규격 목록(정의처 `config_system.DRAINAGE_RECOMMEND_DIAMETERS_MM`)에서
소요 관경 이상인 규격으로 올린다.
"""
if props.get("bridge_required"):
return "물넘이포장"
recommended = props.get("recommended_diameter_mm")
if isinstance(recommended, (int, float)) and recommended > 0:
return f"Φ{int(round(float(recommended)))}mm"
diameter = props.get("pipe_diameter_mm")
if isinstance(diameter, (int, float)) and diameter > 0:
sizes = sorted(DRAINAGE_RECOMMEND_DIAMETERS_MM)
snapped = next((size for size in sizes if size >= float(diameter)), sizes[-1])
return f"Φ{int(snapped)}mm"
return "-"
def _info_box_entities(
drawing_id: str,
number: int,
props: dict[str, Any],
color: str,
origin: tuple[float, float],
interval_m: float,
) -> list[dict[str, Any]]:
"""구역 하나의 정보표 박스(머리행 + 3행)."""
x0, y0 = origin # 박스 좌측 상단
entities: list[dict[str, Any]] = []
rows = len(_BOX_ROWS) + 1
bottom = y0 - rows * _BOX_ROW_H
for index in range(rows + 1):
y = y0 - index * _BOX_ROW_H
entities.append(
_line_entity(
f"{drawing_id}:box:{number}:h:{index}",
(x0, y),
(x0 + _BOX_WIDTH, y),
TABLE_LAYER_ID,
color,
)
)
for index, x in enumerate((x0, x0 + _BOX_LABEL_W, x0 + _BOX_WIDTH)):
entities.append(
_line_entity(
f"{drawing_id}:box:{number}:v:{index}",
(x, y0),
(x, bottom),
TABLE_LAYER_ID,
color,
)
)
head_y = y0 - _BOX_ROW_H / 2.0
station = station_no_label(_number(props.get("chainage_m")), interval_m)
entities.append(
_text_entity(
f"{drawing_id}:box:{number}:head:left",
f"({number}) {station}",
x0 + _BOX_LABEL_W / 2.0,
head_y,
TABLE_LAYER_ID,
_FONT_SIZE,
TABLE_LABEL_COLOR,
)
)
entities.append(
_text_entity(
f"{drawing_id}:box:{number}:head:right",
f"배수규격 {_diameter_label(props)}",
x0 + _BOX_LABEL_W + (_BOX_WIDTH - _BOX_LABEL_W) / 2.0,
head_y,
TABLE_LAYER_ID,
_FONT_SIZE,
TABLE_VALUE_COLOR,
)
)
values = {
"area": f"{_number(props.get('area_m2')) / 10000.0:.2f} ha",
"relief": f"{_format(_number(props.get('relief_m')), 1)} m",
"flow": f"{_format(_number(props.get('flow_length_m')), 1)} m",
}
for index, (label, key) in enumerate(_BOX_ROWS):
center_y = y0 - (index + 1.5) * _BOX_ROW_H
entities.append(
_text_entity(
f"{drawing_id}:box:{number}:label:{key}",
label,
x0 + _BOX_LABEL_W / 2.0,
center_y,
TABLE_LAYER_ID,
_FONT_SIZE,
TABLE_LABEL_COLOR,
)
)
entities.append(
_text_entity(
f"{drawing_id}:box:{number}:value:{key}",
values[key],
x0 + _BOX_LABEL_W + (_BOX_WIDTH - _BOX_LABEL_W) / 2.0,
center_y,
TABLE_LAYER_ID,
_FONT_SIZE,
TABLE_VALUE_COLOR,
)
)
return entities
def build_watershed_drawing(
drawing_id: str,
route_xy: list[tuple[float, float]],
basins: list[dict[str, Any]],
contours: list[list[tuple[float, float]]],
streams: list[list[tuple[float, float]]],
interval_m: float = 20.0,
) -> dict[str, Any]:
"""배수 유역·지형 배경·노선을 유역도 한 장으로 만든다.
좌표는 모두 사업지 CRS(m). `basins` {"ring": [(x, y)...], "props": {...}} 목록으로,
props는 B04 세부유역 피처 속성(chainage_m·area_m2·relief_m·flow_length_m·
pipe_diameter_mm·bridge_required) 그대로 받는다.
"""
if not basins and not route_xy:
raise FileNotFoundError("배수 유역 산출물이 없습니다. B04 배수유역 분석을 먼저 하세요.")
# 콘텐츠 최소점을 원점으로 당겨 종이 mm로 옮긴다(사업지 좌표는 20만 m대라 그대로 쓰면
# 도면 좌표가 지나치게 커진다).
everything = [
*route_xy,
*(point for basin in basins for point in basin.get("ring") or []),
*(point for line in contours for point in line),
*(point for line in streams for point in line),
]
if not everything:
raise FileNotFoundError("유역도에 그릴 좌표가 없습니다.")
min_x = min(x for x, _y in everything)
min_y = min(y for _x, y in everything)
def paper(point: tuple[float, float]) -> tuple[float, float]:
return ((point[0] - min_x) * MM, (point[1] - min_y) * MM)
entities: list[dict[str, Any]] = []
for index, line in enumerate(contours):
contour = polyline_entity(
drawing_id,
[paper(point) for point in line],
CONTOUR_LAYER_ID,
CONTOUR_COLOR,
suffix=f":contour:{index}",
)
if contour:
entities.append(contour)
for index, line in enumerate(streams):
stream = polyline_entity(
drawing_id,
[paper(point) for point in line],
STREAM_LAYER_ID,
STREAM_COLOR,
suffix=f":stream:{index}",
)
if stream:
entities.append(stream)
route = polyline_entity(
drawing_id, [paper(point) for point in route_xy], ROUTE_LAYER_ID, ROUTE_COLOR
)
if route:
entities.append(route)
map_bbox = entities_bbox(entities)
for order, basin in enumerate(basins):
ring = [paper(point) for point in basin.get("ring") or []]
if len(ring) < 3:
continue
props = basin.get("props") or {}
number = int(props.get("index") or order + 1)
color = BASIN_COLORS[(number - 1) % len(BASIN_COLORS)]
outline = polyline_entity(
drawing_id, [*ring, ring[0]], BASIN_LAYER_ID, color, suffix=f":basin:{number}"
)
if outline:
entities.append(outline)
center = _centroid(ring)
entities.append(
_text_entity(
f"{drawing_id}:basin:number:{number}",
f"({number})",
center[0],
center[1],
BASIN_LAYER_ID,
_NUMBER_FONT_SIZE,
color,
)
)
# 정보표는 지형 콘텐츠 오른쪽에 위에서부터 쌓는다.
if map_bbox:
box_x = map_bbox[2] + _BOX_MARGIN
box_y = map_bbox[3]
else:
box_x, box_y = 0.0, 0.0
for order, basin in enumerate(basins):
props = basin.get("props") or {}
number = int(props.get("index") or order + 1)
color = BASIN_COLORS[(number - 1) % len(BASIN_COLORS)]
entities.extend(
_info_box_entities(
drawing_id,
number,
props,
color,
(box_x, box_y - order * ((len(_BOX_ROWS) + 1) * _BOX_ROW_H + _BOX_GAP)),
interval_m,
)
)
bbox = entities_bbox(entities)
if bbox:
min_bx, _min_by, max_bx, max_by = bbox
entities.append(
_text_entity(
f"{drawing_id}:title",
"수 리 집 수 면 적 유 역 도",
(min_bx + max_bx) / 2.0,
max_by + 12.0,
TITLE_LAYER_ID,
_TITLE_FONT_SIZE,
TABLE_LABEL_COLOR,
)
)
entities.append(
_text_entity(
f"{drawing_id}:scale",
f"S = 1/{DRAWING_SCALE_BASIN:,}",
max_bx,
max_by + 5.0,
TITLE_LAYER_ID,
_FONT_SIZE,
TABLE_LABEL_COLOR,
align="right",
)
)
entities.extend(compass_entities(drawing_id, (max_bx - 8.0, max_by - 8.0), 14.0))
entities.extend(frame_entities(drawing_id, entities_bbox(entities) or bbox, fit=False))
return {
"format": DRAWING_FORMAT,
"entities": entities,
"layers": [
_layer(CONTOUR_LAYER_ID, "Contours", locked=True),
_layer(STREAM_LAYER_ID, "Streams", locked=True),
_layer(ROUTE_LAYER_ID, "Route"),
_layer(BASIN_LAYER_ID, "Drainage Basins"),
_layer(TABLE_LAYER_ID, "Basin Info"),
_layer(TITLE_LAYER_ID, "Title"),
_layer(FRAME_LAYER_ID, "Frame", locked=True),
],
}
@@ -0,0 +1,529 @@
"""B07 토적도(유토곡선) CAD 조립 — 확정 종단의 유토곡선 산출물을 도면으로 옮긴다.
납품 양식(2026-08-30 사용자 제시 도면):
- 제목 , 우측 상단 척도 표기(H=1:2,000 / V=1:50,000).
- 좌측 세로축(빨강) + 5,000 눈금·라벨, 누가토량 0 기준선.
- 유토곡선(자홍) + 블록 평형선(흰색) + 경계현(노랑).
- 띠마다 balloon(장비별 육각/타원/사각) `장비 번호 / Q= / L= / EA= / RR= / BR=`.
- 사토·토취 balloon `사토 번호 / Q= / M.N= / EA= / RR= / BR=`.
- 하단 2 테이블: 누가토량(세로쓰기) / 측점(No. 표기).
계산은 하지 않는다. 값은 B06 확정 저장한 `longitudinal_sections.data.mass_haul`
(정의처 `common_util_mass_haul.massHaulPayload`) 그대로 읽어 좌표만 종이 mm로
바꾼다 곡선 보간·토량 배분 로직을 파이썬에 복제하지 않는다.
좌표 규약: x = 누가거리(m) x MM_H, y = 누가토량() / 종이 1mm당 토량.
"""
import math
from typing import Any
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
DRAWING_FORMAT,
FRAME_LAYER_ID,
TABLE_LABEL_COLOR,
TABLE_LINE_COLOR,
TABLE_VALUE_COLOR,
_format,
_layer,
_line_entity,
_stations,
_text_entity,
infer_station_interval,
polyline_entity,
station_no_label,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Template import entities_bbox, frame_entities
from config.config_system import (
DRAWING_SCALE_MASSHAUL_H,
DRAWING_SCALE_MASSHAUL_V_M3_MM,
)
# 도면 좌표 = 종이 mm. 실거리 1 m가 종이에서 차지하는 mm (1/2,000 -> 0.5).
MM_H = 1000.0 / DRAWING_SCALE_MASSHAUL_H
# 토량 1㎥가 종이에서 차지하는 mm (1 mm = 50㎥ -> 0.02).
MM_V = 1.0 / DRAWING_SCALE_MASSHAUL_V_M3_MM
CURVE_LAYER_ID = "b08-masshaul-curve"
CURVE_COLOR = "#ff66ff"
AXIS_LAYER_ID = "b08-masshaul-axis"
AXIS_COLOR = "#ff4d4d"
BAND_LAYER_ID = "b08-masshaul-band"
BAND_COLOR = "#ffe066"
BALANCE_COLOR = "#e8edf4"
TABLE_LAYER_ID = "b08-masshaul-table"
# 세로축 눈금 간격(㎥) — 5,000㎥ = 종이 100 mm.
AXIS_TICK_M3 = 5000.0
_AXIS_TICK_LEN = 1.2
_FONT_SIZE = 2.2
_TITLE_FONT_SIZE = 7.0
_VERTICAL = (0.0, 1.0)
# balloon 치수(mm) — 납품 도면 balloon은 6줄 세로 표기다.
_BALLOON_LINE_H = 3.0
_BALLOON_PAD_X = 2.0
_BALLOON_PAD_Y = 1.5
_BALLOON_CHAR_W = 1.35 # 글자 1개가 차지하는 가로 폭 추정(mm)
_BALLOON_GAP = 10.0 # 곡선과 balloon 사이 최소 틈(mm)
_BALLOON_STAGGER = 3 # 겹침 회피용 층 수
# 장비 키 → 도면 표기(실무 수량산출 용어: 무대·도자·덤프).
EQUIPMENT_LABEL = {
"free_haul": "무대",
"dozer": "도자",
"dump_truck": "덤프",
}
# 하단 테이블 행: (키, 헤더 라벨, 행 높이 mm).
_TABLE_ROWS: tuple[tuple[str, str, float], ...] = (
("cumulative", "누가토량", 12.0),
("station", "측점", 7.0),
)
_TABLE_LABEL_WIDTH = 16.0 # 좌측 행 이름 칸 폭(mm)
_TABLE_TOP_GAP = 6.0 # 그래프 최저점과 테이블 사이 간격(mm)
def _number(value: Any, fallback: float = 0.0) -> float:
return float(value) if isinstance(value, (int, float)) else fallback
def _curve_points(mass_haul: dict[str, Any]) -> list[tuple[float, float]]:
"""유토곡선 점열 (누가거리 m, 누가토량 ㎥). 측점 사이는 직선이다(2026-08-02 확정)."""
points = mass_haul.get("points")
if not isinstance(points, list):
return []
return [
(_number(point.get("chainage_m")), _number(point.get("cumulative_volume_m3")))
for point in points
if isinstance(point, dict) and isinstance(point.get("chainage_m"), (int, float))
]
def _paper(x_m: float, volume_m3: float) -> tuple[float, float]:
return (x_m * MM_H, volume_m3 * MM_V)
def _curve_top_at(curve: list[tuple[float, float]], from_m: float, to_m: float) -> float:
"""구간 안 곡선의 최고 누가토량(㎥). 구간에 점이 없으면 양 끝 값을 쓴다."""
inside = [v for x, v in curve if from_m - 1e-6 <= x <= to_m + 1e-6]
if inside:
return max(inside)
return max((v for _x, v in curve), default=0.0)
def _axis_entities(drawing_id: str, x0: float, min_v: float, max_v: float) -> list[dict[str, Any]]:
"""좌측 세로축(빨강)·눈금·라벨 + 누가토량 0 기준선."""
entities: list[dict[str, Any]] = []
top = _paper(0.0, max_v)[1]
bottom = _paper(0.0, min_v)[1]
entities.append(
_line_entity(f"{drawing_id}:axis:v", (x0, bottom), (x0, top), AXIS_LAYER_ID, AXIS_COLOR)
)
start = int(min_v // AXIS_TICK_M3) * AXIS_TICK_M3
value = start
while value <= max_v + 1e-6:
y = _paper(0.0, value)[1]
entities.append(
_line_entity(
f"{drawing_id}:axis:tick:{value:.0f}",
(x0 - _AXIS_TICK_LEN, y),
(x0, y),
AXIS_LAYER_ID,
AXIS_COLOR,
)
)
entities.append(
_text_entity(
f"{drawing_id}:axis:label:{value:.0f}",
_format(value, 2),
x0 - _AXIS_TICK_LEN - 0.8,
y,
AXIS_LAYER_ID,
_FONT_SIZE,
AXIS_COLOR,
align="right",
)
)
value += AXIS_TICK_M3
return entities
def _balloon_entities(
drawing_id: str,
seed: str,
lines: list[str],
anchor: tuple[float, float],
center: tuple[float, float],
shape: str,
) -> list[dict[str, Any]]:
"""balloon 도형 + 지시선 + 6줄 문자. shape = hexagon | ellipse | rect."""
width = max(len(line) for line in lines) * _BALLOON_CHAR_W + 2 * _BALLOON_PAD_X
height = len(lines) * _BALLOON_LINE_H + 2 * _BALLOON_PAD_Y
cx, cy = center
half_w, half_h = width / 2.0, height / 2.0
if shape == "hexagon":
notch = min(2.5, half_w / 2.0)
outline = [
(cx - half_w, cy),
(cx - half_w + notch, cy + half_h),
(cx + half_w - notch, cy + half_h),
(cx + half_w, cy),
(cx + half_w - notch, cy - half_h),
(cx - half_w + notch, cy - half_h),
(cx - half_w, cy),
]
elif shape == "ellipse":
# 타원은 폴리선 근사(16각) — openwebcad Ellipse 없이도 같은 인상이 난다.
outline = [
(
cx + half_w * math.cos(2 * math.pi * i / 16),
cy + half_h * math.sin(2 * math.pi * i / 16),
)
for i in range(17)
]
else:
outline = [
(cx - half_w, cy - half_h),
(cx - half_w, cy + half_h),
(cx + half_w, cy + half_h),
(cx + half_w, cy - half_h),
(cx - half_w, cy - half_h),
]
entities: list[dict[str, Any]] = []
shape_entity = polyline_entity(
drawing_id, outline, BAND_LAYER_ID, BAND_COLOR, suffix=f":balloon:{seed}"
)
if shape_entity:
entities.append(shape_entity)
# 지시선: balloon 아래(또는 위) 가장자리 → 띠 현 중앙.
edge_y = cy - half_h if anchor[1] < cy else cy + half_h
entities.append(
_line_entity(
f"{drawing_id}:balloon:leader:{seed}",
(cx, edge_y),
anchor,
BAND_LAYER_ID,
BAND_COLOR,
)
)
first_y = cy + half_h - _BALLOON_PAD_Y - _BALLOON_LINE_H * 0.75
for index, line in enumerate(lines):
entities.append(
_text_entity(
f"{drawing_id}:balloon:text:{seed}:{index}",
line,
cx,
first_y - index * _BALLOON_LINE_H,
BAND_LAYER_ID,
_FONT_SIZE,
BAND_COLOR,
)
)
return entities
def _band_entities(
drawing_id: str, plan: dict[str, Any], curve: list[tuple[float, float]]
) -> list[dict[str, Any]]:
"""블록 평형선·띠 경계현·띠 balloon."""
entities: list[dict[str, Any]] = []
blocks = plan.get("blocks")
if not isinstance(blocks, list):
return entities
slot = 0
for block in blocks:
if not isinstance(block, dict):
continue
from_m = _number(block.get("from_m"))
to_m = _number(block.get("to_m"))
base_m3 = _number(block.get("base_m3"))
entities.append(
_line_entity(
f"{drawing_id}:block:{block.get('index')}",
_paper(from_m, base_m3),
_paper(to_m, base_m3),
BAND_LAYER_ID,
BALANCE_COLOR,
)
)
for band in block.get("bands") or []:
if not isinstance(band, dict):
continue
index = band.get("index")
level_base = _number(band.get("level_base_m3"), base_m3)
level_apex = _number(band.get("level_apex_m3"), level_base)
boundary_from = _number(band.get("boundary_from_m"), from_m)
boundary_to = _number(band.get("boundary_to_m"), to_m)
haul_from = _number(band.get("haul_from_m"), boundary_from)
haul_to = _number(band.get("haul_to_m"), boundary_to)
entities.append(
_line_entity(
f"{drawing_id}:band:boundary:{index}",
_paper(boundary_from, level_base),
_paper(boundary_to, level_base),
BAND_LAYER_ID,
BAND_COLOR,
)
)
mid_level = (level_base + level_apex) / 2.0
entities.append(
_line_entity(
f"{drawing_id}:band:haul:{index}",
_paper(haul_from, mid_level),
_paper(haul_to, mid_level),
BAND_LAYER_ID,
BAND_COLOR,
)
)
equipment = str(band.get("equipment") or "")
label = EQUIPMENT_LABEL.get(equipment, "운반")
lines = [
f"{label} {index}",
f"Q={_format(_number(band.get('volume_m3')))}M3",
f"L={_format(_number(band.get('haul_distance_m')))}M",
f"EA={_format(_number(band.get('ea_m3')))}M3",
f"RR={_format(_number(band.get('rr_m3')))}M3",
f"BR={_format(_number(band.get('br_m3')))}M3",
]
anchor = _paper((haul_from + haul_to) / 2.0, mid_level)
top_m3 = _curve_top_at(curve, boundary_from, boundary_to)
height = len(lines) * _BALLOON_LINE_H + 2 * _BALLOON_PAD_Y
center_y = (
_paper(0.0, top_m3)[1] + _BALLOON_GAP + height * (0.5 + slot % _BALLOON_STAGGER)
)
entities.extend(
_balloon_entities(
drawing_id,
f"band:{index}",
lines,
anchor,
(anchor[0], center_y),
{"free_haul": "hexagon", "dozer": "ellipse"}.get(equipment, "rect"),
)
)
slot += 1
return entities
def _residual_entities(
drawing_id: str,
plan: dict[str, Any],
curve: list[tuple[float, float]],
interval_m: float,
) -> list[dict[str, Any]]:
"""사토·토취 balloon — 운반거리 대신 발생 측점(M.N)을 적는다."""
entities: list[dict[str, Any]] = []
residuals = plan.get("residuals")
if not isinstance(residuals, list):
return entities
bottom_m3 = min((v for _x, v in curve), default=0.0)
for slot, residual in enumerate(residuals):
if not isinstance(residual, dict):
continue
index = residual.get("index", slot + 1)
kind = "사토" if residual.get("kind") == "spoil" else "토취"
from_m = _number(residual.get("from_m"))
to_m = _number(residual.get("to_m"), from_m)
level = _number(residual.get("level_from_m3"))
station = station_no_label(from_m, interval_m).removeprefix("No.")
lines = [
f"{kind} {index}",
f"Q={_format(_number(residual.get('volume_m3')))}M3",
f"M.N={station}",
f"EA={_format(_number(residual.get('ea_m3')))}M3",
f"RR={_format(_number(residual.get('rr_m3')))}M3",
f"BR={_format(_number(residual.get('br_m3')))}M3",
]
anchor = _paper((from_m + to_m) / 2.0, level)
height = len(lines) * _BALLOON_LINE_H + 2 * _BALLOON_PAD_Y
center_y = (
_paper(0.0, bottom_m3)[1] - _BALLOON_GAP - height * (0.5 + slot % _BALLOON_STAGGER)
)
entities.extend(
_balloon_entities(
drawing_id, f"residual:{index}", lines, anchor, (anchor[0], center_y), "rect"
)
)
return entities
def _table_entities(
drawing_id: str,
curve: list[tuple[float, float]],
stations: list[dict[str, Any]],
interval_m: float,
top_y: float,
) -> list[dict[str, Any]]:
"""하단 2행 테이블(누가토량 / 측점). 값은 세로쓰기, 측점은 No. 표기."""
entities: list[dict[str, Any]] = []
if not curve:
return entities
cumulative = {round(x, 3): v for x, v in curve}
xs = [x for x, _v in curve]
left = min(xs) * MM_H - _TABLE_LABEL_WIDTH
right = max(xs) * MM_H
y = top_y
boundaries = [y]
for _key, _label, height in _TABLE_ROWS:
y -= height
boundaries.append(y)
for index, line_y in enumerate(boundaries):
entities.append(
_line_entity(
f"{drawing_id}:table:h:{index}",
(left, line_y),
(right, line_y),
TABLE_LAYER_ID,
TABLE_LINE_COLOR,
)
)
for index, x in enumerate((left, left + _TABLE_LABEL_WIDTH, right)):
entities.append(
_line_entity(
f"{drawing_id}:table:v:{index}",
(x, boundaries[0]),
(x, boundaries[-1]),
TABLE_LAYER_ID,
TABLE_LINE_COLOR,
)
)
for row_index, (key, label, height) in enumerate(_TABLE_ROWS):
row_top = boundaries[row_index]
row_bottom = boundaries[row_index + 1]
center_y = (row_top + row_bottom) / 2.0
entities.append(
_text_entity(
f"{drawing_id}:table:name:{key}",
label,
left + _TABLE_LABEL_WIDTH / 2.0,
center_y,
TABLE_LAYER_ID,
_FONT_SIZE,
TABLE_LABEL_COLOR,
)
)
for station in stations:
chainage = _number(station.get("chainage_m"))
x = chainage * MM_H
if x < left + _TABLE_LABEL_WIDTH or x > right:
continue
if key == "station":
text = station_no_label(chainage, interval_m)
y_text = center_y
direction = _VERTICAL
else:
value = cumulative.get(round(chainage, 3))
if value is None:
continue
text = _format(value)
y_text = row_bottom + 0.8
direction = _VERTICAL
entities.append(
_text_entity(
f"{drawing_id}:table:{key}:{chainage:.2f}",
text,
x,
y_text,
TABLE_LAYER_ID,
_FONT_SIZE,
TABLE_VALUE_COLOR,
align="left" if direction == _VERTICAL else "center",
direction=direction,
)
)
return entities
def build_mass_haul_drawing(
longitudinal: dict[str, Any], mass_haul: dict[str, Any], drawing_id: str
) -> dict[str, Any]:
"""확정 종단의 유토곡선 산출물을 토적도 한 장으로 만든다."""
curve = _curve_points(mass_haul)
if len(curve) < 2:
raise FileNotFoundError("확정 종단에 유토곡선 산출물이 없습니다. B06에서 확정하세요.")
all_stations = _stations(longitudinal)
interval_m = infer_station_interval(all_stations)
volumes = [v for _x, v in curve]
min_v, max_v = min(volumes), max(volumes)
entities: list[dict[str, Any]] = []
x0 = min(x for x, _v in curve) * MM_H
entities.extend(_axis_entities(drawing_id, x0, min(min_v, 0.0), max(max_v, 0.0)))
entities.append(
_line_entity(
f"{drawing_id}:axis:zero",
(x0, 0.0),
(max(x for x, _v in curve) * MM_H, 0.0),
AXIS_LAYER_ID,
AXIS_COLOR,
)
)
curve_entity = polyline_entity(
drawing_id, [_paper(x, v) for x, v in curve], CURVE_LAYER_ID, CURVE_COLOR
)
if curve_entity:
entities.append(curve_entity)
plan = mass_haul.get("haul_plan")
if isinstance(plan, dict):
entities.extend(_band_entities(drawing_id, plan, curve))
entities.extend(_residual_entities(drawing_id, plan, curve, interval_m))
# 테이블은 그래프·balloon 어느 것보다도 아래에 둔다 — 사토·토취 balloon이 곡선 밑에
# 깔리므로 그래프 최저점만 보고 자리를 잡으면 표와 겹친다(2026-08-30 화면 실측).
graph_bottom = min(_paper(0.0, min_v)[1], 0.0)
drawn = entities_bbox(entities)
if drawn:
graph_bottom = min(graph_bottom, drawn[1])
entities.extend(
_table_entities(drawing_id, curve, all_stations, interval_m, graph_bottom - _TABLE_TOP_GAP)
)
bbox = entities_bbox(entities)
if bbox:
min_x, _min_y, max_x, max_y = bbox
entities.append(
_text_entity(
f"{drawing_id}:title",
"유 토 곡 선",
(min_x + max_x) / 2.0,
max_y + 12.0,
AXIS_LAYER_ID,
_TITLE_FONT_SIZE,
TABLE_LABEL_COLOR,
)
)
scale_text = (
f"SCALE H=1:{DRAWING_SCALE_MASSHAUL_H:,} "
f"V=1:{int(DRAWING_SCALE_MASSHAUL_V_M3_MM * 1000):,}"
)
entities.append(
_text_entity(
f"{drawing_id}:scale",
scale_text,
max_x,
max_y + 6.0,
AXIS_LAYER_ID,
_FONT_SIZE,
TABLE_LABEL_COLOR,
align="right",
)
)
entities.extend(frame_entities(drawing_id, entities_bbox(entities) or bbox, fit=False))
return {
"format": DRAWING_FORMAT,
"entities": entities,
"layers": [
_layer(CURVE_LAYER_ID, "Mass Haul Curve"),
_layer(AXIS_LAYER_ID, "Graph Axis", locked=True),
_layer(BAND_LAYER_ID, "Haul Bands"),
_layer(TABLE_LAYER_ID, "Station Table"),
_layer(FRAME_LAYER_ID, "Frame", locked=True),
],
}
@@ -26,6 +26,7 @@ _TEMPLATE_DIR = Path(__file__).resolve().parent.parent / "resources" / "template
logger = logging.getLogger(__name__)
A1_TEMPLATE = "00_template_A1"
COMPASS_TEMPLATE = "00_template_compass" # 유역도 등 평면 도면의 방위표
# A1 내부 작도 영역(템플릿 좌표) — 콘텐츠가 이 영역 중앙에 오도록 배치한다.
_A1_INNER = (42.0, 47.0, 812.0, 567.0)
_CONTENT_MARGIN = 0.05 # 내부 작도 영역 대비 콘텐츠 여백 비율(각 방향 5%)
@@ -97,6 +98,34 @@ def _transform_entity(
return out
def compass_entities(
drawing_id: str,
center: tuple[float, float],
size_mm: float,
template_name: str = COMPASS_TEMPLATE,
) -> list[dict[str, Any]]:
"""방위표 템플릿을 지정 위치에 지정 크기로 놓는다(잠금 프레임 레이어).
도각과 달리 콘텐츠를 감쌀 일이 없으므로 변이 size_mm가 되도록 균등 축소한다.
"""
template = _load_template(template_name)
if not template:
return []
entities = template.get("entities", [])
bbox = entities_bbox(entities)
if not bbox:
return []
min_x, min_y, max_x, max_y = bbox
span = max(max_x - min_x, max_y - min_y, 1e-6)
scale = size_mm / span
dx = center[0] - (min_x + max_x) / 2.0 * scale
dy = center[1] - (min_y + max_y) / 2.0 * scale
return [
_transform_entity(entity, f"{drawing_id}:compass:{index}", scale, dx, dy)
for index, entity in enumerate(entities)
]
def usable_area() -> tuple[float, float]:
"""A1 내부 작도 영역에서 여백을 뺀 유효 크기(mm). 척도 고정 도면의 수용 한도."""
ix0, iy0, ix1, iy1 = _A1_INNER
@@ -25,12 +25,15 @@ from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Table import (
extract_quantity_table,
)
from B07_DesignDetail.B07_DesignDetail_Router_Support import (
MASS_HAUL_ID,
WATERSHED_ID,
_cross_sheet_plan,
_drawing_list,
_invalidate_drawing,
_read_drawing,
_recompute_confirmed_design,
_store_confirmed_drawing,
watershed_source,
)
from B07_DesignDetail.B07_DesignDetail_Schema import (
DesignDrawingConfirmRequest,
@@ -39,6 +42,7 @@ from B07_DesignDetail.B07_DesignDetail_Schema import (
DesignDrawingListResponse,
DesignDrawingResponse,
)
from common_util.common_util_drainage_context import load_drainage_context
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_workflow_state import complete_stage, start_stage
from config.config_db import get_db_pool
@@ -127,6 +131,18 @@ async def get_design_drawing(
elif CROSS_SHEET_ID.fullmatch(drawing_id):
# 장은 여러 측점을 담으므로 노선 전체 지정을 한 번에 읽어 넘긴다.
source_design = await _designs_by_chainage(route_id)
elif drawing_id == MASS_HAUL_ID:
# 유토곡선은 B06 확정 시 종단 레코드에 저장해 둔 산출물을 그대로 쓴다.
pool = get_db_pool()
async with pool.acquire() as connection:
section = await get_longitudinal_section(connection, project_id, route_id)
source_design = ((section or {}).get("data") or {}).get("mass_haul")
elif drawing_id == WATERSHED_ID:
# 유역도는 B04 배수유역 산출물 + 도엽 배경을 사업지 CRS로 모아 넘긴다.
context, reason = await load_drainage_context(project_id)
if context is None:
return JSONResponse(status_code=404, content={"status": "error", "message": reason})
source_design = await asyncio.to_thread(watershed_source, context)
kind, label, drawing, confirmed, quantity_table = await asyncio.to_thread(
_read_drawing, project_root, longitudinal_path, drawing_id, source_design
)
@@ -5,10 +5,14 @@
import json
import logging
import math
import re
from pathlib import Path
from typing import Any
from pyproj import Transformer
from B04_PreProcess.B04_PreProcess_Router_Watershed import CONTOUR_FILE, STREAM_FILE
from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files
from B06_Section.B06_Section_Engine_Design import compute_cross_design
from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
@@ -17,10 +21,12 @@ from B07_DesignDetail.B07_DesignDetail_Engine_Cad import (
infer_station_interval,
station_no_label,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Basin import build_watershed_drawing, map_area_mm
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Long import (
build_longitudinal_drawing,
longitudinal_chunks,
)
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_MassHaul import build_mass_haul_drawing
from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Sheet import (
CROSS_SHEET_ID,
build_cross_sheet,
@@ -33,7 +39,9 @@ from B07_DesignDetail.B07_DesignDetail_Engine_Cad_Table import (
from B07_DesignDetail.B07_DesignDetail_Schema import (
DesignDrawingItem,
)
from common_util.common_util_drainage_pipes import detail_basins_path
from common_util.common_util_route_profile import design_elevation_from_longitudinal
from config.config_system import DRAWING_SCALE_BASIN
logger = logging.getLogger(__name__)
@@ -41,6 +49,9 @@ _STAGE_DIR = "B07_DesignDetail"
_CROSS_ID = re.compile(r"^cross_(\d+)m$")
_LONG_ID = re.compile(r"^longitudinal(?:_(\d+))?$")
# 노선 전장에 한 장씩만 나오는 도면 — 라우터가 원본 자료를 따로 실어 넘긴다.
MASS_HAUL_ID = "mass_haul"
WATERSHED_ID = "watershed"
def _read_json(path: Path) -> dict[str, Any]:
@@ -94,6 +105,129 @@ def _write_manifest(project_root: Path, manifest: dict[str, Any]) -> None:
temporary.replace(path)
def _geojson_features(path: Path) -> list[dict[str, Any]]:
"""GeoJSON 피처 목록만 읽는다. 파일이 없거나 깨졌으면 빈 목록."""
if not path.is_file():
return []
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
logger.warning("B07 유역도: GeoJSON을 읽지 못했습니다 — %s", path)
return []
features = payload.get("features") if isinstance(payload, dict) else None
return features if isinstance(features, list) else []
def _geometry_lines(geometry: Any) -> list[list[tuple[float, float]]]:
"""LineString·MultiLineString·Polygon을 점열 목록으로 편다."""
if not isinstance(geometry, dict):
return []
kind = geometry.get("type")
coordinates = geometry.get("coordinates")
if not isinstance(coordinates, list) or not coordinates:
return []
if kind == "LineString":
return [[(float(p[0]), float(p[1])) for p in coordinates if isinstance(p, list)]]
if kind in ("MultiLineString", "Polygon"):
return [
[(float(p[0]), float(p[1])) for p in part if isinstance(p, list)]
for part in coordinates
if isinstance(part, list)
]
return []
def clip_line_to_box(
line: list[tuple[float, float]], box: tuple[float, float, float, float]
) -> list[list[tuple[float, float]]]:
"""범위 안에 든 연속 구간만 조각으로 잘라 낸다(경계 한 점씩 물려 끊긴 티를 줄인다).
도엽 등고선 줄은 도엽 끝까지 이어지므로, 걸치면 통째로 남기면 도면이 A1을 넘는다.
"""
min_x, min_y, max_x, max_y = box
runs: list[list[tuple[float, float]]] = []
current: list[tuple[float, float]] = []
for index, point in enumerate(line):
x, y = point
if min_x <= x <= max_x and min_y <= y <= max_y:
if not current and index > 0:
current.append(line[index - 1]) # 바깥쪽 직전 점까지 물린다
current.append(point)
elif current:
current.append(point)
runs.append(current)
current = []
if current:
runs.append(current)
return [run for run in runs if len(run) >= 2]
def watershed_source(context: Any) -> dict[str, Any]:
"""유역도 입력(노선·세부유역·등고선·세류선)을 사업지 CRS(m)로 모은다.
저장본은 전부 WGS84라 여기서 미터 좌표로 되돌린다(B04가 저장할 때와 반대 방향).
"""
to_metric = Transformer.from_crs("EPSG:4326", f"EPSG:{context.epsg}", always_xy=True)
def metric(point: tuple[float, float]) -> tuple[float, float]:
x, y = to_metric.transform(point[0], point[1])
return (float(x), float(y))
route_xy = [(vertex.x, vertex.y) for vertex in context.vertices]
basins: list[dict[str, Any]] = []
for feature in _geojson_features(detail_basins_path(context.stored_path)):
properties = feature.get("properties") or {}
if properties.get("kind") != "detail_basin":
continue
rings = _geometry_lines(feature.get("geometry"))
if not rings:
continue
basins.append({"ring": [metric(point) for point in rings[0]], "props": properties})
# 배경(등고선·세류선)은 **여러 도엽을 합쳐 받은 뒤 도곽 크기로 절취**한다
# (2026-08-30 사용자 지시 — 노선이 도엽 경계에 걸릴 수 있어 주변 도엽까지 받아 둔다).
# 도엽 등고선 한 줄은 도엽 끝까지 이어지므로 "걸치면 통째로"는 도면이 A1을 넘긴다
# (실측 430x871 mm). 절취 범위 = 도곽 안 지형 영역(정보표·제목 제외)을 축척으로 되돌린 크기.
usable_w_mm, usable_h_mm = map_area_mm()
half_w_m = usable_w_mm / 2.0 * DRAWING_SCALE_BASIN / 1000.0
half_h_m = usable_h_mm / 2.0 * DRAWING_SCALE_BASIN / 1000.0
extent = [*route_xy, *(point for basin in basins for point in basin["ring"])]
if extent:
center_x = (min(x for x, _y in extent) + max(x for x, _y in extent)) / 2.0
center_y = (min(y for _x, y in extent) + max(y for _x, y in extent)) / 2.0
# 노선·유역이 도곽보다 크면 그쪽을 우선한다 — 배경만 잘리고 주제는 다 보인다.
min_x = min(center_x - half_w_m, min(x for x, _y in extent))
max_x = max(center_x + half_w_m, max(x for x, _y in extent))
min_y = min(center_y - half_h_m, min(y for _x, y in extent))
max_y = max(center_y + half_h_m, max(y for _x, y in extent))
else:
min_x = min_y = -math.inf
max_x = max_y = math.inf
box = (min_x, min_y, max_x, max_y)
def clip(line: list[tuple[float, float]]) -> list[list[tuple[float, float]]]:
return clip_line_to_box(line, box)
sheet_dir = Path(context.project_root) / "B04_PreProcess" / "processed"
background: dict[str, list[list[tuple[float, float]]]] = {}
for key, filename in (("contours", CONTOUR_FILE), ("streams", STREAM_FILE)):
lines: list[list[tuple[float, float]]] = []
for feature in _geojson_features(sheet_dir / filename):
for part in _geometry_lines(feature.get("geometry")):
converted = [metric(point) for point in part]
if len(converted) >= 2:
lines.extend(clip(converted))
background[key] = lines
return {
"route_xy": route_xy,
"basins": basins,
"contours": background["contours"],
"streams": background["streams"],
}
def _drawing_list(
project_root: Path,
longitudinal_path: Path,
@@ -127,6 +261,19 @@ def _drawing_list(
confirmed=bool(manifest_drawings.get(sheet["id"], {}).get("confirmed")),
)
)
# 노선 전장 1장짜리 도면 — 자료가 없으면 여는 시점에 404로 알린다(목록에는 항상 둔다).
for drawing_id, kind, label in (
(MASS_HAUL_ID, "mass_haul", "토적도(유토곡선)"),
(WATERSHED_ID, "watershed", "유역도(배수 유역도)"),
):
drawings.append(
DesignDrawingItem(
id=drawing_id,
kind=kind,
label=label,
confirmed=bool(manifest_drawings.get(drawing_id, {}).get("confirmed")),
)
)
return drawings
@@ -306,11 +453,48 @@ def _read_drawing(
# 포맷 버전이 다르면(테이블·레이어 구성 변경 전 저장본) 캐시를 버리고
# 아래에서 원본 기준으로 재생성한다. 확정 상태도 무효로 응답해 재확정 유도.
if saved.get("format") == DRAWING_FORMAT:
kind = "longitudinal" if _LONG_ID.fullmatch(drawing_id) else "cross"
if drawing_id in (MASS_HAUL_ID, WATERSHED_ID):
kind = drawing_id # id와 kind가 같은 단장 도면
else:
kind = "longitudinal" if _LONG_ID.fullmatch(drawing_id) else "cross"
label = str(manifest_entry.get("label") or drawing_id)
stored_table = manifest_entry.get("quantity_table")
table = stored_table if kind == "cross" and isinstance(stored_table, dict) else None
return kind, label, saved, True, table
if drawing_id == MASS_HAUL_ID:
# stored_design = 확정 종단 DB row의 mass_haul 산출물(라우터가 실어 준다).
if not isinstance(stored_design, dict):
raise FileNotFoundError("확정 종단에 유토곡선 산출물이 없습니다.")
longitudinal = _read_json(longitudinal_path)
return (
"mass_haul",
"토적도(유토곡선)",
build_mass_haul_drawing(longitudinal, stored_design, drawing_id),
False,
None,
)
if drawing_id == WATERSHED_ID:
# stored_design = watershed_source()가 모아 준 노선·유역·배경 좌표(사업지 CRS).
if not isinstance(stored_design, dict):
raise FileNotFoundError("배수 유역 산출물이 없습니다.")
longitudinal = _read_json(longitudinal_path)
interval = infer_station_interval(longitudinal.get("stations") or [])
return (
"watershed",
"유역도(배수 유역도)",
build_watershed_drawing(
drawing_id,
stored_design.get("route_xy") or [],
stored_design.get("basins") or [],
stored_design.get("contours") or [],
stored_design.get("streams") or [],
interval,
),
False,
None,
)
if _LONG_ID.fullmatch(drawing_id):
source = _read_json(longitudinal_path)
chunk = next(
+2 -2
View File
@@ -9,7 +9,7 @@ class DesignDrawingItem(BaseModel):
"""B06 확정 산출물에서 노출하는 도면 메타데이터."""
id: str
kind: Literal["longitudinal", "cross"]
kind: Literal["longitudinal", "cross", "mass_haul", "watershed"]
label: str
chainage_m: float | None = None
confirmed: bool = False
@@ -31,7 +31,7 @@ class DesignDrawingResponse(BaseModel):
project_id: str
route_id: int
id: str
kind: Literal["longitudinal", "cross"]
kind: Literal["longitudinal", "cross", "mass_haul", "watershed"]
label: str
drawing: dict[str, Any]
confirmed: bool = False
+3 -3
View File
@@ -42,7 +42,7 @@ import {
/** CAD 앱 수량 패널로 넘기는 설계 컨텍스트 (openwebcad DesignMeta와 동일 형식). */
interface DesignMeta {
kind: "cross" | "longitudinal";
kind: "cross" | "longitudinal" | "mass_haul" | "watershed";
title: string;
info: string;
confirmed: boolean;
@@ -84,8 +84,8 @@ const DRAWING_GROUPS: readonly { label: string; kind?: DesignDrawingItem["kind"]
{ label: "종단면도", kind: "longitudinal" },
{ label: "표준 횡단면도" },
{ label: "횡단면도", kind: "cross" },
{ label: "토적도(유토곡선)" },
{ label: "유역도(배수 유역도)" },
{ label: "토적도(유토곡선)", kind: "mass_haul" },
{ label: "유역도(배수 유역도)", kind: "watershed" },
{ label: "표준도" },
{ label: "용지도" },
];
+1 -1
View File
@@ -40,7 +40,7 @@ export enum HtmlEvent {
/** 부모(B08 페이지)가 도면과 함께 넘기는 설계 컨텍스트 (수량 패널 표시용). */
export interface DesignMeta {
kind: 'cross' | 'longitudinal';
kind: 'cross' | 'longitudinal' | 'mass_haul' | 'watershed';
/** 패널 제목 (측점 라벨, 예: "2+0.0" 또는 "종단도 전체") */
title: string;
/** 측점 부가 정보 (예: "STA.0+050.000") */
@@ -207,10 +207,22 @@ export function haulPlanPayload(plan: HaulPlan): Record<string, unknown> {
ea_m3: round(band.ea_m3),
rr_m3: round(band.rr_m3),
br_m3: round(band.br_m3),
// 띠의 기하 — B07 토적도가 경계현·평균운반거리 현을 이 값으로 그린다.
// 여기서 함께 저장하지 않으면 파이썬 쪽에 곡선 보간·배분 로직을 복제해야 한다.
level_base_m3: round(band.level_base_m3),
level_apex_m3: round(band.level_apex_m3),
boundary_from_m: round(band.boundary_from_m),
boundary_to_m: round(band.boundary_to_m),
haul_from_m: round(band.haul_from_m),
haul_to_m: round(band.haul_to_m),
})),
})),
residuals: plan.residuals.map((residual) => ({
index: residual.index,
kind: residual.kind,
// 평형선 단차 — 토적도 balloon을 곡선 옆 제 높이에 놓는 데 쓴다.
level_from_m3: round(residual.level_from_m3),
level_to_m3: round(residual.level_to_m3),
from_m: round(residual.from_m),
to_m: round(residual.to_m),
volume_m3: round(residual.volume_m3),
+9
View File
@@ -756,6 +756,15 @@ DRAWING_SCALE_LONG_V = 200 # 종단면도 세로 축척 분모
DRAWING_SCALE_CROSS = 100 # 횡단면도 가로·세로 축척 분모
DRAWING_SHEET = "A1" # 용지 규격 (840x594 mm 도각 템플릿)
# 토적도(유토곡선)·유역도 척도 (2026-08-30 사용자 확정 — 납품 도면 표기 기준)
# 유토곡선: 가로 1/2,000 · 세로 1/50,000. 세로축은 길이가 아니라 **토량(㎥)**이라
# 축척분모를 그대로 못 쓴다 — 종이 1 mm가 받는 토량(50 ㎥)으로 적는다.
# 유역도 : 평면 1/6,000 (가로·세로 같은 배율)
# A1 유효 작도영역 693x468 mm를 넘으면 늘리지 않고 경고만 남긴다(척도 보존).
DRAWING_SCALE_MASSHAUL_H = 2000 # 유토곡선 가로 축척 분모 (실거리 1 m = 0.5 mm)
DRAWING_SCALE_MASSHAUL_V_M3_MM = 50.0 # 유토곡선 세로 — 종이 1 mm 당 토량(㎥)
DRAWING_SCALE_BASIN = 6000 # 유역도 평면 축척 분모 (실거리 1 m = 1/6 mm)
# 시스템 리소스 로그 (루트 log 폴더에 단일 파일, 1개월 보관)
LOG_BASE_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "log")
RESOURCE_LOG_PATH = os.path.join(LOG_BASE_DIR, "system_resources.log")