"""Generative Design — Stage 1a geometry: parcel parsing + buildable area + grid. 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 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]: """Backwards-compatible alias. Now delegates to the real :func:`generate`.""" return generate(payload)