Model/backend/documents_parser/tests/test_pashub_sap_accuracy_wythenshawe.py

"""End-to-end accuracy test for the pashub site-notes extractor — WYTHENSHAWE.

Sibling of `test_pashub_sap_accuracy.py` (the Guinness GMCA cohort); see that
module's docstring for the full chain and the meaning of the hybrid `pre_sap`
gauge. This module points at the `pashub_accuracy_wythenshawe/` fixtures built by

    python scripts/build_pashub_accuracy_fixtures.py \\
        --project-code "[Wythenshawe WH:SHF RAs - WH:SF - 240426 - 494493718748]" \\
        --out-dir pashub_accuracy_wythenshawe

from `hubspot_deal_data` (the project_code above) + the `rd_sap_site_note` PDFs
in S3. The ratchets below are calibrated to THIS cohort — do not copy the
Guinness constants across; each portfolio discovers its own floor/ceiling by
running.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from functools import lru_cache
from pathlib import Path
from typing import Optional

import pytest

from backend.documents_parser.parser import parse_site_notes_pdf
from datatypes.epc.domain.mapper import UnmappedPasHubLabel
from domain.sap10_calculator.calculator import Sap10Calculator
from domain.sap10_calculator.exceptions import UnmappedSapCode

# Same known in-progress main-heating-control mapper gap as the Guinness module.
_KNOWN_GAPS: tuple[type[Exception], ...] = (UnmappedSapCode, UnmappedPasHubLabel)

_FIXTURES_DIR = Path(__file__).parent / "fixtures" / "pashub_accuracy_wythenshawe"
_MANIFEST_PATH = _FIXTURES_DIR / "manifest.json"

# --- Ratchets (never loosen at fixed coverage), mirroring the Guinness module.
# First full run (2026-07-16): **149 computed / 19 blocked / 0 errored**
# (within-0.5 76.5% = 114/149, MAE 0.665). The 19 blocks are a spread of
# separately-ticketed mapper gaps — floor type 'Same dwelling below' (×10),
# room-in-roof gable 'None' (×2), 'No thermostatic control...' on 'Room heaters'
# (×2), plus one-off secondary-heating / secondary-fuel / room-heater / HHR
# storage-control (2404-on-402) labels — each its own field-by-field follow-up,
# not tuning. The floor/ceiling below are the observed baseline with a hair of
# headroom; treat them as a regression tripwire, NOT a cohort accuracy figure,
# and re-baseline (coverage growth is not loosening) as further mapper fixes
# unblock fixtures.
_MIN_WITHIN_HALF: float = 0.76
_MAX_SAP_MAE: float = 0.67

_KNOWN_GAP_REASON = (
    "pashub `from_site_notes` mapper does not int-code a main-heating "
    "`main_heating_control` label to a SAP code; fix in progress field-by-field"
)


@dataclass(frozen=True)
class PashubFixture:
    deal_id: str
    uprn: Optional[int]
    pre_sap_raw: str
    pre_sap_score: int
    pdf: str

    @property
    def path(self) -> Path:
        return _FIXTURES_DIR / self.pdf


@dataclass(frozen=True)
class Outcome:
    """Result of running one fixture through the pipeline."""

    blocked: bool  # blocked on the known main-heating-control gap (xfail territory)
    reason: Optional[str]
    sap_continuous: Optional[float]
    diff: Optional[float]  # abs(our SAP - pre_sap)


def _load_manifest() -> list[PashubFixture]:
    if not _MANIFEST_PATH.exists():
        return []
    data = json.loads(_MANIFEST_PATH.read_text())
    return [PashubFixture(**entry) for entry in data["fixtures"]]


_FIXTURES: list[PashubFixture] = _load_manifest()
_BY_ID: dict[str, PashubFixture] = {f.deal_id: f for f in _FIXTURES}
_IDS: list[str] = [f.deal_id for f in _FIXTURES]


@lru_cache(maxsize=None)
def _evaluate(deal_id: str) -> Outcome:
    """Run one fixture end-to-end. Shared across the two tests via the cache."""
    fixture = _BY_ID[deal_id]
    try:
        epc = parse_site_notes_pdf(str(fixture.path), uprn=fixture.uprn)
        result = Sap10Calculator().calculate(epc)
    except _KNOWN_GAPS as exc:
        return Outcome(blocked=True, reason=str(exc), sap_continuous=None, diff=None)
    diff = abs(result.sap_score_continuous - fixture.pre_sap_score)
    return Outcome(
        blocked=False,
        reason=None,
        sap_continuous=result.sap_score_continuous,
        diff=diff,
    )


@pytest.mark.skipif(not _FIXTURES, reason="no pashub_accuracy_wythenshawe fixtures/manifest")
@pytest.mark.parametrize("deal_id", _IDS)
def test_pashub_fixture_computes(deal_id: str) -> None:
    """The extractor must yield a calculable dwelling for each real property."""
    outcome = _evaluate(deal_id)
    if outcome.blocked:
        pytest.xfail(f"{_KNOWN_GAP_REASON}: {outcome.reason}")
    assert outcome.sap_continuous is not None
    assert 1 <= outcome.sap_continuous <= 100


@pytest.mark.skipif(not _FIXTURES, reason="no pashub_accuracy_wythenshawe fixtures/manifest")
def test_pashub_sap_accuracy_aggregate(capsys: pytest.CaptureFixture[str]) -> None:
    """SAP within-0.5 of pashub's `pre_sap`, aggregated over computable fixtures."""
    diffs: list[float] = []
    blocked = 0
    errored = 0
    for deal_id in _IDS:
        try:
            outcome = _evaluate(deal_id)
        except Exception:  # noqa: BLE001 - a per-fixture bug; scored there, not here
            errored += 1
            continue
        if outcome.blocked:
            blocked += 1
            continue
        assert outcome.diff is not None
        diffs.append(outcome.diff)

    if not diffs:
        pytest.xfail(
            f"no fixture computes a SAP score "
            f"({blocked} blocked on the main-heating-control gap, {errored} errored); "
            f"{_KNOWN_GAP_REASON}"
        )

    n = len(diffs)
    within_half = sum(1 for d in diffs if d < 0.5) / n
    sap_mae = sum(diffs) / n

    with capsys.disabled():
        print(
            f"\n[pashub WYTHENSHAWE | {n} computed / {blocked} blocked / "
            f"{errored} errored]"
            f"\n  SAP within-0.5 = {within_half:.1%}   MAE = {sap_mae:.3f}"
        )

    assert within_half >= _MIN_WITHIN_HALF, (
        f"SAP within-0.5 {within_half:.1%} fell below floor "
        f"{_MIN_WITHIN_HALF:.0%}"
    )
    assert sap_mae <= _MAX_SAP_MAE, (
        f"SAP MAE {sap_mae:.3f} exceeded ceiling {_MAX_SAP_MAE}"
    )