Files
Aislo/B09_Estimation/B09_Estimation_MachineOperating.py
T
eomsangdonandClaude Opus 5 fdc30ecf17 feat(B09): 기계 수송비를 세움 — 기계경비의 셋째 몫이 통째로 빠져 있었음
기계경비 = 기계손료 + 운전경비 + 수송비인데 수송비가 없었음. 수송비를 내는 공종
FP-10-04 중기운반은 표가 미판정이라 단가가 아예 안 섰음.

- 산림품셈 10-4 사이클 그대로 — 트레일러(20TON) t1=20·t3=20·t4=0.42 /
  트럭(10.5TON) t1=10·t3=10·t4=5 · ㎝=t1+t2+t3+t4 · N=60×0.9/㎝, 단위는 회당.
- 거리·도로 구분은 설계 입력 — 비면 줄이 안 서고 사유만 남음(임의 거리 금지).
- 원문이 「-」로 둔 칸(트레일러·고속4차선)은 지어내지 않고 사유로 냄.
- ⚠ 우리가 정한 둘을 화면 근거에 적음 — 속도를 8-1-6의 2 나 이동속도표에서 가져온 것,
  운반시간을 왕복으로 본 것.
- 곁다리: 운전경비 표가 페이지에서 잘려 앞자리를 못 이어받아 39 기종이 통째로
  버려지고 있었음(그 안에 수송 차량 2702 가 있었음). 이어받되 카탈로그에 있는
  코드일 때만 받게 함 — 운전경비 줄 92 → 158.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-09 22:23:06 +09:00

