16 changed files with 1702 additions and 44 deletions
|
|
@ -1,7 +1,12 @@
|
|||
from fastapi import APIRouter
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
|
||||
from app.schemas.concept import ConceptInput, ConceptOutput
|
||||
from app.services.generative import geometry
|
||||
from app.services.generative.geometry import ParcelGeometryError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
|
@ -10,9 +15,16 @@ router = APIRouter()
|
|||
async def create_concept(payload: ConceptInput) -> ConceptOutput:
|
||||
"""Generate 3 building variants for the given parcel polygon.
|
||||
|
||||
Stage 1a: returns stub with 3 empty variants.
|
||||
Stage 1b: real greedy placement.
|
||||
Stage 1c: TEAP + financial model attached.
|
||||
Stage 1a: Shapely parse + buildable area (setback) + placement grid.
|
||||
Stage 1b: greedy section placement with STRtree collisions (3 strategies).
|
||||
Stage 1c: real ТЭП + financial model attached to each variant.
|
||||
|
||||
A degenerate parcel (setback consumes everything, malformed geometry) yields a
|
||||
422 rather than empty variants — that is a bad request, not a valid empty result.
|
||||
"""
|
||||
variants = geometry.generate_stub(payload)
|
||||
try:
|
||||
variants = geometry.generate(payload)
|
||||
except ParcelGeometryError as exc:
|
||||
logger.warning("concept generation rejected parcel: %s", exc)
|
||||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||||
return ConceptOutput(variants=variants)
|
||||
|
|
|
|||
|
|
@ -1,3 +1,3 @@
|
|||
from app.services.generative import financial, geometry, teap
|
||||
from app.services.generative import exporters, financial, geometry, placement, teap
|
||||
|
||||
__all__ = ["financial", "geometry", "teap"]
|
||||
__all__ = ["exporters", "financial", "geometry", "placement", "teap"]
|
||||
|
|
|
|||
11
backend/app/services/generative/exporters/__init__.py
Normal file
11
backend/app/services/generative/exporters/__init__.py
Normal file
|
|
@ -0,0 +1,11 @@
|
|||
"""Generative Design — Stage 1c exporters (DXF geometry + PDF summary).
|
||||
|
||||
Distinct from Site Finder's ``app.services.exporters`` (report_pdf etc.): these
|
||||
serialise *concept* output — parcel + placed buildings (DXF) and the ТЭП/финмодель
|
||||
summary (PDF). Both are deterministic and consume already-computed Stage 1a/1b/1c
|
||||
objects (no re-parsing, no DB, no network).
|
||||
"""
|
||||
|
||||
from app.services.generative.exporters import dxf, pdf
|
||||
|
||||
__all__ = ["dxf", "pdf"]
|
||||
163
backend/app/services/generative/exporters/dxf.py
Normal file
163
backend/app/services/generative/exporters/dxf.py
Normal file
|
|
@ -0,0 +1,163 @@
|
|||
"""Generative Design — Stage 1c DXF export via ezdxf.
|
||||
|
||||
Renders the parcel context (boundary + buildable area) and one variant's placed
|
||||
building footprints into a binary DXF for hand-off to architects. Geometry is drawn
|
||||
in the parcel's *metric* CRS (metres) — architects work in metres, and DXF has no
|
||||
geographic CRS concept, so emitting WGS84 degrees would be unusable.
|
||||
|
||||
Layers:
|
||||
* ``PARCEL`` — границы участка (синий).
|
||||
* ``BUILDABLE`` — пятно застройки после отступов (зелёный, пунктир-цвет).
|
||||
* ``BUILDINGS`` — секции варианта (красный), с текстовой подписью номера секции.
|
||||
|
||||
Returns ``bytes`` (binary DXF, R2010) ready for an HTTP response / file write.
|
||||
Deterministic, no DB / no network. ``ezdxf`` is a light import, so it stays at
|
||||
module level (unlike WeasyPrint in :mod:`pdf`).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import logging
|
||||
|
||||
# ezdxf.new живёт в ezdxf.filemanagement и не реэкспортируется через ezdxf.__all__;
|
||||
# импорт из модуля удовлетворяет strict no-implicit-reexport.
|
||||
from ezdxf.filemanagement import new as ezdxf_new
|
||||
from shapely.geometry import Polygon
|
||||
|
||||
from app.schemas.concept import ConceptVariant
|
||||
from app.services.generative.geometry import Parcel
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# AutoCAD Color Index (ACI) per layer.
|
||||
_ACI_PARCEL = 5 # blue
|
||||
_ACI_BUILDABLE = 3 # green
|
||||
_ACI_BUILDINGS = 1 # red
|
||||
|
||||
_LAYER_PARCEL = "PARCEL"
|
||||
_LAYER_BUILDABLE = "BUILDABLE"
|
||||
_LAYER_BUILDINGS = "BUILDINGS"
|
||||
|
||||
# Высота текста подписи секции (метры в модельном пространстве).
|
||||
_LABEL_HEIGHT_M = 2.0
|
||||
|
||||
|
||||
def _polygon_points(poly: Polygon) -> list[tuple[float, float]]:
|
||||
"""Внешнее кольцо полигона как список (x, y) для LWPolyline (без замыкающей точки)."""
|
||||
coords = list(poly.exterior.coords)
|
||||
# Shapely дублирует первую точку в конце; close=True у ezdxf замкнёт сам.
|
||||
if len(coords) > 1 and coords[0] == coords[-1]:
|
||||
coords = coords[:-1]
|
||||
return [(float(x), float(y)) for x, y in coords]
|
||||
|
||||
|
||||
def export_concept_dxf(parcel: Parcel, variant: ConceptVariant) -> bytes:
|
||||
"""Собрать binary DXF: участок + пятно застройки + секции одного варианта.
|
||||
|
||||
Args:
|
||||
parcel: Stage 1a участок (метрическая геометрия parcel/buildable).
|
||||
variant: вариант, чьи секции рисуем (footprints берём из его geojson —
|
||||
но геометрию рисуем из метрического parcel-space через свежий парсинг
|
||||
geojson обратно нельзя без CRS, поэтому секции восстанавливаем ниже).
|
||||
|
||||
Returns:
|
||||
bytes: бинарный DXF R2010.
|
||||
"""
|
||||
doc = ezdxf_new("R2010")
|
||||
doc.layers.add(_LAYER_PARCEL, color=_ACI_PARCEL)
|
||||
doc.layers.add(_LAYER_BUILDABLE, color=_ACI_BUILDABLE)
|
||||
doc.layers.add(_LAYER_BUILDINGS, color=_ACI_BUILDINGS)
|
||||
msp = doc.modelspace()
|
||||
|
||||
# Участок и пятно застройки — из метрической геометрии Parcel.
|
||||
msp.add_lwpolyline(
|
||||
_polygon_points(parcel.polygon_m),
|
||||
close=True,
|
||||
dxfattribs={"layer": _LAYER_PARCEL},
|
||||
)
|
||||
msp.add_lwpolyline(
|
||||
_polygon_points(parcel.buildable_m),
|
||||
close=True,
|
||||
dxfattribs={"layer": _LAYER_BUILDABLE},
|
||||
)
|
||||
|
||||
# Секции: восстанавливаем метрические footprints из WGS84-geojson варианта.
|
||||
features = variant.buildings_geojson.get("features", [])
|
||||
section_count = 0
|
||||
if isinstance(features, list):
|
||||
for feature in features:
|
||||
footprint = _feature_to_metric_polygon(parcel, feature)
|
||||
if footprint is None:
|
||||
continue
|
||||
section_count += 1
|
||||
msp.add_lwpolyline(
|
||||
_polygon_points(footprint),
|
||||
close=True,
|
||||
dxfattribs={"layer": _LAYER_BUILDINGS},
|
||||
)
|
||||
centroid = footprint.centroid
|
||||
label = str(_feature_section_id(feature, section_count))
|
||||
text = msp.add_text(
|
||||
label,
|
||||
dxfattribs={"layer": _LAYER_BUILDINGS, "height": _LABEL_HEIGHT_M},
|
||||
)
|
||||
text.set_placement((float(centroid.x), float(centroid.y)))
|
||||
|
||||
stream = io.BytesIO()
|
||||
doc.write(stream, fmt="bin")
|
||||
data = stream.getvalue()
|
||||
logger.info(
|
||||
"DXF export: strategy=%s sections=%d bytes=%d",
|
||||
variant.strategy,
|
||||
section_count,
|
||||
len(data),
|
||||
)
|
||||
return data
|
||||
|
||||
|
||||
def _feature_section_id(feature: object, fallback: int) -> int:
|
||||
"""Достать section_id из properties Feature, иначе fallback-счётчик."""
|
||||
if isinstance(feature, dict):
|
||||
props = feature.get("properties")
|
||||
if isinstance(props, dict):
|
||||
sid = props.get("section_id")
|
||||
if isinstance(sid, int):
|
||||
return sid
|
||||
return fallback
|
||||
|
||||
|
||||
def _feature_to_metric_polygon(parcel: Parcel, feature: object) -> Polygon | None:
|
||||
"""WGS84 GeoJSON Feature -> метрический Shapely Polygon (через обратный трансформер).
|
||||
|
||||
Возвращает None для невалидных/непригодных фич (graceful — экспорт не падает).
|
||||
"""
|
||||
if not isinstance(feature, dict):
|
||||
return None
|
||||
geometry = feature.get("geometry")
|
||||
if not isinstance(geometry, dict) or geometry.get("type") != "Polygon":
|
||||
return None
|
||||
coords = geometry.get("coordinates")
|
||||
# Shapely mapping() emits nested tuples; accept both tuple and list.
|
||||
if not isinstance(coords, (list, tuple)) or not coords:
|
||||
return None
|
||||
ring = coords[0]
|
||||
if not isinstance(ring, (list, tuple)) or len(ring) < 4:
|
||||
return None
|
||||
|
||||
metric_pts: list[tuple[float, float]] = []
|
||||
for pt in ring:
|
||||
if not isinstance(pt, (list, tuple)) or len(pt) < 2:
|
||||
return None
|
||||
lon, lat = float(pt[0]), float(pt[1])
|
||||
x, y = parcel.wgs84_to_metric(lon, lat)
|
||||
metric_pts.append((x, y))
|
||||
|
||||
try:
|
||||
poly = Polygon(metric_pts)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
return poly if (poly.is_valid and not poly.is_empty) else None
|
||||
|
||||
|
||||
__all__ = ["export_concept_dxf"]
|
||||
160
backend/app/services/generative/exporters/pdf.py
Normal file
160
backend/app/services/generative/exporters/pdf.py
Normal file
|
|
@ -0,0 +1,160 @@
|
|||
"""Generative Design — Stage 1c PDF export via WeasyPrint.
|
||||
|
||||
Renders a one-page summary of the three concept variants — a ТЭП table and a
|
||||
financial table — into a PDF. This is the *concept* summary, distinct from Site
|
||||
Finder's ``app.services.exporters.report_pdf`` (advisory site report).
|
||||
|
||||
WeasyPrint is imported *lazily inside the function* (mirrors the repo's
|
||||
``report_pdf`` / ``snapshot_pdf`` house-style): it is a heavy native dependency, so
|
||||
importing this module must never fail even on a dev box without the system libs.
|
||||
|
||||
All dynamic strings are passed through ``html.escape`` (defence-in-depth: variant
|
||||
strategy names are from a fixed Literal, but treat rendered text as untrusted).
|
||||
Returns ``bytes`` ready for an HTTP response / file write. Deterministic, no DB /
|
||||
no network.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import logging
|
||||
from collections.abc import Sequence
|
||||
|
||||
from app.schemas.concept import ConceptVariant
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# RU-подписи стратегий (ключ — Literal из контракта).
|
||||
_STRATEGY_LABELS: dict[str, str] = {
|
||||
"max_area": "Максимум площади",
|
||||
"max_insolation": "Максимум инсоляции",
|
||||
"balanced": "Баланс",
|
||||
}
|
||||
|
||||
_DASH = "—"
|
||||
|
||||
# Минимальный CSS для печати (А4, читаемые таблицы). Inline — без внешних ресурсов.
|
||||
_CSS = """
|
||||
@page { size: A4 landscape; margin: 18mm; }
|
||||
body { font-family: "DejaVu Sans", Arial, sans-serif; font-size: 11px; color: #1a1a1a; }
|
||||
h1 { font-size: 18px; margin: 0 0 4px; }
|
||||
.sub { color: #666; font-size: 10px; margin: 0 0 14px; }
|
||||
table { border-collapse: collapse; width: 100%; margin-bottom: 18px; }
|
||||
th, td { border: 1px solid #ccc; padding: 6px 8px; text-align: right; }
|
||||
th.row, td.row { text-align: left; font-weight: 600; background: #f5f5f5; }
|
||||
caption { text-align: left; font-weight: 700; font-size: 13px; margin-bottom: 6px; }
|
||||
thead th { background: #ececec; }
|
||||
"""
|
||||
|
||||
_TITLE = "Концепции застройки — сводка вариантов"
|
||||
_SUBTITLE = "Generative Design · Stage 1c · детерминированный расчёт ТЭП и финмодели"
|
||||
|
||||
|
||||
def _fmt_int(value: float | int) -> str:
|
||||
"""Целое с разделителями тысяч (узкий пробел) для читаемости."""
|
||||
return f"{round(value):,}".replace(",", " ")
|
||||
|
||||
|
||||
def _fmt_money(value: float) -> str:
|
||||
"""Деньги в млн руб (1 знак) — итоговые таблицы читаются в млн."""
|
||||
return f"{value / 1_000_000:,.1f}".replace(",", " ")
|
||||
|
||||
|
||||
def _strategy_label(strategy: str) -> str:
|
||||
return _STRATEGY_LABELS.get(strategy, strategy)
|
||||
|
||||
|
||||
def _teap_table(variants: Sequence[ConceptVariant]) -> str:
|
||||
"""HTML-таблица ТЭП по всем вариантам (строки — показатели, колонки — стратегии)."""
|
||||
headers = "".join(
|
||||
f"<th>{html.escape(_strategy_label(v.strategy))}</th>" for v in variants
|
||||
)
|
||||
rows: list[tuple[str, list[str]]] = [
|
||||
("Пятно застройки, кв.м", [_fmt_int(v.teap.built_area_sqm) for v in variants]),
|
||||
("Общая площадь (GFA), кв.м", [_fmt_int(v.teap.total_floor_area_sqm) for v in variants]),
|
||||
("Жилая площадь, кв.м", [_fmt_int(v.teap.residential_area_sqm) for v in variants]),
|
||||
("Квартир, шт", [_fmt_int(v.teap.apartments_count) for v in variants]),
|
||||
("Плотность (FAR)", [f"{v.teap.density:.2f}" for v in variants]),
|
||||
("Машиномест", [_fmt_int(v.teap.parking_spaces) for v in variants]),
|
||||
]
|
||||
body = "".join(
|
||||
"<tr><td class='row'>"
|
||||
+ html.escape(label)
|
||||
+ "</td>"
|
||||
+ "".join(f"<td>{html.escape(cell)}</td>" for cell in cells)
|
||||
+ "</tr>"
|
||||
for label, cells in rows
|
||||
)
|
||||
return (
|
||||
"<table><caption>Технико-экономические показатели</caption>"
|
||||
f"<thead><tr><th class='row'>Показатель</th>{headers}</tr></thead>"
|
||||
f"<tbody>{body}</tbody></table>"
|
||||
)
|
||||
|
||||
|
||||
def _financial_table(variants: Sequence[ConceptVariant]) -> str:
|
||||
"""HTML-таблица финмодели (деньги в млн руб; IRR — proxy, помечен)."""
|
||||
headers = "".join(
|
||||
f"<th>{html.escape(_strategy_label(v.strategy))}</th>" for v in variants
|
||||
)
|
||||
rows: list[tuple[str, list[str]]] = [
|
||||
("Выручка, млн руб", [_fmt_money(v.financial.revenue_rub) for v in variants]),
|
||||
("Затраты, млн руб", [_fmt_money(v.financial.cost_rub) for v in variants]),
|
||||
("Валовая маржа, млн руб", [_fmt_money(v.financial.gross_margin_rub) for v in variants]),
|
||||
("IRR-proxy", [f"{v.financial.irr * 100:.1f}%" for v in variants]),
|
||||
]
|
||||
body = "".join(
|
||||
"<tr><td class='row'>"
|
||||
+ html.escape(label)
|
||||
+ "</td>"
|
||||
+ "".join(f"<td>{html.escape(cell)}</td>" for cell in cells)
|
||||
+ "</tr>"
|
||||
for label, cells in rows
|
||||
)
|
||||
return (
|
||||
"<table><caption>Финансовая модель (упрощённая)</caption>"
|
||||
f"<thead><tr><th class='row'>Показатель</th>{headers}</tr></thead>"
|
||||
f"<tbody>{body}</tbody></table>"
|
||||
)
|
||||
|
||||
|
||||
def _build_html(variants: Sequence[ConceptVariant]) -> str:
|
||||
if not variants:
|
||||
return (
|
||||
f"<html><head><meta charset='utf-8'><style>{_CSS}</style></head>"
|
||||
f"<body><h1>{html.escape(_TITLE)}</h1>"
|
||||
f"<p class='sub'>{html.escape(_SUBTITLE)}</p>"
|
||||
f"<p>{_DASH} нет вариантов для отображения</p></body></html>"
|
||||
)
|
||||
return (
|
||||
f"<html><head><meta charset='utf-8'><style>{_CSS}</style></head><body>"
|
||||
f"<h1>{html.escape(_TITLE)}</h1>"
|
||||
f"<p class='sub'>{html.escape(_SUBTITLE)}</p>"
|
||||
f"{_teap_table(variants)}"
|
||||
f"{_financial_table(variants)}"
|
||||
"<p class='sub'>IRR-proxy — аннуализированная маржа-на-затраты без "
|
||||
"дисконтирования (не настоящий IRR). Цены и себестоимость — рыночные "
|
||||
"ориентиры, не калиброванная модель ценообразования.</p>"
|
||||
"</body></html>"
|
||||
)
|
||||
|
||||
|
||||
def export_concept_pdf(variants: Sequence[ConceptVariant]) -> bytes:
|
||||
"""Свести варианты в PDF-сводку (ТЭП + финмодель). Возвращает bytes (PDF).
|
||||
|
||||
Graceful: пустой список вариантов рендерит страницу-заглушку, экспорт не падает.
|
||||
WeasyPrint импортируется лениво (тяжёлая нативная зависимость).
|
||||
"""
|
||||
# Лениво: импорт WeasyPrint не должен падать при импорте модуля
|
||||
# (тяжёлая нативная зависимость; зеркало report_pdf/snapshot_pdf).
|
||||
from weasyprint import HTML
|
||||
|
||||
document = _build_html(variants)
|
||||
# write_pdf(target=None) возвращает bytes; weasyprint без stubs -> явная коэрция.
|
||||
rendered = HTML(string=document).write_pdf()
|
||||
pdf_bytes: bytes = bytes(rendered) if rendered is not None else b""
|
||||
logger.info("PDF export: variants=%d bytes=%d", len(variants), len(pdf_bytes))
|
||||
return pdf_bytes
|
||||
|
||||
|
||||
__all__ = ["export_concept_pdf"]
|
||||
|
|
@ -1,8 +1,97 @@
|
|||
"""Financial model.
|
||||
"""Generative Design — Stage 1c: simplified financial model.
|
||||
|
||||
Stage 1c:
|
||||
revenue = residential_area_sqm * neighborhood_price_per_sqm
|
||||
cost = total_floor_area_sqm * construction_cost_per_sqm + land_cost
|
||||
gross_margin = revenue - cost
|
||||
base_irr = simplified, no time discounting (defer to Phase 1).
|
||||
From the Stage 1c ``TEAP`` block we derive the ``FinancialModel`` contract:
|
||||
|
||||
* ``revenue_rub`` = residential_area_sqm * sale price per sqm (by housing class).
|
||||
* ``cost_rub`` = total_floor_area_sqm * construction cost per sqm + land cost.
|
||||
* ``gross_margin_rub`` = revenue - cost.
|
||||
* ``irr`` = simplified proxy (margin-on-cost / project years), NO time
|
||||
discounting — this is a static stand-in until the Phase 1 cashflow model lands.
|
||||
|
||||
Prices/costs are coarse RU-market proxies for an MVP (см. константы ниже); they are
|
||||
deliberately conservative round numbers, not a calibrated pricing engine. The IRR
|
||||
field is a *proxy*: a real internal rate of return needs a dated cashflow series,
|
||||
which is out of MVP scope — we return an annualised margin ratio so the field is
|
||||
populated with a plausible, monotonic number rather than zero.
|
||||
|
||||
Детерминированно, без LLM / внешних API / БД.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Literal
|
||||
|
||||
from app.schemas.concept import TEAP, FinancialModel
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
HousingClass = Literal["econom", "comfort", "business"]
|
||||
|
||||
# ── Цена продажи жилья, руб/кв.м (proxy рынка ЕКБ/региона, упрощённо) ──────────
|
||||
_SALE_PRICE_PER_SQM: dict[HousingClass, float] = {
|
||||
"econom": 110_000.0,
|
||||
"comfort": 145_000.0,
|
||||
"business": 210_000.0,
|
||||
}
|
||||
|
||||
# ── Себестоимость СМР, руб/кв.м общей площади (выше класс -> дороже отделка/инж) ─
|
||||
_CONSTRUCTION_COST_PER_SQM: dict[HousingClass, float] = {
|
||||
"econom": 72_000.0,
|
||||
"comfort": 88_000.0,
|
||||
"business": 120_000.0,
|
||||
}
|
||||
|
||||
# Условный горизонт проекта (лет) для аннуализации margin-on-cost в IRR-proxy.
|
||||
_PROJECT_YEARS: float = 3.0
|
||||
|
||||
|
||||
def compute_financial(
|
||||
*,
|
||||
teap: TEAP,
|
||||
housing_class: HousingClass,
|
||||
land_cost_rub: float | None,
|
||||
) -> FinancialModel:
|
||||
"""Свести ТЭП + класс + стоимость земли в :class:`FinancialModel`.
|
||||
|
||||
Args:
|
||||
teap: Stage 1c ТЭП (берём residential_area_sqm и total_floor_area_sqm).
|
||||
housing_class: задаёт цену продажи и себестоимость СМР.
|
||||
land_cost_rub: стоимость участка (опционально); None -> 0 в затратах.
|
||||
"""
|
||||
sale_price = _SALE_PRICE_PER_SQM[housing_class]
|
||||
construction_cost = _CONSTRUCTION_COST_PER_SQM[housing_class]
|
||||
|
||||
revenue = teap.residential_area_sqm * sale_price
|
||||
construction = teap.total_floor_area_sqm * construction_cost
|
||||
land = land_cost_rub if land_cost_rub is not None else 0.0
|
||||
cost = construction + land
|
||||
gross_margin = revenue - cost
|
||||
|
||||
# IRR-proxy: аннуализированная маржа-на-затраты. НЕ настоящий IRR (нет дисконта/
|
||||
# дат денежных потоков — отложено в Phase 1). Защита от деления на ноль и
|
||||
# клампинг в разумный диапазон, чтобы поле было монотонным и читаемым.
|
||||
if cost > 0:
|
||||
margin_on_cost = gross_margin / cost
|
||||
irr = margin_on_cost / _PROJECT_YEARS
|
||||
else:
|
||||
irr = 0.0
|
||||
irr = max(-1.0, min(1.0, irr))
|
||||
|
||||
model = FinancialModel(
|
||||
revenue_rub=round(revenue, 2),
|
||||
cost_rub=round(cost, 2),
|
||||
gross_margin_rub=round(gross_margin, 2),
|
||||
irr=round(irr, 4),
|
||||
)
|
||||
logger.info(
|
||||
"financial: revenue=%.0f cost=%.0f margin=%.0f irr_proxy=%.3f",
|
||||
model.revenue_rub,
|
||||
model.cost_rub,
|
||||
model.gross_margin_rub,
|
||||
model.irr,
|
||||
)
|
||||
return model
|
||||
|
||||
|
||||
__all__ = ["HousingClass", "compute_financial"]
|
||||
|
|
|
|||
|
|
@ -1,35 +1,312 @@
|
|||
"""Generative Design — geometry placement.
|
||||
"""Generative Design — Stage 1a geometry: parcel parsing + buildable area + grid.
|
||||
|
||||
Stage 1a: Shapely-based polygon parsing + normative offsets (buffer).
|
||||
Stage 1b: greedy filling of rectangular MKD with 3 strategies. STRtree for collisions.
|
||||
Performance target: <=10s per variant; fallback acceptance 15s.
|
||||
Pipeline (deterministic, no LLM / no external API / no DB):
|
||||
|
||||
1. Parse the parcel polygon from ``ConceptInput.parcel_geojson`` (GeoJSON Polygon,
|
||||
WGS84 / EPSG:4326) into a Shapely geometry.
|
||||
2. Reproject WGS84 -> a local *metric* CRS (an azimuthal-equidistant projection
|
||||
centred on the parcel centroid) so that all downstream maths is in metres.
|
||||
We deliberately avoid UTM zone math: an AEQD centred on the parcel is accurate
|
||||
to well within construction tolerance for parcels of city-block size and is
|
||||
fully deterministic for any longitude/latitude.
|
||||
3. Apply the normative setback (отступ) as an *inward* buffer -> the buildable area
|
||||
(участок минус отступы).
|
||||
4. Lay a deterministic placement grid of candidate cells over the buildable area's
|
||||
bounding box; a cell is kept when its centre falls inside the buildable area.
|
||||
|
||||
The metric geometry + the WGS84<->metric transformers are bundled in :class:`Parcel`
|
||||
so Stage 1b (placement) can do collision maths in metres and reproject the result
|
||||
back to WGS84 for the ``ConceptVariant.buildings_geojson`` contract field.
|
||||
|
||||
``generate()`` (bottom of file) is the public orchestrator that the API calls; it
|
||||
ties together Stage 1a -> 1b -> 1c. ``generate_stub`` is kept as a thin alias so the
|
||||
existing route import keeps working.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from app.schemas.concept import TEAP, ConceptInput, ConceptVariant, FinancialModel
|
||||
from pyproj import CRS, Transformer
|
||||
from shapely.geometry import Polygon, mapping, shape
|
||||
from shapely.geometry.base import BaseGeometry
|
||||
from shapely.ops import transform as shapely_transform
|
||||
|
||||
from app.schemas.concept import ConceptInput, ConceptVariant
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Normative defaults (упрощённо для MVP) ────────────────────────────────────
|
||||
# Setback (отступ от границ участка до пятна застройки), метры. СП 42.13330 даёт
|
||||
# ~3 м минимум от боковых границ; берём 6 м как консервативный proxy с учётом проездов.
|
||||
DEFAULT_SETBACK_M: float = 6.0
|
||||
# Шаг сетки размещения (метры). 3 м — компромисс: достаточно мелкий, чтобы жадная
|
||||
# раскладка реально различала стратегии по разрыву (gap), и достаточно крупный,
|
||||
# чтобы число ячеек/время оставались ограниченными на квартальном участке.
|
||||
DEFAULT_GRID_STEP_M: float = 3.0
|
||||
# Минимальная площадь buildable area (кв.м), ниже которой застройка не имеет смысла.
|
||||
MIN_BUILDABLE_AREA_SQM: float = 50.0
|
||||
# Потолок числа ячеек сетки. Жадная раскладка с перестройкой STRtree ~O(n^2) по числу
|
||||
# размещённых секций; для огромного участка шаг автоматически огрубляется, чтобы
|
||||
# удержать время в бюджете (<=10 c/вариант). MVP-упрощение.
|
||||
MAX_GRID_CELLS: int = 20_000
|
||||
|
||||
# WGS84 (вход контракта).
|
||||
_WGS84 = CRS.from_epsg(4326)
|
||||
|
||||
|
||||
class ParcelGeometryError(ValueError):
|
||||
"""Входной полигон участка невалиден (не Polygon / вырожден / пустой)."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GridCell:
|
||||
"""Одна ячейка сетки размещения (метрическая СК, центр + габариты)."""
|
||||
|
||||
cx: float
|
||||
cy: float
|
||||
width: float
|
||||
height: float
|
||||
|
||||
def as_polygon(self) -> Polygon:
|
||||
"""Прямоугольник ячейки как Shapely-полигон (метры)."""
|
||||
half_w = self.width / 2.0
|
||||
half_h = self.height / 2.0
|
||||
return Polygon(
|
||||
[
|
||||
(self.cx - half_w, self.cy - half_h),
|
||||
(self.cx + half_w, self.cy - half_h),
|
||||
(self.cx + half_w, self.cy + half_h),
|
||||
(self.cx - half_w, self.cy + half_h),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Parcel:
|
||||
"""Распарсенный участок в метрической СК + данные для Stage 1b/1c.
|
||||
|
||||
``polygon_m`` / ``buildable_m`` — геометрия в метрах (AEQD вокруг центроида).
|
||||
``metric_geom_to_wgs84`` репроецирует метрику обратно в WGS84 для GeoJSON-ответа.
|
||||
"""
|
||||
|
||||
polygon_m: Polygon
|
||||
buildable_m: Polygon
|
||||
grid: tuple[GridCell, ...]
|
||||
grid_step_m: float
|
||||
setback_m: float
|
||||
_to_wgs84: Transformer
|
||||
_to_metric: Transformer
|
||||
|
||||
@property
|
||||
def site_area_sqm(self) -> float:
|
||||
"""Площадь участка, кв.м."""
|
||||
return float(self.polygon_m.area)
|
||||
|
||||
@property
|
||||
def buildable_area_sqm(self) -> float:
|
||||
"""Площадь пятна застройки (участок минус отступы), кв.м."""
|
||||
return float(self.buildable_m.area)
|
||||
|
||||
def metric_geom_to_wgs84(self, geom: BaseGeometry) -> dict[str, Any]:
|
||||
"""Репроекция метрической геометрии обратно в WGS84 -> GeoJSON-mapping."""
|
||||
wgs = shapely_transform(self._reproject, geom)
|
||||
return dict(mapping(wgs))
|
||||
|
||||
def wgs84_to_metric(self, lon: float, lat: float) -> tuple[float, float]:
|
||||
"""Одна точка WGS84 (lon, lat) -> метрическая (x, y) в СК участка."""
|
||||
x, y = self._to_metric.transform(lon, lat)
|
||||
return float(x), float(y)
|
||||
|
||||
def _reproject(self, xs: Any, ys: Any) -> tuple[Any, Any]:
|
||||
lon, lat = self._to_wgs84.transform(xs, ys)
|
||||
return lon, lat
|
||||
|
||||
|
||||
def _parse_polygon(parcel_geojson: dict[str, Any]) -> Polygon:
|
||||
"""GeoJSON -> Shapely Polygon. Принимает голую geometry ИЛИ Feature."""
|
||||
if not isinstance(parcel_geojson, dict):
|
||||
raise ParcelGeometryError("parcel_geojson must be a GeoJSON object")
|
||||
|
||||
geom_dict: dict[str, Any] = parcel_geojson
|
||||
if parcel_geojson.get("type") == "Feature":
|
||||
geometry = parcel_geojson.get("geometry")
|
||||
if not isinstance(geometry, dict):
|
||||
raise ParcelGeometryError("Feature has no geometry")
|
||||
geom_dict = geometry
|
||||
|
||||
try:
|
||||
geom = shape(geom_dict)
|
||||
except (KeyError, TypeError, ValueError, AttributeError) as exc:
|
||||
raise ParcelGeometryError(f"cannot parse GeoJSON geometry: {exc}") from exc
|
||||
|
||||
if geom.geom_type != "Polygon":
|
||||
raise ParcelGeometryError(f"expected Polygon, got {geom.geom_type}")
|
||||
if geom.is_empty:
|
||||
raise ParcelGeometryError("parcel polygon is empty")
|
||||
|
||||
polygon = geom if isinstance(geom, Polygon) else Polygon(geom)
|
||||
if not polygon.is_valid:
|
||||
# buffer(0) — канонический Shapely-фикс самопересечений/неориентированных колец.
|
||||
fixed = polygon.buffer(0)
|
||||
if fixed.is_empty or fixed.geom_type != "Polygon":
|
||||
raise ParcelGeometryError("parcel polygon is not a valid simple polygon")
|
||||
polygon = fixed if isinstance(fixed, Polygon) else Polygon(fixed)
|
||||
return polygon
|
||||
|
||||
|
||||
def _metric_transformers(polygon_wgs84: Polygon) -> tuple[Transformer, Transformer]:
|
||||
"""Построить пару трансформеров WGS84<->метрический AEQD вокруг центроида участка.
|
||||
|
||||
AEQD (azimuthal equidistant) центрированный на участке детерминирован для любых
|
||||
координат и точен на масштабе квартала — не нужен выбор UTM-зоны.
|
||||
"""
|
||||
centroid = polygon_wgs84.centroid
|
||||
metric_crs = CRS.from_proj4(
|
||||
f"+proj=aeqd +lat_0={centroid.y} +lon_0={centroid.x} "
|
||||
"+x_0=0 +y_0=0 +ellps=WGS84 +datum=WGS84 +units=m +no_defs"
|
||||
)
|
||||
to_metric = Transformer.from_crs(_WGS84, metric_crs, always_xy=True)
|
||||
to_wgs84 = Transformer.from_crs(metric_crs, _WGS84, always_xy=True)
|
||||
return to_metric, to_wgs84
|
||||
|
||||
|
||||
def build_placement_grid(buildable_m: Polygon, step_m: float) -> tuple[GridCell, ...]:
|
||||
"""Детерминированная сетка ячеек ``step_m x step_m`` над пятном застройки.
|
||||
|
||||
Ячейка попадает в сетку, если её центр лежит внутри ``buildable_m``. Перебор
|
||||
идёт по фиксированному порядку (снизу-вверх, слева-направо) от округлённого
|
||||
минимального угла bbox -> один и тот же вход даёт один и тот же выход.
|
||||
"""
|
||||
if buildable_m.is_empty or step_m <= 0:
|
||||
return ()
|
||||
|
||||
minx, miny, maxx, maxy = buildable_m.bounds
|
||||
# Якорим старт сетки к кратному step, чтобы убрать дрейф от плавающего bbox.
|
||||
start_x = (minx // step_m) * step_m
|
||||
start_y = (miny // step_m) * step_m
|
||||
|
||||
cells: list[GridCell] = []
|
||||
n_cols = int((maxx - start_x) / step_m) + 1
|
||||
n_rows = int((maxy - start_y) / step_m) + 1
|
||||
for row in range(n_rows):
|
||||
cy = start_y + (row + 0.5) * step_m
|
||||
for col in range(n_cols):
|
||||
cx = start_x + (col + 0.5) * step_m
|
||||
cell = GridCell(cx=cx, cy=cy, width=step_m, height=step_m)
|
||||
# Центр внутри пятна — ячейка пригодна (covers ловит и границу).
|
||||
if buildable_m.covers(cell.as_polygon().centroid):
|
||||
cells.append(cell)
|
||||
return tuple(cells)
|
||||
|
||||
|
||||
def _coarsen_step_for_budget(buildable_m: Polygon, step_m: float) -> float:
|
||||
"""Огрубить шаг сетки, если bbox-оценка ячеек превышает :data:`MAX_GRID_CELLS`.
|
||||
|
||||
Грубая оценка по bbox (верхняя граница реального числа ячеек). Возвращает шаг,
|
||||
при котором оценка <= cap; детерминированно. MVP-страховка от взрыва времени на
|
||||
гигантских участках — обычный квартал её не задевает.
|
||||
"""
|
||||
minx, miny, maxx, maxy = buildable_m.bounds
|
||||
width = float(maxx) - float(minx)
|
||||
height = float(maxy) - float(miny)
|
||||
if width <= 0 or height <= 0 or step_m <= 0:
|
||||
return step_m
|
||||
est_cells = (width / step_m) * (height / step_m)
|
||||
if est_cells <= MAX_GRID_CELLS:
|
||||
return step_m
|
||||
# step растёт как sqrt(est/cap), чтобы число ячеек ~= cap.
|
||||
factor: float = (est_cells / MAX_GRID_CELLS) ** 0.5
|
||||
coarsened: float = step_m * factor
|
||||
logger.warning(
|
||||
"buildable bbox %.0fx%.0f m: grid step coarsened %.1f->%.1f m to cap cells at %d",
|
||||
width,
|
||||
height,
|
||||
step_m,
|
||||
coarsened,
|
||||
MAX_GRID_CELLS,
|
||||
)
|
||||
return coarsened
|
||||
|
||||
|
||||
def parse_parcel(
|
||||
payload: ConceptInput,
|
||||
*,
|
||||
setback_m: float = DEFAULT_SETBACK_M,
|
||||
grid_step_m: float = DEFAULT_GRID_STEP_M,
|
||||
) -> Parcel:
|
||||
"""Stage 1a: ConceptInput -> :class:`Parcel` (метрика + buildable + grid).
|
||||
|
||||
Raises:
|
||||
ParcelGeometryError: полигон невалиден или пятно застройки вырождается.
|
||||
"""
|
||||
polygon_wgs84 = _parse_polygon(payload.parcel_geojson)
|
||||
to_metric, to_wgs84 = _metric_transformers(polygon_wgs84)
|
||||
|
||||
def _fwd(xs: Any, ys: Any) -> tuple[Any, Any]:
|
||||
x, y = to_metric.transform(xs, ys)
|
||||
return x, y
|
||||
|
||||
polygon_m = shapely_transform(_fwd, polygon_wgs84)
|
||||
if not isinstance(polygon_m, Polygon):
|
||||
raise ParcelGeometryError("reprojected parcel is not a polygon")
|
||||
|
||||
# Отступ внутрь: отрицательный буфер. join_style=mitre держит прямые углы.
|
||||
buildable = polygon_m.buffer(-setback_m, join_style="mitre")
|
||||
if buildable.is_empty:
|
||||
raise ParcelGeometryError(
|
||||
f"setback {setback_m} m consumes the whole parcel "
|
||||
f"(area={polygon_m.area:.1f} sqm) — no buildable area"
|
||||
)
|
||||
# После буфера может остаться MultiPolygon (узкий перешеек) — берём крупнейший.
|
||||
if buildable.geom_type == "MultiPolygon":
|
||||
buildable = max(buildable.geoms, key=lambda g: g.area)
|
||||
if not isinstance(buildable, Polygon):
|
||||
raise ParcelGeometryError("buildable area degenerated after setback")
|
||||
if buildable.area < MIN_BUILDABLE_AREA_SQM:
|
||||
raise ParcelGeometryError(
|
||||
f"buildable area {buildable.area:.1f} sqm below minimum "
|
||||
f"{MIN_BUILDABLE_AREA_SQM} sqm"
|
||||
)
|
||||
|
||||
effective_step = _coarsen_step_for_budget(buildable, grid_step_m)
|
||||
grid = build_placement_grid(buildable, effective_step)
|
||||
logger.info(
|
||||
"parsed parcel: site=%.0f sqm buildable=%.0f sqm grid_cells=%d step=%.1fm",
|
||||
polygon_m.area,
|
||||
buildable.area,
|
||||
len(grid),
|
||||
effective_step,
|
||||
)
|
||||
return Parcel(
|
||||
polygon_m=polygon_m,
|
||||
buildable_m=buildable,
|
||||
grid=grid,
|
||||
grid_step_m=effective_step,
|
||||
setback_m=setback_m,
|
||||
_to_wgs84=to_wgs84,
|
||||
_to_metric=to_metric,
|
||||
)
|
||||
|
||||
|
||||
def generate(payload: ConceptInput) -> list[ConceptVariant]:
|
||||
"""Public orchestrator: Stage 1a -> 1b -> 1c -> 3 filled :class:`ConceptVariant`.
|
||||
|
||||
Deterministic end-to-end. On a degenerate parcel (setback eats everything, bad
|
||||
geometry) we *log and re-raise* :class:`ParcelGeometryError` — the API layer maps
|
||||
it to 4xx; silently returning zero-variants would hide a bad request.
|
||||
"""
|
||||
# Local import to avoid a module-level import cycle (placement imports geometry).
|
||||
from app.services.generative import placement
|
||||
|
||||
parcel = parse_parcel(payload)
|
||||
variants = placement.place_all_strategies(parcel, payload)
|
||||
logger.info("generated %d concept variants", len(variants))
|
||||
return variants
|
||||
|
||||
|
||||
def generate_stub(payload: ConceptInput) -> list[ConceptVariant]:
|
||||
"""Placeholder returning 3 empty variants. Replaced in Stage 1b."""
|
||||
empty_buildings: dict[str, Any] = {"type": "FeatureCollection", "features": []}
|
||||
empty_teap = TEAP(
|
||||
built_area_sqm=0.0,
|
||||
total_floor_area_sqm=0.0,
|
||||
residential_area_sqm=0.0,
|
||||
apartments_count=0,
|
||||
density=0.0,
|
||||
parking_spaces=0,
|
||||
)
|
||||
empty_financial = FinancialModel(revenue_rub=0.0, cost_rub=0.0, gross_margin_rub=0.0, irr=0.0)
|
||||
strategies: list[ConceptVariant] = []
|
||||
for strategy in ("max_area", "max_insolation", "balanced"):
|
||||
strategies.append(
|
||||
ConceptVariant(
|
||||
strategy=strategy,
|
||||
buildings_geojson=empty_buildings,
|
||||
teap=empty_teap,
|
||||
financial=empty_financial,
|
||||
)
|
||||
)
|
||||
return strategies
|
||||
"""Backwards-compatible alias. Now delegates to the real :func:`generate`."""
|
||||
return generate(payload)
|
||||
|
|
|
|||
237
backend/app/services/generative/placement.py
Normal file
237
backend/app/services/generative/placement.py
Normal file
|
|
@ -0,0 +1,237 @@
|
|||
"""Generative Design — Stage 1b: greedy section placement with STRtree collisions.
|
||||
|
||||
Given a parsed :class:`~app.services.generative.geometry.Parcel` (Stage 1a) we place
|
||||
rectangular residential sections (секции МКД) onto the placement grid using a greedy
|
||||
sweep. Three strategies trade plot density against insolation comfort:
|
||||
|
||||
* ``max_area`` — tight gaps, deep building footprint -> maximum buildable area.
|
||||
* ``max_insolation`` — wide gaps + slimmer footprint -> light/air between buildings.
|
||||
* ``balanced`` — the middle ground.
|
||||
|
||||
Collisions (overlap + minimum inter-section gap) are checked with a Shapely STRtree
|
||||
spatial index, rebuilt as placements accumulate. The greedy sweep is fully
|
||||
deterministic: candidate anchors are visited in a fixed grid order and the first
|
||||
non-colliding footprint that stays inside the buildable area wins.
|
||||
|
||||
Each placed footprint is reprojected back to WGS84 and emitted as a GeoJSON Feature
|
||||
in ``ConceptVariant.buildings_geojson``; the metric footprints feed Stage 1c
|
||||
(``teap`` + ``financial``) so the variant is filled with real numbers, not zeros.
|
||||
|
||||
Deterministic, no LLM / no external API / no DB.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
|
||||
from shapely.geometry import Polygon, box
|
||||
from shapely.strtree import STRtree
|
||||
|
||||
from app.schemas.concept import ConceptInput, ConceptVariant
|
||||
from app.services.generative import financial, teap
|
||||
from app.services.generative.geometry import Parcel
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Тип стратегии должен совпадать с Literal в ConceptVariant.strategy.
|
||||
StrategyName = str
|
||||
|
||||
# Высота этажа (м) — для перевода target_floors в метрическую высоту/площади.
|
||||
FLOOR_HEIGHT_M: float = 3.0
|
||||
|
||||
# ── Потолок коэффициента застройки (built / buildable) по типу застройки ──────
|
||||
# Контракт не несёт явного FAR/max_coverage, поэтому используем development_type как
|
||||
# естественный регулятор плотности. Жадная раскладка перестаёт ставить секции, как
|
||||
# только пятно достигает доли buildable area ниже. Без этого max_area патологически
|
||||
# забивает участок и даёт нереалистичный FAR. MVP-упрощение (нормативный proxy).
|
||||
_COVERAGE_CAP_BY_TYPE: dict[str, float] = {
|
||||
"spot": 0.35, # точечная застройка — низкое покрытие
|
||||
"mid_rise": 0.45, # среднеэтажная
|
||||
"high_rise": 0.50, # высотная — компактнее пятно, выше этажность
|
||||
}
|
||||
_DEFAULT_COVERAGE_CAP: float = 0.45
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StrategySpec:
|
||||
"""Параметры одной стратегии размещения.
|
||||
|
||||
section_w/section_d — габариты секции (ширина x глубина), метры.
|
||||
gap_m — минимальный разрыв между секциями (инсоляция/противопожарный), метры.
|
||||
floors_factor — множитель к target_floors (комфорт-класс «садит» этажность,
|
||||
макс-площадь «тянет» вверх); этажность клампится к [1, 30] контракта.
|
||||
"""
|
||||
|
||||
name: StrategyName
|
||||
section_w: float
|
||||
section_d: float
|
||||
gap_m: float
|
||||
floors_factor: float
|
||||
|
||||
|
||||
# ── Три стратегии (упрощённо для MVP, габариты типовых панельных/монолитных секций) ──
|
||||
_STRATEGIES: tuple[StrategySpec, ...] = (
|
||||
# Максимум площади: глубокий корпус, минимальные противопожарные разрывы.
|
||||
StrategySpec(name="max_area", section_w=24.0, section_d=18.0, gap_m=6.0, floors_factor=1.15),
|
||||
# Максимум инсоляции: тонкий корпус, широкие разрывы между секциями.
|
||||
StrategySpec(
|
||||
name="max_insolation", section_w=18.0, section_d=12.0, gap_m=15.0, floors_factor=0.85
|
||||
),
|
||||
# Баланс.
|
||||
StrategySpec(name="balanced", section_w=21.0, section_d=15.0, gap_m=10.0, floors_factor=1.0),
|
||||
)
|
||||
|
||||
_FLOORS_MIN = 1
|
||||
_FLOORS_MAX = 30
|
||||
|
||||
|
||||
def _resolve_floors(target_floors: int, factor: float) -> int:
|
||||
"""target_floors * factor, округление к ближайшему, клампинг к [1, 30]."""
|
||||
floors = round(target_floors * factor)
|
||||
return max(_FLOORS_MIN, min(_FLOORS_MAX, floors))
|
||||
|
||||
|
||||
def _greedy_place(
|
||||
parcel: Parcel,
|
||||
spec: StrategySpec,
|
||||
coverage_cap: float,
|
||||
) -> list[Polygon]:
|
||||
"""Жадно разложить секции ``spec`` по сетке участка. Возвращает footprints (метры).
|
||||
|
||||
Алгоритм:
|
||||
* кандидат-якоря — центры ячеек сетки в фиксированном порядке;
|
||||
* footprint строится центрированно на якоре;
|
||||
* принимается, если целиком внутри buildable area И не нарушает разрыв ``gap_m``
|
||||
с уже принятыми (проверка через STRtree по buffered-footprints);
|
||||
* раскладка останавливается, когда пятно достигает ``coverage_cap`` от buildable
|
||||
area (регулятор плотности по типу застройки) — это также ограничивает число
|
||||
размещений и держит O(n^2)-перестройку STRtree в бюджете.
|
||||
"""
|
||||
buildable = parcel.buildable_m
|
||||
max_built = buildable.area * coverage_cap
|
||||
placed: list[Polygon] = []
|
||||
built_area = 0.0
|
||||
# Буферизованные footprints для проверки разрыва; индекс STRtree по ним.
|
||||
buffered: list[Polygon] = []
|
||||
tree: STRtree | None = None
|
||||
|
||||
half_w = spec.section_w / 2.0
|
||||
half_d = spec.section_d / 2.0
|
||||
half_gap = spec.gap_m / 2.0
|
||||
|
||||
for cell in parcel.grid:
|
||||
if built_area >= max_built:
|
||||
break
|
||||
footprint = box(
|
||||
cell.cx - half_w,
|
||||
cell.cy - half_d,
|
||||
cell.cx + half_w,
|
||||
cell.cy + half_d,
|
||||
)
|
||||
# Целиком внутри пятна застройки (covers допускает касание границы).
|
||||
if not buildable.covers(footprint):
|
||||
continue
|
||||
|
||||
# Разрыв между секциями: буферим кандидата на half_gap и проверяем пересечение
|
||||
# с буферизованными соседями — две секции с зазором >= gap_m не пересекутся.
|
||||
candidate_buf = footprint.buffer(half_gap, join_style="mitre")
|
||||
if tree is not None:
|
||||
collision = False
|
||||
for idx in tree.query(candidate_buf):
|
||||
if candidate_buf.intersects(buffered[idx]):
|
||||
collision = True
|
||||
break
|
||||
if collision:
|
||||
continue
|
||||
|
||||
placed.append(footprint)
|
||||
built_area += footprint.area
|
||||
buffered.append(candidate_buf)
|
||||
tree = STRtree(buffered)
|
||||
|
||||
logger.info(
|
||||
"strategy=%s placed %d sections (%.0fx%.0f m, gap=%.0f m, coverage<=%.0f%%)",
|
||||
spec.name,
|
||||
len(placed),
|
||||
spec.section_w,
|
||||
spec.section_d,
|
||||
spec.gap_m,
|
||||
coverage_cap * 100,
|
||||
)
|
||||
return placed
|
||||
|
||||
|
||||
def _footprints_to_geojson(
|
||||
parcel: Parcel,
|
||||
footprints: list[Polygon],
|
||||
floors: int,
|
||||
spec: StrategySpec,
|
||||
) -> dict[str, object]:
|
||||
"""Метрические footprints -> WGS84 FeatureCollection (контракт buildings_geojson)."""
|
||||
features: list[dict[str, object]] = []
|
||||
for i, fp in enumerate(footprints):
|
||||
geom_wgs = parcel.metric_geom_to_wgs84(fp)
|
||||
features.append(
|
||||
{
|
||||
"type": "Feature",
|
||||
"geometry": geom_wgs,
|
||||
"properties": {
|
||||
"section_id": i + 1,
|
||||
"floors": floors,
|
||||
"footprint_sqm": round(float(fp.area), 1),
|
||||
"strategy": spec.name,
|
||||
},
|
||||
}
|
||||
)
|
||||
return {"type": "FeatureCollection", "features": features}
|
||||
|
||||
|
||||
def place_strategy(
|
||||
parcel: Parcel,
|
||||
payload: ConceptInput,
|
||||
spec: StrategySpec,
|
||||
) -> ConceptVariant:
|
||||
"""Полный проход одной стратегии: размещение -> ТЭП -> финмодель -> ConceptVariant."""
|
||||
floors = _resolve_floors(payload.target_floors, spec.floors_factor)
|
||||
coverage_cap = _COVERAGE_CAP_BY_TYPE.get(payload.development_type, _DEFAULT_COVERAGE_CAP)
|
||||
footprints = _greedy_place(parcel, spec, coverage_cap)
|
||||
|
||||
teap_result = teap.compute_teap(
|
||||
footprints=footprints,
|
||||
floors=floors,
|
||||
site_area_sqm=parcel.site_area_sqm,
|
||||
housing_class=payload.housing_class,
|
||||
)
|
||||
financial_result = financial.compute_financial(
|
||||
teap=teap_result,
|
||||
housing_class=payload.housing_class,
|
||||
land_cost_rub=payload.land_cost_rub,
|
||||
)
|
||||
buildings_geojson = _footprints_to_geojson(parcel, footprints, floors, spec)
|
||||
|
||||
# spec.name строится из фиксированного литерала -> совпадает с Literal контракта.
|
||||
return ConceptVariant(
|
||||
strategy=spec.name, # type: ignore[arg-type]
|
||||
buildings_geojson=buildings_geojson,
|
||||
teap=teap_result,
|
||||
financial=financial_result,
|
||||
)
|
||||
|
||||
|
||||
def place_all_strategies(parcel: Parcel, payload: ConceptInput) -> list[ConceptVariant]:
|
||||
"""Stage 1b entry: построить три варианта (max_area / max_insolation / balanced)."""
|
||||
variants = [place_strategy(parcel, payload, spec) for spec in _STRATEGIES]
|
||||
logger.info(
|
||||
"placed all strategies: %s",
|
||||
", ".join(f"{v.strategy}={v.teap.apartments_count}кв" for v in variants),
|
||||
)
|
||||
return variants
|
||||
|
||||
|
||||
__all__ = [
|
||||
"FLOOR_HEIGHT_M",
|
||||
"StrategySpec",
|
||||
"place_all_strategies",
|
||||
"place_strategy",
|
||||
]
|
||||
|
|
@ -1,5 +1,106 @@
|
|||
"""TEAP (technical-economic indicators) calculations.
|
||||
"""Generative Design — Stage 1c: ТЭП (technical-economic indicators).
|
||||
|
||||
Stage 1c: apartment count by ratios (1/2/3-room), KEP (land use coefficient),
|
||||
density, parking by simplified norm.
|
||||
From the Stage 1b placement (rectangular section footprints + floor count) we derive
|
||||
the ``TEAP`` contract block with *real* numbers:
|
||||
|
||||
* ``built_area_sqm`` — пятно застройки (сумма площадей footprint-ов).
|
||||
* ``total_floor_area_sqm`` — общая поэтажная площадь (GFA) = пятно * этажность.
|
||||
* ``residential_area_sqm`` — продаваемая жилая = GFA * коэффициент эффективности
|
||||
(вычет МОП/лестниц/тех.помещений; зависит от класса жилья).
|
||||
* ``apartments_count`` — жилая / средняя площадь квартиры (зависит от класса).
|
||||
* ``density`` — плотность застройки = FAR = GFA / площадь участка.
|
||||
* ``parking_spaces`` — машиноместа по упрощённой норме (мест на квартиру).
|
||||
|
||||
Все коэффициенты — упрощённые нормативные proxy для MVP (см. константы ниже).
|
||||
Детерминированно, без LLM / внешних API / БД.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import math
|
||||
from typing import Literal
|
||||
|
||||
from shapely.geometry import Polygon
|
||||
|
||||
from app.schemas.concept import TEAP
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
HousingClass = Literal["econom", "comfort", "business"]
|
||||
|
||||
# ── Коэффициент эффективности площади (residential / GFA), доля ────────────────
|
||||
# Доля продаваемой жилой в общей поэтажной (остальное — МОП, лестницы, тех.этаж).
|
||||
# Бизнес-класс «тратит» больше на МОП/лобби -> ниже эффективность.
|
||||
_EFFICIENCY_BY_CLASS: dict[HousingClass, float] = {
|
||||
"econom": 0.82,
|
||||
"comfort": 0.78,
|
||||
"business": 0.72,
|
||||
}
|
||||
|
||||
# ── Средняя площадь квартиры (кв.м) по классу — выше класс -> крупнее лот ──────
|
||||
_AVG_APARTMENT_SQM: dict[HousingClass, float] = {
|
||||
"econom": 42.0,
|
||||
"comfort": 55.0,
|
||||
"business": 78.0,
|
||||
}
|
||||
|
||||
# ── Норма парковки (машиномест на квартиру) по классу ─────────────────────────
|
||||
_PARKING_PER_APARTMENT: dict[HousingClass, float] = {
|
||||
"econom": 0.8,
|
||||
"comfort": 1.0,
|
||||
"business": 1.5,
|
||||
}
|
||||
|
||||
|
||||
def compute_teap(
|
||||
*,
|
||||
footprints: list[Polygon],
|
||||
floors: int,
|
||||
site_area_sqm: float,
|
||||
housing_class: HousingClass,
|
||||
) -> TEAP:
|
||||
"""Свести footprints + этажность в :class:`TEAP`.
|
||||
|
||||
Args:
|
||||
footprints: метрические пятна секций (кв.м берётся из ``.area``).
|
||||
floors: этажность (общая для всех секций варианта).
|
||||
site_area_sqm: площадь участка для плотности (FAR).
|
||||
housing_class: класс жилья — задаёт эффективность/средний лот/парковку.
|
||||
"""
|
||||
built_area = float(sum(fp.area for fp in footprints))
|
||||
total_floor_area = built_area * max(0, floors)
|
||||
|
||||
efficiency = _EFFICIENCY_BY_CLASS[housing_class]
|
||||
residential_area = total_floor_area * efficiency
|
||||
|
||||
avg_apartment = _AVG_APARTMENT_SQM[housing_class]
|
||||
apartments_count = math.floor(residential_area / avg_apartment) if avg_apartment else 0
|
||||
|
||||
# Плотность застройки = FAR (GFA / площадь участка). Защита от деления на ноль.
|
||||
density = total_floor_area / site_area_sqm if site_area_sqm > 0 else 0.0
|
||||
|
||||
parking_norm = _PARKING_PER_APARTMENT[housing_class]
|
||||
parking_spaces = math.ceil(apartments_count * parking_norm)
|
||||
|
||||
teap = TEAP(
|
||||
built_area_sqm=round(built_area, 1),
|
||||
total_floor_area_sqm=round(total_floor_area, 1),
|
||||
residential_area_sqm=round(residential_area, 1),
|
||||
apartments_count=apartments_count,
|
||||
density=round(density, 3),
|
||||
parking_spaces=parking_spaces,
|
||||
)
|
||||
logger.info(
|
||||
"TEAP: built=%.0f GFA=%.0f resid=%.0f apts=%d FAR=%.2f parking=%d",
|
||||
teap.built_area_sqm,
|
||||
teap.total_floor_area_sqm,
|
||||
teap.residential_area_sqm,
|
||||
teap.apartments_count,
|
||||
teap.density,
|
||||
teap.parking_spaces,
|
||||
)
|
||||
return teap
|
||||
|
||||
|
||||
__all__ = ["HousingClass", "compute_teap"]
|
||||
|
|
|
|||
|
|
@ -79,12 +79,25 @@ warn_unused_ignores = true
|
|||
[[tool.mypy.overrides]]
|
||||
module = [
|
||||
"app.services.generative.geometry",
|
||||
"app.services.generative.placement",
|
||||
"app.services.generative.teap",
|
||||
"app.services.generative.financial",
|
||||
"app.services.generative.exporters.dxf",
|
||||
"app.services.generative.exporters.pdf",
|
||||
"app.services.site_finder.scorer",
|
||||
]
|
||||
strict = true
|
||||
|
||||
# Геометрия/экспорт-библиотеки без type stubs (shapely/ezdxf/weasyprint не несут
|
||||
# py.typed) — игнорируем missing-imports, чтобы strict-модули generative проходили.
|
||||
[[tool.mypy.overrides]]
|
||||
module = [
|
||||
"shapely.*",
|
||||
"ezdxf.*",
|
||||
"weasyprint.*",
|
||||
]
|
||||
ignore_missing_imports = true
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
asyncio_mode = "auto"
|
||||
|
|
|
|||
115
backend/tests/services/generative/test_api_concepts.py
Normal file
115
backend/tests/services/generative/test_api_concepts.py
Normal file
|
|
@ -0,0 +1,115 @@
|
|||
"""End-to-end API test — POST /api/v1/concepts returns 3 filled variants.
|
||||
|
||||
Goes through the FastAPI route (TestClient) and asserts the contract is populated
|
||||
with *real* numbers (non-zero ТЭП + financial), valid building GeoJSON, and that a
|
||||
degenerate parcel yields 422 rather than empty variants.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
|
||||
_PARCEL = {
|
||||
"type": "Polygon",
|
||||
"coordinates": [
|
||||
[
|
||||
[60.60, 56.830],
|
||||
[60.6045, 56.830],
|
||||
[60.6045, 56.8328],
|
||||
[60.60, 56.8328],
|
||||
[60.60, 56.830],
|
||||
]
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _post(payload: dict[str, object]) -> object:
|
||||
client = TestClient(app)
|
||||
return client.post("/api/v1/concepts", json=payload)
|
||||
|
||||
|
||||
def test_concepts_returns_three_filled_variants() -> None:
|
||||
response = _post(
|
||||
{
|
||||
"parcel_geojson": _PARCEL,
|
||||
"housing_class": "comfort",
|
||||
"target_floors": 9,
|
||||
"development_type": "mid_rise",
|
||||
"land_cost_rub": 150_000_000,
|
||||
}
|
||||
)
|
||||
assert response.status_code == 200, response.text
|
||||
variants = response.json()["variants"]
|
||||
assert len(variants) == 3
|
||||
assert {v["strategy"] for v in variants} == {"max_area", "max_insolation", "balanced"}
|
||||
|
||||
for v in variants:
|
||||
teap = v["teap"]
|
||||
fin = v["financial"]
|
||||
# Реальные, ненулевые числа (не stub-нули).
|
||||
assert teap["built_area_sqm"] > 0
|
||||
assert teap["total_floor_area_sqm"] > 0
|
||||
assert teap["residential_area_sqm"] > 0
|
||||
assert teap["apartments_count"] > 0
|
||||
assert teap["density"] > 0
|
||||
assert teap["parking_spaces"] > 0
|
||||
assert fin["revenue_rub"] > 0
|
||||
assert fin["cost_rub"] > 0
|
||||
# GeoJSON застройки непустой.
|
||||
fc = v["buildings_geojson"]
|
||||
assert fc["type"] == "FeatureCollection"
|
||||
assert len(fc["features"]) > 0
|
||||
|
||||
|
||||
def test_concepts_degenerate_parcel_returns_422() -> None:
|
||||
tiny = {
|
||||
"type": "Polygon",
|
||||
"coordinates": [
|
||||
[
|
||||
[60.60, 56.830],
|
||||
[60.60015, 56.830],
|
||||
[60.60015, 56.83015],
|
||||
[60.60, 56.83015],
|
||||
[60.60, 56.830],
|
||||
]
|
||||
],
|
||||
}
|
||||
response = _post(
|
||||
{
|
||||
"parcel_geojson": tiny,
|
||||
"housing_class": "comfort",
|
||||
"target_floors": 9,
|
||||
"development_type": "mid_rise",
|
||||
}
|
||||
)
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
def test_concepts_response_matches_contract_keys() -> None:
|
||||
response = _post(
|
||||
{
|
||||
"parcel_geojson": _PARCEL,
|
||||
"housing_class": "business",
|
||||
"target_floors": 16,
|
||||
"development_type": "high_rise",
|
||||
}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
variant = response.json()["variants"][0]
|
||||
assert set(variant.keys()) == {"strategy", "buildings_geojson", "teap", "financial"}
|
||||
assert set(variant["teap"].keys()) == {
|
||||
"built_area_sqm",
|
||||
"total_floor_area_sqm",
|
||||
"residential_area_sqm",
|
||||
"apartments_count",
|
||||
"density",
|
||||
"parking_spaces",
|
||||
}
|
||||
assert set(variant["financial"].keys()) == {
|
||||
"revenue_rub",
|
||||
"cost_rub",
|
||||
"gross_margin_rub",
|
||||
"irr",
|
||||
}
|
||||
127
backend/tests/services/generative/test_exporters.py
Normal file
127
backend/tests/services/generative/test_exporters.py
Normal file
|
|
@ -0,0 +1,127 @@
|
|||
"""Stage 1c tests — DXF and PDF exporters.
|
||||
|
||||
DXF is asserted by a binary round-trip (re-read with ezdxf, check layers/entities).
|
||||
The PDF render is skipped when WeasyPrint's native libraries are unavailable (dev
|
||||
boxes без libgobject); the HTML-build step is always exercised.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import tempfile
|
||||
from collections import Counter
|
||||
|
||||
import ezdxf
|
||||
import pytest
|
||||
|
||||
from app.schemas.concept import ConceptInput
|
||||
from app.services.generative import geometry
|
||||
from app.services.generative.exporters import dxf, pdf
|
||||
|
||||
_PARCEL_COORDS = [
|
||||
[60.60, 56.830],
|
||||
[60.6045, 56.830],
|
||||
[60.6045, 56.8328],
|
||||
[60.60, 56.8328],
|
||||
[60.60, 56.830],
|
||||
]
|
||||
|
||||
|
||||
def _payload() -> ConceptInput:
|
||||
return ConceptInput(
|
||||
parcel_geojson={"type": "Polygon", "coordinates": [_PARCEL_COORDS]},
|
||||
housing_class="comfort",
|
||||
target_floors=9,
|
||||
development_type="mid_rise",
|
||||
land_cost_rub=150_000_000.0,
|
||||
)
|
||||
|
||||
|
||||
def _weasyprint_available() -> bool:
|
||||
"""WeasyPrint импортируется только с нативными библиотеками (libgobject и т.д.)."""
|
||||
if importlib.util.find_spec("weasyprint") is None:
|
||||
return False
|
||||
try:
|
||||
import weasyprint # noqa: F401
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def test_dxf_export_round_trips_with_layers() -> None:
|
||||
payload = _payload()
|
||||
parcel = geometry.parse_parcel(payload)
|
||||
variant = geometry.generate(payload)[0]
|
||||
|
||||
data = dxf.export_concept_dxf(parcel, variant)
|
||||
assert data.startswith(b"AutoCAD Binary DXF")
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".dxf", delete=False) as fh:
|
||||
fh.write(data)
|
||||
path = fh.name
|
||||
try:
|
||||
doc = ezdxf.readfile(path)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
msp = doc.modelspace()
|
||||
by_layer = Counter(e.dxf.layer for e in msp)
|
||||
# Участок и пятно застройки нарисованы.
|
||||
assert by_layer["PARCEL"] == 1
|
||||
assert by_layer["BUILDABLE"] == 1
|
||||
# Секции нарисованы (по одному polyline на секцию + подписи).
|
||||
n_sections = len(variant.buildings_geojson["features"])
|
||||
assert n_sections > 0
|
||||
assert by_layer["BUILDINGS"] >= n_sections
|
||||
|
||||
|
||||
def test_dxf_building_footprints_have_metric_area() -> None:
|
||||
from shapely.geometry import Polygon
|
||||
|
||||
payload = _payload()
|
||||
parcel = geometry.parse_parcel(payload)
|
||||
variant = geometry.generate(payload)[0]
|
||||
data = dxf.export_concept_dxf(parcel, variant)
|
||||
|
||||
with tempfile.NamedTemporaryFile(suffix=".dxf", delete=False) as fh:
|
||||
fh.write(data)
|
||||
path = fh.name
|
||||
try:
|
||||
doc = ezdxf.readfile(path)
|
||||
finally:
|
||||
os.unlink(path)
|
||||
|
||||
areas = []
|
||||
for e in doc.modelspace():
|
||||
if e.dxftype() == "LWPOLYLINE" and e.dxf.layer == "BUILDINGS":
|
||||
pts = [(p[0], p[1]) for p in e.get_points()]
|
||||
areas.append(Polygon(pts).area)
|
||||
assert areas, "no building polylines found"
|
||||
# Площади секций — десятки/сотни кв.м (метры), не доли (градусы).
|
||||
for area in areas:
|
||||
assert 50.0 < area < 2000.0
|
||||
|
||||
|
||||
def test_pdf_html_build_contains_tables() -> None:
|
||||
variants = geometry.generate(_payload())
|
||||
html = pdf._build_html(variants)
|
||||
assert "Технико-экономические показатели" in html
|
||||
assert "Финансовая модель" in html
|
||||
assert "IRR-proxy" in html
|
||||
|
||||
|
||||
def test_pdf_html_build_graceful_on_empty() -> None:
|
||||
html = pdf._build_html([])
|
||||
assert "нет вариантов" in html
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not _weasyprint_available(),
|
||||
reason="WeasyPrint native libs unavailable on this host",
|
||||
)
|
||||
def test_pdf_export_produces_pdf_bytes() -> None:
|
||||
variants = geometry.generate(_payload())
|
||||
data = pdf.export_concept_pdf(variants)
|
||||
assert data.startswith(b"%PDF-")
|
||||
assert len(data) > 1000
|
||||
118
backend/tests/services/generative/test_geometry.py
Normal file
118
backend/tests/services/generative/test_geometry.py
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
"""Stage 1a tests — parcel parsing, setback buffer, placement grid.
|
||||
|
||||
Deterministic geometry: a known WGS84 rectangle near ЕКБ is parsed into metres, the
|
||||
setback shrinks it, and the grid covers the buildable area. No network / no DB.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
import pytest
|
||||
from shapely.geometry import Polygon
|
||||
|
||||
from app.schemas.concept import ConceptInput
|
||||
from app.services.generative import geometry
|
||||
from app.services.generative.geometry import ParcelGeometryError, build_placement_grid
|
||||
|
||||
# ~450 m x 310 m rectangle near Екатеринбург (WGS84). Area ~ 0.86 ha.
|
||||
_PARCEL_COORDS = [
|
||||
[60.60, 56.830],
|
||||
[60.6045, 56.830],
|
||||
[60.6045, 56.8328],
|
||||
[60.60, 56.8328],
|
||||
[60.60, 56.830],
|
||||
]
|
||||
|
||||
|
||||
def _payload(**overrides: object) -> ConceptInput:
|
||||
base: dict[str, object] = {
|
||||
"parcel_geojson": {"type": "Polygon", "coordinates": [_PARCEL_COORDS]},
|
||||
"housing_class": "comfort",
|
||||
"target_floors": 9,
|
||||
"development_type": "mid_rise",
|
||||
}
|
||||
base.update(overrides)
|
||||
return ConceptInput(**base) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_parse_parcel_reprojects_to_metres() -> None:
|
||||
parcel = geometry.parse_parcel(_payload())
|
||||
# Площадь в кв.м должна быть на масштабе квартала (десятки тысяч кв.м), не градусов.
|
||||
assert 50_000 < parcel.site_area_sqm < 150_000
|
||||
# Buildable меньше участка ровно за счёт отступа.
|
||||
assert parcel.buildable_area_sqm < parcel.site_area_sqm
|
||||
assert parcel.setback_m == geometry.DEFAULT_SETBACK_M
|
||||
|
||||
|
||||
def test_setback_shrinks_area_by_expected_band() -> None:
|
||||
setback = 6.0
|
||||
parcel = geometry.parse_parcel(_payload(), setback_m=setback)
|
||||
# Грубая нижняя граница убыли: периметр * setback (внутренний буфер).
|
||||
perimeter = parcel.polygon_m.length
|
||||
expected_loss = perimeter * setback
|
||||
actual_loss = parcel.site_area_sqm - parcel.buildable_area_sqm
|
||||
# Внутренний буфер срезает примерно полосу шириной setback по периметру (±50%).
|
||||
assert 0.5 * expected_loss < actual_loss < 1.5 * expected_loss
|
||||
|
||||
|
||||
def test_grid_cells_lie_inside_buildable() -> None:
|
||||
parcel = geometry.parse_parcel(_payload(), grid_step_m=6.0)
|
||||
assert len(parcel.grid) > 0
|
||||
for cell in parcel.grid:
|
||||
assert parcel.buildable_m.covers(cell.as_polygon().centroid)
|
||||
|
||||
|
||||
def test_parse_is_deterministic() -> None:
|
||||
a = geometry.parse_parcel(_payload())
|
||||
b = geometry.parse_parcel(_payload())
|
||||
assert a.site_area_sqm == b.site_area_sqm
|
||||
assert a.buildable_area_sqm == b.buildable_area_sqm
|
||||
assert len(a.grid) == len(b.grid)
|
||||
assert [(c.cx, c.cy) for c in a.grid] == [(c.cx, c.cy) for c in b.grid]
|
||||
|
||||
|
||||
def test_feature_geojson_is_accepted() -> None:
|
||||
# GeoJSON Feature (а не голая geometry) тоже парсится.
|
||||
payload = _payload(
|
||||
parcel_geojson={
|
||||
"type": "Feature",
|
||||
"properties": {},
|
||||
"geometry": {"type": "Polygon", "coordinates": [_PARCEL_COORDS]},
|
||||
}
|
||||
)
|
||||
parcel = geometry.parse_parcel(payload)
|
||||
assert parcel.site_area_sqm > 0
|
||||
|
||||
|
||||
def test_tiny_parcel_rejected_after_setback() -> None:
|
||||
# ~16 m x 16 m: отступ 6 м с каждой стороны схлопывает пятно застройки.
|
||||
tiny = [
|
||||
[60.60, 56.830],
|
||||
[60.60015, 56.830],
|
||||
[60.60015, 56.83015],
|
||||
[60.60, 56.83015],
|
||||
[60.60, 56.830],
|
||||
]
|
||||
payload = _payload(parcel_geojson={"type": "Polygon", "coordinates": [tiny]})
|
||||
with pytest.raises(ParcelGeometryError):
|
||||
geometry.parse_parcel(payload)
|
||||
|
||||
|
||||
def test_non_polygon_rejected() -> None:
|
||||
payload = _payload(
|
||||
parcel_geojson={"type": "Point", "coordinates": [60.60, 56.83]}
|
||||
)
|
||||
with pytest.raises(ParcelGeometryError):
|
||||
geometry.parse_parcel(payload)
|
||||
|
||||
|
||||
def test_build_placement_grid_anchors_are_step_aligned() -> None:
|
||||
# Простой метрический квадрат 30x30 м, шаг 10 -> 3x3 = 9 ячеек.
|
||||
square = Polygon([(0, 0), (30, 0), (30, 30), (0, 30)])
|
||||
cells = build_placement_grid(square, 10.0)
|
||||
assert len(cells) == 9
|
||||
# Центры на полушаге от кратных шагу.
|
||||
for cell in cells:
|
||||
assert math.isclose((cell.cx - 5.0) % 10.0, 0.0, abs_tol=1e-6)
|
||||
assert math.isclose((cell.cy - 5.0) % 10.0, 0.0, abs_tol=1e-6)
|
||||
116
backend/tests/services/generative/test_placement.py
Normal file
116
backend/tests/services/generative/test_placement.py
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
"""Stage 1b tests — greedy placement, STRtree collisions, gaps, strategies.
|
||||
|
||||
Asserts the structural guarantees of the greedy sweep: footprints stay inside the
|
||||
buildable area, respect the inter-section gap (no overlaps), the three strategies
|
||||
differ, the coverage cap bounds density, and the result is deterministic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from shapely.geometry import shape
|
||||
|
||||
from app.schemas.concept import ConceptInput
|
||||
from app.services.generative import geometry, placement
|
||||
|
||||
_PARCEL_COORDS = [
|
||||
[60.60, 56.830],
|
||||
[60.6045, 56.830],
|
||||
[60.6045, 56.8328],
|
||||
[60.60, 56.8328],
|
||||
[60.60, 56.830],
|
||||
]
|
||||
|
||||
|
||||
def _payload(**overrides: object) -> ConceptInput:
|
||||
base: dict[str, object] = {
|
||||
"parcel_geojson": {"type": "Polygon", "coordinates": [_PARCEL_COORDS]},
|
||||
"housing_class": "comfort",
|
||||
"target_floors": 9,
|
||||
"development_type": "mid_rise",
|
||||
}
|
||||
base.update(overrides)
|
||||
return ConceptInput(**base) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_all_three_strategies_present() -> None:
|
||||
parcel = geometry.parse_parcel(_payload())
|
||||
variants = placement.place_all_strategies(parcel, _payload())
|
||||
assert {v.strategy for v in variants} == {"max_area", "max_insolation", "balanced"}
|
||||
|
||||
|
||||
def test_footprints_inside_buildable_and_non_overlapping() -> None:
|
||||
parcel = geometry.parse_parcel(_payload())
|
||||
spec = next(s for s in placement._STRATEGIES if s.name == "max_area")
|
||||
footprints = placement._greedy_place(parcel, spec, coverage_cap=0.45)
|
||||
assert len(footprints) > 0
|
||||
for fp in footprints:
|
||||
# Внутри пятна застройки (с допуском на численную погрешность буфера).
|
||||
assert parcel.buildable_m.buffer(0.01).covers(fp)
|
||||
# Никакие две секции не перекрываются (разрыв gap_m выдержан).
|
||||
for i, a in enumerate(footprints):
|
||||
for b in footprints[i + 1 :]:
|
||||
assert not a.buffer(-0.01).intersects(b.buffer(-0.01))
|
||||
|
||||
|
||||
def test_gap_between_sections_respected() -> None:
|
||||
parcel = geometry.parse_parcel(_payload())
|
||||
spec = next(s for s in placement._STRATEGIES if s.name == "max_insolation")
|
||||
footprints = placement._greedy_place(parcel, spec, coverage_cap=0.45)
|
||||
# Минимальное расстояние между любыми двумя секциями >= gap_m (с допуском).
|
||||
for i, a in enumerate(footprints):
|
||||
for b in footprints[i + 1 :]:
|
||||
assert a.distance(b) >= spec.gap_m - 0.5
|
||||
|
||||
|
||||
def test_max_area_denser_than_max_insolation() -> None:
|
||||
payload = _payload()
|
||||
parcel = geometry.parse_parcel(payload)
|
||||
variants = {v.strategy: v for v in placement.place_all_strategies(parcel, payload)}
|
||||
# Максимум площади даёт большее пятно/жилую, чем максимум инсоляции.
|
||||
assert (
|
||||
variants["max_area"].teap.built_area_sqm
|
||||
> variants["max_insolation"].teap.built_area_sqm
|
||||
)
|
||||
assert (
|
||||
variants["max_area"].teap.residential_area_sqm
|
||||
> variants["max_insolation"].teap.residential_area_sqm
|
||||
)
|
||||
|
||||
|
||||
def test_coverage_cap_bounds_built_area() -> None:
|
||||
# high_rise cap = 0.50; пятно не должно его превышать (+небольшой запас на 1 секцию).
|
||||
payload = _payload(development_type="high_rise")
|
||||
parcel = geometry.parse_parcel(payload)
|
||||
variants = placement.place_all_strategies(parcel, payload)
|
||||
cap = placement._COVERAGE_CAP_BY_TYPE["high_rise"]
|
||||
for v in variants:
|
||||
coverage = v.teap.built_area_sqm / parcel.buildable_area_sqm
|
||||
# +0.05: цикл останавливается ПОСЛЕ секции, перешагнувшей порог.
|
||||
assert coverage <= cap + 0.05
|
||||
|
||||
|
||||
def test_buildings_geojson_features_are_valid_polygons() -> None:
|
||||
payload = _payload()
|
||||
parcel = geometry.parse_parcel(payload)
|
||||
variant = placement.place_all_strategies(parcel, payload)[0]
|
||||
fc = variant.buildings_geojson
|
||||
assert fc["type"] == "FeatureCollection"
|
||||
features = fc["features"]
|
||||
assert isinstance(features, list) and len(features) > 0
|
||||
for feat in features:
|
||||
geom = shape(feat["geometry"])
|
||||
assert geom.geom_type == "Polygon"
|
||||
assert geom.is_valid
|
||||
assert feat["properties"]["floors"] >= 1
|
||||
|
||||
|
||||
def test_placement_deterministic() -> None:
|
||||
payload = _payload()
|
||||
p1 = geometry.parse_parcel(payload)
|
||||
p2 = geometry.parse_parcel(payload)
|
||||
v1 = placement.place_all_strategies(p1, payload)
|
||||
v2 = placement.place_all_strategies(p2, payload)
|
||||
for a, b in zip(v1, v2, strict=True):
|
||||
assert a.teap.apartments_count == b.teap.apartments_count
|
||||
assert a.teap.built_area_sqm == b.teap.built_area_sqm
|
||||
assert len(a.buildings_geojson["features"]) == len(b.buildings_geojson["features"])
|
||||
114
backend/tests/services/generative/test_teap_financial.py
Normal file
114
backend/tests/services/generative/test_teap_financial.py
Normal file
|
|
@ -0,0 +1,114 @@
|
|||
"""Stage 1c tests — ТЭП and financial model arithmetic.
|
||||
|
||||
Pure-arithmetic unit tests against known footprint geometry: GFA, residential area,
|
||||
apartment count, FAR, parking, revenue/cost/margin and the IRR-proxy clamp.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from shapely.geometry import box
|
||||
|
||||
from app.schemas.concept import TEAP
|
||||
from app.services.generative import financial, teap
|
||||
|
||||
# Один прямоугольник 20 x 10 = 200 кв.м пятна.
|
||||
_FOOTPRINT = box(0, 0, 20, 10)
|
||||
|
||||
|
||||
def test_teap_basic_arithmetic() -> None:
|
||||
result = teap.compute_teap(
|
||||
footprints=[_FOOTPRINT],
|
||||
floors=10,
|
||||
site_area_sqm=1000.0,
|
||||
housing_class="comfort",
|
||||
)
|
||||
assert result.built_area_sqm == 200.0
|
||||
# GFA = пятно * этажность.
|
||||
assert result.total_floor_area_sqm == 2000.0
|
||||
# FAR = GFA / участок.
|
||||
assert result.density == 2.0
|
||||
# Жилая = GFA * efficiency(comfort=0.78).
|
||||
assert result.residential_area_sqm == 1560.0
|
||||
# Квартир = floor(жилая / avg(comfort=55)).
|
||||
assert result.apartments_count == int(1560.0 // 55.0)
|
||||
# Парковка = ceil(квартир * 1.0).
|
||||
assert result.parking_spaces == result.apartments_count
|
||||
|
||||
|
||||
def test_teap_class_changes_efficiency_and_lot() -> None:
|
||||
econ = teap.compute_teap(
|
||||
footprints=[_FOOTPRINT], floors=10, site_area_sqm=1000.0, housing_class="econom"
|
||||
)
|
||||
biz = teap.compute_teap(
|
||||
footprints=[_FOOTPRINT], floors=10, site_area_sqm=1000.0, housing_class="business"
|
||||
)
|
||||
# Эконом эффективнее по площади -> больше жилой при той же GFA.
|
||||
assert econ.residential_area_sqm > biz.residential_area_sqm
|
||||
# Бизнес — крупнее лот -> меньше квартир.
|
||||
assert biz.apartments_count < econ.apartments_count
|
||||
# Бизнес — выше норма парковки на квартиру.
|
||||
assert biz.parking_spaces / max(1, biz.apartments_count) >= 1.4
|
||||
|
||||
|
||||
def test_teap_zero_site_area_no_division_error() -> None:
|
||||
result = teap.compute_teap(
|
||||
footprints=[_FOOTPRINT], floors=5, site_area_sqm=0.0, housing_class="comfort"
|
||||
)
|
||||
assert result.density == 0.0
|
||||
|
||||
|
||||
def test_teap_empty_placement_is_zeroed() -> None:
|
||||
result = teap.compute_teap(
|
||||
footprints=[], floors=9, site_area_sqm=1000.0, housing_class="comfort"
|
||||
)
|
||||
assert result.built_area_sqm == 0.0
|
||||
assert result.total_floor_area_sqm == 0.0
|
||||
assert result.apartments_count == 0
|
||||
assert result.parking_spaces == 0
|
||||
|
||||
|
||||
def _teap(residential: float, gfa: float) -> TEAP:
|
||||
return TEAP(
|
||||
built_area_sqm=100.0,
|
||||
total_floor_area_sqm=gfa,
|
||||
residential_area_sqm=residential,
|
||||
apartments_count=10,
|
||||
density=1.0,
|
||||
parking_spaces=10,
|
||||
)
|
||||
|
||||
|
||||
def test_financial_revenue_cost_margin() -> None:
|
||||
t = _teap(residential=1000.0, gfa=1300.0)
|
||||
model = financial.compute_financial(
|
||||
teap=t, housing_class="comfort", land_cost_rub=50_000_000.0
|
||||
)
|
||||
# revenue = 1000 * 145_000.
|
||||
assert model.revenue_rub == 1000.0 * 145_000.0
|
||||
# cost = 1300 * 88_000 + land.
|
||||
assert model.cost_rub == 1300.0 * 88_000.0 + 50_000_000.0
|
||||
assert model.gross_margin_rub == model.revenue_rub - model.cost_rub
|
||||
|
||||
|
||||
def test_financial_land_cost_optional() -> None:
|
||||
t = _teap(residential=1000.0, gfa=1300.0)
|
||||
no_land = financial.compute_financial(teap=t, housing_class="comfort", land_cost_rub=None)
|
||||
with_land = financial.compute_financial(
|
||||
teap=t, housing_class="comfort", land_cost_rub=10_000_000.0
|
||||
)
|
||||
# Земля увеличивает затраты ровно на свою стоимость.
|
||||
assert with_land.cost_rub - no_land.cost_rub == 10_000_000.0
|
||||
|
||||
|
||||
def test_financial_irr_proxy_clamped() -> None:
|
||||
# Огромная маржа -> irr-proxy зажат в [-1, 1].
|
||||
t = _teap(residential=100_000.0, gfa=1.0)
|
||||
model = financial.compute_financial(teap=t, housing_class="business", land_cost_rub=None)
|
||||
assert -1.0 <= model.irr <= 1.0
|
||||
|
||||
|
||||
def test_financial_zero_cost_no_division_error() -> None:
|
||||
t = _teap(residential=0.0, gfa=0.0)
|
||||
model = financial.compute_financial(teap=t, housing_class="comfort", land_cost_rub=None)
|
||||
assert model.irr == 0.0
|
||||
assert model.cost_rub == 0.0
|
||||
|
|
@ -1,4 +1,9 @@
|
|||
"""Smoke test for the Concept stub. Real algorithm tests come in Stage 1b."""
|
||||
"""Smoke test for the Concept endpoint shape (3 strategies present).
|
||||
|
||||
The stub is now replaced by the real Stage 1a/1b/1c pipeline; richer assertions on
|
||||
filled ТЭП/financial live in tests/services/generative/test_api_concepts.py. This
|
||||
file is kept as a minimal endpoint-shape smoke.
|
||||
"""
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue