Merge remote-tracking branch 'origin/sub_desktop_1' into main_desktop_1
This commit is contained in:
@@ -0,0 +1,185 @@
|
||||
"""B09 원가계산 — 기계경비 (PLAN 9-3 의 `S` → `X` 두 단계).
|
||||
|
||||
실무·교본이 같은 두 단계다 (신규 문서 5장 라-1 예제도 같은 모양):
|
||||
|
||||
S 취득가(천원) ──(손료계수)──► X 시간당 중기사용료
|
||||
= 손료(경비) + 연료(재료) + 운전사(노무)
|
||||
|
||||
이 모듈이 내는 것은 **시간당 사용료 한 시간분**이고, 3분할(재료·노무·경비)로 낸다.
|
||||
`B09_Estimation_PriceBook` 의 `MACHINE_BASE`(S) · `MACHINE_HOURLY`(X) 층에 그대로 앉는다.
|
||||
|
||||
데이터 (`resources/data_cost_input_value/mach_base_2026.json`)
|
||||
- `mach_price` **613 기종** — `machine_code` + `specification` 으로 규격이 갈린다.
|
||||
- `mach_loss_coef` **387건** — 시간당 손료계수·내용시간·연간표준시간.
|
||||
- ⚠ `mach_fuel_rate` **0건** · `mach_operator_map` **0건** — **비어 있다.**
|
||||
연료소모량과 기종별 운전사 직종은 품셈 본문에 있고, 아직 뽑히지 않았다.
|
||||
|
||||
⚠ **그래서 이 모듈은 손료(경비)까지만 채우고, 연료·운전사는 「공백」으로 표시한다.**
|
||||
0 으로 때우면 **시간당 사용료가 절반 이하로 나오고 그대로 총액에 섞인다** — 실측
|
||||
비중이 노무 53 % · 재료 20 % · 경비 27 % 라 손료만으로는 4분의 1 남짓이다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from B09_Estimation.B09_Estimation_PriceBook import Money3
|
||||
|
||||
_CATALOG_SUBPATH = ("resources", "data_cost_input_value")
|
||||
_THOUSAND = Decimal(1000)
|
||||
|
||||
|
||||
class MachineCostError(LookupError):
|
||||
"""기계경비를 세울 수 없는 경우. 0 으로 때우지 않고 멈춘다."""
|
||||
|
||||
|
||||
def _project_root() -> str:
|
||||
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
def _read_json(file_name: str) -> dict[str, Any]:
|
||||
with open(os.path.join(_project_root(), *_CATALOG_SUBPATH, file_name), encoding="utf-8") as h:
|
||||
return json.load(h)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MachineSpec:
|
||||
"""기종 한 줄 — 규격까지 붙어야 한 대가 정해진다."""
|
||||
|
||||
machine_code: str
|
||||
name: str
|
||||
specification: str
|
||||
price_thousand_krw: Decimal
|
||||
loss_coefficient_per_hour: Decimal | None = None
|
||||
economic_life_hours: int | None = None
|
||||
annual_standard_hours: int | None = None
|
||||
|
||||
@property
|
||||
def display_name(self) -> str:
|
||||
return f"{self.name} {self.specification}".strip()
|
||||
|
||||
|
||||
@dataclass
|
||||
class MachineCatalog:
|
||||
"""613 기종. **이름만으로는 못 고른다** — 규격이 있어야 한 대가 정해진다."""
|
||||
|
||||
machines: dict[str, MachineSpec] = field(default_factory=dict)
|
||||
|
||||
def by_name(self, name: str) -> list[MachineSpec]:
|
||||
return [m for m in self.machines.values() if m.name == name]
|
||||
|
||||
def resolve(self, name: str, specification: str) -> MachineSpec | None:
|
||||
"""이름 + 규격으로 한 대를 고른다. 규격이 없으면 **고르지 않는다**."""
|
||||
found = self.by_name(name)
|
||||
if not found:
|
||||
return None
|
||||
if len(found) == 1 and not specification:
|
||||
return found[0]
|
||||
narrowed = [m for m in found if m.specification == specification]
|
||||
return narrowed[0] if len(narrowed) == 1 else None
|
||||
|
||||
def get(self, machine_code: str) -> MachineSpec:
|
||||
try:
|
||||
return self.machines[machine_code]
|
||||
except KeyError as exc:
|
||||
raise MachineCostError(f"기종 카탈로그에 없는 코드입니다: {machine_code}") from exc
|
||||
|
||||
|
||||
def load_machine_catalog(file_name: str = "mach_base_2026.json") -> MachineCatalog:
|
||||
"""취득가 613건에 손료계수 387건을 붙여 카탈로그 한 벌을 만든다."""
|
||||
variables = _read_json(file_name)["variables"]
|
||||
coefficients = {
|
||||
r["machine_code"]: r for r in variables.get("mach_loss_coef", {}).get("records", [])
|
||||
}
|
||||
|
||||
catalog = MachineCatalog()
|
||||
for row in variables.get("mach_price", {}).get("records", []):
|
||||
code = row["machine_code"]
|
||||
coefficient = coefficients.get(code, {})
|
||||
catalog.machines[code] = MachineSpec(
|
||||
machine_code=code,
|
||||
name=row["machine_name"],
|
||||
specification=str(row.get("specification", "")),
|
||||
price_thousand_krw=Decimal(str(row["price_thousand_krw"])),
|
||||
loss_coefficient_per_hour=(
|
||||
Decimal(str(coefficient["loss_coefficient_per_hour"]))
|
||||
if "loss_coefficient_per_hour" in coefficient
|
||||
else None
|
||||
),
|
||||
economic_life_hours=coefficient.get("economic_life_hours"),
|
||||
annual_standard_hours=coefficient.get("annual_standard_hours"),
|
||||
)
|
||||
return catalog
|
||||
|
||||
|
||||
def hourly_loss_cost(machine: MachineSpec) -> Decimal:
|
||||
"""시간당 손료 = 취득가 × 손료계수.
|
||||
|
||||
취득가가 **천원 단위**라 원으로 환산한다 — 이 단위를 놓치면 1,000배 틀린다.
|
||||
"""
|
||||
if machine.loss_coefficient_per_hour is None:
|
||||
raise MachineCostError(
|
||||
f"{machine.display_name}: 손료계수가 없습니다 (취득가만 있는 기종 226건 중 하나)"
|
||||
)
|
||||
return machine.price_thousand_krw * _THOUSAND * machine.loss_coefficient_per_hour
|
||||
|
||||
|
||||
@dataclass
|
||||
class HourlyMachineCost:
|
||||
"""시간당 중기사용료 — 3분할과 **채우지 못한 성분 목록**을 함께 낸다."""
|
||||
|
||||
machine: MachineSpec
|
||||
money: Money3
|
||||
gaps: list[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def is_complete(self) -> bool:
|
||||
return not self.gaps
|
||||
|
||||
|
||||
def hourly_machine_cost(
|
||||
machine: MachineSpec,
|
||||
*,
|
||||
fuel_liters_per_hour: Decimal | None = None,
|
||||
fuel_price_per_liter: Decimal | None = None,
|
||||
operator_daily_wage: Decimal | None = None,
|
||||
operator_hours_per_day: int = 8,
|
||||
) -> HourlyMachineCost:
|
||||
"""시간당 사용료 한 시간분.
|
||||
|
||||
손료(경비)는 카탈로그로 바로 나온다. **연료(재료)·운전사(노무)는 값을 주지 않으면
|
||||
0 으로 때우지 않고 `gaps` 에 적어 돌려준다** — 빠진 채로 총액에 섞이는 것이
|
||||
이 자리에서 제일 위험하다.
|
||||
"""
|
||||
gaps: list[str] = []
|
||||
expense = hourly_loss_cost(machine)
|
||||
|
||||
material = Decimal(0)
|
||||
if fuel_liters_per_hour is None or fuel_price_per_liter is None:
|
||||
# TODO(미결 PLAN 9-6): `mach_fuel_rate` 0건 — 연료소모량이 품셈 본문에만 있다.
|
||||
gaps.append("연료소모량(L/hr) 미확보 — 재료비 성분 비어 있음")
|
||||
else:
|
||||
material = fuel_liters_per_hour * fuel_price_per_liter
|
||||
|
||||
labor = Decimal(0)
|
||||
if operator_daily_wage is None:
|
||||
# TODO(미결 PLAN 9-6): `mach_operator_map` 0건 — 기종별 운전사 직종이 품셈 본문에만 있다.
|
||||
gaps.append("운전사 직종 매핑 미확보 — 노무비 성분 비어 있음")
|
||||
else:
|
||||
labor = operator_daily_wage / Decimal(operator_hours_per_day)
|
||||
|
||||
return HourlyMachineCost(
|
||||
machine=machine,
|
||||
money=Money3(material=material, labor=labor, expense=expense),
|
||||
gaps=gaps,
|
||||
)
|
||||
|
||||
|
||||
def catalog_gaps(catalog: MachineCatalog) -> dict[str, int]:
|
||||
"""카탈로그 자체의 공백 — 몇 기종이 손료계수를 못 가졌나."""
|
||||
missing = [m for m in catalog.machines.values() if m.loss_coefficient_per_hour is None]
|
||||
return {"machines": len(catalog.machines), "without_loss_coefficient": len(missing)}
|
||||
@@ -0,0 +1,512 @@
|
||||
"""B09 원가계산 — 자원 축 매칭 (PLAN 8-6 · 9-3).
|
||||
|
||||
메인 창이 낸 **공종 축**(`resources/data_work_item_master/`)을 **읽기만** 하고, 그 위에
|
||||
**자원 축**(`resource_kind`·`resource_code`·`resource_spec`·`amount`·`amount_unit`)을
|
||||
붙여 **별도 파일**로 낸다. 원본은 고치지 않는다 — 메인이 품셈을 다시 돌리면 덮이므로
|
||||
그 안에 섞으면 사라진다.
|
||||
|
||||
지켜야 할 것
|
||||
1. **`pum_form` 을 먼저 본다.** `productivity`(생산량형) = **1 ÷ 값**,
|
||||
`requirement`(소요량형) = **값 ÷ basis_quantity**. ⚠ **뒤집으면 20배 틀린다.**
|
||||
직종 이름부터 보면 「작업능력(㎥/hr)」 표의 비고란 「보통인부 1인/일」에 끌려
|
||||
생산량형을 소요량형으로 읽는다(메인이 실제로 한 번 뒤집혔다가 잡은 자리).
|
||||
2. **`coefficient` · `reference` 는 공종이 아니다.** 일위대가 항목으로 세우지 않는다.
|
||||
`undetermined` 는 **값을 쓰지 않는다.**
|
||||
3. **규격(`resource_spec`)이 없으면 매칭 성공으로 치지 않는다.** 「굴착기」와
|
||||
「굴착기 0.7㎥」는 단가가 다르다 — 이름만 맞추면 조용히 틀린 단가가 붙는다.
|
||||
4. **못 맞춘 것은 빈칸이 아니라 `unmatched` 목록**으로 낸다.
|
||||
5. **자재는 할증 전 값**이다 (PLAN 8-7 ㉠). 할증은 자재총괄 한 곳뿐이다.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from typing import Any
|
||||
|
||||
#: 자원 축을 붙일 수 있는 표 형태. 나머지는 값을 쓰지 않는다.
|
||||
USABLE_FORMS = frozenset({"productivity", "requirement"})
|
||||
#: 공종이 아닌 표 — 일위대가 항목으로 세우지 않는다.
|
||||
NON_WORK_ITEM_FORMS = frozenset({"coefficient", "reference"})
|
||||
#: 형태 판정이 안 된 표 — 값을 쓰지 않는다.
|
||||
UNUSABLE_FORMS = frozenset({"undetermined"})
|
||||
|
||||
_MASTER_SUBPATH = ("resources", "data_work_item_master")
|
||||
_CATALOG_SUBPATH = ("resources", "data_cost_input_value")
|
||||
|
||||
#: 규격이 이름 안에 붙어 있는 흔한 모양 — 「굴착기(0.7㎥)」·「덤프트럭 15톤」.
|
||||
_RE_SPEC = re.compile(r"[(（]([^)）]+)[)）]|(\d+(?:\.\d+)?\s*(?:톤|ton|㎥|m3|㎡|㎜|mm|HP|kW))")
|
||||
_RE_NUMBER = re.compile(r"^-?\d+(?:,\d{3})*(?:\.\d+)?$")
|
||||
|
||||
#: 표 머리글·소계 행의 첫 칸에 흔히 오는 말. 자원이 아니므로 `unmatched` 로도 안 올린다.
|
||||
#: 이것을 안 거르면 못 맞춘 목록이 머리글로 가득 차 **쓸 수 없는 목록**이 된다.
|
||||
_NON_RESOURCE_WORDS = (
|
||||
"구분",
|
||||
"합계",
|
||||
"소계",
|
||||
"계",
|
||||
"단위",
|
||||
"비고",
|
||||
"규격",
|
||||
"명칭",
|
||||
"품명",
|
||||
"종류",
|
||||
"항목",
|
||||
"적용",
|
||||
"기준",
|
||||
"산출",
|
||||
"비율",
|
||||
"할증",
|
||||
"할인",
|
||||
"직접노무비",
|
||||
"재료비",
|
||||
"경비",
|
||||
"위치",
|
||||
"면적",
|
||||
"수량",
|
||||
"공종",
|
||||
"작업",
|
||||
"내역",
|
||||
"총계",
|
||||
"인력",
|
||||
"장비",
|
||||
"기계",
|
||||
)
|
||||
|
||||
|
||||
class ResourceAxisError(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 CatalogEntry:
|
||||
"""단가 카탈로그 한 줄 — 매칭 대상."""
|
||||
|
||||
code: str
|
||||
name: str
|
||||
kind: str
|
||||
spec: str = ""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResourceCatalog:
|
||||
"""이름 → 코드. **같은 이름에 규격이 여럿이면 규격 없이는 못 고른다.**"""
|
||||
|
||||
entries: list[CatalogEntry] = field(default_factory=list)
|
||||
aliases: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
def by_name(self, name: str) -> list[CatalogEntry]:
|
||||
cleaned = _normalize(name)
|
||||
return [e for e in self.entries if _normalize(e.name) == cleaned]
|
||||
|
||||
def resolve(self, name: str, spec: str) -> CatalogEntry | None:
|
||||
"""이름(+규격)으로 한 줄을 고른다. 못 고르면 None — 0 으로 안 때운다."""
|
||||
found = self.by_name(name)
|
||||
if not found:
|
||||
return None
|
||||
if len(found) == 1:
|
||||
return found[0]
|
||||
# 이름이 여럿이면 규격이 있어야 고를 수 있다.
|
||||
if not spec:
|
||||
return None
|
||||
narrowed = [e for e in found if _normalize(e.spec) == _normalize(spec)]
|
||||
return narrowed[0] if len(narrowed) == 1 else None
|
||||
|
||||
|
||||
#: 첫 칸이 자원 이름이 **아닌** 표가 많다 — 규격 구간표(「10∼12」), 기호표(「f」·「E」),
|
||||
#: 치수표 등. 그런 셀을 못 맞춘 목록에 넣으면 목록이 못 쓰게 되므로 먼저 거른다.
|
||||
_RE_RANGE_CELL = re.compile(r"^\d+(?:\.\d+)?\s*[∼~〜\-–]\s*\d+(?:\.\d+)?$")
|
||||
_RE_HANGUL = re.compile(r"[가-힣]")
|
||||
|
||||
|
||||
def is_non_resource_label(cell: str) -> bool:
|
||||
"""표 머리글·소계 행이거나, 애초에 자원 이름이 올 자리가 아닌 셀인가.
|
||||
|
||||
못 맞춘 목록에 이런 것이 섞이면 목록 자체가 못 쓰게 된다. 여기서 먼저 걷어낸다.
|
||||
"""
|
||||
text = _normalize(cell)
|
||||
if not text:
|
||||
return True
|
||||
if text.startswith(("※", "<", "(", "-", "ㆍ", "·")):
|
||||
return True
|
||||
if _RE_RANGE_CELL.match(text): # 규격 구간표의 첫 칸
|
||||
return True
|
||||
# 자원 이름은 숫자로 시작하지 않는다 — 「50이상」·「100m이하」·「2.집재」는 구간·절번호다.
|
||||
if text[0].isdigit():
|
||||
return True
|
||||
# 자원 이름은 한글 두 자 이상이다. 기호(`f`·`E`)·숫자·단위만 있는 칸은 자원이 아니다.
|
||||
if len(_RE_HANGUL.findall(text)) < 2:
|
||||
return True
|
||||
return any(word in text for word in _NON_RESOURCE_WORDS)
|
||||
|
||||
|
||||
def _normalize(text: str) -> str:
|
||||
"""표 셀의 공백·개행 흔들림을 지운다. 「경 암」·「연 암」 같은 것."""
|
||||
return re.sub(r"\s+", "", str(text or "")).strip()
|
||||
|
||||
|
||||
def load_labor_catalog(file_name: str = "labor_const_2026-01-01.json") -> ResourceCatalog:
|
||||
"""노임 카탈로그 132직종 + `aliases`. 코드는 `occupation_code`."""
|
||||
payload = _read_json(*_CATALOG_SUBPATH, file_name)
|
||||
variables = payload["variables"]
|
||||
records = variables["labor_rate"]["records"]
|
||||
entries = [
|
||||
CatalogEntry(code=str(r["occupation_code"]), name=r["occupation_name"], kind="labor")
|
||||
for r in records
|
||||
]
|
||||
return ResourceCatalog(entries=entries, aliases=dict(variables.get("aliases", {})))
|
||||
|
||||
|
||||
def load_machine_catalog_entries(file_name: str = "mach_base_2026.json") -> list[CatalogEntry]:
|
||||
"""기종 카탈로그 613건을 매칭용 항목으로 편다.
|
||||
|
||||
⚠ **규격이 매칭의 일부**다 — 「굴착기(무한궤도)」만 23 규격이라 이름만으로는
|
||||
한 대가 안 정해진다(지시 4번). `CatalogEntry.spec` 에 규격을 실어 둔다.
|
||||
"""
|
||||
from B09_Estimation.B09_Estimation_MachineCost import load_machine_catalog
|
||||
|
||||
catalog = load_machine_catalog(file_name)
|
||||
return [
|
||||
CatalogEntry(code=m.machine_code, name=m.name, kind="machine", spec=m.specification)
|
||||
for m in catalog.machines.values()
|
||||
]
|
||||
|
||||
|
||||
def load_combined_catalog() -> ResourceCatalog:
|
||||
"""노임 + 기종을 한 벌로. 자재는 카탈로그가 아직 없다."""
|
||||
labor = load_labor_catalog()
|
||||
return ResourceCatalog(
|
||||
entries=[*labor.entries, *load_machine_catalog_entries()],
|
||||
aliases=labor.aliases,
|
||||
)
|
||||
|
||||
|
||||
def load_work_item_master(
|
||||
file_name: str = "work_item_master_2026-01-01.json",
|
||||
) -> dict[str, Any]:
|
||||
"""메인 창 산출물 — **읽기 전용**."""
|
||||
return _read_json(*_MASTER_SUBPATH, file_name)
|
||||
|
||||
|
||||
def split_name_and_spec(cell: str) -> tuple[str, str]:
|
||||
"""셀 문자열에서 이름과 규격을 가른다.
|
||||
|
||||
「유압식백호우 (무한궤도,0.7㎥)」 → (`유압식백호우`, `무한궤도,0.7㎥`)
|
||||
「덤프트럭 15톤」 → (`덤프트럭`, `15톤`)
|
||||
"""
|
||||
text = str(cell or "").strip()
|
||||
match = _RE_SPEC.search(text)
|
||||
if not match:
|
||||
return text, ""
|
||||
spec = (match.group(1) or match.group(2) or "").strip()
|
||||
name = (text[: match.start()] + text[match.end() :]).strip(" ,()（）")
|
||||
return name, spec
|
||||
|
||||
|
||||
#: 기종 이름의 괄호 안은 **규격과 형식이 섞여** 있다 —
|
||||
#: 「굴착기(무한궤도, 0.7㎥)」 는 이름 `굴착기(무한궤도)` + 규격 `0.7` 이다.
|
||||
#: 형식(무한궤도·타이어)은 이름의 일부이고, 숫자가 든 조각만 규격이다.
|
||||
_RE_MACHINE = re.compile(r"^(?P<base>[^()（）]+)[(（](?P<inner>[^)）]*)[)）]")
|
||||
_RE_SIZE_TOKEN = re.compile(r"\d+(?:\.\d+)?")
|
||||
|
||||
|
||||
def parse_machine_cell(cell: str) -> tuple[str, str]:
|
||||
"""기종 셀을 카탈로그 이름과 규격으로 가른다.
|
||||
|
||||
「굴착기(무한궤도, 0.7㎥)」 → (`굴착기(무한궤도)`, `0.7`)
|
||||
「굴착기 (무한궤도)」 → (`굴착기(무한궤도)`, ``) ← 규격은 옆 칸에 있다
|
||||
괄호가 없으면 원문 그대로 돌려준다.
|
||||
"""
|
||||
text = _normalize(cell)
|
||||
match = _RE_MACHINE.match(text)
|
||||
if not match:
|
||||
return text, ""
|
||||
base = match.group("base")
|
||||
parts = [p for p in re.split(r"[,·/]", match.group("inner")) if p]
|
||||
form_parts = [p for p in parts if not _RE_SIZE_TOKEN.search(p)]
|
||||
size_parts = [p for p in parts if _RE_SIZE_TOKEN.search(p)]
|
||||
name = f"{base}({','.join(form_parts)})" if form_parts else base
|
||||
spec = ""
|
||||
if size_parts:
|
||||
found = _RE_SIZE_TOKEN.search(size_parts[0])
|
||||
spec = found.group(0) if found else ""
|
||||
return name, spec
|
||||
|
||||
|
||||
def spec_candidates(cells: list[str]) -> list[str]:
|
||||
"""규격이 옆 칸에 있는 표가 많다 — 뒷 칸들에서 규격 후보를 모은다.
|
||||
|
||||
실측 배치: `['굴착기+부착용집게', '0.2㎥', 'hr', '2.71', …]` ·
|
||||
`['굴착기 (무한궤도)', '굴착기(무한궤도,0.2㎥)', 'hr', '0.80', …]`
|
||||
"""
|
||||
found: list[str] = []
|
||||
for cell in cells:
|
||||
text = _normalize(cell)
|
||||
if not text or len(text) > 30:
|
||||
continue
|
||||
_, spec = parse_machine_cell(text)
|
||||
if spec:
|
||||
found.append(spec)
|
||||
continue
|
||||
token = _RE_SIZE_TOKEN.fullmatch(text.rstrip("㎥㎡톤tonm³"))
|
||||
if token:
|
||||
found.append(token.group(0))
|
||||
return found
|
||||
|
||||
|
||||
def parse_amount(cell: str) -> Decimal | None:
|
||||
"""숫자 셀만 값으로 본다. 숫자가 아니면 None — 억지로 읽지 않는다."""
|
||||
text = _normalize(cell)
|
||||
if not _RE_NUMBER.match(text):
|
||||
return None
|
||||
try:
|
||||
return Decimal(text.replace(",", ""))
|
||||
except InvalidOperation: # pragma: no cover - 정규식이 먼저 거른다
|
||||
return None
|
||||
|
||||
|
||||
def convert_amount(raw: Decimal, *, pum_form: str, basis_quantity: Decimal | None) -> Decimal:
|
||||
"""표 형태에 맞춰 소요량으로 환산한다.
|
||||
|
||||
⚠ **여기가 20배 틀리는 자리다.**
|
||||
- `productivity`(생산량형, 예 「㎥/1인/1일」) → **1 ÷ 값**
|
||||
- `requirement`(소요량형, 예 「100㎥당 인부 x인」) → **값 ÷ basis_quantity**
|
||||
"""
|
||||
if pum_form == "productivity":
|
||||
if raw == 0:
|
||||
raise ResourceAxisError("생산량이 0 이라 소요량으로 뒤집을 수 없습니다")
|
||||
return Decimal(1) / raw
|
||||
if pum_form == "requirement":
|
||||
divisor = basis_quantity if basis_quantity else Decimal(1)
|
||||
if divisor == 0:
|
||||
raise ResourceAxisError("기준 수량이 0 입니다")
|
||||
return raw / divisor
|
||||
raise ResourceAxisError(f"자원 축을 붙일 수 없는 표 형태입니다: {pum_form}")
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResourceRow:
|
||||
"""자원 축 한 줄 — 공종(표) 하나에 붙는 자원 하나."""
|
||||
|
||||
work_item_code: str
|
||||
pum_table_id: str
|
||||
pum_form: str
|
||||
resource_kind: str
|
||||
resource_code: str
|
||||
resource_name: str
|
||||
resource_spec: str
|
||||
amount: Decimal
|
||||
amount_unit: str
|
||||
raw_row_index: int
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"work_item_code": self.work_item_code,
|
||||
"pum_table_id": self.pum_table_id,
|
||||
"pum_form": self.pum_form,
|
||||
"resource_kind": self.resource_kind,
|
||||
"resource_code": self.resource_code,
|
||||
"resource_name": self.resource_name,
|
||||
"resource_spec": self.resource_spec,
|
||||
"amount": str(self.amount),
|
||||
"amount_unit": self.amount_unit,
|
||||
"raw_row_index": self.raw_row_index,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class UnmatchedRow:
|
||||
"""못 맞춘 것 — **빈칸으로 두지 않고 여기 모은다**."""
|
||||
|
||||
work_item_code: str
|
||||
pum_table_id: str
|
||||
cell: str
|
||||
reason: str
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"work_item_code": self.work_item_code,
|
||||
"pum_table_id": self.pum_table_id,
|
||||
"cell": self.cell,
|
||||
"reason": self.reason,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class AxisResult:
|
||||
rows: list[ResourceRow] = field(default_factory=list)
|
||||
unmatched: list[UnmatchedRow] = field(default_factory=list)
|
||||
skipped_forms: dict[str, int] = field(default_factory=dict)
|
||||
|
||||
|
||||
def _resolve_cell(catalog: ResourceCatalog, name_cell: str, cells: list[str]):
|
||||
"""셀 하나를 카탈로그 한 줄로 푼다 — 세 가지 모양을 차례로 시도한다.
|
||||
|
||||
① 셀 전체가 곧 이름 (「보통인부」)
|
||||
② 기종 셀 (「굴착기(무한궤도, 0.7㎥)」 → 이름 + 규격)
|
||||
③ 규격이 **옆 칸**에 있는 표 (「굴착기 (무한궤도)」 | 「0.7㎥」)
|
||||
"""
|
||||
machine_name, machine_spec = parse_machine_cell(name_cell)
|
||||
plain_name, plain_spec = split_name_and_spec(name_cell)
|
||||
|
||||
for name, spec in ((machine_name, machine_spec), (plain_name, plain_spec)):
|
||||
if not name:
|
||||
continue
|
||||
entry = catalog.resolve(name, spec)
|
||||
if entry is not None:
|
||||
return entry
|
||||
|
||||
# 이름은 맞는데 규격이 없어 못 고른 경우 — 옆 칸에서 규격을 찾는다.
|
||||
for name in (machine_name, plain_name):
|
||||
if len(catalog.by_name(name)) <= 1:
|
||||
continue
|
||||
for candidate in spec_candidates(cells[1:]):
|
||||
entry = catalog.resolve(name, candidate)
|
||||
if entry is not None:
|
||||
return entry
|
||||
return None
|
||||
|
||||
|
||||
def match_table(
|
||||
node: dict[str, Any],
|
||||
table: dict[str, Any],
|
||||
catalog: ResourceCatalog,
|
||||
result: AxisResult,
|
||||
) -> None:
|
||||
"""표 하나에 자원 축을 붙인다. 값이 안 서면 `unmatched` 로 보낸다."""
|
||||
form = table.get("pum_form", "")
|
||||
if form in NON_WORK_ITEM_FORMS or form in UNUSABLE_FORMS or form not in USABLE_FORMS:
|
||||
result.skipped_forms[form] = result.skipped_forms.get(form, 0) + 1
|
||||
return
|
||||
|
||||
basis = table.get("basis_quantity")
|
||||
basis_quantity = None if basis is None else Decimal(str(basis))
|
||||
unit = table.get("basis_unit") or ""
|
||||
|
||||
for index, row in enumerate(table.get("raw_row", [])):
|
||||
cells = [str(c) for c in row]
|
||||
if not cells:
|
||||
continue
|
||||
name_cell = cells[0]
|
||||
if is_non_resource_label(name_cell):
|
||||
continue
|
||||
|
||||
# 숫자 셀이 없는 행은 자원 줄이 아니다(제목·설명 행) — 목록에 안 올린다.
|
||||
amount_cell = next(
|
||||
(parse_amount(c) for c in cells[1:] if parse_amount(c) is not None), None
|
||||
)
|
||||
if amount_cell is None:
|
||||
continue
|
||||
|
||||
entry = _resolve_cell(catalog, name_cell, cells)
|
||||
if entry is None:
|
||||
name, spec = split_name_and_spec(name_cell)
|
||||
found = catalog.by_name(name)
|
||||
if len(found) > 1:
|
||||
reason = "규격이 없어 같은 이름 여럿 중 고를 수 없음"
|
||||
else:
|
||||
# 지금 가진 카탈로그는 노임뿐이다. 기계·자재는 카탈로그 자체가 없어
|
||||
# 못 맞추는 것이므로 사유를 갈라 적는다 — 「이름이 틀림」과 다르다.
|
||||
reason = "카탈로그에 없는 이름 (기계·자재 카탈로그 미확보 포함)"
|
||||
result.unmatched.append(
|
||||
UnmatchedRow(node["work_item_code"], table["pum_table_id"], name_cell, reason)
|
||||
)
|
||||
continue
|
||||
|
||||
try:
|
||||
amount = convert_amount(amount_cell, pum_form=form, basis_quantity=basis_quantity)
|
||||
except ResourceAxisError as error:
|
||||
result.unmatched.append(
|
||||
UnmatchedRow(node["work_item_code"], table["pum_table_id"], name_cell, str(error))
|
||||
)
|
||||
continue
|
||||
|
||||
result.rows.append(
|
||||
ResourceRow(
|
||||
work_item_code=node["work_item_code"],
|
||||
pum_table_id=table["pum_table_id"],
|
||||
pum_form=form,
|
||||
resource_kind=entry.kind,
|
||||
resource_code=entry.code,
|
||||
resource_name=entry.name,
|
||||
resource_spec=entry.spec,
|
||||
amount=amount,
|
||||
amount_unit=unit,
|
||||
raw_row_index=index,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def build_resource_axis(master: dict[str, Any], catalog: ResourceCatalog) -> AxisResult:
|
||||
"""공종 축 전체를 훑어 자원 축을 만든다."""
|
||||
result = AxisResult()
|
||||
for node in master.get("work_items", []):
|
||||
for table in node.get("tables", []):
|
||||
match_table(node, table, catalog, result)
|
||||
return result
|
||||
|
||||
|
||||
#: 자원 축 산출물이 나가는 자리 — **메인의 `data_work_item_master/` 안에 넣지 않는다.**
|
||||
#: 메인이 품셈을 다시 돌리면 그 폴더가 덮이므로 섞으면 사라진다.
|
||||
OUTPUT_SUBPATH = ("resources", "data_cost_resource_axis")
|
||||
|
||||
|
||||
def write_resource_axis(
|
||||
result: AxisResult,
|
||||
master: dict[str, Any],
|
||||
*,
|
||||
output_dir: str | None = None,
|
||||
) -> dict[str, str]:
|
||||
"""자원 축과 못 맞춘 목록을 파일로 낸다. 만든 파일 경로를 돌려준다."""
|
||||
directory = output_dir or os.path.join(_project_root(), *OUTPUT_SUBPATH)
|
||||
os.makedirs(directory, exist_ok=True)
|
||||
effective_date = master.get("effective_date", "")
|
||||
|
||||
axis_path = os.path.join(directory, f"resource_axis_{effective_date}.json")
|
||||
unmatched_path = os.path.join(directory, f"unmatched_{effective_date}.json")
|
||||
|
||||
axis_payload = {
|
||||
"schema_version": "1.0",
|
||||
"dataset_id": "resource_axis_forest",
|
||||
"effective_date": effective_date,
|
||||
# 어느 공종 축 판에 붙인 것인지 — 세 쪽을 그대로 옮겨 적는다(PLAN 9-2).
|
||||
"source_dataset_version": master.get("dataset_version", {}),
|
||||
"policy": {
|
||||
"axis": "resource_only",
|
||||
"work_item_axis_owner": "B08",
|
||||
"material_amounts_are_before_surcharge": True,
|
||||
},
|
||||
"stats": {
|
||||
"rows": len(result.rows),
|
||||
"unmatched": len(result.unmatched),
|
||||
"skipped_forms": result.skipped_forms,
|
||||
},
|
||||
"rows": [r.as_dict() for r in result.rows],
|
||||
}
|
||||
unmatched_payload = {
|
||||
"schema_version": "1.0",
|
||||
"effective_date": effective_date,
|
||||
"note": (
|
||||
"못 맞춘 자원 이름. 빈칸으로 두지 않고 여기 모은다. "
|
||||
"기계·자재 카탈로그가 아직 없어 그 계열은 전부 여기로 온다."
|
||||
),
|
||||
"rows": [u.as_dict() for u in result.unmatched],
|
||||
}
|
||||
|
||||
for path, payload in ((axis_path, axis_payload), (unmatched_path, unmatched_payload)):
|
||||
with open(path, "w", encoding="utf-8", newline="\n") as handle:
|
||||
json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
|
||||
|
||||
return {"resource_axis": axis_path, "unmatched": unmatched_path}
|
||||
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