mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-19 08:53:17 +00:00
132 lines
4.8 KiB
Python
132 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from datetime import datetime, timezone
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any, cast
|
|
|
|
import openpyxl
|
|
|
|
from domain.magicplan.models import Door, Plan, Room, Window
|
|
from infrastructure.postgres.uploaded_file_table import (
|
|
FileSourceEnum,
|
|
FileTypeEnum,
|
|
UploadedFile,
|
|
)
|
|
from infrastructure.s3.s3_client import S3Client
|
|
|
|
if TYPE_CHECKING:
|
|
from orchestration.audit_generator_unit_of_work import AuditGeneratorUnitOfWork
|
|
|
|
_TEMPLATE_PATH = Path(__file__).parent.parent / "applications" / "audit_generator" / "d1_ventilation_template.xlsx"
|
|
_SHEET_NAME = "D1 Ventilation"
|
|
_DATA_START_ROW = 6
|
|
_MAX_ROWS = 50
|
|
|
|
|
|
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
|
|
|
|
|
|
def _serialise_workbook(wb: Any) -> bytes:
|
|
buf = BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
class AuditGeneratorOrchestrator:
|
|
def __init__(
|
|
self,
|
|
hubspot_deal_id: str,
|
|
s3_client: S3Client,
|
|
uow_factory: Callable[[], "AuditGeneratorUnitOfWork"],
|
|
) -> None:
|
|
self._hubspot_deal_id = hubspot_deal_id
|
|
self._s3_client = s3_client
|
|
self._uow_factory = uow_factory
|
|
|
|
def run(self) -> None:
|
|
with self._uow_factory() as uow:
|
|
uploaded_file = uow.uploaded_file.get_latest_by_hubspot_deal_id(
|
|
self._hubspot_deal_id, FileTypeEnum.MAGIC_PLAN_JSON
|
|
)
|
|
if uploaded_file is None:
|
|
raise ValueError(
|
|
f"No MagicPlan JSON has been uploaded for deal {self._hubspot_deal_id!r}"
|
|
)
|
|
|
|
plan = uow.magic_plan.get_plan_by_uploaded_file_id(cast(int, uploaded_file.id))
|
|
if plan is None:
|
|
raise ValueError(
|
|
f"MagicPlan JSON exists for deal {self._hubspot_deal_id!r} "
|
|
"but the plan is not yet parsed into the database"
|
|
)
|
|
|
|
wb = openpyxl.load_workbook(_TEMPLATE_PATH)
|
|
sheet = wb[_SHEET_NAME]
|
|
_populate_sheet(sheet, plan)
|
|
xlsx_bytes = _serialise_workbook(wb)
|
|
|
|
s3_key = (
|
|
f"documents/hubspot_deal_id/{self._hubspot_deal_id}/ventilation_audit.xlsx"
|
|
)
|
|
self._s3_client.put_object(s3_key, xlsx_bytes)
|
|
|
|
new_row = UploadedFile(
|
|
s3_file_bucket=self._s3_client.bucket,
|
|
s3_file_key=s3_key,
|
|
s3_upload_timestamp=datetime.now(timezone.utc),
|
|
hubspot_deal_id=self._hubspot_deal_id,
|
|
file_type=FileTypeEnum.VENTILATION_AUDIT.value,
|
|
file_source=FileSourceEnum.AUDIT_GENERATOR.value,
|
|
)
|
|
uow.uploaded_file.insert(new_row)
|
|
uow.commit()
|