68 lines
2.7 KiB
Python
68 lines
2.7 KiB
Python
"""Lookup функциональных зон генплана ЕКБ-2045 для analyze_parcel (#1058).
|
||
|
||
Читает ``ekb_genplan_functional_zone`` (м.139) и отдаёт зоны, пересекающие участок.
|
||
Graceful: таблица ещё не задеплоена / ошибка БД / пустой wkt → возвращает [].
|
||
|
||
Analyze-врезка (вшивание в build_ird_analyze_block / analyze_parcel) — отдельный
|
||
follow-up после merge #1131, чтобы не конфликтить с ird-analyze PR.
|
||
|
||
Зеркалит ird_overlay_lookup.py (стиль, graceful handling, CAST psycopg v3).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.exc import OperationalError, ProgrammingError
|
||
from sqlalchemy.orm import Session
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Spatial overlap участка: ST_Intersects через GIST idx_ekb_genplan_fz_geom.
|
||
# Участок передаётся как WKT в 4326. CAST psycopg v3 (НИКОГДА :x::type).
|
||
_ZONE_OVERLAP_SQL = text(
|
||
"""
|
||
SELECT zone_type, status, area_ha, zone_class_code
|
||
FROM ekb_genplan_functional_zone
|
||
WHERE ST_Intersects(geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
|
||
ORDER BY zone_type, status
|
||
"""
|
||
)
|
||
|
||
|
||
def parcel_functional_zones(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
|
||
"""Функциональные зоны генплана, пересекающие участок.
|
||
|
||
Args:
|
||
db: SQLAlchemy-сессия.
|
||
parcel_wkt: геометрия участка в WKT (EPSG:4326). None → пустой результат.
|
||
|
||
Returns:
|
||
Список dict: [{zone_type, status, area_ha, zone_class_code}].
|
||
Пусто при: нет wkt / таблица не задеплоена / нет пересечений / ошибка БД.
|
||
"""
|
||
if not parcel_wkt:
|
||
return []
|
||
try:
|
||
rows = db.execute(_ZONE_OVERLAP_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
|
||
except (OperationalError, ProgrammingError) as exc:
|
||
# Таблица ещё не задеплоена или ошибка БД — graceful degrade (analyze не падает).
|
||
logger.warning(
|
||
"parcel_functional_zones: ekb_genplan_functional_zone недоступна, skip: %s", exc
|
||
)
|
||
return []
|
||
|
||
return [
|
||
{
|
||
"zone_type": r["zone_type"],
|
||
"status": r["status"],
|
||
"area_ha": float(r["area_ha"]) if r["area_ha"] is not None else None,
|
||
"zone_class_code": r["zone_class_code"],
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
__all__ = ["parcel_functional_zones"]
|