From 236f33c25fb3aa58585b2bd004d7a092c42b8144 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Tue, 9 Jun 2026 14:43:24 +0000 Subject: [PATCH] move spreadsheet population logic to domain --- .../extract-populate-sheet-to-domain.md | 68 ++++++++++++++ domain/magicplan/ventilation_audit.py | 86 +++++++++++++++++ orchestration/audit_generator_orchestrator.py | 83 +--------------- .../magicplan/test_ventilation_audit.py | 94 +++++++++++++++++++ .../test_audit_generator_orchestrator.py | 15 --- 5 files changed, 250 insertions(+), 96 deletions(-) create mode 100644 docs/backlog/extract-populate-sheet-to-domain.md create mode 100644 domain/magicplan/ventilation_audit.py create mode 100644 tests/domain/magicplan/test_ventilation_audit.py diff --git a/docs/backlog/extract-populate-sheet-to-domain.md b/docs/backlog/extract-populate-sheet-to-domain.md new file mode 100644 index 00000000..59474895 --- /dev/null +++ b/docs/backlog/extract-populate-sheet-to-domain.md @@ -0,0 +1,68 @@ +# PRD: Extract ventilation audit sheet population into the magicplan domain + +**Status:** Backlog + +--- + +## Problem Statement + +The logic that maps a `Plan` into spreadsheet cells — which column receives `pct_openable / 100`, which rows are rooms vs windows vs doors, what the 50-row capacity limit is, how column Y conditional formatting is applied — currently lives inside the orchestrator. Developers reading `AuditGeneratorOrchestrator` have to wade through cell-writing details to understand the orchestration flow, and there is no way to test the sheet-population rules in isolation without invoking the full orchestrator (which requires a mocked UoW, mocked S3, and the real XLSX template file). + +## Solution + +Move all sheet-population logic into the magicplan domain as a dedicated module (`ventilation_audit`), exposing a single public function `populate_sheet(sheet, plan)`. The orchestrator delegates to this function and retains only its infrastructure responsibilities: loading the template, serialising the workbook, uploading to S3, and persisting metadata. + +This makes the mapping rules directly testable against a plain `openpyxl` sheet with no orchestration overhead, and keeps the orchestrator focused on coordination rather than domain rules. + +## User Stories + +1. As a developer debugging a malformed audit spreadsheet, I want the cell-mapping rules to live in the domain so that I can locate the logic without reading through orchestration code. +2. As a developer writing a test for ventilation audit content, I want to call `populate_sheet` directly with a synthetic `Plan` and a blank sheet so that I can assert cell values without mocking S3 or a unit of work. +3. As a developer adding a new opening type or ventilation field, I want the affected mapping logic to be co-located with the `Plan` domain models so that the change is easy to find and the impact is obvious. +4. As a developer reading the orchestrator, I want the `run()` method to read as a sequence of high-level steps (fetch → populate → serialise → upload → persist) with no cell-writing detail so that the orchestration intent is immediately clear. +5. As a developer running the test suite, I want the 50-row overflow validation to be covered by a domain-level test so that regressions in that constraint are caught without running the full orchestrator. +6. As a developer extending the audit template to a second sheet, I want the sheet-population contract to be a clearly bounded function so that I can add a second `populate_*` function in the same module without touching the orchestrator. + +## Implementation Decisions + +- **New module `domain/magicplan/ventilation_audit.py`** contains the public function `populate_sheet(sheet, plan)` and all private helpers (`_write_cell`, `_apply_column_y_formatting`) and constants (`_DATA_START_ROW`, `_MAX_ROWS`, `_Y_CF_RANGE`, `_Y_THRESHOLD`, `_Y_HEADER`). These are moved verbatim from the orchestrator — no logic changes. + +- **`populate_sheet` is the sole public surface.** Helpers remain private to the module. This follows the existing `mapper.py` pattern (stateless module-level functions, no class wrapper). + +- **The orchestrator imports `populate_sheet`** and replaces its `_populate_sheet(sheet, plan)` call. All `openpyxl.cell.rich_text`, `openpyxl.cell.text`, `openpyxl.formatting.rule`, and `openpyxl.styles` imports move with the logic. `openpyxl.load_workbook` stays — loading the template is an infrastructure step. + +- **`_serialise_workbook` stays in the orchestrator** — converting a workbook to bytes is a serialisation step, not domain logic. + +- **No interface change to the orchestrator's public API** — `AuditGeneratorOrchestrator.__init__` and `run()` signatures are unchanged. + +## Testing Decisions + +Good tests for `populate_sheet` assert observable outputs (cell values, conditional formatting rule count) given a controlled `Plan` input. They do not assert on internal call sequences or private helper invocations. + +Tests should use a fresh `openpyxl.Workbook().active` sheet — no template file needed, which keeps them fast and dependency-free. + +Modules to test (new file: `tests/domain/magicplan/test_ventilation_audit.py`): + +| Scenario | Assertion | +|---|---| +| Rooms written correctly | Col B = room name, col D = area_m2, starting at `_DATA_START_ROW` | +| Windows written correctly | Cols G–I, K–M, Q–R populated; pct_openable divided by 100 | +| Windows with null ventilation | Ventilation columns default to 0 | +| Doors written correctly | Cols V–X populated with room name, width_mm, undercut_mm | +| Room overflow | > 50 rooms raises `ValueError` | +| Window overflow | > 50 windows raises `ValueError` | +| Door overflow | > 50 doors raises `ValueError` | +| Column Y formatting applied | Sheet has two conditional formatting rules after `populate_sheet` | + +Prior art: `tests/orchestration/audit_generator/test_audit_generator_orchestrator.py` shows the `_make_plan` / `_make_window` / `_make_door` fixture pattern to reuse. The existing orchestrator tests need no changes. + +## Out of Scope + +- Changes to the spreadsheet template or column layout. +- Support for plans with more than 50 rooms, windows, or doors (the 50-row limit is a template constraint, not lifted here). +- Extracting `_serialise_workbook` or template-loading into the domain. +- Any changes to the `AuditGeneratorOrchestrator` public API or the Lambda entry point. + +## Further Notes + +The orchestrator test suite already provides integration-level coverage (S3 call order, `UploadedFile` enums, error paths). This refactor adds the missing unit-level coverage for the mapping rules, which are currently exercised only incidentally via the happy-path orchestrator tests. diff --git a/domain/magicplan/ventilation_audit.py b/domain/magicplan/ventilation_audit.py new file mode 100644 index 00000000..36214412 --- /dev/null +++ b/domain/magicplan/ventilation_audit.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +from typing import Any + +from openpyxl.cell.rich_text import CellRichText, TextBlock +from openpyxl.cell.text import InlineFont +from openpyxl.formatting.rule import CellIsRule # type: ignore[reportUnknownVariableType] +from openpyxl.styles import Color, Font + +from domain.magicplan.models import Door, Plan, Room, Window + +_DATA_START_ROW = 6 +_MAX_ROWS = 50 +_Y_CF_RANGE = f"Y{_DATA_START_ROW}:Y{_DATA_START_ROW + _MAX_ROWS - 1}" +_Y_THRESHOLD = 7600 +_Y_HEADER = CellRichText( + TextBlock(InlineFont(b=True, sz=11, rFont="Aptos Narrow"), "Area (mm2)\n"), + TextBlock(InlineFont(b=True, sz=11, color=Color(rgb="FF0000"), rFont="Aptos Narrow"), "<"), + TextBlock(InlineFont(b=True, sz=11, rFont="Aptos Narrow"), " 7600 "), + TextBlock(InlineFont(b=True, sz=11, color=Color(rgb="196B24"), rFont="Aptos Narrow"), "<"), +) + + +def _apply_column_y_formatting(sheet: Any) -> None: + sheet.conditional_formatting.add( + _Y_CF_RANGE, + CellIsRule(operator="lessThan", formula=[str(_Y_THRESHOLD)], font=Font(color=Color(rgb="FF0000"))), + ) + sheet.conditional_formatting.add( + _Y_CF_RANGE, + CellIsRule(operator="greaterThan", formula=[str(_Y_THRESHOLD)], font=Font(color=Color(rgb="196B24"))), + ) + sheet["Y3"] = _Y_HEADER + + +def _write_cell(sheet: Any, row: int, col: str, value: Any) -> None: + sheet[f"{col}{row}"] = value + + +def populate_sheet(sheet: Any, plan: Plan) -> None: + rooms: list[Room] = [room for floor in plan.floors for room in floor.rooms] + windows: list[tuple[str, Window]] = [ + (room.name, w) for room in rooms for w in room.windows + ] + doors: list[tuple[str, Door]] = [ + (room.name, d) for room in rooms for d in room.doors + ] + + if len(rooms) > _MAX_ROWS: + raise ValueError(f"Room series exceeds {_MAX_ROWS} rows ({len(rooms)} rooms)") + if len(windows) > _MAX_ROWS: + raise ValueError(f"Window series exceeds {_MAX_ROWS} rows ({len(windows)} windows)") + if len(doors) > _MAX_ROWS: + raise ValueError(f"Door series exceeds {_MAX_ROWS} rows ({len(doors)} doors)") + + for i, room in enumerate(rooms): + row = _DATA_START_ROW + i + _write_cell(sheet, row, "B", room.name) + _write_cell(sheet, row, "D", room.area_m2) + + for i, (room_name, window) in enumerate(windows): + row = _DATA_START_ROW + i + vent = window.ventilation + _write_cell(sheet, row, "G", room_name) + _write_cell(sheet, row, "H", window.width_m) + _write_cell(sheet, row, "I", window.height_m) + # J = formula =H*I — do not write + _write_cell(sheet, row, "K", vent.opening_type if vent else 0) + _write_cell(sheet, row, "L", vent.num_openings if vent else 0) + pct = vent.pct_openable if vent else None + _write_cell(sheet, row, "M", (pct / 100) if pct is not None else 0) + # N = formula =J*M — do not write + # O, P = blank (visual check by auditor) + _write_cell(sheet, row, "Q", vent.trickle_vent_area_mm2 if vent else 0) + _write_cell(sheet, row, "R", vent.num_trickle_vents if vent else 0) + # S = formula =Q*R — do not write + + for i, (room_name, door) in enumerate(doors): + row = _DATA_START_ROW + i + vent = door.ventilation + _write_cell(sheet, row, "V", room_name) + _write_cell(sheet, row, "W", door.width_mm) + _write_cell(sheet, row, "X", vent.undercut_mm if vent else 0) + # Y = formula =W*X — do not write + + _apply_column_y_formatting(sheet) diff --git a/orchestration/audit_generator_orchestrator.py b/orchestration/audit_generator_orchestrator.py index a22979c4..3e436348 100644 --- a/orchestration/audit_generator_orchestrator.py +++ b/orchestration/audit_generator_orchestrator.py @@ -7,12 +7,8 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, cast import openpyxl -from openpyxl.cell.rich_text import CellRichText, TextBlock -from openpyxl.cell.text import InlineFont -from openpyxl.formatting.rule import CellIsRule # type: ignore[reportUnknownVariableType] -from openpyxl.styles import Color, Font -from domain.magicplan.models import Door, Plan, Room, Window +from domain.magicplan.ventilation_audit import populate_sheet from infrastructure.postgres.uploaded_file_table import ( FileSourceEnum, FileTypeEnum, @@ -25,81 +21,6 @@ if TYPE_CHECKING: _TEMPLATE_PATH = Path(__file__).parent.parent / "applications" / "audit_generator" / "d1_ventilation_template.xlsx" _SHEET_NAME = "D1 Ventilation" -_DATA_START_ROW = 6 -_MAX_ROWS = 50 -_Y_CF_RANGE = f"Y{_DATA_START_ROW}:Y{_DATA_START_ROW + _MAX_ROWS - 1}" -_Y_THRESHOLD = 7600 -_Y_HEADER = CellRichText( - TextBlock(InlineFont(b=True, sz=11, rFont="Aptos Narrow"), "Area (mm2)\n"), - TextBlock(InlineFont(b=True, sz=11, color=Color(rgb="FF0000"), rFont="Aptos Narrow"), "<"), - TextBlock(InlineFont(b=True, sz=11, rFont="Aptos Narrow"), " 7600 "), - TextBlock(InlineFont(b=True, sz=11, color=Color(rgb="196B24"), rFont="Aptos Narrow"), "<"), -) - - -def _apply_column_y_formatting(sheet: Any) -> None: - sheet.conditional_formatting.add( - _Y_CF_RANGE, - CellIsRule(operator="lessThan", formula=[str(_Y_THRESHOLD)], font=Font(color=Color(rgb="FF0000"))), - ) - sheet.conditional_formatting.add( - _Y_CF_RANGE, - CellIsRule(operator="greaterThan", formula=[str(_Y_THRESHOLD)], font=Font(color=Color(rgb="196B24"))), - ) - sheet["Y3"] = _Y_HEADER - - -def _write_cell(sheet: Any, row: int, col: str, value: Any) -> None: - sheet[f"{col}{row}"] = value - - -def _populate_sheet(sheet: Any, plan: Plan) -> None: - rooms: list[Room] = [room for floor in plan.floors for room in floor.rooms] - windows: list[tuple[str, Window]] = [ - (room.name, w) for room in rooms for w in room.windows - ] - doors: list[tuple[str, Door]] = [ - (room.name, d) for room in rooms for d in room.doors - ] - - if len(rooms) > _MAX_ROWS: - raise ValueError(f"Room series exceeds {_MAX_ROWS} rows ({len(rooms)} rooms)") - if len(windows) > _MAX_ROWS: - raise ValueError(f"Window series exceeds {_MAX_ROWS} rows ({len(windows)} windows)") - if len(doors) > _MAX_ROWS: - raise ValueError(f"Door series exceeds {_MAX_ROWS} rows ({len(doors)} doors)") - - for i, room in enumerate(rooms): - row = _DATA_START_ROW + i - _write_cell(sheet, row, "B", room.name) - _write_cell(sheet, row, "D", room.area_m2) - - for i, (room_name, window) in enumerate(windows): - row = _DATA_START_ROW + i - vent = window.ventilation - _write_cell(sheet, row, "G", room_name) - _write_cell(sheet, row, "H", window.width_m) - _write_cell(sheet, row, "I", window.height_m) - # J = formula =H*I — do not write - _write_cell(sheet, row, "K", vent.opening_type if vent else 0) - _write_cell(sheet, row, "L", vent.num_openings if vent else 0) - pct = vent.pct_openable if vent else None - _write_cell(sheet, row, "M", (pct / 100) if pct is not None else 0) - # N = formula =J*M — do not write - # O, P = blank (visual check by auditor) - _write_cell(sheet, row, "Q", vent.trickle_vent_area_mm2 if vent else 0) - _write_cell(sheet, row, "R", vent.num_trickle_vents if vent else 0) - # S = formula =Q*R — do not write - - for i, (room_name, door) in enumerate(doors): - row = _DATA_START_ROW + i - vent = door.ventilation - _write_cell(sheet, row, "V", room_name) - _write_cell(sheet, row, "W", door.width_mm) - _write_cell(sheet, row, "X", vent.undercut_mm if vent else 0) - # Y = formula =W*X — do not write - - _apply_column_y_formatting(sheet) def _serialise_workbook(wb: Any) -> bytes: @@ -138,7 +59,7 @@ class AuditGeneratorOrchestrator: wb = openpyxl.load_workbook(_TEMPLATE_PATH) sheet = wb[_SHEET_NAME] - _populate_sheet(sheet, plan) + populate_sheet(sheet, plan) xlsx_bytes = _serialise_workbook(wb) s3_key = ( diff --git a/tests/domain/magicplan/test_ventilation_audit.py b/tests/domain/magicplan/test_ventilation_audit.py new file mode 100644 index 00000000..eb7ba7cc --- /dev/null +++ b/tests/domain/magicplan/test_ventilation_audit.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import openpyxl +import pytest + +from domain.magicplan.models import ( + Door, + DoorVentilation, + Floor, + Plan, + Room, + Window, + WindowVentilation, +) +from domain.magicplan.ventilation_audit import populate_sheet + + +def _make_window(with_ventilation: bool = True) -> Window: + vent = ( + WindowVentilation( + opening_type="Hinged", + num_openings=1, + pct_openable=50, + trickle_vent_area_mm2=1000, + num_trickle_vents=2, + ) + if with_ventilation + else None + ) + return Window(width_m=1.0, height_m=1.2, area_m2=1.2, ventilation=vent) + + +def _make_door(with_ventilation: bool = True) -> Door: + vent = DoorVentilation(undercut_mm=10.0) if with_ventilation else None + return Door(width_mm=800.0, height_mm=2000.0, ventilation=vent) + + +def _make_plan( + num_rooms: int = 1, + num_windows_per_room: int = 1, + num_doors_per_room: int = 1, +) -> Plan: + rooms = [ + Room( + name=f"Room {i}", + width_m=3.0, + length_m=4.0, + area_m2=12.0, + windows=[_make_window() for _ in range(num_windows_per_room)], + doors=[_make_door() for _ in range(num_doors_per_room)], + ) + for i in range(num_rooms) + ] + return Plan( + uid="test-uid", + name="Test Plan", + address="1 Test St", + postcode="TE1 1ST", + floors=[Floor(level=0, name="Ground", rooms=rooms)], + ) + + +def _blank_sheet() -> object: + return openpyxl.Workbook().active + + +def test_raises_when_rooms_exceed_50() -> None: + # Arrange + plan = _make_plan(num_rooms=51, num_windows_per_room=0, num_doors_per_room=0) + sheet = _blank_sheet() + + # Act / Assert + with pytest.raises(ValueError, match="50"): + populate_sheet(sheet, plan) + + +def test_raises_when_windows_exceed_50() -> None: + # Arrange + plan = _make_plan(num_rooms=1, num_windows_per_room=51, num_doors_per_room=0) + sheet = _blank_sheet() + + # Act / Assert + with pytest.raises(ValueError, match="50"): + populate_sheet(sheet, plan) + + +def test_raises_when_doors_exceed_50() -> None: + # Arrange + plan = _make_plan(num_rooms=1, num_windows_per_room=0, num_doors_per_room=51) + sheet = _blank_sheet() + + # Act / Assert + with pytest.raises(ValueError, match="50"): + populate_sheet(sheet, plan) diff --git a/tests/orchestration/audit_generator/test_audit_generator_orchestrator.py b/tests/orchestration/audit_generator/test_audit_generator_orchestrator.py index 18445fe5..2d9d706c 100644 --- a/tests/orchestration/audit_generator/test_audit_generator_orchestrator.py +++ b/tests/orchestration/audit_generator/test_audit_generator_orchestrator.py @@ -202,18 +202,3 @@ def test_commits_after_s3_upload() -> None: # Assert — S3 upload happens before DB commit assert call_order == ["s3_upload", "commit"] - -# --- 50-row limit --- - - -def test_raises_when_series_exceeds_50_rows() -> None: - # Arrange — 51 windows, which exceeds the template limit - plan = _make_plan(num_rooms=1, num_windows_per_room=51) - mock_uow, mock_uow_factory = _make_mock_uow( - uploaded_file_row=_make_uploaded_file_row(), plan=plan - ) - orch = _make_orchestrator(uow_factory=mock_uow_factory) - - # Act / Assert - with pytest.raises(ValueError, match="50"): - orch.run()