Compare commits

..

No commits in common. "3b84609670b79e44a44e55b2b6c2b1e820a4d69a" and "2ebf7468cd30e4275983b7fef839b06910bf32a5" have entirely different histories.

8 changed files with 45 additions and 955 deletions

View file

@ -15,7 +15,6 @@ from app.schemas.concept import (
MassingRecomputeOutput,
)
from app.services.generative import geometry
from app.services.generative.catalog import available_section_types
from app.services.generative.financial import compute_financial
from app.services.generative.geometry import ParcelGeometryError, _parse_polygon
from app.services.generative.teap import synthesize_teap_from_program
@ -192,26 +191,7 @@ async def create_concept(
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.
Stage 3a (#1965): если задана ``building_program`` (типовые дома из каталога), кладём
РОВНО эту программу вместо жадной max-FAR раскладки (один вариант). Неизвестный
``section_type`` 422 (валидируется тут, до размещения). Если участок не вмещает всю
программу НЕ 422: кладём сколько влезло и отдаём честный ``placed_count`` < requested.
"""
# Stage 3a: валидируем ключи программы по каталогу ДО размещения — неизвестный
# section_type это bad request (422), а не 500 из KeyError в глубине placement.
if payload.building_program:
known = available_section_types()
unknown = sorted({item.section_type for item in payload.building_program} - known)
if unknown:
raise HTTPException(
status_code=422,
detail=(
f"unknown house type(s): {', '.join(unknown)}; "
f"available: {', '.join(sorted(known))}"
),
)
# Рыночную цену продажи жилья считаем ОДИН раз на участок (она едина для всех
# стратегий). DB-lookup — синхронный SQLAlchemy → run_in_threadpool, чтобы не
# блокировать event loop. Падение lookup'а не должно ронять генерацию.
@ -219,7 +199,9 @@ async def create_concept(
price_source: str = "class_norm"
try:
wkt_point = await run_in_threadpool(_parcel_centroid_wkt, payload)
market_price, price_source = await run_in_threadpool(_lookup_market_price, db, wkt_point)
market_price, price_source = await run_in_threadpool(
_lookup_market_price, db, wkt_point
)
except ParcelGeometryError:
# Невалидная геометрия — пусть geometry.generate поднимет её ниже (один 422).
pass

View file

@ -1,27 +1,8 @@
from typing import Annotated, Any, Literal
from typing import Any, Literal
from pydantic import BaseModel, Field
class BuildingProgramItem(BaseModel):
"""Stage 3a (#1965, эпик #1953) — один пункт программы застройки (типовой дом × N).
Пользователь набирает программу из ТИПОВЫХ домов каталога
(``app.services.generative.catalog.HOUSE_TYPES``) вместо max-FAR жадной раскладки:
«поставь ``count`` секций типа ``section_type`` этажностью ``floors``». Габариты
пятна секции берутся из каталога по ключу ``section_type`` (контракт несёт только
ключ, не геометрию единый справочник на бэке).
"""
# Ключ типа дома из каталога (``HouseType.section_type``). Валидируется на слое
# размещения по ``catalog.available_section_types`` — неизвестный ключ → 422.
section_type: str = Field(..., description="Catalog house-type key (HOUSE_TYPES)")
# Этажность этой группы секций; диапазон шире, чем у target_floors (до 40 — башни).
floors: int = Field(..., ge=1, le=40, description="Floors for this house-type group")
# Сколько секций этого типа разместить.
count: int = Field(..., ge=1, le=50, description="Number of sections to place")
class ConceptInput(BaseModel):
"""Stage 1a — input contract. Frozen interface for frontend codegen."""
@ -34,15 +15,6 @@ class ConceptInput(BaseModel):
land_cost_rub: float | None = Field(
None, ge=0, description="Optional land cost for financial model"
)
# Stage 3a (#1965): ОПЦИОНАЛЬНАЯ программа застройки из типовых домов каталога.
# ADDITIVE поле (default None). None → существующая жадная раскладка (Stage 1b,
# byte-for-byte backward-compat). Задано → раскладка кладёт РОВНО эту программу
# (place_program): для каждого пункта ставит count секций каталожного пятна с его
# floors, вместо coverage-cap sweep. См. app.services.generative.placement.
building_program: list[BuildingProgramItem] | None = Field(
None,
description="Optional typed house program; None → greedy max-FAR placement",
)
class TEAP(BaseModel):
@ -175,19 +147,6 @@ class ConceptVariant(BaseModel):
buildings_geojson: dict[str, Any]
teap: TEAP
financial: FinancialModel
# Stage 3a (#1965) — честный сигнал частичного размещения для program-режима.
# Когда задан ``ConceptInput.building_program``: ``requested_count`` — сколько секций
# просили (Σ count по программе), ``placed_count`` — сколько реально влезло в участок.
# placed < requested → участок мал, разместилось N из M (фронт Stage 3b показывает
# «разместилось N из M», без hard-422). Оба None в greedy-режиме (программа не задана)
# → ADDITIVE, backward-compat: старый ответ концепции не меняется. Annotated+`= None`:
# рантайм-дефолт виден mypy (без pydantic-плагина), Field несёт только OpenAPI-описание.
placed_count: Annotated[
int | None, Field(description="Stage 3a: sections actually placed (program mode only)")
] = None
requested_count: Annotated[
int | None, Field(description="Stage 3a: sections requested by program (program mode only)")
] = None
class ConceptOutput(BaseModel):

View file

@ -1,141 +0,0 @@
"""Generative Design — Stage 3a (#1965): каталог типовых домов (house-type catalog).
Контракт Stage 3a: вместо max-FAR жадной раскладки (Stage 1b) пользователь выбирает
ТИПОВЫЕ дома и говорит, сколько секций каждого типа поставить. Этот модуль справочник
таких типов: для каждого ``section_type`` он несёт габариты пятна секции (ширина × глубина,
метры), дефолтную этажность и подходящий класс жилья.
ИСТОЧНИК: это РАЗУМНЫЙ ДЕФОЛТНЫЙ каталог (sane-default), а не выгрузка из БД. Габариты
типовые размеры секций массового жилья РФ (панель/монолит/башня/малоэтажка), округлённые
до реалистичных значений. Каталог намеренно захардкожен в коде на Stage 3a: миграции БД
сейчас нет (см. эпик #1953). Когда понадобится редактируемый застройщиком каталог — его
ПРОДВИГАЮТ в БД-таблицу с тем же контрактом (``section_type`` footprint/floors/class),
а этот модуль станет seed'ом/фолбэком. До тех пор — single source of truth по типам.
Детерминированно, без LLM / внешних API / БД.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Literal
logger = logging.getLogger(__name__)
# Класс жилья — то же Literal-множество, что у ConceptInput / compute_teap (single
# source of truth по допустимым значениям; рассинхрон тут = ошибка типов в mypy-strict).
HousingClass = Literal["econom", "comfort", "business"]
@dataclass(frozen=True)
class HouseType:
"""Один типовой дом каталога (тип секции МКД / малоэтажки).
section_type стабильный машинный КЛЮЧ типа (латиница, snake_case); это и есть
значение, которое фронт кладёт в ``BuildingProgramItem.section_type``.
label_ru человекочитаемый русский лейбл для UI (Stage 3b его показывает).
footprint_w_m ширина пятна секции, метры.
footprint_d_m глубина пятна секции, метры.
default_floors дефолтная этажность типа (UI подставляет, пользователь может менять
в пределах контракта BuildingProgramItem [1, 40]).
housing_class подходящий класс жилья (драйвит ТЭП/финмодель ниже по конвейеру).
"""
section_type: str
label_ru: str
footprint_w_m: float
footprint_d_m: float
default_floors: int
housing_class: HousingClass
@property
def footprint_sqm(self) -> float:
"""Площадь пятна секции, кв.м (ширина × глубина)."""
return self.footprint_w_m * self.footprint_d_m
# ── Каталог типовых домов (sane-default, см. модульный docstring про источник) ──────
# Покрываем основные форматы массового жилья РФ:
# * панель-эконом — длинная неглубокая секция, средняя этажность, эконом-класс;
# * монолит-комфорт— чуть шире/глубже, комфорт-класс, типовая «свечка» 14 этажей;
# * башня-бизнес — компактное квадратное пятно, высотная (точечная) застройка;
# * малоэтажка-комфорт — широкая невысокая секция (3 этажа), низкоплотная застройка;
# * таунхаус — узкое неглубокое пятно блокированной застройки, 3 этажа.
# Габариты — реалистичные типовые размеры; этажность — характерная для формата.
HOUSE_TYPES: tuple[HouseType, ...] = (
HouseType(
section_type="panel_econom",
label_ru="Панельная секция (эконом)",
footprint_w_m=24.0,
footprint_d_m=15.0,
default_floors=9,
housing_class="econom",
),
HouseType(
section_type="monolith_comfort",
label_ru="Монолитная секция (комфорт)",
footprint_w_m=21.0,
footprint_d_m=18.0,
default_floors=14,
housing_class="comfort",
),
HouseType(
section_type="tower_business",
label_ru="Башня (бизнес)",
footprint_w_m=18.0,
footprint_d_m=18.0,
default_floors=25,
housing_class="business",
),
HouseType(
section_type="lowrise_comfort",
label_ru="Малоэтажная секция (комфорт)",
footprint_w_m=30.0,
footprint_d_m=14.0,
default_floors=3,
housing_class="comfort",
),
HouseType(
section_type="townhouse",
label_ru="Таунхаус",
footprint_w_m=12.0,
footprint_d_m=10.0,
default_floors=3,
housing_class="comfort",
),
)
# Индекс по ключу для O(1)-лукапа. Построен один раз при импорте; ключи уникальны
# (assert ниже ловит дубликат типа на старте, а не молча затирает запись).
_BY_KEY: dict[str, HouseType] = {ht.section_type: ht for ht in HOUSE_TYPES}
assert len(_BY_KEY) == len(HOUSE_TYPES), "duplicate section_type key in HOUSE_TYPES"
def get_house_type(section_type: str) -> HouseType:
"""Найти тип дома по ключу ``section_type``. Бросает :class:`KeyError`, если нет.
Вызывающий слой (placement) обязан валидировать ключи программы заранее (см.
:func:`available_section_types`) неизвестный ключ здесь это программная ошибка,
а не пользовательский ввод, поэтому KeyError, а не тихий None.
"""
try:
return _BY_KEY[section_type]
except KeyError as exc:
raise KeyError(
f"unknown house type {section_type!r}; " f"available: {', '.join(sorted(_BY_KEY))}"
) from exc
def available_section_types() -> frozenset[str]:
"""Множество допустимых ключей ``section_type`` каталога (для валидации программы)."""
return frozenset(_BY_KEY)
__all__ = [
"HOUSE_TYPES",
"HouseType",
"HousingClass",
"available_section_types",
"get_house_type",
]

View file

@ -23,19 +23,13 @@ Deterministic, no LLM / no external API / no DB.
from __future__ import annotations
import logging
import math
from dataclasses import dataclass
from shapely.geometry import Polygon, box
from shapely.strtree import STRtree
from app.schemas.concept import (
TEAP,
BuildingProgramItem,
ConceptInput,
ConceptVariant,
)
from app.services.generative import catalog, financial, teap
from app.schemas.concept import ConceptInput, ConceptVariant
from app.services.generative import financial, teap
from app.services.generative.geometry import Parcel, ParcelGeometryError
logger = logging.getLogger(__name__)
@ -98,53 +92,6 @@ def _resolve_floors(target_floors: int, factor: float) -> int:
return max(_FLOORS_MIN, min(_FLOORS_MAX, floors))
class _Placer:
"""Аккумулятор размещённых секций + STRtree-индекс для проверки разрывов.
Извлечён из жадной раскладки, чтобы и :func:`_greedy_place` (coverage-cap sweep), и
:func:`place_program` (фиксированная программа типовых домов, Stage 3a) пользовались
ОДНОЙ И ТОЙ ЖЕ collision/setback-машиной, а не дублировали её. Состояние: принятые
footprints, их буферизованные на ``half_gap`` версии и перестраиваемый по ним STRtree.
"""
def __init__(self) -> None:
self.placed: list[Polygon] = []
self.built_area: float = 0.0
# Буферизованные footprints для проверки разрыва; индекс STRtree по ним.
self._buffered: list[Polygon] = []
self._tree: STRtree | None = None
def try_place(self, footprint: Polygon, buildable: Polygon, half_gap: float) -> bool:
"""Попытаться принять ``footprint``. True — принят, False — не лёг.
Принимается, если целиком внутри ``buildable`` (covers допускает касание границы)
И не нарушает разрыв ``half_gap*2`` с уже принятыми (буферим кандидата на
``half_gap`` и проверяем пересечение с буферизованными соседями через STRtree
две секции с зазором >= gap не пересекутся).
"""
if not buildable.covers(footprint):
return False
candidate_buf = footprint.buffer(half_gap, join_style="mitre")
if self._tree is not None:
for idx in self._tree.query(candidate_buf):
if candidate_buf.intersects(self._buffered[idx]):
return False
self.placed.append(footprint)
self.built_area += footprint.area
self._buffered.append(candidate_buf)
self._tree = STRtree(self._buffered)
return True
def _centered_footprint(cx: float, cy: float, width: float, depth: float) -> Polygon:
"""Прямоугольное пятно секции ``width × depth``, центрированное на (cx, cy), метры."""
half_w = width / 2.0
half_d = depth / 2.0
return box(cx - half_w, cy - half_d, cx + half_w, cy + half_d)
def _greedy_place(
parcel: Parcel,
spec: StrategySpec,
@ -156,32 +103,63 @@ def _greedy_place(
* кандидат-якоря центры ячеек сетки в фиксированном порядке;
* footprint строится центрированно на якоре;
* принимается, если целиком внутри buildable area И не нарушает разрыв ``gap_m``
с уже принятыми (проверка через STRtree, см. :class:`_Placer`);
с уже принятыми (проверка через STRtree по buffered-footprints);
* раскладка останавливается, когда пятно достигает ``coverage_cap`` от buildable
area (регулятор плотности по типу застройки) это также ограничивает число
размещений и держит O(n^2)-перестройку STRtree в бюджете.
"""
buildable = parcel.buildable_m
max_built = buildable.area * coverage_cap
placer = _Placer()
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 placer.built_area >= max_built:
if built_area >= max_built:
break
footprint = _centered_footprint(cell.cx, cell.cy, spec.section_w, spec.section_d)
placer.try_place(footprint, buildable, half_gap)
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(placer.placed),
len(placed),
spec.section_w,
spec.section_d,
spec.gap_m,
coverage_cap * 100,
)
return placer.placed
return placed
def _footprints_to_geojson(
@ -209,263 +187,6 @@ def _footprints_to_geojson(
return {"type": "FeatureCollection", "features": features}
# ── Stage 3a (#1965): program-driven placement (типовые дома вместо max-FAR sweep) ──
# Разрыв между секциями в program-режиме (м). Пункт программы не несёт gap (контракт —
# только тип/этажность/количество), поэтому берём один нормативный противопожарный/
# инсоляционный зазор для всех секций программы — середина диапазона стратегий 1b
# (max_area gap=6 … max_insolation gap=15). Достаточно консервативно для реалистичной
# раскладки, не патологически разрежено.
_PROGRAM_GAP_M: float = 10.0
@dataclass(frozen=True)
class _PlacedSection:
"""Одна размещённая секция программы: пятно + этажность + тип каталога (для GeoJSON)."""
footprint: Polygon
floors: int
section_type: str
@dataclass(frozen=True)
class PlacedProgram:
"""Результат program-раскладки: что легло + честный счётчик «N из M».
``sections`` размещённые секции (пятна + этажность + тип) в порядке программы.
``requested_count`` сколько секций просили (Σ count по программе).
``placed_count`` сколько реально влезло (== len(sections)). placed < requested
участок мал, разместилось N из M (без hard-422 честный сигнал для Stage 3b).
"""
sections: tuple[_PlacedSection, ...]
requested_count: int
placed_count: int
def place_program(
parcel: Parcel,
program: list[BuildingProgramItem],
*,
gap_m: float = _PROGRAM_GAP_M,
) -> PlacedProgram:
"""Stage 3a: разложить РОВНО заданную программу типовых домов на участок.
Для каждого пункта программы (``section_type`` из каталога × ``count`` секций) кладём
до ``count`` секций каталожного пятна на сетку участка, переиспользуя ТУ ЖЕ
collision/STRtree/setback-машину, что и жадная раскладка (:class:`_Placer`) никакого
coverage-cap, стоп-критерий = достигнут ``count`` для пункта или кончились свободные
якоря. Пункты обрабатываются по порядку; накопленные секции участвуют в проверке
разрыва для последующих (общий :class:`_Placer`).
Если участок не вмещает все запрошенные секции, НЕ роняем 422 кладём сколько влезло
и возвращаем честный ``placed_count``/``requested_count`` (разместилось N из M).
Raises:
KeyError: ``section_type`` пункта нет в каталоге (валидируется на API-слое до
размещения; здесь это программная ошибка контракта, не пользовательский ввод).
"""
buildable = parcel.buildable_m
half_gap = gap_m / 2.0
placer = _Placer()
placed_sections: list[_PlacedSection] = []
requested = 0
for item in program:
house = catalog.get_house_type(item.section_type)
requested += item.count
placed_for_item = 0
for cell in parcel.grid:
if placed_for_item >= item.count:
break
footprint = _centered_footprint(
cell.cx, cell.cy, house.footprint_w_m, house.footprint_d_m
)
if placer.try_place(footprint, buildable, half_gap):
placed_sections.append(
_PlacedSection(
footprint=footprint,
floors=item.floors,
section_type=item.section_type,
)
)
placed_for_item += 1
if placed_for_item < item.count:
logger.warning(
"program: type=%s placed %d of %d sections (%.0fx%.0f m) — участок мал",
item.section_type,
placed_for_item,
item.count,
house.footprint_w_m,
house.footprint_d_m,
)
result = PlacedProgram(
sections=tuple(placed_sections),
requested_count=requested,
placed_count=len(placed_sections),
)
logger.info(
"program placed %d of %d sections across %d type(s)",
result.placed_count,
result.requested_count,
len(program),
)
return result
def _placed_program_to_geojson(
parcel: Parcel,
sections: tuple[_PlacedSection, ...],
) -> dict[str, object]:
"""Размещённые секции программы -> WGS84 FeatureCollection (контракт buildings_geojson).
Зеркалит :func:`_footprints_to_geojson`, но каждая секция несёт СВОИ floors и тип
каталога (program-режим смешивает типы/этажности), а ``strategy`` помечается
``"program"`` маркер, что вариант построен из программы, а не из 1b-стратегии.
"""
features: list[dict[str, object]] = []
for i, sec in enumerate(sections):
geom_wgs = parcel.metric_geom_to_wgs84(sec.footprint)
features.append(
{
"type": "Feature",
"geometry": geom_wgs,
"properties": {
"section_id": i + 1,
"floors": sec.floors,
"footprint_sqm": round(float(sec.footprint.area), 1),
"section_type": sec.section_type,
"strategy": "program",
},
}
)
return {"type": "FeatureCollection", "features": features}
def _aggregate_program_teap(
sections: tuple[_PlacedSection, ...],
*,
site_area_sqm: float,
housing_class: teap.HousingClass,
) -> TEAP:
"""Свести размещённую программу (СМЕШАННАЯ этажность) в один :class:`TEAP` — ТОЧНО.
``compute_teap`` берёт ОДНУ этажность на список пятен, поэтому при смешанной по типам
этажности нельзя просто скормить ему все пятна с одним числом (округлённая «средняя»
этажность даёт дрейф GFA ~1%). Вместо этого группируем секции по этажности, считаем
``compute_teap`` для каждой однородной группы и СУММИРУЕМ результаты:
* built / GFA / office / residential аддитивны сумма точна (GFA = Σ площадь_i×floors_i);
* apartments ``Σ floor(жилая_g / avg)`` по группам: физически корректнее, чем
``floor(Σжилая / avg)`` (нельзя «склеивать» дробные квартиры между корпусами);
* parking пересчитываем от ИТОГОВОГО числа квартир по той же норме класса (ceil от
суммы, а не сумма ceil иначе пер-группное округление вверх задвоит места);
* density (FAR) от суммарной GFA и площади участка (защита от деления на ноль).
Единый источник всех нормативных коэффициентов остаётся ``teap``-модуль (классовые
словари), новых магических чисел нет.
"""
if not sections:
return teap.compute_teap(
footprints=[], floors=0, site_area_sqm=site_area_sqm, housing_class=housing_class
)
# Группируем по этажности; внутри группы compute_teap корректен (одна этажность).
groups: dict[int, list[Polygon]] = {}
for sec in sections:
groups.setdefault(sec.floors, []).append(sec.footprint)
built = 0.0
gfa = 0.0
office = 0.0
residential = 0.0
apartments = 0
for floors, fps in groups.items():
# site_area_sqm здесь не важна для аддитивных полей — FAR пересчитаем в конце.
group = teap.compute_teap(
footprints=fps,
floors=floors,
site_area_sqm=site_area_sqm,
housing_class=housing_class,
)
built += group.built_area_sqm
gfa += group.total_floor_area_sqm
office += group.office_area_sqm
residential += group.residential_area_sqm
apartments += group.apartments_count
density = gfa / site_area_sqm if site_area_sqm > 0 else 0.0
parking_norm = teap._PARKING_PER_APARTMENT[housing_class]
parking_spaces = math.ceil(apartments * parking_norm)
return TEAP(
built_area_sqm=round(built, 1),
total_floor_area_sqm=round(gfa, 1),
office_area_sqm=round(office, 1),
residential_area_sqm=round(residential, 1),
apartments_count=apartments,
density=round(density, 3),
parking_spaces=parking_spaces,
)
def place_program_variant(
parcel: Parcel,
payload: ConceptInput,
*,
market_price_per_sqm: float | None = None,
price_source: str = "class_norm",
) -> ConceptVariant | None:
"""Stage 3a: построить ОДИН вариант из ``payload.building_program`` (типовые дома).
Раскладывает программу (:func:`place_program`), сводит размещённые пятна в ТЭП
(:func:`_aggregate_program_teap` точная GFA по группам этажности) и финмодель, и
наклеивает честный сигнал частичного размещения (``placed_count``/``requested_count``).
Возвращает ``None``, если ни одна секция не легла (участок не вмещает даже одну секцию
программы) вызывающий отбракует, как и в жадном пути.
``payload.building_program`` ДОЛЖЕН быть задан (вызывается только из program-ветки).
"""
program = payload.building_program
if not program: # защитный инвариант: эту ветку зовут только при заданной программе
raise ValueError("place_program_variant called without building_program")
placed = place_program(parcel, program)
if placed.placed_count == 0:
logger.warning(
"program placed 0 of %d sections — участок не вмещает программу, отбраковка",
placed.requested_count,
)
return None
teap_result = _aggregate_program_teap(
placed.sections,
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,
market_price_per_sqm=market_price_per_sqm,
price_source=price_source,
development_type=payload.development_type,
)
buildings_geojson = _placed_program_to_geojson(parcel, placed.sections)
# Program-вариант репортуется под "balanced" (контракт strategy — фиксированный
# Literal трёх стратегий 1b; program-режим не вводит новую стратегию, маркер режима
# лежит в properties.strategy="program" каждой фичи GeoJSON). Один вариант на программу.
return ConceptVariant(
strategy="balanced",
buildings_geojson=buildings_geojson,
teap=teap_result,
financial=financial_result,
placed_count=placed.placed_count,
requested_count=placed.requested_count,
)
def place_strategy(
parcel: Parcel,
payload: ConceptInput,
@ -539,31 +260,7 @@ def place_all_strategies(
``market_price_per_sqm`` / ``price_source`` (рыночная калибровка цены жилья, PR-2)
прокидываются неизменными в каждую стратегию цена едина для участка.
Stage 3a (#1965): если задана ``payload.building_program`` — раскладываем РОВНО эту
программу типовых домов (:func:`place_program_variant`, один вариант), а НЕ три жадные
стратегии. ``building_program is None`` существующий жадный путь без изменений.
"""
if payload.building_program:
program_variant = place_program_variant(
parcel,
payload,
market_price_per_sqm=market_price_per_sqm,
price_source=price_source,
)
if program_variant is None:
raise ParcelGeometryError(
"программа застройки не вместила ни одной секции — "
"участок слишком узкий/мелкий для выбранных типов домов"
)
logger.info(
"placed program variant: %d of %d sections, %dкв",
program_variant.placed_count,
program_variant.requested_count,
program_variant.teap.apartments_count,
)
return [program_variant]
variants = [
variant
for spec in _STRATEGIES
@ -591,10 +288,7 @@ def place_all_strategies(
__all__ = [
"FLOOR_HEIGHT_M",
"PlacedProgram",
"StrategySpec",
"place_all_strategies",
"place_program",
"place_program_variant",
"place_strategy",
]

View file

@ -127,18 +127,7 @@ def test_concepts_response_matches_contract_keys() -> None:
)
assert response.status_code == 200
variant = response.json()["variants"][0]
assert set(variant.keys()) == {
"strategy",
"buildings_geojson",
"teap",
"financial",
# Stage 3a (#1965): partial-fit signal (None в greedy-режиме, число в program-режиме).
"placed_count",
"requested_count",
}
# Greedy-режим (программа не задана) → сигнал частичного размещения пуст (backward-compat).
assert variant["placed_count"] is None
assert variant["requested_count"] is None
assert set(variant.keys()) == {"strategy", "buildings_geojson", "teap", "financial"}
assert set(variant["teap"].keys()) == {
"built_area_sqm",
"total_floor_area_sqm",
@ -148,9 +137,6 @@ def test_concepts_response_matches_contract_keys() -> None:
"density",
"parking_spaces",
}
# Stage 3a partial-fit signal присутствует в контракте (None в greedy-режиме).
assert variant["placed_count"] is None
assert variant["requested_count"] is None
assert set(variant["financial"].keys()) == {
# legacy summary (backward-compat)
"revenue_rub",
@ -199,50 +185,3 @@ def test_concepts_response_matches_contract_keys() -> None:
"price_is_calibrated",
"price_source",
}
# ── Stage 3a (#1965): program-driven placement через API ────────────────────────────
def test_concepts_program_places_single_variant_with_signal() -> None:
# building_program задан → один вариант (не три), partial-fit сигнал заполнен.
response = _post(
{
"parcel_geojson": _PARCEL,
"housing_class": "comfort",
"target_floors": 9,
"development_type": "mid_rise",
"building_program": [
{"section_type": "monolith_comfort", "floors": 14, "count": 3},
{"section_type": "panel_econom", "floors": 9, "count": 2},
],
}
)
assert response.status_code == 200, response.text
variants = response.json()["variants"]
assert len(variants) == 1
v = variants[0]
assert v["requested_count"] == 5
assert v["placed_count"] == 5
assert v["teap"]["built_area_sqm"] > 0
assert v["teap"]["apartments_count"] > 0
# GeoJSON-фичи помечены program-режимом и несут тип секции.
feats = v["buildings_geojson"]["features"]
assert len(feats) == 5
assert {f["properties"]["strategy"] for f in feats} == {"program"}
def test_concepts_program_unknown_type_returns_422() -> None:
response = _post(
{
"parcel_geojson": _PARCEL,
"housing_class": "comfort",
"target_floors": 9,
"development_type": "mid_rise",
"building_program": [
{"section_type": "does_not_exist", "floors": 10, "count": 1},
],
}
)
assert response.status_code == 422
assert "unknown house type" in response.json()["detail"].lower()

View file

@ -1,56 +0,0 @@
"""Stage 3a tests (#1965) — house-type catalog integrity + lookup.
Asserts the hardcoded ``HOUSE_TYPES`` catalog is well-formed (unique keys, sane
footprints, valid housing classes) and the lookup helpers behave (hit / miss).
"""
from __future__ import annotations
import pytest
from app.services.generative import catalog
def test_catalog_non_empty_and_keys_unique() -> None:
assert len(catalog.HOUSE_TYPES) >= 4
keys = [ht.section_type for ht in catalog.HOUSE_TYPES]
assert len(keys) == len(set(keys)), "section_type keys must be unique"
def test_catalog_fields_are_sane() -> None:
valid_classes = {"econom", "comfort", "business"}
for ht in catalog.HOUSE_TYPES:
assert ht.section_type and ht.section_type.isascii()
assert ht.label_ru, "RU label must be present"
assert ht.footprint_w_m > 0 and ht.footprint_d_m > 0
assert 1 <= ht.default_floors <= 40
assert ht.housing_class in valid_classes
# footprint_sqm helper == w * d.
assert ht.footprint_sqm == ht.footprint_w_m * ht.footprint_d_m
def test_catalog_covers_expected_formats() -> None:
# Каталог должен покрывать основные форматы массового жилья РФ.
keys = {ht.section_type for ht in catalog.HOUSE_TYPES}
assert {"panel_econom", "monolith_comfort", "tower_business"} <= keys
# Есть и высотный (>=20 этажей), и малоэтажный (<=4) формат.
floors = [ht.default_floors for ht in catalog.HOUSE_TYPES]
assert max(floors) >= 20
assert min(floors) <= 4
def test_get_house_type_hit() -> None:
ht = catalog.get_house_type("monolith_comfort")
assert ht.section_type == "monolith_comfort"
assert ht.housing_class == "comfort"
def test_get_house_type_miss_raises_keyerror() -> None:
with pytest.raises(KeyError):
catalog.get_house_type("does_not_exist")
def test_available_section_types_matches_catalog() -> None:
assert catalog.available_section_types() == frozenset(
ht.section_type for ht in catalog.HOUSE_TYPES
)

View file

@ -1,240 +0,0 @@
"""Stage 3a tests (#1965) — program-driven placement (типовые дома каталога).
Covers the three core guarantees of the ``building_program`` path:
(a) ``building_program=None`` reproduces today's greedy output unchanged (backward-compat).
(b) A 2-item program places EXACTLY the requested sections (when they fit) and the TEAP
reflects them (GFA = Σ(площадь_i × floors_i), apartments tied to placed footprints).
(c) An over-packed program reports partial placement (placed < requested) WITHOUT raising.
Plus structural checks: program footprints respect the collision/setback machinery, the
per-feature GeoJSON carries the section's own floors + catalog type, and an empty-fit
program is rejected (degenerate parcel) rather than returning a lying zero variant.
"""
from __future__ import annotations
import pytest
from shapely.geometry import shape
from app.schemas.concept import BuildingProgramItem, ConceptInput
from app.services.generative import catalog, geometry, placement
from app.services.generative.geometry import ParcelGeometryError
# Крупный квартальный участок (~7.9 га buildable) — вмещает десятки секций.
_BIG_PARCEL = [
[60.60, 56.830],
[60.6045, 56.830],
[60.6045, 56.8328],
[60.60, 56.8328],
[60.60, 56.830],
]
# Маленький участок (~3.8 тыс. кв.м buildable) — вмещает лишь ~2 крупные секции.
_SMALL_PARCEL = [
[60.600, 56.8300],
[60.6010, 56.8300],
[60.6010, 56.8308],
[60.600, 56.8308],
[60.600, 56.8300],
]
def _payload(coords: list[list[float]], **overrides: object) -> ConceptInput:
base: dict[str, object] = {
"parcel_geojson": {"type": "Polygon", "coordinates": [coords]},
"housing_class": "comfort",
"target_floors": 9,
"development_type": "mid_rise",
}
base.update(overrides)
return ConceptInput(**base) # type: ignore[arg-type]
# ── (a) backward-compat: greedy output is byte-identical after the _Placer refactor ──
# GOLDEN PIN — детерминированный жадный выход на фиксированном участке _BIG_PARCEL
# (comfort / 9 эт. / mid_rise). Захвачен ПОСЛЕ извлечения _Placer и сверен ревьювером
# как идентичный до-рефактора. Любой будущий ДРЕЙФ жадной геометрии (число секций /
# пятно / GFA / квартиры) уронит этот тест — это и есть anti-regression guard, которого
# не давали инвариантные тесты в test_placement.py. Формат на стратегию:
# (features, built_area_sqm, total_floor_area_sqm, apartments_count).
_GREEDY_GOLDEN: dict[str, tuple[int, float, float, int]] = {
"max_area": (83, 35856.0, 358560.0, 4830),
"max_insolation": (70, 15120.0, 120960.0, 1629),
"balanced": (88, 27720.0, 249480.0, 3361),
}
def test_greedy_output_matches_golden_pin() -> None:
# Реальный golden-pin: жадный путь (building_program не задан) должен выдать РОВНО
# замороженные литералы. Так фиксируем, что _Placer-рефактор не сдвинул раскладку.
payload = _payload(_BIG_PARCEL)
parcel = geometry.parse_parcel(payload)
variants = {v.strategy: v for v in placement.place_all_strategies(parcel, payload)}
assert set(variants) == set(_GREEDY_GOLDEN)
for name, (features, built, gfa, apts) in _GREEDY_GOLDEN.items():
v = variants[name]
assert len(v.buildings_geojson["features"]) == features, name
assert round(v.teap.built_area_sqm, 1) == built, name
assert round(v.teap.total_floor_area_sqm, 1) == gfa, name
assert v.teap.apartments_count == apts, name
# Greedy-режим не выставляет partial-fit сигнал (остаётся None).
assert v.placed_count is None
assert v.requested_count is None
def test_explicit_none_program_equals_default_greedy() -> None:
# building_program=None (явно) и дефолт (поле опущено) идут одним greedy-путём →
# должны совпасть покомпонентно. Дополняет golden-pin: ветка None не разветвляется.
payload_default = _payload(_BIG_PARCEL)
payload_none = _payload(_BIG_PARCEL, building_program=None)
variants_default = placement.place_all_strategies(
geometry.parse_parcel(payload_default), payload_default
)
variants_none = placement.place_all_strategies(
geometry.parse_parcel(payload_none), payload_none
)
by_default = {v.strategy: v for v in variants_default}
by_none = {v.strategy: v for v in variants_none}
assert set(by_default) == set(by_none)
for name, dv in by_default.items():
nv = by_none[name]
assert dv.teap == nv.teap
assert dv.financial == nv.financial
assert len(dv.buildings_geojson["features"]) == len(nv.buildings_geojson["features"])
# ── (b) 2-item program places EXACTLY the requested sections; TEAP reflects them ─────
def test_two_item_program_places_exactly_requested() -> None:
program = [
BuildingProgramItem(section_type="monolith_comfort", floors=14, count=3),
BuildingProgramItem(section_type="tower_business", floors=20, count=2),
]
payload = _payload(_BIG_PARCEL, building_program=program)
parcel = geometry.parse_parcel(payload)
placed = placement.place_program(parcel, program)
# Все 5 секций влезли в крупный участок — разместилось ровно столько, сколько просили.
assert placed.requested_count == 5
assert placed.placed_count == 5
assert len(placed.sections) == 5
# Каждая секция несёт свой тип/этажность из программы.
monolith = [s for s in placed.sections if s.section_type == "monolith_comfort"]
tower = [s for s in placed.sections if s.section_type == "tower_business"]
assert len(monolith) == 3 and all(s.floors == 14 for s in monolith)
assert len(tower) == 2 and all(s.floors == 20 for s in tower)
def test_two_item_program_variant_teap_reflects_program() -> None:
program = [
BuildingProgramItem(section_type="monolith_comfort", floors=14, count=3),
BuildingProgramItem(section_type="tower_business", floors=20, count=2),
]
payload = _payload(_BIG_PARCEL, building_program=program)
parcel = geometry.parse_parcel(payload)
variant = placement.place_program_variant(parcel, payload)
assert variant is not None
# Partial-fit сигнал заполнен (program-режим): разместилось 5 из 5.
assert variant.placed_count == 5
assert variant.requested_count == 5
# ТЭП отражает именно размещённую программу:
# built = Σ площадей пятен каталога (3×monolith 21×18 + 2×tower 18×18).
mono = catalog.get_house_type("monolith_comfort")
tower = catalog.get_house_type("tower_business")
expected_built = 3 * mono.footprint_sqm + 2 * tower.footprint_sqm
assert variant.teap.built_area_sqm == pytest.approx(expected_built, abs=1.0)
# GFA = Σ(площадь_i × floors_i) — ТОЧНО (per-floor-group аггрегация, без дрейфа
# от округления «средней» этажности).
expected_gfa = 3 * mono.footprint_sqm * 14 + 2 * tower.footprint_sqm * 20
assert variant.teap.total_floor_area_sqm == pytest.approx(expected_gfa, abs=1.0)
assert variant.teap.apartments_count > 0
assert variant.financial.revenue_rub > 0
# Один program-вариант (не три жадные стратегии) идёт через place_all_strategies.
variants = placement.place_all_strategies(parcel, payload)
assert len(variants) == 1
assert variants[0].placed_count == 5
def test_program_geojson_features_carry_own_floors_and_type() -> None:
program = [
BuildingProgramItem(section_type="monolith_comfort", floors=14, count=2),
BuildingProgramItem(section_type="townhouse", floors=3, count=2),
]
payload = _payload(_BIG_PARCEL, building_program=program)
parcel = geometry.parse_parcel(payload)
variant = placement.place_program_variant(parcel, payload)
assert variant is not None
features = variant.buildings_geojson["features"]
assert isinstance(features, list)
floors_seen = set()
types_seen = set()
for feat in features:
geom = shape(feat["geometry"])
assert geom.geom_type == "Polygon" and geom.is_valid
props = feat["properties"]
assert props["strategy"] == "program"
assert props["section_type"] in {"monolith_comfort", "townhouse"}
floors_seen.add(props["floors"])
types_seen.add(props["section_type"])
# Смешанная этажность/типы сохранены пер-секционно в GeoJSON.
assert floors_seen == {14, 3}
assert types_seen == {"monolith_comfort", "townhouse"}
def test_program_footprints_non_overlapping_inside_buildable() -> None:
program = [BuildingProgramItem(section_type="monolith_comfort", floors=14, count=6)]
payload = _payload(_BIG_PARCEL, building_program=program)
parcel = geometry.parse_parcel(payload)
placed = placement.place_program(parcel, program)
fps = [s.footprint for s in placed.sections]
assert len(fps) > 1
for fp in fps:
assert parcel.buildable_m.buffer(0.01).covers(fp)
for i, a in enumerate(fps):
for b in fps[i + 1 :]:
assert not a.buffer(-0.01).intersects(b.buffer(-0.01))
# ── (c) over-packed program reports partial placement WITHOUT raising ───────────────
def test_overpacked_program_reports_partial_without_raising() -> None:
# Маленький участок вмещает ~2 крупные секции, просим 20 — НЕ 422, честный N<M.
program = [BuildingProgramItem(section_type="monolith_comfort", floors=14, count=20)]
payload = _payload(_SMALL_PARCEL, building_program=program)
parcel = geometry.parse_parcel(payload)
placed = placement.place_program(parcel, program)
assert placed.requested_count == 20
assert 0 < placed.placed_count < 20 # разместилось N из M
# И через полный конвейер: вариант строится, partial-fit сигнал проброшен, без raise.
variants = placement.place_all_strategies(parcel, payload)
assert len(variants) == 1
v = variants[0]
assert v.requested_count == 20
assert 0 < v.placed_count < 20
assert v.teap.apartments_count > 0
def test_program_with_zero_fit_raises_parcel_error() -> None:
# Участок настолько мал, что НИ ОДНА секция программы не влезает → ParcelGeometryError
# (API мапит в 422) — лучше отказ, чем лживый нулевой вариант.
tiny = [
[60.60, 56.830],
[60.60015, 56.830],
[60.60015, 56.83015],
[60.60, 56.83015],
[60.60, 56.830],
]
program = [BuildingProgramItem(section_type="tower_business", floors=25, count=5)]
payload = _payload(tiny, building_program=program)
with pytest.raises(ParcelGeometryError):
parcel = geometry.parse_parcel(payload)
placement.place_all_strategies(parcel, payload)

View file

@ -26,11 +26,6 @@ export interface paths {
*
* 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.
*
* Stage 3a (#1965): если задана ``building_program`` (типовые дома из каталога), кладём
* РОВНО эту программу вместо жадной max-FAR раскладки (один вариант). Неизвестный
* ``section_type`` 422 (валидируется тут, до размещения). Если участок не вмещает всю
* программу НЕ 422: кладём сколько влезло и отдаём честный ``placed_count`` < requested.
*/
post: operations["create_concept_api_v1_concepts_post"];
delete?: never;
@ -2988,33 +2983,6 @@ export interface components {
recommendation_for_tz: components["schemas"]["LayoutTzRecommendation"];
data_quality: components["schemas"]["LayoutDataQuality"];
};
/**
* BuildingProgramItem
* @description Stage 3a (#1965, эпик #1953) один пункт программы застройки (типовой дом × N).
*
* Пользователь набирает программу из ТИПОВЫХ домов каталога
* (``app.services.generative.catalog.HOUSE_TYPES``) вместо max-FAR жадной раскладки:
* «поставь ``count`` секций типа ``section_type`` этажностью ``floors``». Габариты
* пятна секции берутся из каталога по ключу ``section_type`` (контракт несёт только
* ключ, не геометрию единый справочник на бэке).
*/
BuildingProgramItem: {
/**
* Section Type
* @description Catalog house-type key (HOUSE_TYPES)
*/
section_type: string;
/**
* Floors
* @description Floors for this house-type group
*/
floors: number;
/**
* Count
* @description Number of sections to place
*/
count: number;
};
/**
* BulkGeoEnqueueRequest
* @description Параметры для параллельного backfill geo по Свердловской обл.
@ -3340,11 +3308,6 @@ export interface components {
* @description Optional land cost for financial model
*/
land_cost_rub?: number | null;
/**
* Building Program
* @description Optional typed house program; None greedy max-FAR placement
*/
building_program?: components["schemas"]["BuildingProgramItem"][] | null;
};
/** ConceptOutput */
ConceptOutput: {
@ -3364,16 +3327,6 @@ export interface components {
};
teap: components["schemas"]["TEAP"];
financial: components["schemas"]["FinancialModel"];
/**
* Placed Count
* @description Stage 3a: sections actually placed (program mode only)
*/
placed_count?: number | null;
/**
* Requested Count
* @description Stage 3a: sections requested by program (program mode only)
*/
requested_count?: number | null;
};
/** ConnectionPointsResponse */
ConnectionPointsResponse: {