448 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""B09 원가계산 — 운전경비(연료·잡재료·조종원)를 품셈 8-4 에서 뽑는다.
`mach_base_2026.json` 의 `mach_fuel_rate`·`mach_operator_map` 이 **0건**이라 시간당
중기사용료의 **재료비·노무비 성분이 비어 있었다**(PLAN 9-3). 그 값은 건설품셈
**8-4 운전경비 산정** 표에 있다.
**파생 파일로 낸다** — 기준자료(`resources/data_cost_input_value/`)를 고치지 않는다.
그 파일은 `_manifest.json` 지문으로 재현성을 거는 자리이고 원천 xlsx 에서 다시
생성되는 물건이라, 값을 채워 넣으면 ① 이미 그 판으로 계산한 스냅샷과 대조가 깨지고
② 다음 재생성 때 날아가고 ③ 주인이 겹친다.
⚠ **길게 보면 기준자료 원본이 채워지는 것이 맞다** — `mach_fuel_rate` 가 그 파일 안에
**빈 변수로 선언**돼 있다는 것이 「원래 거기 들어갈 값」이라는 뜻이다. 원천 재생성 몫이라
미결로 올려 둔다.
⚠ **표가 열 단위로 뭉쳐 있다.** PDF 에서 뽑히며 한 열이 한 칸에 공백으로 이어 붙었다 —
`분류번호` 칸 하나에 코드 31개, `주연료` 칸 하나에 값 31개가 들어 있다. 그래서
**위치로 짝짓고, 짝이 안 맞으면 그 표를 통째로 버린다.** 어긋난 채로 짝지으면 다른
기종의 연료값이 조용히 붙는다.
"""
from __future__ import annotations
import json
import os
import re
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any
_CATALOG_SUBPATH = ("resources", "data_cost_input_value")
_OUTPUT_SUBPATH = ("resources", "data_cost_machine_operating")
#: 품셈 8-4 운전경비 표의 머리글 — 이 여섯이 다 있어야 그 표로 본다.
_REQUIRED_HEADERS = ("분류번호", "기계명", "규격", "주연료", "잡재료", "조종원")
_RE_CODE_FULL = re.compile(r"^(\d{4})-(\d{4})$")
_RE_CODE_TAIL = re.compile(r"^\d{4}$")
_RE_DECIMAL = re.compile(r"^\d+(?:\.\d+)?$")
#: 연료 종류가 값 앞에 붙는 경우 — 「휘발유0.7」·「중유487.2」.
_RE_FUEL_WITH_KIND = re.compile(r"^(휘발유|중유|경유)?(\d+(?:\.\d+)?)$")
#: 조종원 직종 — 품셈 표는 「인/일」 수만 주고 직종명을 안 준다.
#: TODO(미결 PLAN 9-6): 기종별 직종이 품셈 다른 장에 있다. 아래는 `aliases` 기반 **잠정**이며
#: 결과에 `operator_mapping_is_provisional: true` 로 드러난다.
_OPERATOR_TRUCK_WORDS = ("덤프트럭", "트럭", "트레일러", "화물")
_OPERATOR_ALIAS_TRUCK = "labor_op_truck"
_OPERATOR_ALIAS_CONSTRUCTION = "labor_op_const"
class OperatingCostError(ValueError):
"""운전경비 표를 못 읽은 경우. 어긋난 채로 짝짓지 않는다."""
def _project_root() -> str:
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def _read_json(*parts: str) -> dict[str, Any]:
with open(os.path.join(_project_root(), *parts), encoding="utf-8") as handle:
return json.load(handle)
@dataclass(frozen=True)
class OperatingRecord:
"""기종 하나의 운전경비 원단위."""
machine_code: str
machine_name: str
specification: str
fuel_liters_per_hour: Decimal | None
fuel_kind: str
misc_material_percent: Decimal | None
operator_person_days: Decimal | None
operator_occupation_code: str = ""
operator_mapping_is_provisional: bool = True
def as_dict(self) -> dict[str, Any]:
return {
"machine_code": self.machine_code,
"machine_name": self.machine_name,
"specification": self.specification,
"fuel_liters_per_hour": (
None if self.fuel_liters_per_hour is None else str(self.fuel_liters_per_hour)
),
"fuel_kind": self.fuel_kind,
"misc_material_percent": (
None if self.misc_material_percent is None else str(self.misc_material_percent)
),
"operator_person_days": (
None if self.operator_person_days is None else str(self.operator_person_days)
),
"operator_occupation_code": self.operator_occupation_code,
"operator_mapping_is_provisional": self.operator_mapping_is_provisional,
}
@dataclass
class OperatingParseResult:
records: list[OperatingRecord] = field(default_factory=list)
#: 짝이 안 맞아 버린 표 — 조용히 넘기지 않고 센다.
dropped_tables: list[str] = field(default_factory=list)
def _tokens(cell: str) -> list[str]:
return [t for t in re.split(r"\s+", str(cell or "").strip()) if t]
def expand_codes(tokens: list[str], carry_prefix: str = "") -> list[str]:
"""`0201-0012 0020 0040` → `0201-0012 · 0201-0020 · 0201-0040`.
뒤 코드는 앞 코드의 **앞 네 자리를 이어받는다**. 이어받을 앞자리가 없으면 버린다.
⚠ `carry_prefix` — **앞 표의 마지막 앞자리**. 원문이 페이지에서 잘리면 이어지는 판이
앞자리를 다시 안 적는다(8-4-3 이 `0080 0100 …` 로 시작한다). 그것을 못 이어받아
**운반·하역기계 39 기종의 운전경비가 통째로 버려지고 있었다** — 그 안에 수송비를 내는
**트럭 트랙터 및 평판트레일러(2702)** 가 들어 있다(2026-09-09 밤 실측).
⚠ 이어받은 코드는 **카탈로그에 있는지 확인한 뒤에만** 쓴다(`parse_operating_tables`).
"""
codes: list[str] = []
prefix = carry_prefix
for token in tokens:
full = _RE_CODE_FULL.match(token)
if full:
prefix = full.group(1)
codes.append(token)
continue
if _RE_CODE_TAIL.match(token) and prefix:
codes.append(f"{prefix}-{token}")
continue
return [] # 코드 열이 아닌 표
return codes
def _parse_fuel(token: str) -> tuple[Decimal | None, str]:
"""「11.6」·「휘발유0.7」·「-」 를 값과 연료 종류로 가른다."""
text = token.strip()
if text in ("-", "", ""):
return None, ""
match = _RE_FUEL_WITH_KIND.match(text)
if not match:
return None, ""
kind = match.group(1) or "경유"
return Decimal(match.group(2)), kind
def _parse_percent(token: str) -> Decimal | None:
text = token.strip().rstrip("%")
return Decimal(text) if _RE_DECIMAL.match(text) else None
def _parse_person_days(token: str) -> Decimal | None:
text = token.strip()
return Decimal(text) if _RE_DECIMAL.match(text) else None
def _operator_code(machine_name: str, aliases: dict[str, str]) -> str:
"""기종 이름으로 운전사 직종을 고른다 — **잠정 규칙**.
품셈 8-4 표는 「조종원 인/일」 수만 주고 직종명을 안 준다. 노임표의 `aliases` 가
운전사 직종 셋(`labor_op_const`·`labor_op_truck`·`labor_op_general`)을 들고 있어
트럭 계열만 화물차운전사로, 나머지는 건설기계운전사로 **잠정** 매핑한다.
"""
key = (
_OPERATOR_ALIAS_TRUCK
if any(word in machine_name for word in _OPERATOR_TRUCK_WORDS)
else _OPERATOR_ALIAS_CONSTRUCTION
)
return aliases.get(key, "")
def parse_operating_tables(
pum: dict[str, Any],
aliases: dict[str, str],
known_codes: set[str] | None = None,
) -> OperatingParseResult:
"""품셈 표 뭉치에서 8-4 운전경비 표만 골라 기종별 원단위를 만든다.
⚠ `known_codes` — 기종 카탈로그의 코드. **앞 표에서 앞자리를 이어받을 때만** 쓰며,
이어받아 만든 코드가 하나라도 카탈로그에 없으면 **그 줄을 통째로 버린다.**
이어받기는 원문이 페이지에서 잘린 자리를 잇는 것이지 코드를 지어내는 것이 아니다.
"""
result = OperatingParseResult()
carry_prefix = ""
for table in pum.get("tables", []):
headers = table.get("headers") or []
joined = " ".join(headers)
if not all(word in joined for word in _REQUIRED_HEADERS):
continue
for row in table.get("rows") or []:
if len(row) < 6:
continue
codes = expand_codes(_tokens(row[0]))
if not codes and carry_prefix:
carried = expand_codes(_tokens(row[0]), carry_prefix)
# ⚠ 이어받은 코드가 **전부 카탈로그에 있을 때만** 받는다 — 하나라도 없으면
# 앞자리를 잘못 물어 온 것이므로 종전대로 버린다.
if carried and known_codes and all(code in known_codes for code in carried):
codes = carried
if codes:
carry_prefix = codes[-1][:4]
names = _tokens(row[1])
specs = _tokens(row[2])
fuels = _tokens(row[3])
miscs = _tokens(row[4])
operators = _tokens(row[5])
# ⚠ 위치로 짝짓는다 — 개수가 안 맞으면 그 줄을 통째로 버린다.
if not codes or not (len(codes) == len(specs) == len(fuels) == len(operators)):
result.dropped_tables.append(f"{table.get('section', '')[:40]} (개수 불일치)")
continue
# ⚠ 이름 칸은 **표 전체의 기종명이 한 덩어리로** 들어온다
# (「불도저(무한궤도)불도저(타이어)습지불도저굴착기(무한궤도)…」).
# 코드별로 못 가르므로 **이름은 여기서 안 쓴다** — 613 기종 카탈로그에서
# `machine_code` 로 찾아 붙인다(`enrich_with_catalog`).
_ = names
for index, code in enumerate(codes):
fuel, kind = _parse_fuel(fuels[index])
misc = _parse_percent(miscs[index]) if index < len(miscs) else None
result.records.append(
OperatingRecord(
machine_code=code,
machine_name="", # 카탈로그에서 채운다
specification=specs[index],
fuel_liters_per_hour=fuel,
fuel_kind=kind,
misc_material_percent=misc,
operator_person_days=_parse_person_days(operators[index]),
operator_occupation_code="", # 이름을 안 뒤 정한다
)
)
return result
def enrich_with_catalog(
result: OperatingParseResult,
catalog: Any,
aliases: dict[str, str],
) -> OperatingParseResult:
"""코드로 613 기종 카탈로그에서 **이름·규격을 가져와** 채운다.
품셈 표의 이름 칸은 통째로 뭉쳐 와 못 쓴다. 코드가 정본이다.
카탈로그에 없는 코드는 **버리고 목록에 남긴다** — 이름 없는 기종은 단가가 못 붙는다.
"""
enriched: list[OperatingRecord] = []
for record in result.records:
machine = catalog.machines.get(record.machine_code)
if machine is None:
result.dropped_tables.append(f"{record.machine_code} (기종 카탈로그에 없음)")
continue
enriched.append(
OperatingRecord(
machine_code=record.machine_code,
machine_name=machine.name,
specification=machine.specification or record.specification,
fuel_liters_per_hour=record.fuel_liters_per_hour,
fuel_kind=record.fuel_kind,
misc_material_percent=record.misc_material_percent,
operator_person_days=record.operator_person_days,
operator_occupation_code=_operator_code(machine.name, aliases),
)
)
result.records = enriched
return result
def load_operating_records(
pum_file: str = "pum_const_2026.json",
labor_file: str = "labor_const_2026-01-01.json",
) -> OperatingParseResult:
from B09_Estimation.B09_Estimation_MachineCost import load_machine_catalog
pum = _read_json(*_CATALOG_SUBPATH, pum_file)["variables"]["pum"]
aliases = _read_json(*_CATALOG_SUBPATH, labor_file)["variables"].get("aliases", {})
catalog = load_machine_catalog()
parsed = parse_operating_tables(pum, aliases, set(catalog.machines))
return enrich_with_catalog(parsed, catalog, aliases)
def write_operating_records(
result: OperatingParseResult,
*,
source_dataset_version: dict[str, str],
output_dir: str | None = None,
) -> str:
"""파생 파일로 낸다. **어느 기준자료 판에서 파생됐는지**를 함께 적는다."""
directory = output_dir or os.path.join(_project_root(), *_OUTPUT_SUBPATH)
os.makedirs(directory, exist_ok=True)
path = os.path.join(directory, "machine_operating_2026.json")
payload = {
"schema_version": "1.0",
"dataset_id": "machine_operating_derived",
"derived_from": source_dataset_version,
"note": (
"건설품셈 8-4 운전경비 산정에서 뽑은 파생본. 기준자료가 아니다 — "
"`mach_fuel_rate`·`mach_operator_map` 이 채워지면 이 파일은 걷어낸다."
),
"policy": {
"operator_mapping_is_provisional": True,
"operator_mapping_rule": "트럭 계열 = 화물차운전사, 그 밖 = 건설기계운전사 (잠정)",
},
"stats": {"records": len(result.records), "dropped": len(result.dropped_tables)},
"records": [r.as_dict() for r in result.records],
"dropped_tables": result.dropped_tables,
}
with open(path, "w", encoding="utf-8", newline="\n") as handle:
json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
return path
#: 시도별 유가 판 — 품셈 8-1-7 5호 「유류가격은 **해당지역의 가격**으로 한다」.
#: ⚠ **파일이 있을 때만 지역을 고를 수 있다** — 없으면 전국평균 한 벌로 돈다(코드로 막지 않음).
REGIONAL_OIL_FILE = "oil_regional_2026-09-09.json"
def load_regional_fuel_table(oil_file: str = REGIONAL_OIL_FILE) -> tuple[dict[str, dict], dict]:
"""(시도코드 → {이름·값}, 판 신원). 판이 없으면 **빈 표**를 돌려준다.
⚠ 원문에 있는 코드 `00`(전국)은 **지역 선택지에서 뺀다** — 그 자리는 전국평균 판이
맡고, 두 판이 같은 이름으로 서면 어느 값으로 섰는지 못 가린다.
"""
try:
payload = _read_json(*_CATALOG_SUBPATH, oil_file)
except FileNotFoundError:
return {}, {}
diesel = payload["variables"]["oil_diesel"]
table = {
str(record["sido_code"]): {
"name": str(record["sido_name"]),
"value": Decimal(str(record["value"])),
}
for record in diesel.get("records", [])
if str(record["sido_code"]) != "00"
}
meta = {
"dataset_id": payload.get("dataset_id", ""),
"effective_date": payload.get("effective_date", ""),
"scope": diesel.get("scope", ""),
}
return table, meta
def load_fuel_price(
oil_file: str = "oil_2026-08-14.json", region: str | None = None
) -> tuple[Decimal, dict[str, str]]:
"""경유 단가와 그 판의 신원. `region`(시도코드)을 주면 **그 지역 값**으로 선다.
품셈 8-1-7 5호가 「유류가격은 **해당 지역의 가격**」이라 규정한다.
⚠ **안 주면 전국평균**이다 — 현장 소재지를 임의로 찍지 않는다(프로젝트가 고름).
⚠ 준 지역이 판에 없으면 **조용히 전국평균으로 눕지 않고** 그 사실을 신원에 적는다.
"""
payload = _read_json(*_CATALOG_SUBPATH, oil_file)
diesel = payload["variables"]["oil_diesel"]
meta = {
"dataset_id": payload.get("dataset_id", ""),
"effective_date": payload.get("effective_date", ""),
"scope": diesel.get("scope", ""),
"region": "",
"region_name": "",
}
if not region:
return Decimal(str(diesel["value"])), meta
table, region_meta = load_regional_fuel_table()
picked = table.get(str(region))
if picked is None:
meta["region"] = str(region)
meta["region_missing"] = "그 지역 값이 판에 없어 전국평균으로 섰습니다"
return Decimal(str(diesel["value"])), meta
return picked["value"], {
"dataset_id": region_meta.get("dataset_id", ""),
"effective_date": region_meta.get("effective_date", ""),
"scope": region_meta.get("scope", ""),
"region": str(region),
"region_name": picked["name"],
}
def load_operator_wages(labor_file: str = "labor_const_2026-01-01.json") -> dict[str, Decimal]:
"""직종코드 → 일당."""
records = _read_json(*_CATALOG_SUBPATH, labor_file)["variables"]["labor_rate"]["records"]
return {
str(r["occupation_code"]): Decimal(str(r["daily_wage_krw"]))
for r in records
if "daily_wage_krw" in r
}
#: 노임 신뢰도 기호 — 원문 「임금적용요령」이 직종 옆에 붙여 둔 표시.
#: `*` 조사현장 5개 미만 (적용 시 유의)
#: `**` 미조사 — 유사 직종 단가에 준하여 적용(재경원 회계 45101-45)
#: ⚠ **값이 없는 것과 다르다** — 단가는 서 있되 **표본이 얇다**는 뜻이라 막지 않고 알린다.
#: 지식DB `노임단가_적용 §2-3` 이 「프로그램에서 단가 채택 시 플래그 유지 필요」로 두었고,
#: `원가_입력변수_사전` 도 「해당 단가 사용 시 경고 표시」라 적었는데 노임 층에 없었다
#: (2026-09-09에 이음).
LABOR_RELIABILITY_LABEL = {
"*": "조사현장 5개 미만 — 표본이 얇습니다",
"**": "미조사 직종 — 유사 직종 단가에 준해 적용(재경원 회계 45101-45)",
}
def load_labor_reliability(labor_file: str = "labor_const_2026-01-01.json") -> dict[str, str]:
"""직종코드 → 신뢰도 기호(`*`·`**`). 정상 공표 직종은 아예 안 담는다."""
records = _read_json(*_CATALOG_SUBPATH, labor_file)["variables"]["labor_rate"]["records"]
return {
str(r["occupation_code"]): str(r["reliability"])
for r in records
if str(r.get("reliability") or "").strip()
}
def hourly_cost_of(machine_code: str, *, region: str | None = None):
"""기종 하나의 **시간당 사용료 3분할**을 완성해 돌려준다.
재료비 = 주연료 × 유가 + 잡재료(주연료의 %) / 노무비 = 조종원 일당 ÷ 8시간 /
경비 = 손료. `region`(시도코드)을 주면 **그 지역 유가**로 선다(품셈 8-1-7 5호).
"""
from B09_Estimation.B09_Estimation_MachineCost import hourly_machine_cost, load_machine_catalog
catalog = load_machine_catalog()
machine = catalog.get(machine_code)
records = {r.machine_code: r for r in load_operating_records().records}
record = records.get(machine_code)
if record is None:
return hourly_machine_cost(machine)
fuel_price, _ = load_fuel_price(region=region)
wages = load_operator_wages()
liters = record.fuel_liters_per_hour
if liters is not None and record.misc_material_percent is not None:
# 잡재료는 **주연료의 %** 라 유가와 같이 움직인다(PLAN 8-18 유가 민감분).
liters = liters * (Decimal(1) + record.misc_material_percent / Decimal(100))
wage = wages.get(record.operator_occupation_code)
if record.operator_person_days is not None and wage is not None:
wage = wage * record.operator_person_days
return hourly_machine_cost(
machine,
fuel_liters_per_hour=liters,
fuel_price_per_liter=fuel_price if liters is not None else None,
operator_daily_wage=wage,
)