feat(modelling): flat per-property CSV for the inspection report

format_report_csv emits one comma-safe row per property: the calculator-error
fields (lodged/calculated/Δ/flag), the Plan headline figures (baseline+post
SAP/band, measures, cost+contingency, bill & CO2 savings, valuation %), the
flattened measure triggers, and any captured error — sortable in a spreadsheet
for a large dump.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Khalim Conn-Kowlessar 2026-06-04 11:16:52 +00:00
parent 1c00708ecd
commit ae267070b1
2 changed files with 97 additions and 0 deletions

View file

@ -330,3 +330,65 @@ def format_report_markdown(reports: list[PropertyReport]) -> str:
*_measures_triggers_section(reports),
]
return "\n".join(sections).rstrip() + "\n"
_CSV_HEADER: Final[str] = (
"cert,lodged_sap,calculated_sap,sap_error,sap_error_flag,"
"baseline_sap,post_sap,baseline_band,post_band,measures,measure_types,"
"cost_of_works,contingency,bill_savings,co2_savings,valuation_pct,"
"triggers,error"
)
def _csv_cell(value: object) -> str:
"""Render a CSV cell, rounding floats and keeping the row comma-safe
(commas in any value become ';' so the column count never changes)."""
if value is None:
return ""
if isinstance(value, float):
return f"{value:.2f}"
return str(value).replace(",", ";")
def _csv_triggers(report: PropertyReport) -> str:
"""Flatten the fired measures and their triggers into one comma-safe cell:
`type(field=value;field=value)|type(field=value)`."""
return "|".join(
f"{trigger.measure_type}("
+ ";".join(f"{field}={value}" for field, value in trigger.triggers.items())
+ ")"
for trigger in report.measure_triggers
)
def format_report_csv(reports: list[PropertyReport]) -> str:
"""Render the report as a flat CSV — one row per property, browsable and
sortable in a spreadsheet for a large dump. The calculator-error fields, the
Plan headline figures, and the flattened triggers all share one row."""
rows: list[str] = [_CSV_HEADER]
for report in reports:
plan: Optional[Plan] = report.plan
cells: list[object] = [
report.name,
report.lodged_sap,
report.calculated_sap,
report.sap_error,
1 if report.sap_error_exceeds_threshold else 0,
None if plan is None else plan.baseline.sap_continuous,
None if plan is None else plan.post_sap_continuous,
None if plan is None else plan.baseline_epc_rating.value,
None if plan is None else plan.post_epc_rating.value,
None if plan is None else len(plan.measures),
None
if plan is None
else ";".join(measure.measure_type for measure in plan.measures),
None if plan is None else plan.cost_of_works,
None if plan is None else plan.contingency_cost,
None if plan is None else plan.energy_bill_savings,
None if plan is None else plan.co2_savings_kg_per_yr,
None if plan is None else plan.valuation.average_pct * 100,
_csv_triggers(report),
report.calculator_error or report.plan_error,
]
rows.append(",".join(_csv_cell(cell) for cell in cells))
return "\n".join(rows)

View file

@ -11,6 +11,7 @@ from harness.report import (
PropertyReport,
build_property_report,
build_property_reports,
format_report_csv,
format_report_markdown,
parity_report_for,
)
@ -188,6 +189,40 @@ def test_markdown_renders_the_three_sections(tmp_path: Path) -> None:
assert "floor_construction_type" in markdown
def test_csv_has_one_row_per_property_with_flags_and_triggers(tmp_path: Path) -> None:
# Arrange
bad: Path = tmp_path / "broken.json"
bad.write_text(json.dumps({"not": "an epc"}))
reports: list[PropertyReport] = build_property_reports(
[
_GOLDEN / f"{_WITHIN_TOLERANCE}.json",
_GOLDEN / f"{_DIVERGENT}.json",
bad,
]
)
# Act
csv: str = format_report_csv(reports)
# Assert — header plus one row per property.
lines: list[str] = csv.splitlines()
assert lines[0].startswith("cert,")
assert len(lines) == len(reports) + 1
# Every data row is comma-safe (no row splits into extra columns).
column_count: int = len(lines[0].split(","))
assert all(len(line.split(",")) == column_count for line in lines[1:])
rows: dict[str, str] = {line.split(",")[0]: line for line in lines[1:]}
# The divergent cert carries the |Δ| > 0.5 flag and the within-tolerance one doesn't.
flag_index: int = lines[0].split(",").index("sap_error_flag")
assert rows[_DIVERGENT].split(",")[flag_index] == "1"
assert rows[_WITHIN_TOLERANCE].split(",")[flag_index] == "0"
# The measure-bearing cert flattens its triggers into the row.
assert "solid_floor_insulation" in rows[_WITHIN_TOLERANCE]
assert "floor_construction_type=Solid" in rows[_WITHIN_TOLERANCE]
# The unscorable cert keeps its error.
assert "ValueError" in rows["broken"]
def test_unparseable_cert_is_captured_not_raised(tmp_path: Path) -> None:
# Arrange — a payload the mapper rejects must not abort the report.
bad: Path = tmp_path / "broken.json"