move spreadsheet population logic to domain

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.

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)

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 = (

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)

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()