Merge remote-tracking branch 'origin/main_desktop_1' into main_laptop_1
This commit is contained in:
@@ -0,0 +1,348 @@
|
||||
"""산림사업 표준품셈 476표 → 공종 마스터 정규화 (B08 일감 1번 · 공종 축).
|
||||
|
||||
무엇을 만드나
|
||||
`resources/data_work_item_master/work_item_master_<effective_date>.json` — 공종 계층 + 표 귀속.
|
||||
같은 폴더에 `form_undetermined_*.json`(형태 판정 실패분)과 `_manifest.json` 을 함께 낸다.
|
||||
|
||||
왜 이렇게 나누나 (PLAN 8-7 담당 경계)
|
||||
이 파일은 **공종 축만** 만든다. 자원 축(`resource_kind`·`resource_code`·`amount`)은
|
||||
단가표를 아는 B09 가 뒤 패스로 채운다. 그래서 각 표의 **원문 셀(`raw_row`)을 그대로 실어
|
||||
보낸다** — 버리면 B09 가 476표를 다시 열어야 한다.
|
||||
|
||||
공종의 정체 = 품셈의 **절 번호**다
|
||||
품셈은 「9-3-1. 인력」처럼 절 제목이 공종이고, 표의 행은 그 공종의 **조건별 변형**
|
||||
(토질·암종·규격)이다. 그래서 계층·정렬은 목차표(`F0001`)에서 나오고, 표는 그 절에 붙는다.
|
||||
|
||||
⚠ 최대 함정 — `pum_form` (PLAN 8-6)
|
||||
같은 숫자라도 뜻이 반대다.
|
||||
productivity(생산량형) : 「㎥/hr」·「㎥/1인/1일」 → 품 = 1 ÷ 값
|
||||
requirement(소요량형) : 「100㎥당 x인」 → 품 = 값 ÷ 밑수
|
||||
태그가 없으면 뒤집힌 값이 조용히 들어간다. **판정 못 한 표는 빈칸으로 두지 않고
|
||||
`form_undetermined` 목록으로 뽑아** 사람이 보게 한다.
|
||||
|
||||
실행
|
||||
./venv/Scripts/python.exe B08_Quantity/B08_Quantity_Build_WorkItemMaster.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
SOURCE = ROOT / "resources" / "data_cost_input_value" / "pum_forest_2026.json"
|
||||
OUT_DIR = ROOT / "resources" / "data_work_item_master"
|
||||
|
||||
SCHEMA_VERSION = "1.0"
|
||||
CODE_PREFIX = "FP" # Forest Pumsem — 공종코드 접두. 품셈 절 번호를 그대로 싣는다.
|
||||
|
||||
# ── 형태 판정 ────────────────────────────────────────────────────────────
|
||||
# 헤더·비고 문자열에서 찾는 표지. 앞의 것이 먼저 걸린다(생산량형이 더 좁은 표현이라 우선).
|
||||
PRODUCTIVITY_MARKS = (
|
||||
"㎥/hr",
|
||||
"m3/hr",
|
||||
"㎡/hr",
|
||||
"본/hr",
|
||||
"/1인/1일",
|
||||
"/인/1일",
|
||||
"인/1일",
|
||||
"작업능력",
|
||||
"ha당 평균 작업량",
|
||||
"ha당 집재재적",
|
||||
)
|
||||
REQUIREMENT_MARKS = (
|
||||
"소요인력",
|
||||
"소요량",
|
||||
"당 주입량",
|
||||
"단위수량",
|
||||
"수 량",
|
||||
"수량",
|
||||
"인/km",
|
||||
"인/ha",
|
||||
"㏊당",
|
||||
"ha당",
|
||||
"당 소요",
|
||||
)
|
||||
# 값이 공종 품이 아니라 계산식 파라미터인 표. 공종으로 세우지 않는다.
|
||||
COEFFICIENT_MARKS = ("손료계수", "시간당손료계수", "기계손료")
|
||||
# 기계 시공능력 공식(Q = 3600·qo·K·f·E / Cm)의 파라미터 기호. 이 기호만 있는 표는 계수표다.
|
||||
COEFFICIENT_ROW_KEYS = {"K", "f", "E", "Cm", "㎝(sec)", "q", "qo", "Q", "V", "L"}
|
||||
# 품셈 1장(적용기준)·할증·공제·단위 표준 — 공종이 아니라 **기준표**다.
|
||||
REFERENCE_MARKS = ("할증률", "할인", "공제율", "적재량", "단 위", "지위")
|
||||
# 값 대신 「어디를 따르라」고만 적은 표. 품이 아니므로 공종으로 세우지 않는다.
|
||||
REFERENCE_CELL_MARKS = ("별도계상", "구역화물", "적용한다", "따른다", "준용")
|
||||
# 직종이 값의 주인인 표 = 소요량형. `보통인부(인)`·`콘크리트공(인)` 처럼 `(인)` 이 붙는다.
|
||||
# ⚠ 원문에 `인 부` 처럼 낱말 사이 공백이 있어, 비교 전에 공백을 모두 지운다(`squeeze`).
|
||||
OCCUPATION_RE = re.compile(
|
||||
r"\((?:인|조|인/일|인·일)\)|인부|기능공|운전사|특별인부|보통인부|콘크리트공|철근공|石工|석공|목공|용접공"
|
||||
)
|
||||
# 재료 소요량표의 값 단위. 직종이 없어도 이 단위가 값 열에 오면 소요량형이다.
|
||||
MATERIAL_UNIT_MARKS = ("(kg)", "(㎏)", "(개)", "(매)", "(본)", "(ℓ)", "(L)", "(㎥)", "(㎡)", "(m)")
|
||||
|
||||
|
||||
def sha256_of(path: Path) -> str:
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def norm(text: Any) -> str:
|
||||
"""공백을 하나로 줄인 문자열. `None` 은 빈 문자열."""
|
||||
return " ".join(str(text).split()) if text is not None else ""
|
||||
|
||||
|
||||
def parse_toc(rows: list[list[str]]) -> list[dict[str, Any]]:
|
||||
"""목차표(F0001) → 계층 목록. 장(제n장)·절(n-n)·항(n-n-n)·목(n-n-n-n)."""
|
||||
nodes: list[dict[str, Any]] = []
|
||||
order = 0
|
||||
for raw in rows:
|
||||
cells = [norm(c) for c in raw]
|
||||
cells = [c for c in cells if c]
|
||||
if not cells:
|
||||
continue
|
||||
key = cells[0]
|
||||
name = cells[1] if len(cells) > 1 else ""
|
||||
if m := re.fullmatch(r"제(\d+)장", key):
|
||||
number = m.group(1)
|
||||
elif re.fullmatch(r"\d+(?:-\d+){1,3}", key):
|
||||
number = key
|
||||
else:
|
||||
continue # 부록 등 — 공종 계층이 아니다.
|
||||
order += 256 # STmate 의 SORTCODE 관례(256 간격) — 중간 삽입 여유.
|
||||
parts = number.split("-")
|
||||
nodes.append(
|
||||
{
|
||||
"work_item_code": f"{CODE_PREFIX}-" + "-".join(p.zfill(2) for p in parts),
|
||||
"number": number,
|
||||
"name": name,
|
||||
"level": len(parts),
|
||||
"parent_code": (
|
||||
f"{CODE_PREFIX}-" + "-".join(p.zfill(2) for p in parts[:-1])
|
||||
if len(parts) > 1
|
||||
else None
|
||||
),
|
||||
"sort_order": order,
|
||||
"tables": [],
|
||||
}
|
||||
)
|
||||
return nodes
|
||||
|
||||
|
||||
def section_number(section: str) -> str | None:
|
||||
"""`"9-3-1. 인력"` → `"9-3-1"`. 번호가 없으면 `None`."""
|
||||
m = re.match(r"^\s*(\d+(?:-\d+){0,3})[.\s]", section + " ")
|
||||
return m.group(1) if m else None
|
||||
|
||||
|
||||
def detect_form(table: dict[str, Any], chapter: str | None) -> tuple[str, str]:
|
||||
"""`(pum_form, 근거 문구)`. 판정 못 하면 `("undetermined", 이유)`.
|
||||
|
||||
순서가 뜻을 가진다 — 좁은 표지부터 본다. 계수·기준표를 먼저 걸러 내야
|
||||
「작업능력」 같은 흔한 낱말이 기준표를 공종으로 오인하지 않는다.
|
||||
"""
|
||||
hay = " ".join(norm(h) for h in table.get("headers", []))
|
||||
keys = [norm(r[0]) for r in table.get("rows", []) if r]
|
||||
head_rows = [norm(c) for r in table.get("rows", [])[:3] for c in r]
|
||||
first_rows = " ".join(head_rows)
|
||||
both = f"{hay} || {first_rows}"
|
||||
squeezed = both.replace(" ", "") # `인 부` → `인부`. 원문 자간 공백을 지운 뒤 비교한다.
|
||||
|
||||
for mark in COEFFICIENT_MARKS:
|
||||
if mark in hay:
|
||||
return "coefficient", f"헤더 '{mark}'"
|
||||
if keys and set(keys) <= COEFFICIENT_ROW_KEYS:
|
||||
return "coefficient", f"행 키가 시공능력 공식 기호뿐 {sorted(set(keys))}"
|
||||
if keys and all(re.fullmatch(r"[a-zA-Zqf][₀-₉0-9]?", k) for k in keys):
|
||||
return "coefficient", f"행 키가 기호뿐 {sorted(set(keys))}"
|
||||
|
||||
# 1장은 적용기준 장 자체다 — 공종이 아니라 기준표로 둔다(PLAN 8-14 「목록은 규정에서」).
|
||||
if chapter == "1":
|
||||
return "reference", "품셈 제1장(적용기준)"
|
||||
for mark in REFERENCE_MARKS:
|
||||
if mark in hay:
|
||||
return "reference", f"헤더 '{mark}'"
|
||||
for mark in REFERENCE_CELL_MARKS:
|
||||
if mark in squeezed:
|
||||
return "reference", f"'{mark}' — 값이 아니라 참조 지시"
|
||||
|
||||
# ⚠ 생산량형을 **직종보다 먼저** 본다. 「작업능력(㎥/hr)」 표의 비고란에 흔히
|
||||
# 「보통인부 1인/일」이 붙어 있어, 직종을 먼저 보면 생산량형이 소요량형으로 뒤집힌다.
|
||||
# 그 뒤집힘이 곧 PLAN 8-6 이 경고한 「값이 조용히 반대로 들어가는」 사고다.
|
||||
for mark in PRODUCTIVITY_MARKS:
|
||||
if mark in hay:
|
||||
return "productivity", f"헤더 '{mark}'"
|
||||
|
||||
# 직종이 값의 주인이면 소요량형이다 — 「보통인부(인) 0.16」 은 ㎥당 품이다.
|
||||
if OCCUPATION_RE.search(squeezed):
|
||||
return "requirement", "직종 표기((인)·인부·공)"
|
||||
for mark in MATERIAL_UNIT_MARKS:
|
||||
if mark in hay:
|
||||
return "requirement", f"값 단위 '{mark}'"
|
||||
|
||||
for mark in PRODUCTIVITY_MARKS:
|
||||
if mark in both:
|
||||
return "productivity", f"본문 '{mark}'"
|
||||
for mark in REQUIREMENT_MARKS:
|
||||
if mark in both:
|
||||
return "requirement", f"'{mark}'"
|
||||
return "undetermined", "헤더·첫 행에 단위·밑수·직종 표지 없음"
|
||||
|
||||
|
||||
BASIS_RE = re.compile(r"(\d[\d,.]*)\s*(㎥|m3|㎡|m2|㏊|ha|km|㎞|m|인|본|개|kg|㎏|톤|ton)\s*당")
|
||||
|
||||
|
||||
def detect_basis(table: dict[str, Any]) -> tuple[float | None, str | None]:
|
||||
"""「100㎥당」 같은 밑수. 없으면 `(None, None)` — 단위당 1 로 단정하지 않는다."""
|
||||
hay = " ".join(norm(h) for h in table.get("headers", []))
|
||||
hay += " " + " ".join(norm(c) for r in table.get("rows", [])[:2] for c in r)
|
||||
if m := BASIS_RE.search(hay):
|
||||
try:
|
||||
return float(m.group(1).replace(",", "")), m.group(2)
|
||||
except ValueError:
|
||||
return None, m.group(2)
|
||||
return None, None
|
||||
|
||||
|
||||
def variant_axis(table: dict[str, Any]) -> list[str]:
|
||||
"""행이 갈리는 축 — 표의 첫 열 값들(토질·암종·규격). 값 열은 뺀다."""
|
||||
seen: list[str] = []
|
||||
for row in table.get("rows", []):
|
||||
key = norm(row[0]) if row else ""
|
||||
if key and key not in seen:
|
||||
seen.append(key)
|
||||
return seen[:24]
|
||||
|
||||
|
||||
def build() -> dict[str, Any]:
|
||||
data = json.loads(SOURCE.read_text(encoding="utf-8"))
|
||||
tables = data["variables"]["pum"]["tables"]
|
||||
toc_table = next(t for t in tables if t["table_id"] == "F0001")
|
||||
nodes = parse_toc(toc_table["rows"])
|
||||
by_number = {n["number"]: n for n in nodes}
|
||||
|
||||
attached = 0
|
||||
orphans: list[dict[str, Any]] = []
|
||||
undetermined: list[dict[str, Any]] = []
|
||||
|
||||
for table in tables:
|
||||
if table["table_id"] == "F0001":
|
||||
continue # 목차 자신은 공종이 아니다.
|
||||
section = norm(table.get("section"))
|
||||
number = section_number(section)
|
||||
chapter = number.split("-")[0] if number else None
|
||||
form, why = detect_form(table, chapter)
|
||||
basis_qty, basis_unit = detect_basis(table)
|
||||
entry = {
|
||||
"pum_table_id": table["table_id"],
|
||||
"section": section,
|
||||
"source_line": table.get("line"),
|
||||
"pum_form": form,
|
||||
"form_basis": why,
|
||||
"basis_quantity": basis_qty,
|
||||
"basis_unit": basis_unit,
|
||||
"variant_key": variant_axis(table),
|
||||
"condition_note": [norm(h) for h in table.get("headers", []) if norm(h)],
|
||||
"raw_row": table.get("rows", []), # 원문 셀 — B09 자원 축이 읽는다.
|
||||
}
|
||||
if form == "undetermined":
|
||||
undetermined.append(
|
||||
{
|
||||
"pum_table_id": table["table_id"],
|
||||
"section": section,
|
||||
"headers": entry["condition_note"],
|
||||
"first_rows": table.get("rows", [])[:2],
|
||||
"reason": why,
|
||||
}
|
||||
)
|
||||
node = by_number.get(number) if number else None
|
||||
if node is None:
|
||||
orphans.append({"pum_table_id": table["table_id"], "section": section})
|
||||
continue
|
||||
node["tables"].append(entry)
|
||||
attached += 1
|
||||
|
||||
src_meta = {
|
||||
"dataset_id": data["dataset_id"],
|
||||
"effective_date": data["effective_date"],
|
||||
"sha256": sha256_of(SOURCE),
|
||||
"file": SOURCE.name,
|
||||
}
|
||||
return {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"dataset_id": "work_item_master_forest",
|
||||
"effective_date": data["effective_date"],
|
||||
"generated_at": datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds"),
|
||||
"dataset_version": src_meta,
|
||||
"policy": {
|
||||
"axis": "work_item_only",
|
||||
"resource_axis_owner": "B09",
|
||||
"no_invented_values": True,
|
||||
"raw_row_preserved": True,
|
||||
},
|
||||
"stats": {
|
||||
"toc_nodes": len(nodes),
|
||||
"tables_total": len(tables) - 1,
|
||||
"tables_attached": attached,
|
||||
"tables_orphan": len(orphans),
|
||||
"form_undetermined": len(undetermined),
|
||||
},
|
||||
"orphan_tables": orphans,
|
||||
"work_items": nodes,
|
||||
}, undetermined
|
||||
|
||||
|
||||
def main() -> None:
|
||||
master, undetermined = build()
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
date = master["effective_date"]
|
||||
master_path = OUT_DIR / f"work_item_master_{date}.json"
|
||||
undet_path = OUT_DIR / f"form_undetermined_{date}.json"
|
||||
|
||||
master_path.write_text(json.dumps(master, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
undet_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"dataset_id": "work_item_master_form_undetermined",
|
||||
"effective_date": date,
|
||||
"note": "형태를 못 정한 표. 사람이 보고 productivity/requirement/coefficient 로 확정할 것.",
|
||||
"items": undetermined,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=1,
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
manifest = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"dataset_id": "data_work_item_master_manifest",
|
||||
"generated_at": master["generated_at"],
|
||||
"built_by": "B08_Quantity/B08_Quantity_Build_WorkItemMaster.py",
|
||||
"source": master["dataset_version"],
|
||||
"files": [
|
||||
{
|
||||
"file": p.name,
|
||||
"sha256": sha256_of(p),
|
||||
"size_bytes": p.stat().st_size,
|
||||
}
|
||||
for p in (master_path, undet_path)
|
||||
],
|
||||
}
|
||||
(OUT_DIR / "_manifest.json").write_text(
|
||||
json.dumps(manifest, ensure_ascii=False, indent=1), encoding="utf-8"
|
||||
)
|
||||
|
||||
s = master["stats"]
|
||||
print(f"목차 계층 {s['toc_nodes']}")
|
||||
print(
|
||||
f"표 귀속 {s['tables_attached']} / {s['tables_total']} (미귀속 {s['tables_orphan']})"
|
||||
)
|
||||
print(f"형태 미판정 {s['form_undetermined']}")
|
||||
print(f"산출 {master_path.relative_to(ROOT)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,209 @@
|
||||
"""토적표 — 측점별 단면적을 평균단면적법으로 체적화한다 (B08 일감 2 · PLAN 8-4b).
|
||||
|
||||
무엇을 만드나
|
||||
실무 토적표의 열 구성 그대로다. 거창 실무 워크북 `토적표` 시트와 오솔길 `1.BOM` 36열이
|
||||
서로 1:1 로 맞물리는 것을 확인해 열 이름을 그대로 옮겼다(PLAN 8-4b).
|
||||
|
||||
측점 · 거리 · 절토[토사·암 각 (단면적·입적·보정량)] · 측구터파기[토사·암 각 3칸]
|
||||
· 보정량계 · 성토[단면적·입적] · 유용토 · 차인토량 · 누가토량
|
||||
|
||||
사면 계열(층따기·면고르기·법면보호공·지장목제거)은 사면길이가 아직 없어 일감 3에서 붙인다.
|
||||
|
||||
평균단면적법 (신규 문서 5장 「다. 공사수량의 산출」)
|
||||
체적 = (앞 측점 단면적 + 현 측점 단면적) ÷ 2 × 두 측점 사이 거리.
|
||||
첫 측점은 앞이 없으므로 체적이 없다(거창 실무 토적표도 첫 행 체적이 비어 있다).
|
||||
|
||||
보정량 = 체적 × 토량환산계수(다짐)
|
||||
절취한 흙이 다져지면 줄거나 부푼다. 성토에 쓸 수 있는 양으로 환산한 것이 보정량이다.
|
||||
계수의 유일한 정의처는 `config.config_system_design.EARTHWORK_CONVERSION_FACTORS` 이며
|
||||
여기서 값을 다시 적지 않는다.
|
||||
|
||||
⚠ 숫자는 자르지 않는다 (PLAN 8-16)
|
||||
품셈 1-2-2 의 소수 자리는 **표기 규칙**이다. 계산은 전정밀로 두고 화면·출력에서만
|
||||
반올림한다. 원가 쪽(줄마다 원 단위 절사)과 규칙이 반대이므로 섞지 말 것.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Iterable
|
||||
|
||||
from config.config_system_design import EARTHWORK_CONVERSION_FACTORS
|
||||
|
||||
# 절토 암을 어느 환산계수로 볼지 — 측점의 `cut_rock_kind` 를 그대로 쓴다.
|
||||
# 값이 없으면 리핑암으로 본다(발파암보다 보수적으로 적은 쪽).
|
||||
_DEFAULT_ROCK_KIND = "ripping_rock"
|
||||
|
||||
|
||||
def _factor(kind: str) -> float:
|
||||
"""지반유형 → 다짐 환산계수. 모르는 유형이면 토사로 본다."""
|
||||
entry = EARTHWORK_CONVERSION_FACTORS.get(kind) or EARTHWORK_CONVERSION_FACTORS["soil"]
|
||||
return float(entry["compacted"])
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class StationArea:
|
||||
"""토적표 한 줄이 필요로 하는 측점 값. B06 설계 결과에서 그대로 옮겨 담는다."""
|
||||
|
||||
chainage_m: float
|
||||
cut_soil_area_m2: float = 0.0
|
||||
cut_rock_area_m2: float = 0.0
|
||||
fill_area_m2: float = 0.0
|
||||
ditch_area_m2: float = 0.0
|
||||
cut_rock_kind: str | None = None
|
||||
|
||||
@classmethod
|
||||
def from_design(cls, chainage_m: float, design: dict[str, Any]) -> "StationArea":
|
||||
def num(key: str) -> float:
|
||||
value = design.get(key)
|
||||
return float(value) if isinstance(value, (int, float)) else 0.0
|
||||
|
||||
return cls(
|
||||
chainage_m=float(chainage_m),
|
||||
cut_soil_area_m2=num("cut_soil_area_m2"),
|
||||
cut_rock_area_m2=num("cut_rock_area_m2"),
|
||||
fill_area_m2=num("fill_area_m2"),
|
||||
ditch_area_m2=num("ditch_area_m2"),
|
||||
cut_rock_kind=design.get("cut_rock_kind") or None,
|
||||
)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class EarthworkRow:
|
||||
"""토적표 한 줄. 열 이름은 실무 토적표(PLAN 8-4b)를 따른다."""
|
||||
|
||||
chainage_m: float
|
||||
distance_m: float = 0.0
|
||||
cut_soil_area_m2: float = 0.0
|
||||
cut_soil_volume_m3: float = 0.0
|
||||
cut_soil_adjusted_m3: float = 0.0
|
||||
cut_rock_area_m2: float = 0.0
|
||||
cut_rock_volume_m3: float = 0.0
|
||||
cut_rock_adjusted_m3: float = 0.0
|
||||
ditch_soil_area_m2: float = 0.0
|
||||
ditch_soil_volume_m3: float = 0.0
|
||||
ditch_soil_adjusted_m3: float = 0.0
|
||||
ditch_rock_area_m2: float = 0.0
|
||||
ditch_rock_volume_m3: float = 0.0
|
||||
ditch_rock_adjusted_m3: float = 0.0
|
||||
adjusted_total_m3: float = 0.0
|
||||
fill_area_m2: float = 0.0
|
||||
fill_volume_m3: float = 0.0
|
||||
diverted_m3: float = 0.0
|
||||
balance_m3: float = 0.0
|
||||
cumulative_m3: float = 0.0
|
||||
notes: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
def _split_ditch(area: StationArea) -> tuple[float, float]:
|
||||
"""측구터파기 단면적을 토사·암으로 가른다.
|
||||
|
||||
⚠ TODO(미결 · PLAN 8-4b) — 설계가 측구를 토사·암으로 나눠 주지 않는다(`ditch_area_m2`
|
||||
한 값뿐). 실무 토적표는 둘로 갈라 적으므로, **그 측점의 절토 토사:암 면적비로 안분**한다.
|
||||
측구는 절토부에 파므로 같은 지반을 만난다는 것이 근거다. 설계가 측구 지반을 따로 내주게
|
||||
되면 이 함수만 갈아끼운다.
|
||||
"""
|
||||
ditch = area.ditch_area_m2
|
||||
if ditch <= 0:
|
||||
return 0.0, 0.0
|
||||
soil, rock = area.cut_soil_area_m2, area.cut_rock_area_m2
|
||||
total = soil + rock
|
||||
if total <= 0:
|
||||
return ditch, 0.0 # 절토가 없으면 토사로 본다.
|
||||
return ditch * soil / total, ditch * rock / total
|
||||
|
||||
|
||||
def build_rows(stations: Iterable[StationArea]) -> list[EarthworkRow]:
|
||||
"""측점 목록 → 토적표 줄 목록. 측점은 이정 순으로 정렬해 받는다."""
|
||||
ordered = sorted(stations, key=lambda s: s.chainage_m)
|
||||
rows: list[EarthworkRow] = []
|
||||
previous: StationArea | None = None
|
||||
previous_ditch: tuple[float, float] = (0.0, 0.0)
|
||||
cumulative = 0.0
|
||||
|
||||
for station in ordered:
|
||||
ditch_soil, ditch_rock = _split_ditch(station)
|
||||
soil_factor = _factor("soil")
|
||||
rock_factor = _factor(station.cut_rock_kind or _DEFAULT_ROCK_KIND)
|
||||
row = EarthworkRow(
|
||||
chainage_m=station.chainage_m,
|
||||
cut_soil_area_m2=station.cut_soil_area_m2,
|
||||
cut_rock_area_m2=station.cut_rock_area_m2,
|
||||
ditch_soil_area_m2=ditch_soil,
|
||||
ditch_rock_area_m2=ditch_rock,
|
||||
fill_area_m2=station.fill_area_m2,
|
||||
)
|
||||
if previous is not None:
|
||||
distance = station.chainage_m - previous.chainage_m
|
||||
row.distance_m = distance
|
||||
|
||||
def mean_volume(before: float, now: float) -> float:
|
||||
return (before + now) / 2.0 * distance
|
||||
|
||||
row.cut_soil_volume_m3 = mean_volume(
|
||||
previous.cut_soil_area_m2, station.cut_soil_area_m2
|
||||
)
|
||||
row.cut_rock_volume_m3 = mean_volume(
|
||||
previous.cut_rock_area_m2, station.cut_rock_area_m2
|
||||
)
|
||||
row.ditch_soil_volume_m3 = mean_volume(previous_ditch[0], ditch_soil)
|
||||
row.ditch_rock_volume_m3 = mean_volume(previous_ditch[1], ditch_rock)
|
||||
row.fill_volume_m3 = mean_volume(previous.fill_area_m2, station.fill_area_m2)
|
||||
|
||||
row.cut_soil_adjusted_m3 = row.cut_soil_volume_m3 * soil_factor
|
||||
row.cut_rock_adjusted_m3 = row.cut_rock_volume_m3 * rock_factor
|
||||
row.ditch_soil_adjusted_m3 = row.ditch_soil_volume_m3 * soil_factor
|
||||
row.ditch_rock_adjusted_m3 = row.ditch_rock_volume_m3 * rock_factor
|
||||
|
||||
row.adjusted_total_m3 = (
|
||||
row.cut_soil_adjusted_m3
|
||||
+ row.cut_rock_adjusted_m3
|
||||
+ row.ditch_soil_adjusted_m3
|
||||
+ row.ditch_rock_adjusted_m3
|
||||
)
|
||||
# 유용토 = 그 측점에서 절취분과 성토분이 서로 만나는 몫.
|
||||
row.diverted_m3 = min(row.adjusted_total_m3, row.fill_volume_m3)
|
||||
row.balance_m3 = row.adjusted_total_m3 - row.fill_volume_m3
|
||||
cumulative += row.balance_m3
|
||||
row.cumulative_m3 = cumulative
|
||||
|
||||
rows.append(row)
|
||||
previous = station
|
||||
previous_ditch = (ditch_soil, ditch_rock)
|
||||
return rows
|
||||
|
||||
|
||||
def totals(rows: list[EarthworkRow]) -> dict[str, float]:
|
||||
"""합계 행. 단면적은 합이 뜻이 없어 싣지 않는다(실무 토적표도 비워 둔다)."""
|
||||
keys = (
|
||||
"distance_m",
|
||||
"cut_soil_volume_m3",
|
||||
"cut_soil_adjusted_m3",
|
||||
"cut_rock_volume_m3",
|
||||
"cut_rock_adjusted_m3",
|
||||
"ditch_soil_volume_m3",
|
||||
"ditch_soil_adjusted_m3",
|
||||
"ditch_rock_volume_m3",
|
||||
"ditch_rock_adjusted_m3",
|
||||
"adjusted_total_m3",
|
||||
"fill_volume_m3",
|
||||
"diverted_m3",
|
||||
"balance_m3",
|
||||
)
|
||||
return {key: sum(getattr(row, key) for row in rows) for key in keys}
|
||||
|
||||
|
||||
def build_table(stations: Iterable[StationArea]) -> dict[str, Any]:
|
||||
"""화면·API 가 그대로 쓰는 모양. 값은 자르지 않는다(PLAN 8-16)."""
|
||||
rows = build_rows(stations)
|
||||
return {
|
||||
"method": "average_end_area",
|
||||
"conversion_factors": EARTHWORK_CONVERSION_FACTORS,
|
||||
"rows": [row.__dict__ if not hasattr(row, "__slots__") else _as_dict(row) for row in rows],
|
||||
"totals": totals(rows),
|
||||
"station_count": len(rows),
|
||||
}
|
||||
|
||||
|
||||
def _as_dict(row: EarthworkRow) -> dict[str, Any]:
|
||||
return {name: getattr(row, name) for name in EarthworkRow.__slots__}
|
||||
@@ -0,0 +1,64 @@
|
||||
"""B08 토적표 조회 라우터 (일감 2 · PLAN 8-4b).
|
||||
|
||||
값은 어디서 오나
|
||||
측점별 단면적은 **B06 이 이미 낸 정본**이다(`cross_sections.data.design` 의
|
||||
`cut_soil_area_m2`·`cut_rock_area_m2`·`fill_area_m2`·`ditch_area_m2`).
|
||||
B08 은 그것을 다시 재지 않고 **평균단면적법으로 체적화만** 한다.
|
||||
|
||||
계산 자리 (CLAUDE.md 5장)
|
||||
초기값은 서버가 한 번 계산해 영구저장한다. 여기서는 저장된 단면적을 읽어 표를 만든다 —
|
||||
새 수량을 낳지 않으므로 캐시·조작 경로가 따로 필요 없다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from B06_Section.B06_Section_Repository import (
|
||||
get_cross_section_designs,
|
||||
get_workflow_route_context,
|
||||
)
|
||||
from B08_Quantity.B08_Quantity_Engine_EarthworkTable import StationArea, build_table
|
||||
from config.config_db import run_with_connection
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/projects", tags=["B08 Quantity"])
|
||||
|
||||
|
||||
def _stations(designs: list[dict[str, Any]]) -> list[StationArea]:
|
||||
return [
|
||||
StationArea.from_design(item["chainage_m"], item.get("design") or {}) for item in designs
|
||||
]
|
||||
|
||||
|
||||
@router.get("/{project_id}/quantity/{route_id}/earthwork-table")
|
||||
async def get_earthwork_table(project_id: UUID, route_id: int) -> JSONResponse:
|
||||
"""토적표 — 측점별 단면적을 평균단면적법으로 체적화한 표."""
|
||||
try:
|
||||
designs = await run_with_connection(get_cross_section_designs, route_id)
|
||||
except Exception:
|
||||
logger.exception("B08 토적표 조회 실패: project_id=%s route_id=%s", project_id, route_id)
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content={"status": "error", "message": "토적표를 만들지 못했습니다."},
|
||||
)
|
||||
table = build_table(_stations(designs))
|
||||
table["route_id"] = route_id
|
||||
return JSONResponse(content=table)
|
||||
|
||||
|
||||
@router.get("/{project_id}/quantity/earthwork-table")
|
||||
async def get_earthwork_table_for_current_route(project_id: UUID) -> JSONResponse:
|
||||
"""경로를 안 주면 워크플로가 보고 있는 경로로 낸다 — 화면이 route_id 를 모를 때 쓴다."""
|
||||
context = await run_with_connection(get_workflow_route_context, project_id)
|
||||
if not context or not context.get("route_id"):
|
||||
return JSONResponse(
|
||||
status_code=404,
|
||||
content={"status": "error", "message": "이 프로젝트에 확정된 노선이 없습니다."},
|
||||
)
|
||||
return await get_earthwork_table(project_id, int(context["route_id"]))
|
||||
@@ -0,0 +1,234 @@
|
||||
/* =============================================================================
|
||||
* B08_Quantity_UI_EarthworkGrid.ts
|
||||
* 토적표 그리드 — 실무 토적표(3단 머리글)를 그대로 그린다 (PLAN 8-4b).
|
||||
*
|
||||
* 왜 실무 서식 그대로인가
|
||||
* 이 화면의 첫 사용자는 「프로그램이 맞나」를 확인하려는 설계자다. 보기 좋게 재배치하면
|
||||
* 실무 산출서와 눈으로 대조를 못 한다. 열 순서·머리글 문구를 실무 시트에 맞춘다.
|
||||
*
|
||||
* ⚠ 소수 자리는 표기 규칙일 뿐이다 (PLAN 8-16)
|
||||
* 서버는 전정밀 값을 준다. 자르는 것은 여기(화면)뿐이다. 실무 시트 관측 그대로
|
||||
* 단면적·체적 2자리 · 보정량계·유용토·차인·누가 1자리 · 거리 정수로 보인다.
|
||||
* 원가 쪽(줄마다 원 단위 절사)과 규칙이 반대이므로 그 코드를 여기로 옮기지 말 것.
|
||||
* ========================================================================== */
|
||||
|
||||
/** 서버가 주는 토적표 한 줄. 이름은 엔진(`B08_Quantity_Engine_EarthworkTable.py`)과 같다. */
|
||||
export interface EarthworkRow {
|
||||
chainage_m: number;
|
||||
distance_m: number;
|
||||
cut_soil_area_m2: number;
|
||||
cut_soil_volume_m3: number;
|
||||
cut_soil_adjusted_m3: number;
|
||||
cut_rock_area_m2: number;
|
||||
cut_rock_volume_m3: number;
|
||||
cut_rock_adjusted_m3: number;
|
||||
ditch_soil_area_m2: number;
|
||||
ditch_soil_volume_m3: number;
|
||||
ditch_soil_adjusted_m3: number;
|
||||
ditch_rock_area_m2: number;
|
||||
ditch_rock_volume_m3: number;
|
||||
ditch_rock_adjusted_m3: number;
|
||||
adjusted_total_m3: number;
|
||||
fill_area_m2: number;
|
||||
fill_volume_m3: number;
|
||||
diverted_m3: number;
|
||||
balance_m3: number;
|
||||
cumulative_m3: number;
|
||||
}
|
||||
|
||||
export interface EarthworkTable {
|
||||
method: string;
|
||||
station_count: number;
|
||||
route_id?: number;
|
||||
rows: EarthworkRow[];
|
||||
totals: Record<string, number>;
|
||||
}
|
||||
|
||||
/** 열 하나. `digits` 는 **표기 자리**이며 값 자체는 자르지 않는다. */
|
||||
interface Column {
|
||||
key: keyof EarthworkRow;
|
||||
digits: number;
|
||||
/** 합계행에 낼지 — 단면적은 합이 뜻이 없어 비운다(실무 시트도 비어 있다). */
|
||||
sum?: boolean;
|
||||
}
|
||||
|
||||
/** 실무 토적표 3단 머리글. 대분류 → 중분류 → 소분류 순서가 곧 열 순서다. */
|
||||
const GROUPS: { label: string; sub: { label: string; cols: Column[] }[] }[] = [
|
||||
{ label: "", sub: [{ label: "측 점", cols: [{ key: "chainage_m", digits: 0 }] }] },
|
||||
{ label: "", sub: [{ label: "거 리", cols: [{ key: "distance_m", digits: 0, sum: true }] }] },
|
||||
{
|
||||
label: "절 토",
|
||||
sub: [
|
||||
{
|
||||
label: "토 사",
|
||||
cols: [
|
||||
{ key: "cut_soil_area_m2", digits: 2 },
|
||||
{ key: "cut_soil_volume_m3", digits: 2, sum: true },
|
||||
{ key: "cut_soil_adjusted_m3", digits: 2, sum: true },
|
||||
],
|
||||
},
|
||||
{
|
||||
label: "암 석",
|
||||
cols: [
|
||||
{ key: "cut_rock_area_m2", digits: 2 },
|
||||
{ key: "cut_rock_volume_m3", digits: 2, sum: true },
|
||||
{ key: "cut_rock_adjusted_m3", digits: 2, sum: true },
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
label: "측 구 터 파 기",
|
||||
sub: [
|
||||
{
|
||||
label: "토 사",
|
||||
cols: [
|
||||
{ key: "ditch_soil_area_m2", digits: 2 },
|
||||
{ key: "ditch_soil_volume_m3", digits: 2, sum: true },
|
||||
{ key: "ditch_soil_adjusted_m3", digits: 2, sum: true },
|
||||
],
|
||||
},
|
||||
{
|
||||
label: "암 석",
|
||||
cols: [
|
||||
{ key: "ditch_rock_area_m2", digits: 2 },
|
||||
{ key: "ditch_rock_volume_m3", digits: 2, sum: true },
|
||||
{ key: "ditch_rock_adjusted_m3", digits: 2, sum: true },
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
label: "",
|
||||
sub: [{ label: "보정량계", cols: [{ key: "adjusted_total_m3", digits: 1, sum: true }] }],
|
||||
},
|
||||
{
|
||||
label: "성 토",
|
||||
sub: [
|
||||
{
|
||||
label: "",
|
||||
cols: [
|
||||
{ key: "fill_area_m2", digits: 2 },
|
||||
{ key: "fill_volume_m3", digits: 2, sum: true },
|
||||
],
|
||||
},
|
||||
],
|
||||
},
|
||||
{ label: "", sub: [{ label: "유 용 토", cols: [{ key: "diverted_m3", digits: 1, sum: true }] }] },
|
||||
{ label: "", sub: [{ label: "차인토량", cols: [{ key: "balance_m3", digits: 1, sum: true }] }] },
|
||||
{ label: "", sub: [{ label: "누가토량", cols: [{ key: "cumulative_m3", digits: 1 }] }] },
|
||||
];
|
||||
|
||||
/** 소분류가 「단면적·입적·보정량」 삼중조일 때 붙는 3단 머리글 문구. */
|
||||
const TRIPLE_LABELS = ["단면적", "입 적", "보정량"];
|
||||
const PAIR_LABELS = ["단면적", "입 적"];
|
||||
|
||||
const flatColumns = (): Column[] =>
|
||||
GROUPS.flatMap((g) => g.sub.flatMap((s) => s.cols));
|
||||
|
||||
/** 측점 표기 — `20` → `NO.1`, `25` → `NO.1+5`. 실무 토적표가 이 모양이다. */
|
||||
function stationLabel(chainage: number, interval = 20): string {
|
||||
const no = Math.floor(chainage / interval);
|
||||
const plus = chainage - no * interval;
|
||||
const rounded = Math.round(plus * 100) / 100;
|
||||
return rounded === 0 ? `NO.${no}` : `NO.${no}+${rounded}`;
|
||||
}
|
||||
|
||||
function cell(value: number | undefined, digits: number): string {
|
||||
if (value === undefined || value === null || Number.isNaN(value)) return "";
|
||||
if (value === 0) return "";
|
||||
return value.toLocaleString("ko-KR", {
|
||||
minimumFractionDigits: digits,
|
||||
maximumFractionDigits: digits,
|
||||
});
|
||||
}
|
||||
|
||||
function buildHead(): HTMLTableSectionElement {
|
||||
const head = document.createElement("thead");
|
||||
const r1 = document.createElement("tr");
|
||||
const r2 = document.createElement("tr");
|
||||
const r3 = document.createElement("tr");
|
||||
|
||||
for (const group of GROUPS) {
|
||||
const span = group.sub.reduce((n, s) => n + s.cols.length, 0);
|
||||
if (group.label) {
|
||||
const th = document.createElement("th");
|
||||
th.colSpan = span;
|
||||
th.textContent = group.label;
|
||||
r1.append(th);
|
||||
for (const sub of group.sub) {
|
||||
const th2 = document.createElement("th");
|
||||
th2.colSpan = sub.cols.length;
|
||||
th2.textContent = sub.label;
|
||||
r2.append(th2);
|
||||
const labels = sub.cols.length === 3 ? TRIPLE_LABELS : PAIR_LABELS;
|
||||
sub.cols.forEach((_, index) => {
|
||||
const th3 = document.createElement("th");
|
||||
th3.textContent = labels[index] ?? "";
|
||||
r3.append(th3);
|
||||
});
|
||||
}
|
||||
continue;
|
||||
}
|
||||
// 대분류가 없는 열(측점·거리·보정량계·유용토·…)은 세 줄을 하나로 합친다.
|
||||
for (const sub of group.sub) {
|
||||
const th = document.createElement("th");
|
||||
th.colSpan = sub.cols.length;
|
||||
th.rowSpan = 3;
|
||||
th.textContent = sub.label;
|
||||
r1.append(th);
|
||||
}
|
||||
}
|
||||
head.append(r1, r2, r3);
|
||||
return head;
|
||||
}
|
||||
|
||||
function buildBody(rows: EarthworkRow[]): HTMLTableSectionElement {
|
||||
const body = document.createElement("tbody");
|
||||
const columns = flatColumns();
|
||||
for (const row of rows) {
|
||||
const tr = document.createElement("tr");
|
||||
columns.forEach((column, index) => {
|
||||
const td = document.createElement("td");
|
||||
td.textContent =
|
||||
index === 0 ? stationLabel(row.chainage_m) : cell(row[column.key], column.digits);
|
||||
if (index === 0) td.className = "b08-grid__station";
|
||||
tr.append(td);
|
||||
});
|
||||
body.append(tr);
|
||||
}
|
||||
return body;
|
||||
}
|
||||
|
||||
function buildFoot(totals: Record<string, number>): HTMLTableSectionElement {
|
||||
const foot = document.createElement("tfoot");
|
||||
const tr = document.createElement("tr");
|
||||
flatColumns().forEach((column, index) => {
|
||||
const td = document.createElement("td");
|
||||
if (index === 0) td.textContent = "계";
|
||||
else if (column.sum) td.textContent = cell(totals[column.key], column.digits);
|
||||
tr.append(td);
|
||||
});
|
||||
foot.append(tr);
|
||||
return foot;
|
||||
}
|
||||
|
||||
/** 토적표 하나를 그린다. 넓은 표라 스스로 가로 스크롤한다. */
|
||||
export function renderEarthworkGrid(table: EarthworkTable): HTMLElement {
|
||||
const wrap = document.createElement("div");
|
||||
wrap.className = "b08-grid";
|
||||
|
||||
const caption = document.createElement("p");
|
||||
caption.className = "b08-grid__caption";
|
||||
caption.textContent = `측점 ${table.station_count}곳 · 평균단면적법`;
|
||||
wrap.append(caption);
|
||||
|
||||
const scroller = document.createElement("div");
|
||||
scroller.className = "b08-grid__scroll";
|
||||
const element = document.createElement("table");
|
||||
element.className = "b08-grid__table";
|
||||
element.append(buildHead(), buildBody(table.rows), buildFoot(table.totals));
|
||||
scroller.append(element);
|
||||
wrap.append(scroller);
|
||||
return wrap;
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
{
|
||||
"schema_version": "1.0",
|
||||
"dataset_id": "data_work_item_master_manifest",
|
||||
"generated_at": "2026-09-07T19:59:00+09:00",
|
||||
"built_by": "B08_Quantity/B08_Quantity_Build_WorkItemMaster.py",
|
||||
"source": {
|
||||
"dataset_id": "pum_forest",
|
||||
"effective_date": "2026-01-01",
|
||||
"sha256": "111208fd482dcfc3b8c8dd58004b153382fc46555d0d00985c19b45ebbc2e0dd",
|
||||
"file": "pum_forest_2026.json"
|
||||
},
|
||||
"files": [
|
||||
{
|
||||
"file": "work_item_master_2026-01-01.json",
|
||||
"sha256": "ab6afec24867df51374122efb3fc10416a48b611c2b488c9e3aaa48c2035bedc",
|
||||
"size_bytes": 725962
|
||||
},
|
||||
{
|
||||
"file": "form_undetermined_2026-01-01.json",
|
||||
"sha256": "e43b39bfb844f1d280fb43066ebcf09f9c129eabcbfb2190d249084994d06942",
|
||||
"size_bytes": 40578
|
||||
}
|
||||
]
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user