portfolio-optimiser/src/portfolio_optimiser/reference_domain.py

95 lines
3.6 KiB
Python

"""Synthetic reference domain (D4): a small, fictional set of "anleggskostnad"
(construction-cost) projects with dummy data.
This is the framework's bundled example input — a portfolio of *independent*
projects the optimiser runs against. The framework finds cost-savings INSIDE
each project (Enhet B), so every project carries cost line items where a savings
measure could later be proposed and then deterministically validated.
It is a synthetic FIXTURE — not real data, and not the validated IR. The
deliberate data-source *contract* (JSON-Schema-validated config, B5) is a Fase 2
concern; here we keep a plain, typed loader over a bundled JSON file.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from importlib.resources import files
_DATA_RESOURCE = "data/reference_projects.json"
@dataclass(frozen=True)
class CostItem:
"""One cost line in a project's estimate."""
code: str
description: str
quantity: float
unit: str
unit_cost: float # NOK per unit
@property
def total_cost(self) -> float:
return self.quantity * self.unit_cost
@dataclass(frozen=True)
class Project:
"""One independent construction-cost project (Enhet B operates inside this)."""
id: str
name: str
description: str
currency: str
cost_items: tuple[CostItem, ...]
docs_dir: str # absolute path to this project's bundled cost-docs folder (config-driven)
verdict_input: dict[str, str] # SYNTHETIC Layer-2 expert decision/rationale (config-driven)
bundle_dir: str | None = None # abs path to an OKF bundle backing this project (Fase 2a S2.0)
verdict_dir: str | None = None # abs path to this project's async verdict inbox (Fase 2a S2.0)
@property
def total_cost(self) -> float:
return sum((item.total_cost for item in self.cost_items), 0.0)
def load_reference_projects() -> tuple[Project, ...]:
"""Load the bundled synthetic reference projects (D4).
Each project's ``docs_dir`` is stored in the JSON relative to the package ``data/`` root
and resolved here to an absolute filesystem path; ``verdict_input`` carries the SYNTHETIC
Layer-2 expert decision/rationale. Missing keys raise ``KeyError`` (fail-fast, matching the
existing loader contract)."""
resource = files("portfolio_optimiser").joinpath(_DATA_RESOURCE)
raw = json.loads(resource.read_text(encoding="utf-8"))
def _resolve(rel: str | None) -> str | None:
"""Optional config paths (bundle_dir/verdict_dir) resolve to absolute like docs_dir when set,
else stay None. Read via ``p.get(...)`` — never fail-fast: the shipped JSON omits both, so
every existing row keeps loading (backward-compatible; contrast the required-key reads)."""
return str(files("portfolio_optimiser").joinpath(f"data/{rel}")) if rel else None
return tuple(
Project(
id=p["id"],
name=p["name"],
description=p["description"],
currency=p["currency"],
cost_items=tuple(
CostItem(
code=c["code"],
description=c["description"],
quantity=c["quantity"],
unit=c["unit"],
unit_cost=c["unit_cost"],
)
for c in p["cost_items"]
),
docs_dir=str(files("portfolio_optimiser").joinpath(f"data/{p['docs_dir']}")),
verdict_input=p["verdict_input"],
bundle_dir=_resolve(p.get("bundle_dir")),
verdict_dir=_resolve(p.get("verdict_dir")),
)
for p in raw["projects"]
)